Source code for designate.sink.service

# Copyright 2012 Managed I.T.
# Copyright 2014 Hewlett-Packard Development Company, L.P.
# Author: Kiall Mac Innes <>
# 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, 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.
from oslo_log import log as logging
import oslo_messaging as messaging

import designate.conf
from designate import notification_handler
from designate import rpc
from designate import service

CONF = designate.conf.CONF
LOG = logging.getLogger(__name__)

[docs] class Service(service.Service): def __init__(self): super().__init__( self.service_name, threads=CONF['service:sink'].threads ) # Initialize extensions self._server = None self.handlers = self._init_extensions() self.subscribers = self._get_subscribers() @property def service_name(self): return 'sink' @staticmethod def _init_extensions(): """Loads and prepares all enabled extensions""" enabled_notification_handlers = ( CONF['service:sink'].enabled_notification_handlers ) notification_handlers = notification_handler.get_notification_handlers( enabled_notification_handlers) if len(notification_handlers) == 0: LOG.warning('No designate-sink handlers enabled or loaded') return notification_handlers def _get_subscribers(self): subscriptions = {} for handler in self.handlers: for et in handler.get_event_types(): subscriptions.setdefault(et, []) subscriptions[et].append(handler) return subscriptions
[docs] def start(self): super().start() # Setup notification subscriptions and start consuming targets = self._get_targets() # TODO(ekarlso): Change this is to endpoint objects rather then # ourselves? if targets: self._server = rpc.get_notification_listener( targets, [self], pool=CONF['service:sink'].listener_pool_name ) self._server.start()
[docs] def stop(self, graceful=True): # Try to shut the connection down, but if we get any sort of # errors, go ahead and ignore them.. as we're shutting down anyway try: if self._server: self._server.stop() except Exception as e: LOG.warning( 'Unable to gracefully stop the notification listener: %s', e ) super().stop(graceful)
def _get_targets(self): """ Set's up subscriptions for the various exchange+topic combinations that we have a handler for. """ targets = [] for handler in self.handlers: exchange, topics = handler.get_exchange_topics() for topic in topics: target = messaging.Target(exchange=exchange, topic=topic) targets.append(target) return targets def _get_handler_event_types(self): """return a dict - keys are the event types we can handle""" return self.subscribers
[docs] def info(self, context, publisher_id, event_type, payload, metadata): """ Processes an incoming notification, offering each extension the opportunity to handle it. """ # NOTE(zykes): Only bother to actually do processing if there's any # matching events, skips logging of things like compute.exists etc. if event_type in self._get_handler_event_types(): for handler in self.handlers: if event_type in handler.get_event_types(): LOG.debug('Found handler for: %s', event_type) handler.process_notification(context, event_type, payload)