b605943796
The IPA sends heartbeats to the conductor periodically and when requested, e.g. at the end of asynchronous commands. In order to avoid to send such notifications in too quick succession, e.g. when two asynchronous commands finish at the same time or when the periodic heartbeat was just sent right before a command ended, this patch proposes to coalesce heartbeats which are close together timewise and send only one for all of them in a time interval of 5 seconds. Co-Authored-By: Arne Wiebalck <arne.wiebalck@cern.ch> Story: #2008983 Task: 42633 Change-Id: Idfbce44065e1e5a8b730b94741b2604c51f0ab14
503 lines
20 KiB
Python
503 lines
20 KiB
Python
# 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
|
|
#
|
|
# 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 collections
|
|
import ipaddress
|
|
import random
|
|
import socket
|
|
import threading
|
|
import time
|
|
from urllib import parse as urlparse
|
|
|
|
import eventlet
|
|
from ironic_lib import exception as lib_exc
|
|
from ironic_lib import mdns
|
|
from oslo_concurrency import processutils
|
|
from oslo_config import cfg
|
|
from oslo_log import log
|
|
import pkg_resources
|
|
|
|
from ironic_python_agent.api import app
|
|
from ironic_python_agent import config
|
|
from ironic_python_agent import encoding
|
|
from ironic_python_agent import errors
|
|
from ironic_python_agent.extensions import base
|
|
from ironic_python_agent import hardware
|
|
from ironic_python_agent import inspector
|
|
from ironic_python_agent import ironic_api_client
|
|
from ironic_python_agent import utils
|
|
|
|
LOG = log.getLogger(__name__)
|
|
|
|
# Time(in seconds) to wait for any of the interfaces to be up
|
|
# before lookup of the node is attempted
|
|
NETWORK_WAIT_TIMEOUT = 60
|
|
|
|
# Time(in seconds) to wait before reattempt
|
|
NETWORK_WAIT_RETRY = 5
|
|
|
|
cfg.CONF.import_group('metrics', 'ironic_lib.metrics_utils')
|
|
cfg.CONF.import_group('metrics_statsd', 'ironic_lib.metrics_statsd')
|
|
|
|
Host = collections.namedtuple('Host', ['hostname', 'port'])
|
|
|
|
|
|
def _time():
|
|
"""Wraps time.time() for simpler testing."""
|
|
return time.time()
|
|
|
|
|
|
class IronicPythonAgentStatus(encoding.Serializable):
|
|
"""Represents the status of an agent."""
|
|
|
|
serializable_fields = ('started_at', 'version')
|
|
|
|
def __init__(self, started_at, version):
|
|
self.started_at = started_at
|
|
self.version = version
|
|
|
|
|
|
class IronicPythonAgentHeartbeater(threading.Thread):
|
|
"""Thread that periodically heartbeats to Ironic."""
|
|
|
|
# If we could wait at most N seconds between heartbeats (or in case of an
|
|
# error) we will instead wait r x N seconds, where r is a random value
|
|
# between these multipliers.
|
|
min_jitter_multiplier = 0.3
|
|
max_jitter_multiplier = 0.6
|
|
|
|
def __init__(self, agent):
|
|
"""Initialize the heartbeat thread.
|
|
|
|
:param agent: an :class:`ironic_python_agent.agent.IronicPythonAgent`
|
|
instance.
|
|
"""
|
|
super(IronicPythonAgentHeartbeater, self).__init__()
|
|
self.agent = agent
|
|
self.stop_event = threading.Event()
|
|
self.api = agent.api_client
|
|
self.interval = 0
|
|
self.heartbeat_forced = False
|
|
self.previous_heartbeat = 0
|
|
|
|
def run(self):
|
|
"""Start the heartbeat thread."""
|
|
# The first heartbeat happens immediately
|
|
LOG.info('Starting heartbeater')
|
|
self.agent.set_agent_advertise_addr()
|
|
|
|
while not self.stop_event.wait(min(self.interval, 5)):
|
|
if self._heartbeat_expected():
|
|
self.do_heartbeat()
|
|
eventlet.sleep(0)
|
|
|
|
def _heartbeat_expected(self):
|
|
# Normal heartbeating
|
|
if _time() > self.previous_heartbeat + self.interval:
|
|
return True
|
|
|
|
# Forced heartbeating, but once in 5 seconds
|
|
if (self.heartbeat_forced
|
|
and _time() > self.previous_heartbeat + 5):
|
|
return True
|
|
|
|
def do_heartbeat(self):
|
|
"""Send a heartbeat to Ironic."""
|
|
try:
|
|
self.api.heartbeat(
|
|
uuid=self.agent.get_node_uuid(),
|
|
advertise_address=self.agent.advertise_address,
|
|
advertise_protocol=self.agent.advertise_protocol,
|
|
generated_cert=self.agent.generated_cert,
|
|
)
|
|
LOG.info('heartbeat successful')
|
|
self.heartbeat_forced = False
|
|
self.previous_heartbeat = _time()
|
|
except errors.HeartbeatConflictError:
|
|
LOG.warning('conflict error sending heartbeat to {}'.format(
|
|
self.agent.api_url))
|
|
except Exception:
|
|
LOG.exception('error sending heartbeat to {}'.format(
|
|
self.agent.api_url))
|
|
finally:
|
|
interval_multiplier = random.uniform(self.min_jitter_multiplier,
|
|
self.max_jitter_multiplier)
|
|
self.interval = self.agent.heartbeat_timeout * interval_multiplier
|
|
log_msg = 'sleeping before next heartbeat, interval: {0}'
|
|
LOG.info(log_msg.format(self.interval))
|
|
|
|
def force_heartbeat(self):
|
|
self.heartbeat_forced = True
|
|
|
|
def stop(self):
|
|
"""Stop the heartbeat thread."""
|
|
LOG.info('stopping heartbeater')
|
|
self.stop_event.set()
|
|
return self.join()
|
|
|
|
|
|
class IronicPythonAgent(base.ExecuteCommandMixin):
|
|
"""Class for base agent functionality."""
|
|
|
|
def __init__(self, api_url, advertise_address, listen_address,
|
|
ip_lookup_attempts, ip_lookup_sleep, network_interface,
|
|
lookup_timeout, lookup_interval, standalone, agent_token,
|
|
hardware_initialization_delay=0, advertise_protocol='http'):
|
|
super(IronicPythonAgent, self).__init__()
|
|
if bool(cfg.CONF.keyfile) != bool(cfg.CONF.certfile):
|
|
LOG.warning("Only one of 'keyfile' and 'certfile' options is "
|
|
"defined in config file. Its value will be ignored.")
|
|
self.ext_mgr = base.init_ext_manager(self)
|
|
self.api_url = api_url
|
|
if (not self.api_url or self.api_url == 'mdns') and not standalone:
|
|
try:
|
|
self.api_url, params = mdns.get_endpoint('baremetal')
|
|
except lib_exc.ServiceLookupFailure:
|
|
if self.api_url:
|
|
# mDNS explicitly requested, report failure.
|
|
raise
|
|
else:
|
|
# implicit fallback to mDNS, do not fail (maybe we're only
|
|
# running inspection).
|
|
LOG.warning('Could not get baremetal endpoint from mDNS, '
|
|
'will not heartbeat')
|
|
else:
|
|
config.override(params)
|
|
|
|
if self.api_url:
|
|
self.api_client = ironic_api_client.APIClient(self.api_url)
|
|
self.heartbeater = IronicPythonAgentHeartbeater(self)
|
|
self.listen_address = listen_address
|
|
self.advertise_address = advertise_address
|
|
self.advertise_protocol = advertise_protocol
|
|
self.version = pkg_resources.get_distribution('ironic-python-agent')\
|
|
.version
|
|
self.api = app.Application(self, cfg.CONF)
|
|
self.heartbeat_timeout = None
|
|
self.started_at = None
|
|
self.node = None
|
|
# lookup timeout in seconds
|
|
self.lookup_timeout = lookup_timeout
|
|
self.lookup_interval = lookup_interval
|
|
self.ip_lookup_attempts = ip_lookup_attempts
|
|
self.ip_lookup_sleep = ip_lookup_sleep
|
|
self.network_interface = network_interface
|
|
self.standalone = standalone
|
|
self.hardware_initialization_delay = hardware_initialization_delay
|
|
# IPA will stop serving requests and exit after this is set to False
|
|
self.serve_api = True
|
|
self.agent_token = agent_token
|
|
# Allows this to be turned on by the conductor while running,
|
|
# in the event of long running ramdisks where the conductor
|
|
# got upgraded somewhere along the way.
|
|
self.agent_token_required = cfg.CONF.agent_token_required
|
|
self.generated_cert = None
|
|
|
|
def get_status(self):
|
|
"""Retrieve a serializable status.
|
|
|
|
:returns: a :class:`ironic_python_agent.agent.IronicPythonAgent`
|
|
instance describing the agent's status.
|
|
"""
|
|
return IronicPythonAgentStatus(
|
|
started_at=self.started_at,
|
|
version=self.version
|
|
)
|
|
|
|
def validate_agent_token(self, token):
|
|
# We did not get a token, i.e. None and
|
|
# we've previously seen a token, which is
|
|
# a mid-cluster upgrade case with long-running ramdisks.
|
|
if (not token and self.agent_token
|
|
and not self.agent_token_required):
|
|
# TODO(TheJulia): Rip this out during or after the V cycle.
|
|
LOG.warning('Agent token for requests are not required '
|
|
'by the conductor, yet we received a token. '
|
|
'Cluster may be mid-upgrade. Support to '
|
|
'not fail in this condition will be removed in '
|
|
'the Victoria development cycle.')
|
|
# Tell the API everything is okay.
|
|
return True
|
|
|
|
return self.agent_token == token
|
|
|
|
def _get_route_source(self, dest):
|
|
"""Get the IP address to send packages to destination."""
|
|
try:
|
|
out, _err = utils.execute('ip', 'route', 'get', dest)
|
|
except (EnvironmentError, processutils.ProcessExecutionError) as e:
|
|
LOG.warning('Cannot get route to host %(dest)s: %(err)s',
|
|
{'dest': dest, 'err': e})
|
|
return
|
|
|
|
try:
|
|
source = out.strip().split('\n')[0].split('src')[1].split()[0]
|
|
except IndexError:
|
|
LOG.warning('No route to host %(dest)s, route record: %(rec)s',
|
|
{'dest': dest, 'rec': out})
|
|
return
|
|
|
|
try:
|
|
if ipaddress.ip_address(source).is_link_local:
|
|
LOG.info('Ignoring link-local source to %(dest)s: %(rec)s',
|
|
{'dest': dest, 'rec': out})
|
|
return
|
|
except ValueError as exc:
|
|
LOG.warning('Invalid IP address %(addr)s returned as a route '
|
|
'to host %(dest)s: %(err)s',
|
|
{'dest': dest, 'addr': source, 'err': exc})
|
|
|
|
return source
|
|
|
|
def set_agent_advertise_addr(self):
|
|
"""Set advertised IP address for the agent, if not already set.
|
|
|
|
If agent's advertised IP address is still default (None), try to
|
|
find a better one. If the agent's network interface is None, replace
|
|
that as well.
|
|
|
|
:raises: LookupAgentIPError if an IP address could not be found
|
|
"""
|
|
if self.advertise_address.hostname is not None:
|
|
return
|
|
|
|
found_ip = None
|
|
if self.network_interface is not None:
|
|
# TODO(dtantsur): deprecate this
|
|
found_ip = hardware.dispatch_to_managers('get_ipv4_addr',
|
|
self.network_interface)
|
|
else:
|
|
url = urlparse.urlparse(self.api_url)
|
|
ironic_host = url.hostname
|
|
# Try resolving it in case it's not an IP address
|
|
try:
|
|
ironic_host = socket.gethostbyname(ironic_host)
|
|
except socket.gaierror:
|
|
LOG.debug('Count not resolve %s, maybe no DNS', ironic_host)
|
|
|
|
for attempt in range(self.ip_lookup_attempts):
|
|
found_ip = self._get_route_source(ironic_host)
|
|
if found_ip:
|
|
break
|
|
|
|
time.sleep(self.ip_lookup_sleep)
|
|
|
|
if found_ip:
|
|
self.advertise_address = Host(hostname=found_ip,
|
|
port=self.advertise_address.port)
|
|
else:
|
|
raise errors.LookupAgentIPError('Agent could not find a valid IP '
|
|
'address.')
|
|
|
|
def get_node_uuid(self):
|
|
"""Get UUID for Ironic node.
|
|
|
|
If the agent has not yet heartbeated to Ironic, it will not have
|
|
the UUID and this will raise an exception.
|
|
|
|
:returns: A string containing the UUID for the Ironic node.
|
|
:raises: UnknownNodeError if UUID is unknown.
|
|
"""
|
|
if self.node is None or 'uuid' not in self.node:
|
|
raise errors.UnknownNodeError()
|
|
return self.node['uuid']
|
|
|
|
def list_command_results(self):
|
|
"""Get a list of command results.
|
|
|
|
:returns: list of :class:`ironic_python_agent.extensions.base.
|
|
BaseCommandResult` objects.
|
|
"""
|
|
return list(self.command_results.values())
|
|
|
|
def get_command_result(self, result_id):
|
|
"""Get a specific command result by ID.
|
|
|
|
:returns: a :class:`ironic_python_agent.extensions.base.
|
|
BaseCommandResult` object.
|
|
:raises: RequestedObjectNotFoundError if command with the given ID
|
|
is not found.
|
|
"""
|
|
try:
|
|
return self.command_results[result_id]
|
|
except KeyError:
|
|
raise errors.RequestedObjectNotFoundError('Command Result',
|
|
result_id)
|
|
|
|
def force_heartbeat(self):
|
|
if not self.standalone:
|
|
self.heartbeater.force_heartbeat()
|
|
|
|
def _wait_for_interface(self):
|
|
"""Wait until at least one interface is up."""
|
|
|
|
wait_till = time.time() + NETWORK_WAIT_TIMEOUT
|
|
while time.time() < wait_till:
|
|
interfaces = hardware.dispatch_to_managers(
|
|
'list_network_interfaces')
|
|
if not any(ifc.mac_address for ifc in interfaces):
|
|
LOG.debug('Network is not up yet. '
|
|
'No valid interfaces found, retrying ...')
|
|
time.sleep(NETWORK_WAIT_RETRY)
|
|
else:
|
|
break
|
|
|
|
else:
|
|
LOG.warning("No valid network interfaces found. "
|
|
"Node lookup will probably fail.")
|
|
|
|
def _start_auto_tls(self):
|
|
# NOTE(dtantsur): if listen_tls is True, assume static TLS
|
|
# configuration and don't auto-generate anything.
|
|
if cfg.CONF.listen_tls or not cfg.CONF.enable_auto_tls:
|
|
LOG.debug('Automated TLS is disabled')
|
|
return None, None
|
|
|
|
if not self.api_url or not self.api_client.supports_auto_tls():
|
|
LOG.warning('Ironic does not support automated TLS')
|
|
return None, None
|
|
|
|
self.set_agent_advertise_addr()
|
|
|
|
LOG.info('Generating TLS parameters automatically for IP %s',
|
|
self.advertise_address.hostname)
|
|
tls_info = hardware.dispatch_to_managers(
|
|
'generate_tls_certificate', self.advertise_address.hostname)
|
|
self.generated_cert = tls_info.text
|
|
self.advertise_protocol = 'https'
|
|
return tls_info.path, tls_info.private_key_path
|
|
|
|
def serve_ipa_api(self):
|
|
"""Serve the API until an extension terminates it."""
|
|
cert_file, key_file = self._start_auto_tls()
|
|
self.api.start(cert_file, key_file)
|
|
if not self.standalone and self.api_url:
|
|
# Don't start heartbeating until the server is listening
|
|
self.heartbeater.start()
|
|
try:
|
|
while self.serve_api:
|
|
eventlet.sleep(0.1)
|
|
except KeyboardInterrupt:
|
|
LOG.info('Caught keyboard interrupt, exiting')
|
|
self.api.stop()
|
|
|
|
def process_lookup_data(self, content):
|
|
"""Update agent configuration from lookup data."""
|
|
|
|
self.node = content['node']
|
|
LOG.info('Lookup succeeded, node UUID is %s',
|
|
self.node['uuid'])
|
|
hardware.cache_node(self.node)
|
|
self.heartbeat_timeout = content['config']['heartbeat_timeout']
|
|
|
|
# Update config with values from Ironic
|
|
config = content.get('config', {})
|
|
if config.get('metrics'):
|
|
for opt, val in config.items():
|
|
setattr(cfg.CONF.metrics, opt, val)
|
|
if config.get('metrics_statsd'):
|
|
for opt, val in config.items():
|
|
setattr(cfg.CONF.metrics_statsd, opt, val)
|
|
if config.get('agent_token_required'):
|
|
self.agent_token_required = True
|
|
token = config.get('agent_token')
|
|
if token:
|
|
if len(token) >= 32:
|
|
LOG.debug('Agent token recorded as designated by '
|
|
'the ironic installation.')
|
|
self.agent_token = token
|
|
# set with-in the API client.
|
|
if not self.standalone:
|
|
self.api_client.agent_token = token
|
|
elif token == '******':
|
|
LOG.warning('The agent token has already been '
|
|
'retrieved. IPA may not operate as '
|
|
'intended and the deployment may fail '
|
|
'depending on settings in the ironic '
|
|
'deployment.')
|
|
if not self.agent_token and self.agent_token_required:
|
|
LOG.error('Ironic is signaling that agent tokens '
|
|
'are required, however we do not have '
|
|
'a token on file. '
|
|
'This is likely **FATAL**.')
|
|
else:
|
|
LOG.info('An invalid token was received.')
|
|
if self.agent_token and not self.standalone:
|
|
# Explicitly set the token in our API client before
|
|
# starting heartbeat operations.
|
|
self.api_client.agent_token = self.agent_token
|
|
|
|
def run(self):
|
|
"""Run the Ironic Python Agent."""
|
|
LOG.info('Starting ironic-python-agent version: %s',
|
|
self.version)
|
|
# Get the UUID so we can heartbeat to Ironic. Raises LookupNodeError
|
|
# if there is an issue (uncaught, restart agent)
|
|
self.started_at = _time()
|
|
# Attempt to sync the software clock
|
|
utils.sync_clock(ignore_errors=True)
|
|
|
|
# Cached hw managers at runtime, not load time. See bug 1490008.
|
|
hardware.get_managers()
|
|
# Operator-settable delay before hardware actually comes up.
|
|
# Helps with slow RAID drivers - see bug 1582797.
|
|
if self.hardware_initialization_delay > 0:
|
|
LOG.info('Waiting %d seconds before proceeding',
|
|
self.hardware_initialization_delay)
|
|
time.sleep(self.hardware_initialization_delay)
|
|
|
|
if not self.standalone:
|
|
# Inspection should be started before call to lookup, otherwise
|
|
# lookup will fail due to unknown MAC.
|
|
uuid = None
|
|
# We can't try to inspect or heartbeat until we have valid
|
|
# interfaces to perform those actions over.
|
|
self._wait_for_interface()
|
|
|
|
if cfg.CONF.inspection_callback_url:
|
|
try:
|
|
# Attempt inspection. This may fail, and previously
|
|
# an error would be logged.
|
|
uuid = inspector.inspect()
|
|
except errors.InspectionError as e:
|
|
LOG.error('Failed to perform inspection: %s', e)
|
|
|
|
if self.api_url:
|
|
content = self.api_client.lookup_node(
|
|
hardware_info=hardware.list_hardware_info(use_cache=True),
|
|
timeout=self.lookup_timeout,
|
|
starting_interval=self.lookup_interval,
|
|
node_uuid=uuid)
|
|
LOG.debug('Received lookup results: %s', content)
|
|
self.process_lookup_data(content)
|
|
# Save the API url in case we need it later.
|
|
hardware.save_api_client(
|
|
self.api_client, self.lookup_timeout,
|
|
self.lookup_interval)
|
|
|
|
elif cfg.CONF.inspection_callback_url:
|
|
LOG.info('No ipa-api-url configured, Heartbeat and lookup '
|
|
'skipped for inspector.')
|
|
else:
|
|
# NOTE(TheJulia): Once communication flow capability is
|
|
# able to be driven solely from the conductor, this is no
|
|
# longer a major issue.
|
|
LOG.error('Neither ipa-api-url nor inspection_callback_url'
|
|
'found, please check your pxe append parameters.')
|
|
|
|
self.serve_ipa_api()
|
|
|
|
if not self.standalone and self.api_url:
|
|
self.heartbeater.stop()
|