/usr/lib/python2.7/dist-packages/neutron_lbaas/drivers/octavia/octavia_messaging_consumer.py is in python-neutron-lbaas 1:9.1.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | # Copyright 2016 Rackspace
#
# 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.
from neutron_lbaas._i18n import _
from neutron_lbaas._i18n import _LI
from oslo_config import cfg
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_service import service
oslo_messaging_opts = [
cfg.StrOpt('event_stream_topic',
default='neutron_lbaas_event',
help=_('topic name for receiving events from a queue'))
]
cfg.CONF.register_opts(oslo_messaging_opts, group='oslo_messaging')
LOG = logging.getLogger(__name__)
class InfoContainer(object):
@staticmethod
def from_dict(dict_obj):
return InfoContainer(dict_obj['info_type'],
dict_obj['info_id'],
dict_obj['info_payload'])
def __init__(self, info_type, info_id, info_payload):
self.info_type = info_type
self.info_id = info_id
self.info_payload = info_payload
def to_dict(self):
return {'info_type': self.info_type,
'info_id': self.info_id,
'info_payload': self.info_payload}
def __eq__(self, other):
if not isinstance(other, InfoContainer):
return False
if self.info_type != other.info_type:
return False
if self.info_id != other.info_id:
return False
if self.info_payload != other.info_payload:
return False
return True
def __ne__(self, other):
return not self == other
class ConsumerEndPoint(object):
target = messaging.Target(namespace="control", version='1.0')
def __init__(self, driver):
self.driver = driver
def update_info(self, ctx, container):
LOG.debug("Received event from stream %s", container)
container_inst = InfoContainer.from_dict(container)
self.driver.handle_streamed_event(container_inst)
class OctaviaConsumer(service.Service):
def __init__(self, driver, **kwargs):
super(OctaviaConsumer, self).__init__(**kwargs)
topic = cfg.CONF.oslo_messaging.event_stream_topic
server = cfg.CONF.host
self.driver = driver
self.transport = messaging.get_transport(cfg.CONF)
self.target = messaging.Target(topic=topic, server=server,
exchange="common", fanout=False)
self.endpoints = [ConsumerEndPoint(self.driver)]
self.server = None
def start(self):
super(OctaviaConsumer, self).start()
LOG.info(_LI("Starting octavia consumer..."))
self.server = messaging.get_rpc_server(self.transport, self.target,
self.endpoints,
executor='eventlet')
self.server.start()
def stop(self, graceful=False):
if self.server:
LOG.info(_LI('Stopping consumer...'))
self.server.stop()
if graceful:
LOG.info(
_LI('Consumer successfully stopped. Waiting for final '
'messages to be processed...'))
self.server.wait()
super(OctaviaConsumer, self).stop(graceful=graceful)
def reset(self):
if self.server:
self.server.reset()
super(OctaviaConsumer, self).reset()
|