50afa71853
1.Upgrade pylint to 2.4.4, add exclusions to the tests, and fix some lint errors in the code 2. Fix user creation with GRANT in MySQL 8.0(Ubuntu Focal) In Ubuntu Bionic (18.04) mysql 5.7 version used to create the user implicitly when using using the GRANT. Ubuntu Focal (20.04) has mysql 8.0 and with mysql 8.0 there is no implicit user creation with GRANT. We need to create the user first before using GRANT command. See also commit I97b0dcbb88c6ef7c22e3c55970211bed792bbd0d 3. Remove fwaas from the zuul.yaml 4. Remove DB migration test which is failing ue to FWaaS migration with py38 5. Fix cover tests python version in .tox 6. fix requirememnts Change-Id: I22654a5d5ccaad3185ae3365a90afba1ce870695
604 lines
25 KiB
Python
604 lines
25 KiB
Python
# Copyright 2018 VMware, Inc.
|
|
# All Rights Reserved
|
|
#
|
|
# 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 copy
|
|
import socket
|
|
import time
|
|
|
|
from oslo_config import cfg
|
|
from oslo_log import helpers as log_helpers
|
|
from oslo_log import log as logging
|
|
import oslo_messaging as messaging
|
|
from oslo_messaging.rpc import dispatcher
|
|
|
|
from octavia.api.drivers import utils as oct_utils
|
|
from octavia.db import api as db_apis
|
|
from octavia.db import repositories
|
|
from octavia_lib.api.drivers import driver_lib
|
|
from octavia_lib.api.drivers import exceptions
|
|
|
|
from octavia_lib.api.drivers import provider_base as driver_base
|
|
|
|
from vmware_nsx.services.lbaas.octavia import constants as d_const
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
cfg.CONF.import_group('oslo_messaging', 'octavia.common.config')
|
|
|
|
TRANSPORT = None
|
|
RPC_SERVER = None
|
|
|
|
|
|
def get_transport():
|
|
global TRANSPORT
|
|
if not TRANSPORT:
|
|
TRANSPORT = messaging.get_rpc_transport(cfg.CONF)
|
|
return TRANSPORT
|
|
|
|
|
|
def get_rpc_server(target, endpoints, access_policy):
|
|
global RPC_SERVER
|
|
if not RPC_SERVER:
|
|
RPC_SERVER = messaging.get_rpc_server(
|
|
TRANSPORT, target, endpoints, executor='threading',
|
|
access_policy=access_policy)
|
|
return RPC_SERVER
|
|
|
|
|
|
# List of keys per object type that will not be sent to the listener
|
|
unsupported_keys = {'Loadbalancer': ['vip_qos_policy_id'],
|
|
'Listener': ['sni_container_refs',
|
|
'insert_headers',
|
|
'timeout_client_data',
|
|
'timeout_member_connect',
|
|
'timeout_member_data',
|
|
'timeout_tcp_inspect'],
|
|
'HealthMonitor': ['max_retries_down'],
|
|
'Member': ['monitor_address', 'monitor_port']}
|
|
|
|
|
|
class NSXOctaviaDriver(driver_base.ProviderDriver):
|
|
@log_helpers.log_method_call
|
|
def __init__(self):
|
|
super(NSXOctaviaDriver, self).__init__()
|
|
self._init_rpc_messaging()
|
|
self.repositories = repositories.Repositories()
|
|
|
|
@log_helpers.log_method_call
|
|
def _init_rpc_messaging(self):
|
|
topic = d_const.OCTAVIA_TO_DRIVER_TOPIC
|
|
transport = get_transport()
|
|
target = messaging.Target(topic=topic, exchange="common",
|
|
namespace='control', fanout=False,
|
|
version='1.0')
|
|
self.client = messaging.RPCClient(transport, target)
|
|
|
|
def get_obj_project_id(self, obj_type, obj_dict):
|
|
if obj_dict.get('project_id'):
|
|
return obj_dict['project_id']
|
|
if obj_dict.get('tenant_id'):
|
|
return obj_dict['tenant_id']
|
|
|
|
# look for the project id of the attached objects
|
|
project_id = None
|
|
if obj_dict.get('loadbalancer_id'):
|
|
db_lb = self.repositories.load_balancer.get(
|
|
db_apis.get_session(), id=obj_dict['loadbalancer_id'])
|
|
if db_lb:
|
|
project_id = db_lb.project_id
|
|
if not project_id and obj_dict.get('pool_id'):
|
|
db_pool = self.repositories.pool.get(
|
|
db_apis.get_session(), id=obj_dict['pool_id'])
|
|
if db_pool:
|
|
project_id = db_pool.load_balancer.project_id
|
|
if not project_id and obj_dict.get('listener_id'):
|
|
db_list = self.repositories.listener.get(
|
|
db_apis.get_session(), id=obj_dict['listener_id'])
|
|
if db_list:
|
|
project_id = db_list.load_balancer.project_id
|
|
if not project_id and obj_dict.get('l7policy_id'):
|
|
db_policy = self.repositories.l7policy.get(
|
|
db_apis.get_session(), id=obj_dict['l7policy_id'])
|
|
if db_policy:
|
|
if db_policy.listener:
|
|
db_lb = db_policy.listener.load_balancer
|
|
elif db_policy.redirect_pool:
|
|
db_lb = db_policy.redirect_pool.load_balancer
|
|
if db_lb:
|
|
project_id = db_lb.project_id
|
|
|
|
if not project_id:
|
|
LOG.warning("Could not find the tenant id for %(type)s "
|
|
"%(obj)s", {'type': obj_type, 'obj': obj_dict})
|
|
return project_id
|
|
|
|
def _get_load_balancer_dict(self, loadbalancer_id):
|
|
if not loadbalancer_id:
|
|
return
|
|
db_lb = self.repositories.load_balancer.get(
|
|
db_apis.get_session(), id=loadbalancer_id)
|
|
if not db_lb:
|
|
return
|
|
lb_dict = {'name': db_lb.name, 'id': loadbalancer_id}
|
|
if db_lb.vip:
|
|
lb_dict['vip_port_id'] = db_lb.vip.port_id
|
|
lb_dict['vip_address'] = db_lb.vip.ip_address
|
|
lb_dict['vip_port_id'] = db_lb.vip.port_id
|
|
lb_dict['vip_network_id'] = db_lb.vip.network_id
|
|
lb_dict['vip_subnet_id'] = db_lb.vip.subnet_id
|
|
return lb_dict
|
|
|
|
def _get_listener_in_pool_dict(self, pool_dict, is_update):
|
|
if 'listener' not in pool_dict:
|
|
if pool_dict.get('listener_id'):
|
|
db_listener = self.repositories.listener.get(
|
|
db_apis.get_session(), id=pool_dict['listener_id'])
|
|
listener_obj = oct_utils.db_listener_to_provider_listener(
|
|
db_listener)
|
|
listener_dict = listener_obj.to_dict(
|
|
recurse=False, render_unsets=True)
|
|
listener_dict['id'] = listener_dict['listener_id']
|
|
listener_dict['l7_policies'] = listener_dict['l7policies']
|
|
# Add the loadbalancer to the listener dict
|
|
if pool_dict.get('loadbalancer_id'):
|
|
# Generate a loadbalancer object
|
|
listener_dict['loadbalancer'] = (
|
|
self._get_load_balancer_dict(
|
|
pool_dict['loadbalancer_id']))
|
|
pool_dict['listener'] = listener_dict
|
|
if 'listeners' not in pool_dict:
|
|
# multiple listeners is not really supported yet
|
|
pool_dict['listeners'] = [listener_dict]
|
|
# Do not add listener in update situation, as we want to use
|
|
# the original listener of this pool
|
|
elif not is_update:
|
|
pool_dict['listener'] = None
|
|
if 'listeners' not in pool_dict:
|
|
pool_dict['listeners'] = []
|
|
|
|
def _get_pool_dict(self, pool_id, is_update, parent_project_id=None):
|
|
if not pool_id:
|
|
return {}
|
|
db_pool = self.repositories.pool.get(db_apis.get_session(), id=pool_id)
|
|
if not db_pool:
|
|
return {}
|
|
pool_obj = oct_utils.db_pool_to_provider_pool(db_pool)
|
|
pool_dict = pool_obj.to_dict(recurse=True, render_unsets=True)
|
|
pool_dict['id'] = pool_id
|
|
# Get the load balancer object
|
|
if pool_dict.get('loadbalancer_id'):
|
|
# Generate a loadbalancer object
|
|
pool_dict['loadbalancer'] = self._get_load_balancer_dict(
|
|
pool_dict['loadbalancer_id'])
|
|
if 'listener' not in pool_dict:
|
|
self._get_listener_in_pool_dict(pool_dict, is_update)
|
|
# make sure this pool has a project id
|
|
if not pool_dict.get('project_id'):
|
|
project_id = self.get_obj_project_id('Pool', pool_dict)
|
|
if project_id is None:
|
|
project_id = parent_project_id
|
|
pool_dict['tenant_id'] = pool_dict['project_id'] = project_id
|
|
|
|
return pool_dict
|
|
|
|
def _get_hm_dict(self, hm_id, is_update):
|
|
if not hm_id:
|
|
return {}
|
|
db_hm = self.repositories.health_monitor.get(
|
|
db_apis.get_session(), id=hm_id)
|
|
if not db_hm:
|
|
return {}
|
|
hm_obj = oct_utils.db_HM_to_provider_HM(db_hm)
|
|
hm_dict = hm_obj.to_dict(recurse=True, render_unsets=True)
|
|
hm_dict['id'] = hm_id
|
|
# Get the pol object
|
|
if hm_dict.get('pool_id'):
|
|
hm_dict['pool'] = self._get_pool_dict(
|
|
hm_dict['pool_id'], is_update)
|
|
return hm_dict
|
|
|
|
def update_policy_dict(self, policy_dict, policy_obj, is_update=False):
|
|
if policy_dict.get('listener_id'):
|
|
db_list = self.repositories.listener.get(
|
|
db_apis.get_session(), id=policy_dict['listener_id'])
|
|
list_obj = oct_utils.db_listener_to_provider_listener(db_list)
|
|
list_dict = list_obj.to_dict(recurse=True, render_unsets=True)
|
|
list_dict['id'] = policy_dict['listener_id']
|
|
policy_dict['listener'] = list_dict
|
|
if policy_obj.rules:
|
|
policy_dict['rules'] = []
|
|
for rule in policy_obj.rules:
|
|
if isinstance(rule, dict):
|
|
rule_dict = rule
|
|
else:
|
|
rule_dict = rule.to_dict(recurse=False, render_unsets=True)
|
|
rule_dict['id'] = rule_dict['l7rule_id']
|
|
policy_dict['rules'].append(rule_dict)
|
|
elif not is_update:
|
|
policy_dict['rules'] = []
|
|
|
|
def _remove_unsupported_keys(self, obj_type, obj_dict):
|
|
for key in unsupported_keys.get(obj_type, []):
|
|
if key in obj_dict:
|
|
if obj_dict.get(key):
|
|
LOG.warning("Ignoring %(key)s:%(val)s in %(type)s as the "
|
|
"NSX plugin does not currently support it",
|
|
{'key': key, 'val': obj_dict[key],
|
|
'type': obj_type})
|
|
del obj_dict[key]
|
|
|
|
def obj_to_dict(self, obj, is_update=False, project_id=None):
|
|
obj_type = obj.__class__.__name__
|
|
# create a dictionary out of the object
|
|
render_unsets = bool(not is_update)
|
|
obj_dict = obj.to_dict(recurse=True, render_unsets=render_unsets)
|
|
|
|
# Update the dictionary to match what the nsx driver expects
|
|
if not project_id:
|
|
project_id = self.get_obj_project_id(obj_type, obj_dict)
|
|
obj_dict['tenant_id'] = obj_dict['project_id'] = project_id
|
|
|
|
if 'id' not in obj_dict:
|
|
obj_dict['id'] = obj_dict.get('%s_id' % obj_type.lower())
|
|
|
|
if not obj_dict.get('name') and not is_update:
|
|
obj_dict['name'] = ""
|
|
|
|
self._remove_unsupported_keys(obj_type, obj_dict)
|
|
|
|
if obj_type == 'LoadBalancer':
|
|
# clean listeners and pools for update case:
|
|
if 'listeners' in obj_dict:
|
|
if is_update and not obj_dict['listeners']:
|
|
del obj_dict['listeners']
|
|
else:
|
|
if obj_dict['listeners'] is None:
|
|
obj_dict['listeners'] = []
|
|
for listener in obj_dict['listeners']:
|
|
listener['id'] = listener['listener_id']
|
|
for policy in listener.get('l7policies', []):
|
|
policy['id'] = policy['l7policy_id']
|
|
for rule in policy.get('rules', []):
|
|
rule['id'] = rule['l7rule_id']
|
|
if 'pools' in obj_dict:
|
|
if is_update and not obj_dict['pools']:
|
|
del obj_dict['pools']
|
|
else:
|
|
if obj_dict['pools'] is None:
|
|
obj_dict['pools'] = []
|
|
for pool in obj_dict['pools']:
|
|
pool['id'] = pool['pool_id']
|
|
for member in pool.get('members', []):
|
|
member['id'] = member['member_id']
|
|
if pool.get('healthmonitor'):
|
|
pool['healthmonitor'] = self._get_hm_dict(
|
|
pool['healthmonitor']['healthmonitor_id'],
|
|
is_update)
|
|
pool['tenant_id'] = project_id
|
|
|
|
elif obj_type == 'Listener':
|
|
if 'l7policies' in obj_dict:
|
|
obj_dict['l7_policies'] = obj_dict['l7policies']
|
|
if obj_dict.get('loadbalancer_id'):
|
|
# Generate a loadbalancer object
|
|
obj_dict['loadbalancer'] = self._get_load_balancer_dict(
|
|
obj_dict['loadbalancer_id'])
|
|
if obj_dict.get('default_pool_id'):
|
|
# Generate the default pool object
|
|
obj_dict['default_pool'] = self._get_pool_dict(
|
|
obj_dict['default_pool_id'], is_update, project_id)
|
|
|
|
elif obj_type == 'Pool':
|
|
if 'listener' not in obj_dict:
|
|
self._get_listener_in_pool_dict(obj_dict, is_update)
|
|
if obj_dict.get('healthmonitor'):
|
|
obj_dict['healthmonitor']['id'] = obj_dict[
|
|
'healthmonitor']['healthmonitor_id']
|
|
|
|
elif obj_type == 'Member':
|
|
# Get the pool object
|
|
if obj_dict.get('pool_id'):
|
|
obj_dict['pool'] = self._get_pool_dict(
|
|
obj_dict['pool_id'], is_update)
|
|
obj_dict['loadbalancer'] = None
|
|
if 'loadbalancer' in obj_dict['pool']:
|
|
obj_dict['loadbalancer'] = obj_dict['pool']['loadbalancer']
|
|
if not obj_dict.get('subnet_id'):
|
|
# Use the parent vip_subnet_id instead
|
|
obj_dict['subnet_id'] = obj_dict['loadbalancer'][
|
|
'vip_subnet_id']
|
|
elif not is_update:
|
|
# Do not set pool & LB if in update situation, as we want to
|
|
# use the original data of this member
|
|
obj_dict['pool'] = None
|
|
obj_dict['loadbalancer'] = None
|
|
|
|
elif obj_type == 'HealthMonitor':
|
|
# Get the pool object
|
|
if obj_dict.get('pool_id'):
|
|
obj_dict['pool'] = self._get_pool_dict(
|
|
obj_dict['pool_id'], is_update)
|
|
|
|
elif obj_type == 'L7Policy':
|
|
self.update_policy_dict(obj_dict, obj, is_update=is_update)
|
|
|
|
elif obj_type == 'L7Rule':
|
|
# Get the L7 policy object
|
|
if obj_dict.get('l7policy_id'):
|
|
db_policy = self.repositories.l7policy.get(
|
|
db_apis.get_session(), id=obj_dict['l7policy_id'])
|
|
policy_obj = oct_utils.db_l7policy_to_provider_l7policy(
|
|
db_policy)
|
|
policy_dict = policy_obj.to_dict(
|
|
recurse=True, render_unsets=True)
|
|
policy_dict['id'] = obj_dict['l7policy_id']
|
|
self.update_policy_dict(
|
|
policy_dict, policy_obj, is_update=is_update)
|
|
obj_dict['policy'] = policy_dict
|
|
|
|
LOG.debug("Translated %(type)s to dictionary: %(obj)s",
|
|
{'type': obj_type, 'obj': obj_dict})
|
|
return obj_dict
|
|
|
|
# Load Balancer
|
|
@log_helpers.log_method_call
|
|
def create_vip_port(self, loadbalancer_id, project_id, vip_dictionary):
|
|
raise exceptions.NotImplementedError()
|
|
|
|
@log_helpers.log_method_call
|
|
def loadbalancer_create(self, loadbalancer):
|
|
kw = {'loadbalancer': self.obj_to_dict(loadbalancer)}
|
|
self.client.cast({}, 'loadbalancer_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def loadbalancer_delete(self, loadbalancer, cascade=False):
|
|
kw = {'loadbalancer': self.obj_to_dict(loadbalancer),
|
|
'cascade': cascade}
|
|
self.client.cast({}, 'loadbalancer_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def loadbalancer_failover(self, loadbalancer_id):
|
|
LOG.error('Loadbalancer failover is handled by platform')
|
|
raise exceptions.NotImplementedError()
|
|
|
|
@log_helpers.log_method_call
|
|
def loadbalancer_update(self, old_loadbalancer, new_loadbalancer):
|
|
old_dict = self.obj_to_dict(old_loadbalancer)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_dict.update(self.obj_to_dict(
|
|
new_loadbalancer, is_update=True,
|
|
project_id=old_dict.get('project_id')))
|
|
kw = {'old_loadbalancer': old_dict,
|
|
'new_loadbalancer': new_dict}
|
|
self.client.cast({}, 'loadbalancer_update', **kw)
|
|
|
|
def _create_lb_certificate(self, listener_dict):
|
|
# Extract Octavia certificate data into a dict which is readable by
|
|
# the listener_mgr
|
|
if listener_dict.get('default_tls_container_ref'):
|
|
cert_data = listener_dict.get('default_tls_container_data', {})
|
|
return {'ref': listener_dict.get('default_tls_container_ref'),
|
|
'certificate': cert_data.get('certificate'),
|
|
'private_key': cert_data.get('private_key'),
|
|
'passphrase': cert_data.get('passphrase')}
|
|
|
|
# Listener
|
|
@log_helpers.log_method_call
|
|
def listener_create(self, listener):
|
|
dict_list = self.obj_to_dict(listener)
|
|
cert = self._create_lb_certificate(dict_list)
|
|
kw = {'listener': dict_list, 'cert': cert}
|
|
self.client.cast({}, 'listener_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def listener_delete(self, listener):
|
|
kw = {'listener': self.obj_to_dict(listener)}
|
|
self.client.cast({}, 'listener_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def listener_update(self, old_listener, new_listener):
|
|
old_dict = self.obj_to_dict(old_listener)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_dict.update(self.obj_to_dict(
|
|
new_listener, is_update=True,
|
|
project_id=old_dict.get('project_id')))
|
|
cert = self._create_lb_certificate(new_dict)
|
|
kw = {'old_listener': old_dict,
|
|
'new_listener': new_dict,
|
|
'cert': cert}
|
|
self.client.cast({}, 'listener_update', **kw)
|
|
|
|
# Pool
|
|
@log_helpers.log_method_call
|
|
def pool_create(self, pool):
|
|
kw = {'pool': self.obj_to_dict(pool)}
|
|
self.client.cast({}, 'pool_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def pool_delete(self, pool):
|
|
kw = {'pool': self.obj_to_dict(pool)}
|
|
self.client.cast({}, 'pool_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def pool_update(self, old_pool, new_pool):
|
|
old_dict = self.obj_to_dict(old_pool)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_pool_dict = self.obj_to_dict(
|
|
new_pool, is_update=True, project_id=old_dict.get('project_id'))
|
|
new_dict.update(new_pool_dict)
|
|
kw = {'old_pool': old_dict,
|
|
'new_pool': new_dict}
|
|
self.client.cast({}, 'pool_update', **kw)
|
|
|
|
# Member
|
|
@log_helpers.log_method_call
|
|
def member_create(self, member):
|
|
kw = {'member': self.obj_to_dict(member)}
|
|
self.client.cast({}, 'member_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def member_delete(self, member):
|
|
kw = {'member': self.obj_to_dict(member)}
|
|
self.client.cast({}, 'member_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def member_update(self, old_member, new_member):
|
|
old_dict = self.obj_to_dict(old_member)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_dict.update(self.obj_to_dict(
|
|
new_member, is_update=True, project_id=old_dict.get('project_id')))
|
|
kw = {'old_member': old_dict,
|
|
'new_member': new_dict}
|
|
self.client.cast({}, 'member_update', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def member_batch_update(self, members):
|
|
raise NotImplementedError()
|
|
|
|
# Health Monitor
|
|
@log_helpers.log_method_call
|
|
def health_monitor_create(self, healthmonitor):
|
|
kw = {'healthmonitor': self.obj_to_dict(healthmonitor)}
|
|
self.client.cast({}, 'healthmonitor_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def health_monitor_delete(self, healthmonitor):
|
|
kw = {'healthmonitor': self.obj_to_dict(healthmonitor)}
|
|
self.client.cast({}, 'healthmonitor_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def health_monitor_update(self, old_healthmonitor, new_healthmonitor):
|
|
old_dict = self.obj_to_dict(old_healthmonitor)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_dict.update(self.obj_to_dict(
|
|
new_healthmonitor, is_update=True,
|
|
project_id=old_dict.get('project_id')))
|
|
kw = {'old_healthmonitor': old_dict,
|
|
'new_healthmonitor': new_dict}
|
|
self.client.cast({}, 'healthmonitor_update', **kw)
|
|
|
|
# L7 Policy
|
|
@log_helpers.log_method_call
|
|
def l7policy_create(self, l7policy):
|
|
kw = {'l7policy': self.obj_to_dict(l7policy)}
|
|
self.client.cast({}, 'l7policy_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def l7policy_delete(self, l7policy):
|
|
kw = {'l7policy': self.obj_to_dict(l7policy)}
|
|
self.client.cast({}, 'l7policy_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def l7policy_update(self, old_l7policy, new_l7policy):
|
|
old_dict = self.obj_to_dict(old_l7policy)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_dict.update(self.obj_to_dict(
|
|
new_l7policy, is_update=True,
|
|
project_id=old_dict.get('project_id')))
|
|
kw = {'old_l7policy': old_dict,
|
|
'new_l7policy': new_dict}
|
|
self.client.cast({}, 'l7policy_update', **kw)
|
|
|
|
# L7 Rule
|
|
@log_helpers.log_method_call
|
|
def l7rule_create(self, l7rule):
|
|
kw = {'l7rule': self.obj_to_dict(l7rule)}
|
|
self.client.cast({}, 'l7rule_create', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def l7rule_delete(self, l7rule):
|
|
kw = {'l7rule': self.obj_to_dict(l7rule)}
|
|
self.client.cast({}, 'l7rule_delete', **kw)
|
|
|
|
@log_helpers.log_method_call
|
|
def l7rule_update(self, old_l7rule, new_l7rule):
|
|
old_dict = self.obj_to_dict(old_l7rule)
|
|
new_dict = copy.deepcopy(old_dict)
|
|
new_dict.update(self.obj_to_dict(
|
|
new_l7rule, is_update=True, project_id=old_dict.get('project_id')))
|
|
kw = {'old_l7rule': old_dict,
|
|
'new_l7rule': new_dict}
|
|
self.client.cast({}, 'l7rule_update', **kw)
|
|
|
|
# Flavor
|
|
@log_helpers.log_method_call
|
|
def get_supported_flavor_metadata(self):
|
|
raise exceptions.NotImplementedError()
|
|
|
|
@log_helpers.log_method_call
|
|
def validate_flavor(self, flavor_metadata):
|
|
raise exceptions.NotImplementedError()
|
|
|
|
|
|
class NSXOctaviaDriverEndpoint(driver_lib.DriverLibrary):
|
|
target = messaging.Target(namespace="control", version='1.0')
|
|
|
|
def __init__(self, status_socket=driver_lib.DEFAULT_STATUS_SOCKET,
|
|
stats_socket=driver_lib.DEFAULT_STATS_SOCKET, **kwargs):
|
|
super(NSXOctaviaDriverEndpoint, self).__init__(
|
|
status_socket, stats_socket, **kwargs)
|
|
self.repositories = repositories.Repositories()
|
|
|
|
@log_helpers.log_method_call
|
|
def update_loadbalancer_status(self, ctxt, status):
|
|
# refresh the driver lib session
|
|
self.db_session = db_apis.get_session()
|
|
for member in status.get('members', []):
|
|
if member.get('id'):
|
|
pass
|
|
elif member.get('member_ip') and member.get('pool_id'):
|
|
db_member = self.repositories.member.get(
|
|
self.db_session,
|
|
pool_id=member['pool_id'],
|
|
ip_address=member['member_ip'])
|
|
if db_member:
|
|
member['id'] = db_member.id
|
|
try:
|
|
return super(NSXOctaviaDriverEndpoint,
|
|
self).update_loadbalancer_status(status)
|
|
except exceptions.UpdateStatusError as e:
|
|
LOG.error("Failed to update Octavia loadbalancer status. "
|
|
"Status %s, Error %s", status, e.fault_string)
|
|
|
|
@log_helpers.log_method_call
|
|
def update_listener_statistics(self, ctxt, statistics):
|
|
# refresh the driver lib session
|
|
self.db_session = db_apis.get_session()
|
|
try:
|
|
return super(NSXOctaviaDriverEndpoint,
|
|
self).update_listener_statistics(statistics)
|
|
except exceptions.UpdateStatisticsError as e:
|
|
LOG.error("Failed to update Octavia listener statistics. "
|
|
"Stats %s, Error %s", statistics, e.fault_string)
|
|
|
|
|
|
@log_helpers.log_method_call
|
|
def vmware_nsx_provider_agent(exit_event):
|
|
# Initialize RPC listener
|
|
topic = d_const.DRIVER_TO_OCTAVIA_TOPIC
|
|
server = socket.gethostname()
|
|
target = messaging.Target(topic=topic, server=server,
|
|
exchange="common", fanout=False)
|
|
endpoints = [NSXOctaviaDriverEndpoint()]
|
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
|
get_transport()
|
|
octavia_server = get_rpc_server(target, endpoints, access_policy)
|
|
octavia_server.start()
|
|
|
|
LOG.info('VMware NSX Octavia provider agent has started.')
|
|
while not exit_event.is_set():
|
|
time.sleep(1)
|
|
LOG.info('VMware NSX Octavia provider agent is exiting.')
|