Source code for ironic_python_agent.ironic_api_client

# Copyright 2013 Rackspace, Inc.
# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import time

from oslo_config import cfg
from oslo_log import log
import requests
import tenacity

from ironic_python_agent import encoding
from ironic_python_agent import errors
from ironic_python_agent import netutils
from ironic_python_agent import utils
from ironic_python_agent import version

LOG = log.getLogger(__name__)

# TODO(TheJulia): This should be increased at some point.
# NOTE(dtantsur): change this constant every time you add support for more
# versions to ensure that we send the highest version we know about.

CONNECT_EXCEPTIONS = (requests.exceptions.Timeout,

[docs] class APIClient(object): api_version = 'v1' lookup_api = '/%s/lookup' % api_version heartbeat_api = '/%s/heartbeat/{uuid}' % api_version _ironic_api_version = None agent_token = None lookup_lock_pause = 0 def __init__(self, api_urls): if isinstance(api_urls, str): api_urls = [api_urls] self.api_urls = [url.rstrip('/') for url in api_urls] # Only keep alive a maximum of 2 connections to the API. More will be # opened if they are needed, but they will be closed immediately after # use. adapter = requests.adapters.HTTPAdapter(pool_connections=2, pool_maxsize=2) self.session = requests.Session() self.session.mount('https://', adapter) self.session.mount('http://', adapter) self.encoder = encoding.RESTJSONEncoder() def _request(self, method, path, data=None, headers=None, **kwargs): if data is not None: data = self.encoder.encode(data) headers = headers or {} headers.update({ 'Content-Type': 'application/json', 'Accept': 'application/json', }) if CONF.global_request_id: headers["X-OpenStack-Request-ID"] = CONF.global_request_id verify, cert = utils.get_ssl_client_options(CONF) for idx, api_url in enumerate(self.api_urls): request_url = f'{api_url}{path}' try: resp = self.session.request(method, request_url, headers=headers, data=data, verify=verify, cert=cert, timeout=CONF.http_request_timeout, **kwargs) # Make sure the working URL is on the top, so that the next # time we start from it. Also allows us to log self.api_urls[0] # as the currently used URL. self.api_urls = self.api_urls[idx:] + self.api_urls[:idx] return resp except CONNECT_EXCEPTIONS as exc: if idx == len(self.api_urls) - 1: raise LOG.warning("Connection error when accessing %s, trying the " "next URL. Error: %s", request_url, exc) def _get_ironic_api_version_header(self, version=None): if version is None: ironic_version = self._get_ironic_api_version() version = min(ironic_version, AGENT_TOKEN_IRONIC_VERSION) return {'X-OpenStack-Ironic-API-Version': '%d.%d' % version} def _get_ironic_api_version(self): if self._ironic_api_version: return self._ironic_api_version if CONF.ironic_api_version is not None: try: version = CONF.ironic_api_version.split('.') self._ironic_api_version = (int(version[0]), int(version[1])) return self._ironic_api_version except Exception: LOG.exception("An error occurred while attempting to parse" "the ironic_api_version. Will fall back to " "auto-detection") try: response = self._request('GET', '/') data = json.loads(response.content) version = data['default_version']['version'].split('.') self._ironic_api_version = (int(version[0]), int(version[1])) return self._ironic_api_version except Exception: LOG.exception("An error occurred while attempting to discover " "the available Ironic API versions, falling " "back to using version %s", ".".join(map(str, MIN_IRONIC_VERSION))) return MIN_IRONIC_VERSION
[docs] def supports_auto_tls(self): return self._get_ironic_api_version() >= AGENT_VERIFY_CA_IRONIC_VERSION
def _error_from_response(self, response): try: body = response.json() except ValueError: text = response.text else: body = body.get('error_message', body) if not isinstance(body, dict): # Old ironic format try: body = json.loads(body) except json.decoder.JSONDecodeError: body = {} text = (body.get('faultstring') or body.get('title') or response.text) return 'Error %d: %s' % (response.status_code, text)
[docs] def heartbeat(self, uuid, advertise_address, advertise_protocol='http', generated_cert=None): path = self.heartbeat_api.format(uuid=uuid) data = {'callback_url': self._get_agent_url(advertise_address, advertise_protocol)} api_ver = self._get_ironic_api_version() if api_ver >= AGENT_TOKEN_IRONIC_VERSION: data['agent_token'] = self.agent_token if api_ver >= AGENT_VERSION_IRONIC_VERSION: data['agent_version'] = version.__version__ if api_ver >= AGENT_VERIFY_CA_IRONIC_VERSION and generated_cert: data['agent_verify_ca'] = generated_cert api_ver = min(MAX_KNOWN_VERSION, api_ver) headers = self._get_ironic_api_version_header(api_ver) LOG.debug('Heartbeat: announcing callback URL %s, API version is ' '%d.%d', data['callback_url'], *api_ver) try: response = self._request('POST', path, data=data, headers=headers) except requests.exceptions.ConnectionError as e: raise errors.HeartbeatConnectionError(str(e)) except Exception as e: raise errors.HeartbeatError(str(e)) if response.status_code == error = self._error_from_response(response) raise errors.HeartbeatConflictError(error) elif response.status_code != error = self._error_from_response(response) raise errors.HeartbeatError(error)
[docs] def lookup_node(self, hardware_info, timeout, starting_interval, node_uuid=None, max_interval=60): retry = tenacity.retry( retry=tenacity.retry_if_result(lambda r: r is False), stop=tenacity.stop_after_delay(timeout), wait=tenacity.wait_random_exponential(min=starting_interval, max=max_interval), reraise=True) try: return retry(self._do_lookup)(hardware_info=hardware_info, node_uuid=node_uuid) except tenacity.RetryError: raise errors.LookupNodeError('Could not look up node info. Check ' 'logs for details.')
def _do_lookup(self, hardware_info, node_uuid): """The actual call to lookup a node.""" params = { 'addresses': ','.join(iface.mac_address for iface in hardware_info['interfaces'] if iface.mac_address) } if node_uuid: params['node_uuid'] = node_uuid LOG.debug('Looking up node with addresses %r and UUID %s at %s', params['addresses'], node_uuid, self.api_urls) try: response = self._request( 'GET', self.lookup_api, headers=self._get_ironic_api_version_header(), params=params) except CONNECT_EXCEPTIONS as err: # Report the last URL, there are warnings for the rest already LOG.warning( 'Error detected while attempting to perform lookup ' 'with %s, retrying. Error: %s', self.api_urls[-1], err ) return False except Exception as err: # NOTE(TheJulia): If you're looking here, and you're wondering # why the retry logic is not working or your investigating a weird # error or even IPA just exiting, # See!/story/2007968 # To be clear, we're going to try to provide as much detail as # possible in the exit handling msg = ('Unhandled error looking up node with addresses {} at ' '{}: {}'.format(params['addresses'], self.api_urls, err)) # No matter what we do at this point, IPA is going to exit. # This is because we don't know why the exception occurred and # we likely should not try to retry as such. # We will attempt to provide as much detail to the logs as # possible as to what occurred, although depending on the logging # subsystem, additional errors can occur, thus the additional # handling below. try: LOG.exception(msg) return False except Exception as exc_err: LOG.error(msg) exc_msg = ('Unexpected exception occurred while trying to ' 'log additional detail. Error: {}'.format(exc_err)) LOG.error(exc_msg) raise errors.LookupNodeError(msg) if response.status_code == if self.lookup_lock_pause == 0: self.lookup_lock_pause = 5 elif self.lookup_lock_pause == 5: self.lookup_lock_pause = 10 elif self.lookup_lock_pause == 10: # If we're reaching this point, we've got a long held # persistent lock, which means things can go very sideways # or the ironic deployment is downright grumpy. Either way, # we need to slow things down. self.lookup_lock_pause = 30 LOG.warning( 'Ironic has responded with a conflict, signaling the ' 'node is locked. We will wait %(time)s seconds before trying ' 'again. %(err)s', {'time': self.lookup_lock_pause, 'error': self._error_from_response(response)} ) time.sleep(self.lookup_lock_pause) return False if response.status_code != LOG.warning( 'Failed looking up node with addresses %r at %s. ' 'Check if inspection has completed? %s', params['addresses'], self.api_urls[0], self._error_from_response(response) ) return False try: content = json.loads(response.content) except json.decoder.JSONDecodeError as e: LOG.warning('Error decoding response: %s', e) return False # Check for valid response data if 'node' not in content or 'uuid' not in content['node']: LOG.warning( 'Got invalid node data in response to query for node ' 'with addresses %r from %s: %s', params['addresses'], self.api_urls[0], content, ) return False if 'config' not in content: # Old API try: content['config'] = {'heartbeat_timeout': content.pop('heartbeat_timeout')} except KeyError: LOG.warning('Got invalid heartbeat from the API: %s', content) return False # Got valid content return content def _get_agent_url(self, advertise_address, advertise_protocol='http'): return '{}://{}:{}'.format(advertise_protocol, netutils.wrap_ipv6(advertise_address[0]), advertise_address[1])