x
/
novajoin
Code Issues Proposed changes

Add debug messages 

We are having a hard time keeping track of which operations
correspond to which request.  This patch adds the ability to track
operations in the notifier with the message_id of the notification
being processed.  This message_id (which is generated by oslo is
a uuid

For the server, we could also set the message_id to the request_id
of the python-requests object received, but this is already
logged as part of the server logs.

Change-Id: Ie8b885a2b5cba6684e92c49eed4a99d24621402e
This commit is contained in:
Ade Lee 2019-08-16 17:35:22 -04:00 committed by Grzegorz Grasza
parent 9750c363f6
commit ade787b90c
3 changed files with 185 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

175
novajoin/ipa.py
View File

@ -62,6 +62,7 @@ class IPANovaJoinBase(object):
self.ntries = CONF.connect_retries self.ntries = CONF.connect_retries
self.initial_backoff = CONF.connect_backoff self.initial_backoff = CONF.connect_backoff
self.ccache = "MEMORY:" + str(uuid.uuid4()) self.ccache = "MEMORY:" + str(uuid.uuid4())
LOG.debug("cache: %s", self.ccache)
os.environ['KRB5CCNAME'] = self.ccache os.environ['KRB5CCNAME'] = self.ccache
if self._ipa_client_configured() and not api.isdone('finalize'): if self._ipa_client_configured() and not api.isdone('finalize'):
(hostname, realm) = self.get_host_and_realm() (hostname, realm) = self.get_host_and_realm()
@ -126,23 +127,24 @@ class IPANovaJoinBase(object):
return (hostname, realm) return (hostname, realm)
def __backoff(self): def __backoff(self, message_id):
LOG.debug("Backing off %s seconds", self.backoff) LOG.debug("[%s] Backing off %s seconds", message_id, self.backoff)
time.sleep(self.backoff) time.sleep(self.backoff)
if self.backoff < 512: if self.backoff < 512:
self.backoff = self.backoff * 2 self.backoff = self.backoff * 2
def __reset_backoff(self): def __reset_backoff(self, message_id):
if self.backoff > self.initial_backoff: if self.backoff > self.initial_backoff:
LOG.debug("Resetting backoff to %d", self.initial_backoff) LOG.debug("[%s] Resetting backoff to %d",
message_id, self.initial_backoff)
self.backoff = self.initial_backoff self.backoff = self.initial_backoff
def __get_connection(self): def __get_connection(self, message_id):
"""Make a connection to IPA or raise an error.""" """Make a connection to IPA or raise an error."""
tries = 0 tries = 0
while (tries <= self.ntries): while (tries <= self.ntries):
LOG.debug("Attempt %d of %d", tries, self.ntries) LOG.debug("[%s] Attempt %d of %d", message_id, tries, self.ntries)
if api.Backend.rpcclient.isconnected(): if api.Backend.rpcclient.isconnected():
api.Backend.rpcclient.disconnect() api.Backend.rpcclient.disconnect()
try: try:
@ -154,7 +156,7 @@ class IPANovaJoinBase(object):
errors.TicketExpired, errors.TicketExpired,
errors.KerberosError) as e: errors.KerberosError) as e:
tries += 1 tries += 1
LOG.debug("kinit again: %s", e) LOG.debug("[%s] kinit again: %s", message_id, e)
# pylint: disable=no-member # pylint: disable=no-member
try: try:
kinit_keytab(str('nova/%s@%s' % kinit_keytab(str('nova/%s@%s' %
@ -162,35 +164,37 @@ class IPANovaJoinBase(object):
CONF.keytab, CONF.keytab,
self.ccache) self.ccache)
except GSSError as e: except GSSError as e:
LOG.debug("kinit failed: %s", e) LOG.debug("[%s] kinit failed: %s", message_id, e)
if self.backoff: if self.backoff:
self.__backoff() self.__backoff(message_id)
except errors.NetworkError: except errors.NetworkError:
tries += 1 tries += 1
if self.backoff: if self.backoff:
self.__backoff() self.__backoff(message_id)
except http_client.ResponseNotReady: except http_client.ResponseNotReady:
# NOTE(xek): This means that the server closed the socket, # NOTE(xek): This means that the server closed the socket,
# so keep-alive ended and we can't use that connection. # so keep-alive ended and we can't use that connection.
api.Backend.rpcclient.disconnect() api.Backend.rpcclient.disconnect()
tries += 1 tries += 1
if self.backoff: if self.backoff:
self.__backoff() self.__backoff(message_id)
else: else:
# successful connection # successful connection
self.__reset_backoff() self.__reset_backoff(message_id)
return return
LOG.error("Failed to connect to IPA after %d attempts", self.ntries) LOG.error("[%s] Failed to connect to IPA after %d attempts",
message_id, self.ntries)
raise exception.IPAConnectionError(tries=self.ntries) raise exception.IPAConnectionError(tries=self.ntries)
def start_batch_operation(self): def start_batch_operation(self, message_id=0):
"""Start a batch operation. """Start a batch operation.
IPA method calls will be collected in a batch job IPA method calls will be collected in a batch job
and submitted to IPA once all the operations have collected and submitted to IPA once all the operations have collected
by a call to _flush_batch_operation(). by a call to _flush_batch_operation().
""" """
LOG.debug("[%s] start batch operation", message_id)
self.batch_args = list() self.batch_args = list()
def _add_batch_operation(self, command, *args, **kw): def _add_batch_operation(self, command, *args, **kw):
@ -200,21 +204,21 @@ class IPANovaJoinBase(object):
"params": [args, kw], "params": [args, kw],
}) })
def flush_batch_operation(self): def flush_batch_operation(self, message_id=0):
"""Make an IPA batch call.""" """Make an IPA batch call."""
LOG.debug("flush_batch_operation") LOG.debug("[%] flush_batch_operation", message_id)
if not self.batch_args: if not self.batch_args:
return None return None
kw = {} kw = {}
LOG.debug(self.batch_args) LOG.debug("[%s] %s", message_id, self.batch_args)
return self._call_ipa('batch', *self.batch_args, **kw) return self._call_ipa(message_id, 'batch', *self.batch_args, **kw)
def _call_ipa(self, command, *args, **kw): def _call_ipa(self, message_id, command, *args, **kw):
"""Make an IPA call.""" """Make an IPA call."""
if not api.Backend.rpcclient.isconnected(): if not api.Backend.rpcclient.isconnected():
self.__get_connection() self.__get_connection(message_id)
if 'version' not in kw: if 'version' not in kw:
kw['version'] = u'2.146' # IPA v4.2.0 for compatibility kw['version'] = u'2.146' # IPA v4.2.0 for compatibility
@ -226,11 +230,11 @@ class IPANovaJoinBase(object):
except (errors.CCacheError, except (errors.CCacheError,
errors.TicketExpired, errors.TicketExpired,
errors.KerberosError): errors.KerberosError):
LOG.debug("Refresh authentication") LOG.debug("[%s] Refresh authentication", message_id)
self.__get_connection() self.__get_connection(message_id)
except errors.NetworkError: except errors.NetworkError:
if self.backoff: if self.backoff:
self.__backoff() self.__backoff(message_id)
else: else:
raise raise
except http_client.ResponseNotReady: except http_client.ResponseNotReady:
@ -238,7 +242,7 @@ class IPANovaJoinBase(object):
# so keep-alive ended and we can't use that connection. # so keep-alive ended and we can't use that connection.
api.Backend.rpcclient.disconnect() api.Backend.rpcclient.disconnect()
if self.backoff: if self.backoff:
self.__backoff() self.__backoff(message_id)
else: else:
raise raise
@ -259,7 +263,8 @@ class IPAClient(IPANovaJoinBase):
host_cache = cachetools.TTLCache(maxsize=512, ttl=30) host_cache = cachetools.TTLCache(maxsize=512, ttl=30)
service_cache = cachetools.TTLCache(maxsize=512, ttl=30) service_cache = cachetools.TTLCache(maxsize=512, ttl=30)
def add_host(self, hostname, ipaotp, metadata=None, image_metadata=None): def add_host(self, hostname, ipaotp, metadata=None, image_metadata=None,
message_id=0):
"""Add a host to IPA. """Add a host to IPA.
If requested in the metadata, add a host to IPA. The assumption If requested in the metadata, add a host to IPA. The assumption
@ -270,10 +275,10 @@ class IPAClient(IPANovaJoinBase):
and if that fails due to NotFound the host is added. and if that fails due to NotFound the host is added.
""" """
LOG.debug('Adding ' + hostname + ' to IPA.') LOG.debug("[%s] Adding %s to IPA.", message_id, hostname)
if not self._ipa_client_configured(): if not self._ipa_client_configured():
LOG.debug('IPA is not configured') LOG.debug('[%s] IPA is not configured', message_id)
return False return False
# There's no use in doing any operations if ipalib hasn't been # There's no use in doing any operations if ipalib hasn't been
@ -287,7 +292,7 @@ class IPAClient(IPANovaJoinBase):
image_metadata = {} image_metadata = {}
if hostname in self.host_cache: if hostname in self.host_cache:
LOG.debug('Host ' + hostname + ' found in cache.') LOG.debug("[%s] Host %s found in cache.", message_id, hostname)
return self.host_cache[hostname] return self.host_cache[hostname]
params = [hostname] params = [hostname]
@ -316,11 +321,11 @@ class IPAClient(IPANovaJoinBase):
} }
try: try:
self._call_ipa('host_mod', *params, **modargs) self._call_ipa(message_id, 'host_mod', *params, **modargs)
self.host_cache[hostname] = six.text_type(ipaotp) self.host_cache[hostname] = six.text_type(ipaotp)
except errors.NotFound: except errors.NotFound:
try: try:
self._call_ipa('host_add', *params, **hostargs) self._call_ipa(message_id, 'host_add', *params, **hostargs)
self.host_cache[hostname] = six.text_type(ipaotp) self.host_cache[hostname] = six.text_type(ipaotp)
except errors.DuplicateEntry: except errors.DuplicateEntry:
# We have no idea what the OTP is for the existing host. # We have no idea what the OTP is for the existing host.
@ -337,30 +342,30 @@ class IPAClient(IPANovaJoinBase):
return self.host_cache.get(hostname, False) return self.host_cache.get(hostname, False)
def add_subhost(self, hostname): def add_subhost(self, hostname, message_id=0):
"""Add a subhost to IPA. """Add a subhost to IPA.
Servers can have multiple network interfaces, and therefore can Servers can have multiple network interfaces, and therefore can
have multiple aliases. Moreover, they can part of a service using have multiple aliases. Moreover, they can part of a service using
a virtual host (VIP). These aliases are denoted 'subhosts', a virtual host (VIP). These aliases are denoted 'subhosts',
""" """
LOG.debug('Adding subhost: ' + hostname) LOG.debug('[%s] Adding subhost: %s', message_id, hostname)
if hostname not in self.host_cache: if hostname not in self.host_cache:
params = [hostname] params = [hostname]
hostargs = {'force': True} hostargs = {'force': True}
self._add_batch_operation('host_add', *params, **hostargs) self._add_batch_operation('host_add', *params, **hostargs)
self.host_cache[hostname] = True self.host_cache[hostname] = True
else: else:
LOG.debug('subhost ' + hostname + ' found in cache.') LOG.debug("[%s] subhost %s found in cache.", message_id, hostname)
def delete_subhost(self, hostname, batch=True): def delete_subhost(self, hostname, batch=True, message_id=0):
"""Delete a subhost from IPA. """Delete a subhost from IPA.
Servers can have multiple network interfaces, and therefore can Servers can have multiple network interfaces, and therefore can
have multiple aliases. Moreover, they can part of a service using have multiple aliases. Moreover, they can part of a service using
a virtual host (VIP). These aliases are denoted 'subhosts', a virtual host (VIP). These aliases are denoted 'subhosts',
""" """
LOG.debug('Deleting subhost: ' + hostname) LOG.debug(" [%s] Deleting subhost: %s", message_id, hostname)
host_params = [hostname] host_params = [hostname]
(hn, domain) = self.split_hostname(hostname) (hn, domain) = self.split_hostname(hostname)
@ -381,19 +386,20 @@ class IPAClient(IPANovaJoinBase):
else: else:
if hostname in self.host_cache: if hostname in self.host_cache:
del self.host_cache[hostname] del self.host_cache[hostname]
self._call_ipa('host_del', *host_params, **host_kw) self._call_ipa(message_id, 'host_del', *host_params, **host_kw)
try: try:
self._call_ipa('dnsrecord_del', *dns_params, **dns_kw) self._call_ipa(message_id, 'dnsrecord_del',
*dns_params, **dns_kw)
except (errors.NotFound, errors.ACIError): except (errors.NotFound, errors.ACIError):
# Ignore DNS deletion errors # Ignore DNS deletion errors
pass pass
def delete_host(self, hostname, metadata=None): def delete_host(self, hostname, metadata=None, message_id=0):
"""Delete a host from IPA and remove all related DNS entries.""" """Delete a host from IPA and remove all related DNS entries."""
LOG.debug('Deleting ' + hostname + ' from IPA.') LOG.debug("[%s] Deleting %s from IPA", message_id, hostname)
if not self._ipa_client_configured(): if not self._ipa_client_configured():
LOG.debug('IPA is not configured') LOG.debug('[%s] IPA is not configured', message_id)
return return
if metadata is None: if metadata is None:
@ -409,7 +415,7 @@ class IPAClient(IPANovaJoinBase):
try: try:
if hostname in self.host_cache: if hostname in self.host_cache:
del self.host_cache[hostname] del self.host_cache[hostname]
self._call_ipa('host_del', *params, **kw) self._call_ipa(message_id, 'host_del', *params, **kw)
except (errors.NotFound, errors.ACIError): except (errors.NotFound, errors.ACIError):
# Trying to delete a host that doesn't exist will raise an ACIError # Trying to delete a host that doesn't exist will raise an ACIError
# to hide whether the entry exists or not # to hide whether the entry exists or not
@ -422,27 +428,28 @@ class IPAClient(IPANovaJoinBase):
dns_kw = {'del_all': True, } dns_kw = {'del_all': True, }
try: try:
self._call_ipa('dnsrecord_del', *dns_params, **dns_kw) self._call_ipa(message_id, 'dnsrecord_del', *dns_params, **dns_kw)
except (errors.NotFound, errors.ACIError): except (errors.NotFound, errors.ACIError):
# Ignore DNS deletion errors # Ignore DNS deletion errors
pass pass
def add_service(self, principal): def add_service(self, principal, message_id=0):
if principal not in self.service_cache: if principal not in self.service_cache:
try: try:
(service, hostname, realm) = self.split_principal(principal) (service, hostname, realm) = self.split_principal(principal)
except errors.MalformedServicePrincipal as e: except errors.MalformedServicePrincipal as e:
LOG.error("Unable to split principal %s: %s", principal, e) LOG.error("[%s] Unable to split principal %s: %s",
message_id, principal, e)
raise raise
LOG.debug('Adding service: ' + principal) LOG.debug("[%s] Adding service: %s", message_id, principal)
params = [principal] params = [principal]
service_args = {'force': True} service_args = {'force': True}
self._add_batch_operation('service_add', *params, **service_args) self._add_batch_operation('service_add', *params, **service_args)
self.service_cache[principal] = [hostname] self.service_cache[principal] = [hostname]
else: else:
LOG.debug('Service ' + principal + ' found in cache.') LOG.debug("[%s] Service %s found in cache", message_id, principal)
def service_add_host(self, service_principal, host): def service_add_host(self, service_principal, host, message_id=0):
"""Add a host to a service. """Add a host to a service.
In IPA there is a relationship between a host and the services for In IPA there is a relationship between a host and the services for
@ -454,8 +461,8 @@ class IPAClient(IPANovaJoinBase):
in IPA this is done using the service-add-host API. in IPA this is done using the service-add-host API.
""" """
if host not in self.service_cache.get(service_principal, []): if host not in self.service_cache.get(service_principal, []):
LOG.debug('Adding principal ' + service_principal + LOG.debug("[%s] Adding principal %s to host %s",
' to host ' + host) message_id, service_principal, host)
params = [service_principal] params = [service_principal]
service_args = {'host': (host,)} service_args = {'host': (host,)}
self._add_batch_operation('service_add_host', *params, self._add_batch_operation('service_add_host', *params,
@ -463,17 +470,19 @@ class IPAClient(IPANovaJoinBase):
self.service_cache[service_principal] = self.service_cache.get( self.service_cache[service_principal] = self.service_cache.get(
service_principal, []) + [host] service_principal, []) + [host]
else: else:
LOG.debug('Host ' + host + ' managing ' + service_principal + LOG.debug("[%s] Host %s managing %s found in cache",
' found in cache.') message_id, host, service_principal)
def service_has_hosts(self, service_principal): def service_has_hosts(self, service_principal, message_id=0):
"""Return True if hosts other than parent manages this service""" """Return True if hosts other than parent manages this service"""
LOG.debug('Checking if principal ' + service_principal + ' has hosts') LOG.debug("[%s] Checking if principal %s has hosts",
message_id, service_principal)
params = [service_principal] params = [service_principal]
service_args = {} service_args = {}
try: try:
result = self._call_ipa('service_show', *params, **service_args) result = self._call_ipa(message_id, 'service_show',
*params, **service_args)
except errors.NotFound: except errors.NotFound:
raise KeyError raise KeyError
serviceresult = result['result'] serviceresult = result['result']
@ -483,7 +492,8 @@ class IPAClient(IPANovaJoinBase):
service_principal service_principal
) )
except errors.MalformedServicePrincipal as e: except errors.MalformedServicePrincipal as e:
LOG.error("Unable to split principal %s: %s", service_principal, e) LOG.error("[%s] Unable to split principal %s: %s",
message_id, service_principal, e)
raise raise
for candidate in serviceresult.get('managedby_host', []): for candidate in serviceresult.get('managedby_host', []):
@ -491,28 +501,30 @@ class IPAClient(IPANovaJoinBase):
return True return True
return False return False
def host_get_services(self, service_host): def host_get_services(self, service_host, message_id=0):
"""Return list of services this host manages""" """Return list of services this host manages"""
LOG.debug('Checking host ' + service_host + ' services') LOG.debug("[%s] Checking host %s services", message_id, service_host)
params = [] params = []
service_args = {'man_by_host': six.text_type(service_host)} service_args = {'man_by_host': six.text_type(service_host)}
result = self._call_ipa('service_find', *params, **service_args) result = self._call_ipa(message_id, 'service_find',
*params, **service_args)
return [service['krbprincipalname'][0] for service in result['result']] return [service['krbprincipalname'][0] for service in result['result']]
def host_has_services(self, service_host): def host_has_services(self, service_host, message_id=0):
"""Return True if this host manages any services""" """Return True if this host manages any services"""
return len(self.host_get_services(service_host)) > 0 return len(self.host_get_services(service_host, message_id)) > 0
def find_host(self, hostname): def find_host(self, hostname, message_id=0):
"""Return True if this host exists""" """Return True if this host exists"""
LOG.debug('Checking if host ' + hostname + ' exists') LOG.debug("[%s] Checking if host %s exists", message_id, hostname)
params = [] params = []
service_args = {'fqdn': six.text_type(hostname)} service_args = {'fqdn': six.text_type(hostname)}
result = self._call_ipa('host_find', *params, **service_args) result = self._call_ipa(message_id, 'host_find',
*params, **service_args)
return result['count'] > 0 return result['count'] > 0
def delete_service(self, principal, batch=True): def delete_service(self, principal, batch=True, message_id=0):
LOG.debug('Deleting service: ' + principal) LOG.debug("[%s] Deleting service: %s", message_id, principal)
params = [principal] params = [principal]
service_args = {} service_args = {}
if batch: if batch:
@ -522,14 +534,15 @@ class IPAClient(IPANovaJoinBase):
else: else:
if principal in self.service_cache: if principal in self.service_cache:
del self.service_cache[principal] del self.service_cache[principal]
return self._call_ipa('service_del', *params, **service_args) return self._call_ipa(message_id, 'service_del',
*params, **service_args)
def add_ip(self, hostname, floating_ip): def add_ip(self, hostname, floating_ip, message_id=0):
"""Add a floating IP to a given hostname.""" """Add a floating IP to a given hostname."""
LOG.debug('In add_ip') LOG.debug("[%s] In add_ip", message_id)
if not self._ipa_client_configured(): if not self._ipa_client_configured():
LOG.debug('IPA is not configured') LOG.debug('[%s] IPA is not configured', message_id)
return return
params = [six.text_type(get_domain() + '.'), params = [six.text_type(get_domain() + '.'),
@ -537,35 +550,39 @@ class IPAClient(IPANovaJoinBase):
kw = {'a_part_ip_address': six.text_type(floating_ip)} kw = {'a_part_ip_address': six.text_type(floating_ip)}
try: try:
self._call_ipa('dnsrecord_add', *params, **kw) self._call_ipa(message_id, 'dnsrecord_add', *params, **kw)
except (errors.DuplicateEntry, errors.ValidationError): except (errors.DuplicateEntry, errors.ValidationError):
pass pass
def find_record(self, floating_ip): def find_record(self, floating_ip, message_id=0):
"""Find DNS A record for floating IP address""" """Find DNS A record for floating IP address"""
LOG.debug('looking up host for floating ip' + floating_ip) LOG.debug("[%s] looking up host for floating ip %s",
message_id, floating_ip)
params = [six.text_type(get_domain() + '.')] params = [six.text_type(get_domain() + '.')]
service_args = {'arecord': six.text_type(floating_ip)} service_args = {'arecord': six.text_type(floating_ip)}
result = self._call_ipa('dnsrecord_find', *params, **service_args) result = self._call_ipa(message_id, 'dnsrecord_find',
*params, **service_args)
if result['count'] == 0: if result['count'] == 0:
return return
assert(result['count'] == 1) assert(result['count'] == 1)
return result['result'][0]['idnsname'][0].to_unicode() return result['result'][0]['idnsname'][0].to_unicode()
def remove_ip(self, floating_ip): def remove_ip(self, floating_ip, message_id=0):
"""Remove a floating IP from a given hostname.""" """Remove a floating IP from a given hostname."""
LOG.debug('In remove_ip') LOG.debug("[%s] In remove_ip", message_id)
if not self._ipa_client_configured(): if not self._ipa_client_configured():
LOG.debug('IPA is not configured') LOG.debug("[%s] IPA is not configured", message_id)
return return
hostname = self.find_record(floating_ip) hostname = self.find_record(floating_ip, message_id)
if not hostname: if not hostname:
LOG.debug('floating IP record not found') LOG.debug("[%s] floating IP record not found for %s",
message_id, floating_ip)
return return
params = [six.text_type(get_domain() + '.'), hostname] params = [six.text_type(get_domain() + '.'), hostname]
service_args = {'arecord': six.text_type(floating_ip)} service_args = {'arecord': six.text_type(floating_ip)}
self._call_ipa('dnsrecord_del', *params, **service_args) self._call_ipa(message_id, 'dnsrecord_del',
*params, **service_args)

41
novajoin/join.py
View File

@ -108,6 +108,12 @@ class JoinController(Controller):
Options passed in but as yet-unused are and user-data. Options passed in but as yet-unused are and user-data.
""" """
# Set message id to zero for now.
# We could set it to the request_id in the python-request,
# but this is already logged as part of the server logs.
message_id = 0
if not body: if not body:
LOG.error('No body in create request') LOG.error('No body in create request')
raise base.Fault(webob.exc.HTTPBadRequest()) raise base.Fault(webob.exc.HTTPBadRequest())
@ -191,7 +197,8 @@ class JoinController(Controller):
try: try:
data['ipaotp'] = self.ipaclient.add_host(data['hostname'], ipaotp, data['ipaotp'] = self.ipaclient.add_host(data['hostname'], ipaotp,
metadata, image_metadata) metadata, image_metadata,
message_id)
if not data['ipaotp']: if not data['ipaotp']:
# OTP was not added to host, don't return one # OTP was not added to host, don't return one
del data['ipaotp'] del data['ipaotp']
@ -199,24 +206,26 @@ class JoinController(Controller):
LOG.error('adding host failed %s', e) LOG.error('adding host failed %s', e)
LOG.error(traceback.format_exc()) LOG.error(traceback.format_exc())
self.ipaclient.start_batch_operation() self.ipaclient.start_batch_operation(message_id)
# key-per-service # key-per-service
managed_services = [metadata[key] for key in metadata.keys() managed_services = [metadata[key] for key in metadata.keys()
if key.startswith('managed_service_')] if key.startswith('managed_service_')]
if managed_services: if managed_services:
self.add_managed_services(data['hostname'], managed_services) self.add_managed_services(
data['hostname'], managed_services, message_id)
compact_services = util.get_compact_services(metadata) compact_services = util.get_compact_services(metadata)
if compact_services: if compact_services:
self.add_compact_services(hostname_short, compact_services) self.add_compact_services(
hostname_short, compact_services, message_id)
self.ipaclient.flush_batch_operation() self.ipaclient.flush_batch_operation(message_id)
return data return data
def add_managed_services(self, base_host, services): def add_managed_services(self, base_host, services, message_id=0):
"""Make any host/principal assignments passed into metadata.""" """Make any host/principal assignments passed into metadata."""
LOG.debug("In add_managed_services") LOG.debug("[%s] In add_managed_services", message_id)
hosts_found = list() hosts_found = list()
services_found = list() services_found = list()
@ -226,17 +235,18 @@ class JoinController(Controller):
# add host if not present # add host if not present
if principal_host not in hosts_found: if principal_host not in hosts_found:
self.ipaclient.add_subhost(principal_host) self.ipaclient.add_subhost(principal_host, message_id)
hosts_found.append(principal_host) hosts_found.append(principal_host)
# add service if not present # add service if not present
if principal not in services_found: if principal not in services_found:
self.ipaclient.add_service(principal) self.ipaclient.add_service(principal, message_id)
services_found.append(principal) services_found.append(principal)
self.ipaclient.service_add_host(principal, base_host) self.ipaclient.service_add_host(principal, base_host, message_id)
def add_compact_services(self, base_host_short, service_repr): def add_compact_services(self, base_host_short, service_repr,
message_id=0):
"""Make any host/principal assignments passed from metadata """Make any host/principal assignments passed from metadata
This takes a dictionary representation of the services and networks This takes a dictionary representation of the services and networks
@ -270,7 +280,7 @@ class JoinController(Controller):
This attempts to do a more compact representation since the nova This attempts to do a more compact representation since the nova
metadta entries have a limit of 255 characters. metadta entries have a limit of 255 characters.
""" """
LOG.debug("In add_compact_services") LOG.debug("[%s] In add_compact_services", message_id)
hosts_found = list() hosts_found = list()
services_found = list() services_found = list()
@ -284,12 +294,13 @@ class JoinController(Controller):
# add host if not present # add host if not present
if principal_host not in hosts_found: if principal_host not in hosts_found:
self.ipaclient.add_subhost(principal_host) self.ipaclient.add_subhost(principal_host, message_id)
hosts_found.append(principal_host) hosts_found.append(principal_host)
# add service if not present # add service if not present
if principal not in services_found: if principal not in services_found:
self.ipaclient.add_service(principal) self.ipaclient.add_service(principal, message_id)
services_found.append(principal) services_found.append(principal)
self.ipaclient.service_add_host(principal, base_host) self.ipaclient.service_add_host(
principal, base_host, message_id)

112
novajoin/notifications.py
View File

@ -65,9 +65,10 @@ class Registry(dict):
def __call__(self, name, version=None, service='nova'): def __call__(self, name, version=None, service='nova'):
def register_event(fun): def register_event(fun):
if version: if version:
def check_event(sself, payload): def check_event(sself, payload, message_id):
self.check_version(payload, version, service) self.check_version(payload, version, service)
return fun(sself, payload[service + '_object.data']) return fun(sself, payload[service + '_object.data'],
message_id)
self[name] = check_event self[name] = check_event
return check_event return check_event
self[name] = fun self[name] = fun
@ -124,19 +125,22 @@ class NotificationEndpoint(object):
LOG.debug("publisher: %s, event: %s, metadata: %s", publisher_id, LOG.debug("publisher: %s, event: %s, metadata: %s", publisher_id,
event_type, metadata) event_type, metadata)
message_id = metadata['message_id']
event_handler = self.event_handlers.get( event_handler = self.event_handlers.get(
event_type, lambda payload: LOG.error("Status update or unknown")) event_type, lambda payload: LOG.error("Status update or unknown"))
# run event handler for received notification type # run event handler for received notification type
event_handler(self, payload) event_handler(self, payload, message_id)
@event_handlers('compute.instance.create.end') @event_handlers('compute.instance.create.end')
def compute_instance_create(self, payload): def compute_instance_create(self, payload, message_id):
hostname = self._generate_hostname(payload.get('hostname')) hostname = self._generate_hostname(payload.get('hostname'))
instance_id = payload['instance_id'] instance_id = payload['instance_id']
LOG.info("Add new host %s (%s)", instance_id, hostname) LOG.info("[%s] Add new host %s (%s)",
message_id, instance_id, hostname)
@event_handlers('compute.instance.update') @event_handlers('compute.instance.update')
def compute_instance_update(self, payload): def compute_instance_update(self, payload, message_id):
ipa = ipaclient() ipa = ipaclient()
join_controller = join.JoinController(ipa) join_controller = join.JoinController(ipa)
hostname_short = payload['hostname'] hostname_short = payload['hostname']
@ -149,33 +153,37 @@ class NotificationEndpoint(object):
enroll = payload_metadata.get('ipa_enroll', '') enroll = payload_metadata.get('ipa_enroll', '')
image_enroll = image_metadata.get('ipa_enroll', '') image_enroll = image_metadata.get('ipa_enroll', '')
if enroll.lower() != 'true' and image_enroll.lower() != 'true': if enroll.lower() != 'true' and image_enroll.lower() != 'true':
LOG.info('IPA enrollment not requested, skipping update of %s', LOG.info(
hostname) '[%s] IPA enrollment not requested, skipping update of %s',
message_id, hostname)
return return
# Ensure this instance exists in nova # Ensure this instance exists in nova
instance = get_instance(instance_id) instance = get_instance(instance_id)
if instance is None: if instance is None:
msg = 'No such instance-id, %s' % instance_id msg = '[%s] No such instance-id, %s' % (message_id, instance_id)
LOG.error(msg) LOG.error(msg)
return return
ipa.start_batch_operation() LOG.info("[%s] compute instance update for %s", message_id, hostname)
ipa.start_batch_operation(message_id)
# key-per-service # key-per-service
managed_services = [ managed_services = [
payload_metadata[key] for key in payload_metadata.keys() payload_metadata[key] for key in payload_metadata.keys()
if key.startswith('managed_service_')] if key.startswith('managed_service_')]
if managed_services: if managed_services:
join_controller.add_managed_services(hostname, managed_services) join_controller.add_managed_services(
hostname, managed_services, message_id)
compact_services = util.get_compact_services(payload_metadata) compact_services = util.get_compact_services(payload_metadata)
if compact_services: if compact_services:
join_controller.add_compact_services( join_controller.add_compact_services(
hostname_short, compact_services) hostname_short, compact_services, message_id)
ipa.flush_batch_operation() ipa.flush_batch_operation(message_id)
@event_handlers('compute.instance.delete.end') @event_handlers('compute.instance.delete.end')
def compute_instance_delete(self, payload): def compute_instance_delete(self, payload, message_id):
hostname_short = payload['hostname'] hostname_short = payload['hostname']
instance_id = payload['instance_id'] instance_id = payload['instance_id']
payload_metadata = payload['metadata'] payload_metadata = payload['metadata']
@ -187,49 +195,51 @@ class NotificationEndpoint(object):
image_enroll = image_metadata.get('ipa_enroll', '') image_enroll = image_metadata.get('ipa_enroll', '')
if enroll.lower() != 'true' and image_enroll.lower() != 'true': if enroll.lower() != 'true' and image_enroll.lower() != 'true':
LOG.info('IPA enrollment not requested, skipping delete of %s', LOG.info(
hostname) '[%s] IPA enrollment not requested, skipping delete of %s',
message_id, hostname)
return return
LOG.info("Delete host %s (%s)", instance_id, hostname) LOG.info("[%s] Delete host %s (%s)", message_id, instance_id, hostname)
try: try:
ipa = ipaclient() ipa = ipaclient()
ipa.delete_host(hostname, {}) ipa.delete_host(hostname, {}, message_id)
self.delete_subhosts(ipa, hostname_short, payload_metadata) self.delete_subhosts(ipa, hostname_short, payload_metadata)
except exception.IPAConnectionError: except exception.IPAConnectionError:
LOG.error("IPA Connection Error when deleting host %s (%s). " LOG.error("[%s] IPA Connection Error when deleting host %s (%s). "
"Manual cleanup may be required in the IPA server.", "Manual cleanup may be required in the IPA server.",
instance_id, hostname) message_id, instance_id, hostname)
@event_handlers('network.floating_ip.associate') @event_handlers('network.floating_ip.associate')
def floaitng_ip_associate(self, payload): def floaitng_ip_associate(self, payload, message_id):
floating_ip = payload['floating_ip'] floating_ip = payload['floating_ip']
LOG.info("Associate floating IP %s" % floating_ip) LOG.info("[%s] Associate floating IP %s" % (message_id, floating_ip))
ipa = ipaclient() ipa = ipaclient()
nova = novaclient() nova = novaclient()
server = nova.servers.get(payload['instance_id']) server = nova.servers.get(payload['instance_id'])
if server: if server:
ipa.add_ip(server.name, floating_ip) ipa.add_ip(server.name, floating_ip, message_id)
else: else:
LOG.error("Could not resolve %s into a hostname", LOG.error("[%s] Could not resolve %s into a hostname",
payload['instance_id']) message_id, payload['instance_id'])
@event_handlers('network.floating_ip.disassociate') @event_handlers('network.floating_ip.disassociate')
def floating_ip_disassociate(self, payload): def floating_ip_disassociate(self, payload, message_id):
floating_ip = payload['floating_ip'] floating_ip = payload['floating_ip']
LOG.info("Disassociate floating IP %s" % floating_ip) LOG.info("[%s] Disassociate floating IP %s", message_id, floating_ip)
ipa = ipaclient() ipa = ipaclient()
ipa.remove_ip(floating_ip) ipa.remove_ip(floating_ip, message_id)
@event_handlers('floatingip.update.end') @event_handlers('floatingip.update.end')
def floating_ip_update(self, payload): def floating_ip_update(self, payload, message_id):
"""Neutron event""" """Neutron event"""
floatingip = payload['floatingip'] floatingip = payload['floatingip']
floating_ip = floatingip['floating_ip_address'] floating_ip = floatingip['floating_ip_address']
port_id = floatingip['port_id'] port_id = floatingip['port_id']
ipa = ipaclient() ipa = ipaclient()
if port_id: if port_id:
LOG.info("Neutron floating IP associate: %s" % floating_ip) LOG.info("[%s] Neutron floating IP associate: %s",
message_id, floating_ip)
nova = novaclient() nova = novaclient()
neutron = neutronclient() neutron = neutronclient()
search_opts = {'id': port_id} search_opts = {'id': port_id}
@ -239,14 +249,16 @@ class NotificationEndpoint(object):
if device_id: if device_id:
server = nova.servers.get(device_id) server = nova.servers.get(device_id)
if server: if server:
ipa.add_ip(server.name, floating_ip) ipa.add_ip(server.name, floating_ip, message_id)
else: else:
LOG.error("Expected 1 port, got %d", len(ports)) LOG.error("[%s] Expected 1 port, got %d",
message_id, len(ports))
else: else:
LOG.info("Neutron floating IP disassociate: %s" % floating_ip) LOG.info("[%s] Neutron floating IP disassociate: %s",
ipa.remove_ip(floating_ip) message_id, floating_ip)
ipa.remove_ip(floating_ip, message_id)
def delete_subhosts(self, ipa, hostname_short, metadata): def delete_subhosts(self, ipa, hostname_short, metadata, message_id):
"""Delete subhosts and remove VIPs if possible. """Delete subhosts and remove VIPs if possible.
Servers can have multiple network interfaces, and therefore can Servers can have multiple network interfaces, and therefore can
@ -269,13 +281,15 @@ class NotificationEndpoint(object):
compact_services = util.get_compact_services(metadata) compact_services = util.get_compact_services(metadata)
if compact_services: if compact_services:
self.delete_compact_services(ipa, hostname_short, self.delete_compact_services(ipa, hostname_short,
compact_services) compact_services,
message_id)
managed_services = [metadata[key] for key in metadata.keys() managed_services = [metadata[key] for key in metadata.keys()
if key.startswith('managed_service_')] if key.startswith('managed_service_')]
if managed_services: if managed_services:
self.delete_managed_services(ipa, managed_services) self.delete_managed_services(ipa, managed_services, message_id)
def delete_compact_services(self, ipa, host_short, service_repr): def delete_compact_services(self, ipa, host_short, service_repr,
message_id):
"""Reconstructs and removes subhosts for compact services. """Reconstructs and removes subhosts for compact services.
Data looks like this: Data looks like this:
@ -289,10 +303,10 @@ class NotificationEndpoint(object):
services to be automatically deleted through IPA referential services to be automatically deleted through IPA referential
integrity. integrity.
""" """
LOG.debug("In delete compact services") LOG.debug("[%s] In delete compact services", message_id)
hosts_found = list() hosts_found = list()
ipa.start_batch_operation() ipa.start_batch_operation(message_id)
for service_name, net_list in service_repr.items(): for service_name, net_list in service_repr.items():
for network in net_list: for network in net_list:
host = "%s.%s" % (host_short, network) host = "%s.%s" % (host_short, network)
@ -302,15 +316,15 @@ class NotificationEndpoint(object):
if principal_host not in hosts_found: if principal_host not in hosts_found:
ipa.delete_subhost(principal_host) ipa.delete_subhost(principal_host)
hosts_found.append(principal_host) hosts_found.append(principal_host)
ipa.flush_batch_operation() ipa.flush_batch_operation(message_id)
def delete_managed_services(self, ipa, services): def delete_managed_services(self, ipa, services, message_id):
"""Delete any managed services if possible. """Delete any managed services if possible.
Checks to see if the managed service subhost has no managed hosts Checks to see if the managed service subhost has no managed hosts
associations and if so, deletes the host. associations and if so, deletes the host.
""" """
LOG.debug("In delete_managed_services") LOG.debug("[%s] In delete_managed_services", message_id)
hosts_deleted = list() hosts_deleted = list()
services_deleted = list() services_deleted = list()
@ -353,15 +367,15 @@ class VersionedNotificationEndpoint(NotificationEndpoint):
event_handlers = Registry(NotificationEndpoint.event_handlers) event_handlers = Registry(NotificationEndpoint.event_handlers)
@event_handlers('instance.create.end', '1.10') @event_handlers('instance.create.end', '1.10')
def instance_create(self, payload): def instance_create(self, payload, message_id):
newpayload = { newpayload = {
'hostname': payload['host_name'], 'hostname': payload['host_name'],
'instance_id': payload['uuid'], 'instance_id': payload['uuid'],
} }
self.compute_instance_create(newpayload) self.compute_instance_create(newpayload, message_id)
@event_handlers('instance.update', '1.8') @event_handlers('instance.update', '1.8')
def instance_update(self, payload): def instance_update(self, payload, message_id):
glance = glanceclient() glance = glanceclient()
newpayload = { newpayload = {
'hostname': payload['host_name'], 'hostname': payload['host_name'],
@ -369,10 +383,10 @@ class VersionedNotificationEndpoint(NotificationEndpoint):
'metadata': payload['metadata'], 'metadata': payload['metadata'],
'image_meta': glance.images.get(payload['image_uuid']) 'image_meta': glance.images.get(payload['image_uuid'])
} }
self.compute_instance_update(newpayload) self.compute_instance_update(newpayload, message_id)
@event_handlers('instance.delete.end', '1.7') @event_handlers('instance.delete.end', '1.7')
def instance_delete(self, payload): def instance_delete(self, payload, message_id):
glance = glanceclient() glance = glanceclient()
newpayload = { newpayload = {
'hostname': payload['host_name'], 'hostname': payload['host_name'],
@ -380,7 +394,7 @@ class VersionedNotificationEndpoint(NotificationEndpoint):
'metadata': payload['metadata'], 'metadata': payload['metadata'],
'image_meta': glance.images.get(payload['image_uuid']) 'image_meta': glance.images.get(payload['image_uuid'])
} }
self.compute_instance_delete(newpayload) self.compute_instance_delete(newpayload, message_id)
def main(): def main():