# Copyright 2012 VMware, Inc.
#
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import abc
import time
from oslo_config import cfg
from oslo_log import log as logging
import six
from six.moves import http_client as httplib
from vmware_nsx import api_client
LOG = logging.getLogger(__name__)
GENERATION_ID_TIMEOUT = -1
DEFAULT_CONCURRENT_CONNECTIONS = 3
DEFAULT_CONNECT_TIMEOUT = 5
@six.add_metaclass(abc.ABCMeta)
class ApiClientBase(object):
"""An abstract baseclass for all API client implementations."""
def _create_connection(self, host, port, is_ssl):
if is_ssl:
return httplib.HTTPSConnection(host, port,
timeout=self._connect_timeout)
return httplib.HTTPConnection(host, port,
timeout=self._connect_timeout)
@staticmethod
def _conn_params(http_conn):
is_ssl = isinstance(http_conn, httplib.HTTPSConnection)
return (http_conn.host, http_conn.port, is_ssl)
@property
def user(self):
return self._user
@property
def password(self):
return self._password
@property
def config_gen(self):
# If NSX_gen_timeout is not -1 then:
# Maintain a timestamp along with the generation ID. Hold onto the
# ID long enough to be useful and block on sequential requests but
# not long enough to persist when Onix db is cleared, which resets
# the generation ID, causing the DAL to block indefinitely with some
# number that's higher than the cluster's value.
if self._gen_timeout != -1:
ts = self._config_gen_ts
if ts is not None:
if (time.time() - ts) > self._gen_timeout:
return None
return self._config_gen
@config_gen.setter
def config_gen(self, value):
if self._config_gen != value:
if self._gen_timeout != -1:
self._config_gen_ts = time.time()
self._config_gen = value
def auth_cookie(self, conn):
cookie = None
data = self._get_provider_data(conn)
if data:
cookie = data[1]
return cookie
def set_auth_cookie(self, conn, cookie):
data = self._get_provider_data(conn)
if data:
self._set_provider_data(conn, (data[0], cookie))
def acquire_connection(self, auto_login=True, headers=None, rid=-1):
'''Check out an available HTTPConnection instance.
Blocks until a connection is available.
:auto_login: automatically logins before returning conn
:headers: header to pass on to login attempt
:param rid: request id passed in from request eventlet.
:returns: An available HTTPConnection instance or None if no
api_providers are configured.
'''
if not self._api_providers:
LOG.warning("[%d] no API providers currently available.", rid)
return None
if self._conn_pool.empty():
LOG.debug("[%d] Waiting to acquire API client connection.", rid)
priority, conn = self._conn_pool.get()
now = time.time()
if getattr(conn, 'last_used', now) < now - cfg.CONF.conn_idle_timeout:
LOG.info("[%(rid)d] Connection %(conn)s idle for %(sec)0.2f "
"seconds; reconnecting.",
{'rid': rid, 'conn': api_client.ctrl_conn_to_str(conn),
'sec': now - conn.last_used})
conn = self._create_connection(*self._conn_params(conn))
conn.last_used = now
conn.priority = priority # stash current priority for release
qsize = self._conn_pool.qsize()
LOG.debug("[%(rid)d] Acquired connection %(conn)s. %(qsize)d "
"connection(s) available.",
{'rid': rid, 'conn': api_client.ctrl_conn_to_str(conn),
'qsize': qsize})
if auto_login and self.auth_cookie(conn) is None:
self._wait_for_login(conn, headers)
return conn
def release_connection(self, http_conn, bad_state=False,
service_unavail=False, rid=-1):
'''Mark HTTPConnection instance as available for check-out.
:param http_conn: An HTTPConnection instance obtained from this
instance.
:param bad_state: True if http_conn is known to be in a bad state
(e.g. connection fault.)
:service_unavail: True if http_conn returned 503 response.
:param rid: request id passed in from request eventlet.
'''
conn_params = self._conn_params(http_conn)
if self._conn_params(http_conn) not in self._api_providers:
LOG.debug("[%(rid)d] Released connection %(conn)s is not an "
"API provider for the cluster",
{'rid': rid,
'conn': api_client.ctrl_conn_to_str(http_conn)})
return
elif hasattr(http_conn, "no_release"):
return
priority = http_conn.priority
if bad_state:
# Reconnect to provider.
LOG.warning("[%(rid)d] Connection returned in bad state, "
"reconnecting to %(conn)s",
{'rid': rid,
'conn': api_client.ctrl_conn_to_str(http_conn)})
http_conn = self._create_connection(*self._conn_params(http_conn))
elif service_unavail:
# http_conn returned a service unaviable response, put other
# connections to the same controller at end of priority queue,
conns = []
while not self._conn_pool.empty():
priority, conn = self._conn_pool.get()
if self._conn_params(conn) == conn_params:
priority = self._next_conn_priority
self._next_conn_priority += 1
conns.append((priority, conn))
for priority, conn in conns:
self._conn_pool.put((priority, conn))
# put http_conn at end of queue also
priority = self._next_conn_priority
self._next_conn_priority += 1
self._conn_pool.put((priority, http_conn))
LOG.debug("[%(rid)d] Released connection %(conn)s. %(qsize)d "
"connection(s) available.",
{'rid': rid, 'conn': api_client.ctrl_conn_to_str(http_conn),
'qsize': self._conn_pool.qsize()})
def _wait_for_login(self, conn, headers=None):
'''Block until a login has occurred for the current API provider.'''
data = self._get_provider_data(conn)
if data is None:
LOG.error("Login request for an invalid connection: '%s'",
api_client.ctrl_conn_to_str(conn))
return
provider_sem = data[0]
if provider_sem.acquire(blocking=False):
try:
cookie = self._login(conn, headers)
self.set_auth_cookie(conn, cookie)
finally:
provider_sem.release()
else:
LOG.debug("Waiting for auth to complete")
# Wait until we can acquire then release
provider_sem.acquire(blocking=True)
provider_sem.release()
def _get_provider_data(self, conn_or_conn_params, default=None):
"""Get data for specified API provider.
Args:
conn_or_conn_params: either a HTTP(S)Connection object or the
resolved conn_params tuple returned by self._conn_params().
default: conn_params if ones passed aren't known
Returns: Data associated with specified provider
"""
conn_params = self._normalize_conn_params(conn_or_conn_params)
return self._api_provider_data.get(conn_params, default)
def _set_provider_data(self, conn_or_conn_params, data):
"""Set data for specified API provider.
Args:
conn_or_conn_params: either a HTTP(S)Connection object or the
resolved conn_params tuple returned by self._conn_params().
data: data to associate with API provider
"""
conn_params = self._normalize_conn_params(conn_or_conn_params)
if data is None:
del self._api_provider_data[conn_params]
else:
self._api_provider_data[conn_params] = data
def _normalize_conn_params(self, conn_or_conn_params):
"""Normalize conn_param tuple.
Args:
conn_or_conn_params: either a HTTP(S)Connection object or the
resolved conn_params tuple returned by self._conn_params().
Returns: Normalized conn_param tuple
"""
if (not isinstance(conn_or_conn_params, tuple) and
not isinstance(conn_or_conn_params, httplib.HTTPConnection)):
LOG.debug("Invalid conn_params value: '%s'",
str(conn_or_conn_params))
return conn_or_conn_params
if isinstance(conn_or_conn_params, httplib.HTTPConnection):
conn_params = self._conn_params(conn_or_conn_params)
else:
conn_params = conn_or_conn_params
host, port, is_ssl = conn_params
if port is None:
port = 443 if is_ssl else 80
return (host, port, is_ssl)