d7493a223f
Most of this code will probably stay with us for quite some time, so let's make things easier and consider them as our way of doing RPC. blueprint oslo-messaging Change-Id: Iaf353b23f9c54b82d1e02a6bd5a5960cec827c88
337 lines
13 KiB
Python
337 lines
13 KiB
Python
# Copyright 2013 New Dream Network, LLC (DreamHost)
|
|
#
|
|
# 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.
|
|
#
|
|
# @author: Mark McClain, DreamHost
|
|
|
|
from oslo.config import cfg
|
|
|
|
from neutron.agent import rpc as agent_rpc
|
|
from neutron.common import constants as n_const
|
|
from neutron.common import exceptions as n_exc
|
|
from neutron.common import rpc as n_rpc
|
|
from neutron.common import topics
|
|
from neutron import context
|
|
from neutron.openstack.common import importutils
|
|
from neutron.openstack.common import log as logging
|
|
from neutron.openstack.common import loopingcall
|
|
from neutron.openstack.common import periodic_task
|
|
from neutron.plugins.common import constants
|
|
from neutron.services.loadbalancer.agent import agent_api
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
OPTS = [
|
|
cfg.MultiStrOpt(
|
|
'device_driver',
|
|
default=['neutron.services.loadbalancer.drivers'
|
|
'.haproxy.namespace_driver.HaproxyNSDriver'],
|
|
help=_('Drivers used to manage loadbalancing devices'),
|
|
),
|
|
]
|
|
|
|
|
|
class DeviceNotFoundOnAgent(n_exc.NotFound):
|
|
msg = _('Unknown device with pool_id %(pool_id)s')
|
|
|
|
|
|
class LbaasAgentManager(n_rpc.RpcCallback, periodic_task.PeriodicTasks):
|
|
|
|
RPC_API_VERSION = '2.0'
|
|
# history
|
|
# 1.0 Initial version
|
|
# 1.1 Support agent_updated call
|
|
# 2.0 Generic API for agent based drivers
|
|
# - modify/reload/destroy_pool methods were removed;
|
|
# - added methods to handle create/update/delete for every lbaas
|
|
# object individually;
|
|
|
|
def __init__(self, conf):
|
|
super(LbaasAgentManager, self).__init__()
|
|
self.conf = conf
|
|
self.context = context.get_admin_context_without_session()
|
|
self.plugin_rpc = agent_api.LbaasAgentApi(
|
|
topics.LOADBALANCER_PLUGIN,
|
|
self.context,
|
|
self.conf.host
|
|
)
|
|
self._load_drivers()
|
|
|
|
self.agent_state = {
|
|
'binary': 'neutron-lbaas-agent',
|
|
'host': conf.host,
|
|
'topic': topics.LOADBALANCER_AGENT,
|
|
'configurations': {'device_drivers': self.device_drivers.keys()},
|
|
'agent_type': n_const.AGENT_TYPE_LOADBALANCER,
|
|
'start_flag': True}
|
|
self.admin_state_up = True
|
|
|
|
self._setup_state_rpc()
|
|
self.needs_resync = False
|
|
# pool_id->device_driver_name mapping used to store known instances
|
|
self.instance_mapping = {}
|
|
|
|
def _load_drivers(self):
|
|
self.device_drivers = {}
|
|
for driver in self.conf.device_driver:
|
|
try:
|
|
driver_inst = importutils.import_object(
|
|
driver,
|
|
self.conf,
|
|
self.plugin_rpc
|
|
)
|
|
except ImportError:
|
|
msg = _('Error importing loadbalancer device driver: %s')
|
|
raise SystemExit(msg % driver)
|
|
|
|
driver_name = driver_inst.get_name()
|
|
if driver_name not in self.device_drivers:
|
|
self.device_drivers[driver_name] = driver_inst
|
|
else:
|
|
msg = _('Multiple device drivers with the same name found: %s')
|
|
raise SystemExit(msg % driver_name)
|
|
|
|
def _setup_state_rpc(self):
|
|
self.state_rpc = agent_rpc.PluginReportStateAPI(
|
|
topics.LOADBALANCER_PLUGIN)
|
|
report_interval = self.conf.AGENT.report_interval
|
|
if report_interval:
|
|
heartbeat = loopingcall.FixedIntervalLoopingCall(
|
|
self._report_state)
|
|
heartbeat.start(interval=report_interval)
|
|
|
|
def _report_state(self):
|
|
try:
|
|
instance_count = len(self.instance_mapping)
|
|
self.agent_state['configurations']['instances'] = instance_count
|
|
self.state_rpc.report_state(self.context,
|
|
self.agent_state)
|
|
self.agent_state.pop('start_flag', None)
|
|
except Exception:
|
|
LOG.exception(_("Failed reporting state!"))
|
|
|
|
def initialize_service_hook(self, started_by):
|
|
self.sync_state()
|
|
|
|
@periodic_task.periodic_task
|
|
def periodic_resync(self, context):
|
|
if self.needs_resync:
|
|
self.needs_resync = False
|
|
self.sync_state()
|
|
|
|
@periodic_task.periodic_task(spacing=6)
|
|
def collect_stats(self, context):
|
|
for pool_id, driver_name in self.instance_mapping.items():
|
|
driver = self.device_drivers[driver_name]
|
|
try:
|
|
stats = driver.get_stats(pool_id)
|
|
if stats:
|
|
self.plugin_rpc.update_pool_stats(pool_id, stats)
|
|
except Exception:
|
|
LOG.exception(_('Error updating statistics on pool %s'),
|
|
pool_id)
|
|
self.needs_resync = True
|
|
|
|
def sync_state(self):
|
|
known_instances = set(self.instance_mapping.keys())
|
|
try:
|
|
ready_instances = set(self.plugin_rpc.get_ready_devices())
|
|
|
|
for deleted_id in known_instances - ready_instances:
|
|
self._destroy_pool(deleted_id)
|
|
|
|
for pool_id in ready_instances:
|
|
self._reload_pool(pool_id)
|
|
|
|
except Exception:
|
|
LOG.exception(_('Unable to retrieve ready devices'))
|
|
self.needs_resync = True
|
|
|
|
self.remove_orphans()
|
|
|
|
def _get_driver(self, pool_id):
|
|
if pool_id not in self.instance_mapping:
|
|
raise DeviceNotFoundOnAgent(pool_id=pool_id)
|
|
|
|
driver_name = self.instance_mapping[pool_id]
|
|
return self.device_drivers[driver_name]
|
|
|
|
def _reload_pool(self, pool_id):
|
|
try:
|
|
logical_config = self.plugin_rpc.get_logical_device(pool_id)
|
|
driver_name = logical_config['driver']
|
|
if driver_name not in self.device_drivers:
|
|
LOG.error(_('No device driver '
|
|
'on agent: %s.'), driver_name)
|
|
self.plugin_rpc.update_status(
|
|
'pool', pool_id, constants.ERROR)
|
|
return
|
|
|
|
self.device_drivers[driver_name].deploy_instance(logical_config)
|
|
self.instance_mapping[pool_id] = driver_name
|
|
self.plugin_rpc.pool_deployed(pool_id)
|
|
except Exception:
|
|
LOG.exception(_('Unable to deploy instance for pool: %s'), pool_id)
|
|
self.needs_resync = True
|
|
|
|
def _destroy_pool(self, pool_id):
|
|
driver = self._get_driver(pool_id)
|
|
try:
|
|
driver.undeploy_instance(pool_id)
|
|
del self.instance_mapping[pool_id]
|
|
self.plugin_rpc.pool_destroyed(pool_id)
|
|
except Exception:
|
|
LOG.exception(_('Unable to destroy device for pool: %s'), pool_id)
|
|
self.needs_resync = True
|
|
|
|
def remove_orphans(self):
|
|
for driver_name in self.device_drivers:
|
|
pool_ids = [pool_id for pool_id in self.instance_mapping
|
|
if self.instance_mapping[pool_id] == driver_name]
|
|
try:
|
|
self.device_drivers[driver_name].remove_orphans(pool_ids)
|
|
except NotImplementedError:
|
|
pass # Not all drivers will support this
|
|
|
|
def _handle_failed_driver_call(self, operation, obj_type, obj_id, driver):
|
|
LOG.exception(_('%(operation)s %(obj)s %(id)s failed on device driver '
|
|
'%(driver)s'),
|
|
{'operation': operation.capitalize(), 'obj': obj_type,
|
|
'id': obj_id, 'driver': driver})
|
|
self.plugin_rpc.update_status(obj_type, obj_id, constants.ERROR)
|
|
|
|
def create_vip(self, context, vip):
|
|
driver = self._get_driver(vip['pool_id'])
|
|
try:
|
|
driver.create_vip(vip)
|
|
except Exception:
|
|
self._handle_failed_driver_call('create', 'vip', vip['id'],
|
|
driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status('vip', vip['id'], constants.ACTIVE)
|
|
|
|
def update_vip(self, context, old_vip, vip):
|
|
driver = self._get_driver(vip['pool_id'])
|
|
try:
|
|
driver.update_vip(old_vip, vip)
|
|
except Exception:
|
|
self._handle_failed_driver_call('update', 'vip', vip['id'],
|
|
driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status('vip', vip['id'], constants.ACTIVE)
|
|
|
|
def delete_vip(self, context, vip):
|
|
driver = self._get_driver(vip['pool_id'])
|
|
driver.delete_vip(vip)
|
|
|
|
def create_pool(self, context, pool, driver_name):
|
|
if driver_name not in self.device_drivers:
|
|
LOG.error(_('No device driver on agent: %s.'), driver_name)
|
|
self.plugin_rpc.update_status('pool', pool['id'], constants.ERROR)
|
|
return
|
|
|
|
driver = self.device_drivers[driver_name]
|
|
try:
|
|
driver.create_pool(pool)
|
|
except Exception:
|
|
self._handle_failed_driver_call('create', 'pool', pool['id'],
|
|
driver.get_name())
|
|
else:
|
|
self.instance_mapping[pool['id']] = driver_name
|
|
self.plugin_rpc.update_status('pool', pool['id'], constants.ACTIVE)
|
|
|
|
def update_pool(self, context, old_pool, pool):
|
|
driver = self._get_driver(pool['id'])
|
|
try:
|
|
driver.update_pool(old_pool, pool)
|
|
except Exception:
|
|
self._handle_failed_driver_call('update', 'pool', pool['id'],
|
|
driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status('pool', pool['id'], constants.ACTIVE)
|
|
|
|
def delete_pool(self, context, pool):
|
|
driver = self._get_driver(pool['id'])
|
|
driver.delete_pool(pool)
|
|
del self.instance_mapping[pool['id']]
|
|
|
|
def create_member(self, context, member):
|
|
driver = self._get_driver(member['pool_id'])
|
|
try:
|
|
driver.create_member(member)
|
|
except Exception:
|
|
self._handle_failed_driver_call('create', 'member', member['id'],
|
|
driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status('member', member['id'],
|
|
constants.ACTIVE)
|
|
|
|
def update_member(self, context, old_member, member):
|
|
driver = self._get_driver(member['pool_id'])
|
|
try:
|
|
driver.update_member(old_member, member)
|
|
except Exception:
|
|
self._handle_failed_driver_call('update', 'member', member['id'],
|
|
driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status('member', member['id'],
|
|
constants.ACTIVE)
|
|
|
|
def delete_member(self, context, member):
|
|
driver = self._get_driver(member['pool_id'])
|
|
driver.delete_member(member)
|
|
|
|
def create_pool_health_monitor(self, context, health_monitor, pool_id):
|
|
driver = self._get_driver(pool_id)
|
|
assoc_id = {'pool_id': pool_id, 'monitor_id': health_monitor['id']}
|
|
try:
|
|
driver.create_pool_health_monitor(health_monitor, pool_id)
|
|
except Exception:
|
|
self._handle_failed_driver_call(
|
|
'create', 'health_monitor', assoc_id, driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status(
|
|
'health_monitor', assoc_id, constants.ACTIVE)
|
|
|
|
def update_pool_health_monitor(self, context, old_health_monitor,
|
|
health_monitor, pool_id):
|
|
driver = self._get_driver(pool_id)
|
|
assoc_id = {'pool_id': pool_id, 'monitor_id': health_monitor['id']}
|
|
try:
|
|
driver.update_pool_health_monitor(old_health_monitor,
|
|
health_monitor,
|
|
pool_id)
|
|
except Exception:
|
|
self._handle_failed_driver_call(
|
|
'update', 'health_monitor', assoc_id, driver.get_name())
|
|
else:
|
|
self.plugin_rpc.update_status(
|
|
'health_monitor', assoc_id, constants.ACTIVE)
|
|
|
|
def delete_pool_health_monitor(self, context, health_monitor, pool_id):
|
|
driver = self._get_driver(pool_id)
|
|
driver.delete_pool_health_monitor(health_monitor, pool_id)
|
|
|
|
def agent_updated(self, context, payload):
|
|
"""Handle the agent_updated notification event."""
|
|
if payload['admin_state_up'] != self.admin_state_up:
|
|
self.admin_state_up = payload['admin_state_up']
|
|
if self.admin_state_up:
|
|
self.needs_resync = True
|
|
else:
|
|
for pool_id in self.instance_mapping.keys():
|
|
LOG.info(_("Destroying pool %s due to agent disabling"),
|
|
pool_id)
|
|
self._destroy_pool(pool_id)
|
|
LOG.info(_("Agent_updated by server side %s!"), payload)
|