x
/
vmware-nsxlib
Code Issues Proposed changes
A common library that interfaces with VMware NSX.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1116 Commits
19 Branches
13 Tags
16 MiB
 
 
Tree: 253a76c527
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '253a76c527'
${ noResults }
vmware-nsxlib/vmware_nsxlib/v3/policy/core_resources.py

4901 lines
203 KiB
Raw Blame History

# Copyright 2017 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 abc
from distutils import version
import sys
import decorator
import eventlet
from oslo_log import log as logging
from oslo_utils import uuidutils
from vmware_nsxlib._i18n import _
from vmware_nsxlib.v3 import exceptions
from vmware_nsxlib.v3 import nsx_constants
from vmware_nsxlib.v3 import utils
from vmware_nsxlib.v3.policy import constants
from vmware_nsxlib.v3.policy import core_defs
from vmware_nsxlib.v3.policy import transaction as trans
from vmware_nsxlib.v3.policy import utils as p_utils
LOG = logging.getLogger(__name__)
# Sentitel object to indicate unspecified attribute value
# None value in attribute would indicate "unset" functionality,
# while "ignore" means that the value not be present in request
# body
IGNORE = object()
DEFAULT_MAP_ID = 'DEFAULT'
@decorator.decorator
def check_allowed_passthrough(f, *args, **kwargs):
resource_api = args[0]
if not resource_api.nsx_api:
caller = sys._getframe(1).f_code.co_name
LOG.error("%s failed: Passthrough api is disabled", caller)
return
return f(*args, **kwargs)
class NsxPolicyResourceBase(object, metaclass=abc.ABCMeta):
"""Abstract class for NSX policy resources
declaring the basic apis each policy resource should support,
and implement some common apis and utilities
"""
SINGLE_ENTRY_ID = 'entry'
def __init__(self, policy_api, nsx_api, version, nsxlib_config):
self.policy_api = policy_api
self.nsx_api = nsx_api
self.version = version
self.nsxlib_config = nsxlib_config
@property
def entry_def(self):
pass
@abc.abstractmethod
def list(self, *args, **kwargs):
pass
@abc.abstractmethod
def get(self, uuid, *args, **kwargs):
pass
@abc.abstractmethod
def delete(self, uuid, *args, **kwargs):
pass
@abc.abstractmethod
def create_or_overwrite(self, *args, **kwargs):
"""Create new or overwrite existing resource
Create would list keys and attributes, set defaults and
perform necessary validations.
If object with same IDs exists on backend, it will
be overridden.
"""
pass
@abc.abstractmethod
def update(self, *args, **kwargs):
"""Update existing resource
Update is different from create since it specifies only
attributes that need changing. Non-updateble attributes
should not be listed as update arguments.
Create_or_overwrite is not
good enough since it sets defaults, and thus would return
non-default values to default if not specified in kwargs.
"""
pass
def _any_arg_set(self, *args):
"""Helper to identify if user specified any of args"""
for arg in args:
if arg != IGNORE:
return True
return False
def _get_user_args(self, **kwargs):
return {key: value for key, value in kwargs.items()
if value != IGNORE}
def _init_def(self, **kwargs):
"""Helper for update function - ignore attrs without explicit value"""
args = self._get_user_args(**kwargs)
return self.entry_def(nsx_version=self.version, **args)
def _init_parent_def(self, **kwargs):
"""Helper for update function - ignore attrs without explicit value"""
args = self._get_user_args(**kwargs)
return self.parent_entry_def(**args)
def _get_and_update_def(self, **kwargs):
"""Helper for update function - ignore attrs without explicit value"""
args = self._get_user_args(**kwargs)
resource_def = self.entry_def(nsx_version=self.version, **args)
body = self.policy_api.get(resource_def)
if body:
resource_def.set_obj_dict(body)
return resource_def
def _update(self, allow_partial_updates=True, force=False, **kwargs):
"""Helper for update function - ignore attrs without explicit value"""
@utils.retry_upon_exception(
exceptions.StaleRevision,
max_attempts=self.policy_api.client.max_attempts)
def _do_update_with_retry():
if (allow_partial_updates and
self.policy_api.partial_updates_supported()):
policy_def = self._init_def(**kwargs)
partial_updates = True
else:
policy_def = self._get_and_update_def(**kwargs)
partial_updates = False
if policy_def.bodyless():
# Nothing to update - only keys provided in kwargs
return
self.policy_api.create_or_update(
policy_def, partial_updates=partial_updates, force=force)
return _do_update_with_retry()
@staticmethod
def _init_obj_uuid(obj_uuid):
if not obj_uuid:
# generate a random id
obj_uuid = str(uuidutils.generate_uuid())
return obj_uuid
def _canonize_name(self, name):
# remove spaces and slashes from objects names
return name.replace(' ', '_').replace('/', '_')
def get_by_name(self, name, *args, **kwargs):
# Return first match by name
resources_list = self.list(*args, **kwargs)
for obj in resources_list:
if obj.get('display_name') == name:
return obj
def _get_realization_info(self, resource_def, entity_type=None,
silent=False):
entities = []
try:
path = resource_def.get_resource_full_path()
entities = self.policy_api.get_realized_entities(
path, silent=silent)
if entities:
if entity_type:
# look for the entry with the right entity_type
for entity in entities:
if entity.get('entity_type') == entity_type:
return entity
else:
# return the first realization entry
# (Useful for resources with single realization entity)
return entities[0]
except exceptions.ResourceNotFound:
pass
# If we got here the resource was not deployed yet
if silent:
LOG.debug("No realization info found for %(path)s type %(type)s: "
"%(entities)s",
{"path": path, "type": entity_type,
"entities": entities})
else:
LOG.warning("No realization info found for %(path)s type %(type)s",
{"path": path, "type": entity_type})
def _get_realized_state(self, resource_def, entity_type=None,
realization_info=None):
if not realization_info:
realization_info = self._get_realization_info(
resource_def, entity_type=entity_type)
if realization_info and realization_info.get('state'):
return realization_info['state']
def _get_realized_id(self, resource_def, entity_type=None,
realization_info=None):
if not realization_info:
realization_info = self._get_realization_info(
resource_def, entity_type=entity_type)
if (realization_info and
realization_info.get('realization_specific_identifier')):
return realization_info['realization_specific_identifier']
def _get_realization_error_message_and_code(self, info):
error_msg = 'unknown'
error_code = None
related_error_codes = []
if info.get('alarms'):
alarm = info['alarms'][0]
error_msg = alarm.get('message')
if alarm.get('error_details'):
error_code = alarm['error_details'].get('error_code')
if alarm['error_details'].get('related_errors'):
related = alarm['error_details']['related_errors']
for err_obj in related:
error_msg = '%s: %s' % (error_msg,
err_obj.get('error_message'))
if err_obj.get('error_code'):
related_error_codes.append(err_obj['error_code'])
return error_msg, error_code, related_error_codes
def _wait_until_realized(self, resource_def, entity_type=None,
sleep=None, max_attempts=None):
"""Wait until the resource has been realized
Return the realization info, or raise an error
"""
if sleep is None:
sleep = self.nsxlib_config.realization_wait_sec
if max_attempts is None:
max_attempts = self.nsxlib_config.realization_max_attempts
info = {}
@utils.retry_upon_none_result(max_attempts, delay=sleep, random=True)
def get_info():
info = self._get_realization_info(
resource_def, entity_type=entity_type, silent=True)
if info:
if info['state'] == constants.STATE_REALIZED:
return info
if info['state'] == constants.STATE_ERROR:
error_msg, error_code, related_error_codes = \
self._get_realization_error_message_and_code(info)
# There could be a delay between setting NSX-T
# Error realization state and updating the realization
# entity with alarms. Retry should be perform upon None
# error code to avoid 'Unknown' RealizationErrorStateError
# exception
if error_code is None:
return
raise exceptions.RealizationErrorStateError(
resource_type=resource_def.resource_type(),
resource_id=resource_def.get_id(),
error=error_msg, error_code=error_code,
related_error_codes=related_error_codes)
try:
return get_info()
except exceptions.RealizationError as e:
raise e
except Exception:
# max retries reached
LOG.error("_wait_until_realized maxed-out for "
"resource: %s. Last realization info was %s",
resource_def.get_resource_full_path(), info)
raise exceptions.RealizationTimeoutError(
resource_type=resource_def.resource_type(),
resource_id=resource_def.get_id(),
attempts=max_attempts,
sleep=sleep)
def _wait_until_state_successful(self, res_def,
sleep=None, max_attempts=None,
with_refresh=False):
res_path = res_def.get_resource_full_path()
state = {}
if sleep is None:
sleep = self.nsxlib_config.realization_wait_sec
if max_attempts is None:
max_attempts = self.nsxlib_config.realization_max_attempts
@utils.retry_upon_none_result(max_attempts, delay=sleep, random=True)
def get_state():
state = self.policy_api.get_intent_consolidated_status(
res_path, silent=True)
if state and state.get('consolidated_status'):
con_state = state['consolidated_status'].get(
'consolidated_status')
if con_state == 'SUCCESS':
return True
if con_state == 'ERROR':
LOG.error("_wait_until_state_successful errored for "
"resource: %s. Last consolidated_status result "
"was %s", res_path, state)
raise exceptions.RealizationErrorStateError(
resource_type=res_def.resource_type(),
resource_id=res_def.get_id(),
error="Unknown")
if with_refresh:
# Refresh the consolidated state for the next time
# (if not, it will be refreshed at the policy level after a
# refresh cycle)
self.policy_api.refresh_realized_state(res_path)
try:
return get_state()
except exceptions.RealizationError as e:
raise e
except Exception:
# max retries reached
LOG.error("_wait_until_state_successful maxed-out for "
"resource: %s. Last consolidated_status result was %s",
res_path, state)
raise exceptions.RealizationTimeoutError(
resource_type=res_def.resource_type(),
resource_id=res_def.get_id(),
attempts=max_attempts,
sleep=sleep)
@check_allowed_passthrough
def _get_realized_id_using_search(self, policy_resource_path,
mp_resource_type, resource_def=None,
entity_type=None, silent=False,
sleep=None, max_attempts=None):
"""Wait until the policy path will be found using search api
And return the NSX ID of the MP resource that was found
"""
if sleep is None:
sleep = self.nsxlib_config.realization_wait_sec
if max_attempts is None:
max_attempts = self.nsxlib_config.realization_max_attempts
check_status = 3
tag = [{'scope': 'policyPath',
'tag': utils.escape_tag_data(policy_resource_path)}]
resources = []
test_num = 0
while test_num < max_attempts:
# Use the search api to find the realization id of this entity.
resources = self.nsx_api.search_by_tags(
tags=tag, resource_type=mp_resource_type,
silent=silent)['results']
if resources:
# If status exists, make sure the state is successful
if (not resources[0].get('status') or
resources[0]['status'].get('state') == 'success'):
return resources[0]['id']
# From time to time also check the Policy realization state,
# as if it is in ERROR waiting should be avoided.
if resource_def and test_num % check_status == (check_status - 1):
info = self._get_realization_info(resource_def,
entity_type=entity_type)
if info and info['state'] == constants.STATE_ERROR:
error_msg, error_code, related_error_codes = \
self._get_realization_error_message_and_code(info)
LOG.error("_get_realized_id_using_search Failed for "
"resource: %s. Got error in realization info %s",
policy_resource_path, info)
raise exceptions.RealizationErrorStateError(
resource_type=resource_def.resource_type(),
resource_id=resource_def.get_id(),
error=error_msg, error_code=error_code,
related_error_codes=related_error_codes)
if (info and info['state'] == constants.STATE_REALIZED and
info.get('realization_specific_identifier')):
LOG.warning("Realization ID for %s was not found via "
"search api although it was realized",
policy_resource_path)
return info['realization_specific_identifier']
eventlet.sleep(sleep)
test_num += 1
# max retries reached
LOG.error("_get_realized_id_using_search maxed-out for "
"resource: %s. Last search result was %s",
policy_resource_path, resources)
raise exceptions.RealizationTimeoutError(
resource_type=mp_resource_type,
resource_id=policy_resource_path,
attempts=max_attempts,
sleep=sleep)
def _get_extended_attr_from_realized_info(self, realization_info,
requested_attr):
# Returns a list. In case a single value is expected,
# caller must extract the first index to retrieve the value
if realization_info:
try:
for attr in realization_info.get('extended_attributes', []):
if attr.get('key') == requested_attr:
return attr.get('values')
except IndexError:
return
def _list(self, obj_def, silent=False):
return self.policy_api.list(obj_def, silent=silent).get('results', [])
def _create_or_store(self, policy_def, child_def=None):
transaction = trans.NsxPolicyTransaction.get_current()
if transaction:
# Store this def for batch apply for this transaction
transaction.store_def(policy_def, self.policy_api.client)
if child_def and not policy_def.mandatory_child_def:
transaction.store_def(child_def, self.policy_api.client)
else:
# No transaction - apply now
# In case the same object was just deleted, or depends on another
# resource, create may need to be retried.
@utils.retry_upon_exception(
(exceptions.NsxPendingDelete, exceptions.StaleRevision),
max_attempts=self.policy_api.client.max_attempts)
def _do_create_with_retry():
if child_def:
self.policy_api.create_with_parent(policy_def, child_def)
else:
self.policy_api.create_or_update(policy_def)
_do_create_with_retry()
def _delete_or_store(self, policy_def):
transaction = trans.NsxPolicyTransaction.get_current()
if transaction:
# Mark this resource is about to be deleted
policy_def.set_delete()
# Set some mandatory default values to avoid failure
# TODO(asarfaty): This can be removed once platform bug is fixed
policy_def.set_default_mandatory_vals()
# Store this def for batch apply for this transaction
transaction.store_def(policy_def, self.policy_api.client)
else:
# No transaction - apply now
self._delete_with_retry(policy_def)
def _delete_with_retry(self, policy_def):
@utils.retry_upon_exception(
exceptions.StaleRevision,
max_attempts=self.policy_api.client.max_attempts)
def do_delete():
self.policy_api.delete(policy_def)
do_delete()
class NsxPolicyDomainApi(NsxPolicyResourceBase):
"""NSX Policy Domain."""
@property
def entry_def(self):
return core_defs.DomainDef
def create_or_overwrite(self, name, domain_id=None,
description=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
domain_id = self._init_obj_uuid(domain_id)
domain_def = self._init_def(domain_id=domain_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
self._create_or_store(domain_def)
return domain_id
def delete(self, domain_id, tenant=constants.POLICY_INFRA_TENANT):
domain_def = core_defs.DomainDef(domain_id=domain_id, tenant=tenant)
self._delete_with_retry(domain_def)
def get(self, domain_id, tenant=constants.POLICY_INFRA_TENANT,
silent=False):
domain_def = core_defs.DomainDef(domain_id=domain_id, tenant=tenant)
return self.policy_api.get(domain_def, silent=silent)
def list(self, tenant=constants.POLICY_INFRA_TENANT):
domain_def = core_defs.DomainDef(tenant=tenant)
return self._list(domain_def)
def update(self, domain_id, name=IGNORE,
description=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(domain_id=domain_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
class NsxPolicyGroupApi(NsxPolicyResourceBase):
"""NSX Policy Group (under a Domain) with condition/s"""
@property
def entry_def(self):
return core_defs.GroupDef
def create_or_overwrite(
self, name, domain_id, group_id=None,
description=IGNORE,
cond_val=None,
cond_key=constants.CONDITION_KEY_TAG,
cond_op=constants.CONDITION_OP_EQUALS,
cond_member_type=constants.CONDITION_MEMBER_PORT,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
"""Create a group with/without a condition.
Empty condition value will result a group with no condition.
"""
group_id = self._init_obj_uuid(group_id)
# Prepare the condition
if cond_val is not None:
condition = core_defs.Condition(value=cond_val,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
conditions = [condition]
else:
conditions = []
group_def = self._init_def(domain_id=domain_id,
group_id=group_id,
name=name,
description=description,
conditions=conditions,
tags=tags,
tenant=tenant)
self._create_or_store(group_def)
return group_id
def build_condition(
self, cond_val=None,
cond_key=constants.CONDITION_KEY_TAG,
cond_op=constants.CONDITION_OP_EQUALS,
cond_member_type=constants.CONDITION_MEMBER_PORT):
return core_defs.Condition(value=cond_val,
key=cond_key,
operator=cond_op,
member_type=cond_member_type)
def build_ip_address_expression(self, ip_addresses):
return core_defs.IPAddressExpression(ip_addresses)
def build_path_expression(self, paths):
return core_defs.PathExpression(paths)
def build_union_condition(self, operator=constants.CONDITION_OP_OR,
conditions=None):
# NSX don't allow duplicate expressions in expression list
# of a group -> (ERROR: Duplicate expressions specified)
# Members of input conditions is either instance of Condition
# or NestedExpression class.
expressions = []
if conditions:
conditions = list(set(conditions))
expressions = []
for cond in conditions:
if len(expressions):
expressions.append(core_defs.ConjunctionOperator(
operator=operator))
expressions.append(cond)
return expressions
def build_nested_condition(
self, operator=constants.CONDITION_OP_AND,
conditions=None):
expressions = self.build_union_condition(
operator=operator, conditions=conditions)
return core_defs.NestedExpression(expressions=expressions)
def create_or_overwrite_with_conditions(
self, name, domain_id, group_id=None,
description=IGNORE,
conditions=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
"""Create a group with a list of conditions.
To build the conditions in the list, build_condition
or build_nested_condition can be used
"""
group_id = self._init_obj_uuid(group_id)
if not conditions:
conditions = []
group_def = self._init_def(domain_id=domain_id,
group_id=group_id,
name=name,
description=description,
conditions=conditions,
tags=tags,
tenant=tenant)
self._create_or_store(group_def)
return group_id
def delete(self, domain_id, group_id,
tenant=constants.POLICY_INFRA_TENANT):
group_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=tenant)
self._delete_with_retry(group_def)
def get(self, domain_id, group_id,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
group_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=tenant)
return self.policy_api.get(group_def, silent=silent)
def list(self, domain_id,
tenant=constants.POLICY_INFRA_TENANT):
"""List all the groups of a specific domain."""
group_def = core_defs.GroupDef(domain_id=domain_id,
tenant=tenant)
return self._list(group_def)
def get_by_name(self, domain_id, name,
tenant=constants.POLICY_INFRA_TENANT):
"""Return first group matched by name of this domain"""
return super(NsxPolicyGroupApi, self).get_by_name(name, domain_id,
tenant=tenant)
def update(self, domain_id, group_id,
name=IGNORE, description=IGNORE,
tags=IGNORE, tenant=constants.POLICY_INFRA_TENANT):
self._update(domain_id=domain_id,
group_id=group_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
def update_with_conditions(
self, domain_id, group_id,
name=IGNORE, description=IGNORE, conditions=IGNORE,
tags=IGNORE, tenant=constants.POLICY_INFRA_TENANT,
update_payload_cbk=None):
group_def = self._init_def(domain_id=domain_id,
group_id=group_id,
name=name,
description=description,
conditions=conditions,
tags=tags,
tenant=tenant)
group_path = group_def.get_resource_path()
@utils.retry_upon_exception(
exceptions.StaleRevision,
max_attempts=self.policy_api.client.max_attempts)
def _update():
# Get the current data of group
group = self.policy_api.get(group_def)
if update_payload_cbk:
# The update_payload_cbk function takes two arguments.
# The first one is the result from the internal GET request.
# The second one is a dict of user-provided attributes,
# which can be changed inside the callback function and
# used as the new payload for the following PUT request.
# For example, users want to combine the new conditions
# passed to update_with_conditions() with the original
# conditions retrieved from the internal GET request
# instead of overriding the original conditions.
update_payload_cbk(group, group_def.attrs)
group_def.set_obj_dict(group)
body = group_def.get_obj_dict()
# Update the entire group at the NSX
self.policy_api.client.update(group_path, body)
_update()
def get_realized_state(self, domain_id, group_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
group_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=tenant)
return self._get_realized_state(group_def, entity_type=entity_type,
realization_info=realization_info)
def get_realized_id(self, domain_id, group_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
group_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=tenant)
return self._get_realized_id(group_def, entity_type=entity_type,
realization_info=realization_info)
def get_realization_info(self, domain_id, group_id, entity_type=None,
silent=False,
tenant=constants.POLICY_INFRA_TENANT):
group_def = core_defs.GroupDef(domain_id=domain_id,
group_id=group_id,
tenant=tenant)
return self._get_realization_info(group_def, entity_type=entity_type,
silent=silent)
def get_path(self, domain_id, group_id,
tenant=constants.POLICY_INFRA_TENANT):
group_def = self.entry_def(domain_id=domain_id,
group_id=group_id,
tenant=tenant)
return group_def.get_resource_full_path()
def wait_until_realized(self, domain_id, group_id,
entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
sleep=None, max_attempts=None):
group_def = self.entry_def(domain_id=domain_id, group_id=group_id,
tenant=tenant)
return self._wait_until_realized(group_def, entity_type=entity_type,
sleep=sleep,
max_attempts=max_attempts)
class NsxPolicyServiceBase(NsxPolicyResourceBase):
"""Base class for NSX Policy Service with a single entry.
Note the nsx-policy backend supports multiple service entries per service.
At this point this is not supported here.
"""
@property
def parent_entry_def(self):
return core_defs.ServiceDef
def delete(self, service_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Delete the service with all its entries"""
service_def = core_defs.ServiceDef(service_id=service_id,
tenant=tenant)
self._delete_with_retry(service_def)
def get(self, service_id,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
service_def = core_defs.ServiceDef(service_id=service_id,
tenant=tenant)
return self.policy_api.get(service_def, silent=silent)
def list(self, tenant=constants.POLICY_INFRA_TENANT):
service_def = core_defs.ServiceDef(tenant=tenant)
return self._list(service_def)
def get_realized_state(self, service_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
service_def = core_defs.ServiceDef(service_id=service_id,
tenant=tenant)
return self._get_realized_state(service_def, entity_type=entity_type,
realization_info=realization_info)
def get_realized_id(self, service_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
service_def = core_defs.ServiceDef(service_id=service_id,
tenant=tenant)
return self._get_realized_id(service_def, entity_type=entity_type,
realization_info=realization_info)
def get_realization_info(self, service_id, entity_type=None,
silent=False,
tenant=constants.POLICY_INFRA_TENANT):
service_def = core_defs.ServiceDef(service_id=service_id,
tenant=tenant)
return self._get_realization_info(service_def,
entity_type=entity_type,
silent=silent)
class NsxPolicyL4ServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with a single L4 service entry.
Note the nsx-policy backend supports multiple service entries per service.
At this point this is not supported here.
"""
@property
def entry_def(self):
return core_defs.L4ServiceEntryDef
def create_or_overwrite(self, name, service_id=None,
description=IGNORE,
protocol=constants.TCP,
dest_ports=IGNORE,
source_ports=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
service_id = self._init_obj_uuid(service_id)
service_def = self._init_parent_def(service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
entry_def = self._init_def(service_id=service_id,
entry_id=self.SINGLE_ENTRY_ID,
name=self.SINGLE_ENTRY_ID,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=tenant)
service_def.mandatory_child_def = entry_def
self._create_or_store(service_def, entry_def)
return service_id
def update(self, service_id,
name=IGNORE, description=IGNORE,
protocol=IGNORE, dest_ports=IGNORE, source_ports=IGNORE,
tags=IGNORE, tenant=constants.POLICY_INFRA_TENANT):
parent_def = self._init_parent_def(
service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
entry_def = self._get_and_update_def(
service_id=service_id,
entry_id=self.SINGLE_ENTRY_ID,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=tenant)
self.policy_api.create_with_parent(parent_def, entry_def)
def build_entry(self, name, service_id, entry_id,
description=None, protocol=None,
dest_ports=None, source_ports=None,
tags=None, tenant=constants.POLICY_INFRA_TENANT):
return self._init_def(service_id=service_id,
entry_id=entry_id,
name=name,
description=description,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tags=tags,
tenant=tenant)
class NsxPolicyIcmpServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with a single ICMP service entry.
Note the nsx-policy backend supports multiple service entries per service.
At this point this is not supported here.
"""
@property
def entry_def(self):
return core_defs.IcmpServiceEntryDef
def create_or_overwrite(self, name, service_id=None,
description=IGNORE,
version=4, icmp_type=IGNORE, icmp_code=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
service_id = self._init_obj_uuid(service_id)
service_def = self._init_parent_def(service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
entry_def = self._init_def(
service_id=service_id,
entry_id=self.SINGLE_ENTRY_ID,
name=self.SINGLE_ENTRY_ID,
version=version,
icmp_type=icmp_type,
icmp_code=icmp_code,
tenant=tenant)
service_def.mandatory_child_def = entry_def
self._create_or_store(service_def, entry_def)
return service_id
def update(self, service_id,
name=IGNORE, description=IGNORE,
version=IGNORE, icmp_type=IGNORE,
icmp_code=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
parent_def = self._init_parent_def(
service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
entry_def = self._get_and_update_def(
service_id=service_id,
entry_id=self.SINGLE_ENTRY_ID,
version=version,
icmp_type=icmp_type,
icmp_code=icmp_code,
tenant=tenant)
return self.policy_api.create_with_parent(parent_def, entry_def)
def build_entry(self, name, service_id, entry_id,
description=None, version=4,
icmp_type=None, icmp_code=None,
tags=None, tenant=constants.POLICY_INFRA_TENANT):
return self._init_def(service_id=service_id,
entry_id=entry_id,
name=name,
description=description,
version=version,
icmp_type=icmp_type,
icmp_code=icmp_code,
tags=tags,
tenant=tenant)
class NsxPolicyIPProtocolServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with a single IPProtocol service entry.
Note the nsx-policy backend supports multiple service entries per service.
At this point this is not supported here.
"""
@property
def entry_def(self):
return core_defs.IPProtocolServiceEntryDef
def create_or_overwrite(self, name, service_id=None,
description=IGNORE,
protocol_number=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
service_id = self._init_obj_uuid(service_id)
service_def = self._init_parent_def(service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
entry_def = self._init_def(
service_id=service_id,
entry_id=self.SINGLE_ENTRY_ID,
name=self.SINGLE_ENTRY_ID,
protocol_number=protocol_number,
tenant=tenant)
service_def.mandatory_child_def = entry_def
self._create_or_store(service_def, entry_def)
return service_id
def update(self, service_id,
name=IGNORE, description=IGNORE,
protocol_number=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
parent_def = self._init_parent_def(
service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
entry_def = self._get_and_update_def(
service_id=service_id,
entry_id=self.SINGLE_ENTRY_ID,
protocol_number=protocol_number,
tenant=tenant)
return self.policy_api.create_with_parent(parent_def, entry_def)
def build_entry(self, name, service_id, entry_id,
description=None, protocol_number=None,
tags=None, tenant=constants.POLICY_INFRA_TENANT):
return self._init_def(service_id=service_id,
entry_id=entry_id,
name=name,
protocol_number=protocol_number,
tags=tags,
tenant=tenant)
class NsxPolicyMixedServiceApi(NsxPolicyServiceBase):
"""NSX Policy Service with mixed service entries."""
def create_or_overwrite(self, name, service_id,
description=IGNORE,
entries=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
service_def = self._init_parent_def(service_id=service_id,
name=name,
description=description,
entries=entries,
tags=tags,
tenant=tenant)
if entries != IGNORE:
self._create_or_store(service_def, entries)
else:
self._create_or_store(service_def)
return service_id
def update(self, service_id,
name=IGNORE, description=IGNORE,
entries=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
parent_def = self._init_parent_def(
service_id=service_id,
name=name,
description=description,
tags=tags,
tenant=tenant)
if entries != IGNORE:
self.policy_api.create_with_parent(parent_def, entries)
else:
self.policy_api.create_or_update(parent_def)
class NsxPolicyTier1Api(NsxPolicyResourceBase):
"""NSX Tier1 API """
LOCALE_SERVICE_SUFF = '-0'
@property
def entry_def(self):
return core_defs.Tier1Def
def build_route_advertisement(self, static_routes=False, subnets=False,
nat=False, lb_vip=False, lb_snat=False,
ipsec_endpoints=False):
return core_defs.RouteAdvertisement(static_routes=static_routes,
subnets=subnets,
nat=nat,
lb_vip=lb_vip,
lb_snat=lb_snat,
ipsec_endpoints=ipsec_endpoints)
def create_or_overwrite(self, name, tier1_id=None,
description=IGNORE,
tier0=IGNORE,
force_whitelisting=IGNORE,
failover_mode=constants.NON_PREEMPTIVE,
route_advertisement=IGNORE,
dhcp_config=IGNORE,
disable_firewall=IGNORE,
ipv6_ndra_profile_id=IGNORE,
pool_allocation=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
tier1_id = self._init_obj_uuid(tier1_id)
tier1_def = self._init_def(tier1_id=tier1_id,
name=name,
description=description,
tier0=tier0,
force_whitelisting=force_whitelisting,
tags=tags,
failover_mode=failover_mode,
route_advertisement=route_advertisement,
dhcp_config=dhcp_config,
disable_firewall=disable_firewall,
ipv6_ndra_profile_id=ipv6_ndra_profile_id,
pool_allocation=pool_allocation,
tenant=tenant)
self._create_or_store(tier1_def)
return tier1_id
def delete(self, tier1_id, tenant=constants.POLICY_INFRA_TENANT):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
self._delete_with_retry(tier1_def)
def get(self, tier1_id, tenant=constants.POLICY_INFRA_TENANT,
silent=False):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
return self.policy_api.get(tier1_def, silent=silent)
def get_path(self, tier1_id, tenant=constants.POLICY_INFRA_TENANT):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
return tier1_def.get_resource_full_path()
def list(self, tenant=constants.POLICY_INFRA_TENANT):
tier1_def = self.entry_def(tenant=tenant)
return self._list(tier1_def)
def update(self, tier1_id, name=IGNORE, description=IGNORE,
force_whitelisting=IGNORE,
failover_mode=IGNORE, tier0=IGNORE,
dhcp_config=IGNORE, tags=IGNORE,
enable_standby_relocation=IGNORE,
disable_firewall=IGNORE,
ipv6_ndra_profile_id=IGNORE,
route_advertisement=IGNORE,
route_advertisement_rules=IGNORE,
pool_allocation=IGNORE,
tenant=constants.POLICY_INFRA_TENANT,
current_body=None):
self._update(tier1_id=tier1_id,
name=name,
description=description,
force_whitelisting=force_whitelisting,
failover_mode=failover_mode,
dhcp_config=dhcp_config,
tier0=tier0,
enable_standby_relocation=enable_standby_relocation,
disable_firewall=disable_firewall,
ipv6_ndra_profile_id=ipv6_ndra_profile_id,
route_advertisement=route_advertisement,
route_advertisement_rules=route_advertisement_rules,
pool_allocation=pool_allocation,
tags=tags,
tenant=tenant)
def update_route_advertisement(
self, tier1_id,
static_routes=None,
subnets=None,
nat=None,
lb_vip=None,
lb_snat=None,
ipsec_endpoints=None,
tier0=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
tier1_dict = self.get(tier1_id, tenant)
route_adv = self.entry_def.get_route_adv(tier1_dict)
route_adv.update(static_routes=static_routes,
subnets=subnets,
nat=nat,
lb_vip=lb_vip,
lb_snat=lb_snat,
ipsec_endpoints=ipsec_endpoints)
self.update(tier1_id,
route_advertisement=route_adv,
tier0=tier0,
tenant=tenant)
def add_advertisement_rule(
self, tier1_id, name, action=None, prefix_operator=None,
route_advertisement_types=None, subnets=None,
tenant=constants.POLICY_INFRA_TENANT):
tier1_dict = self.get(tier1_id, tenant)
adv_rules = tier1_dict.get('route_advertisement_rules', [])
adv_rules = [r for r in adv_rules if r.get('name') != name]
adv_rule = core_defs.RouteAdvertisementRule(
name=name, action=action, prefix_operator=prefix_operator,
route_advertisement_types=route_advertisement_types,
subnets=subnets)
adv_rules.append(adv_rule)
self.update(tier1_id,
route_advertisement_rules=adv_rules,
tenant=tenant,
current_body=tier1_dict)
def remove_advertisement_rule(self, tier1_id, name,
tenant=constants.POLICY_INFRA_TENANT):
tier1_dict = self.get(tier1_id, tenant)
adv_rules = tier1_dict.get('route_advertisement_rules', [])
updated_adv_rules = [r for r in adv_rules if r.get('name') != name]
if updated_adv_rules != adv_rules:
self.update(tier1_id,
route_advertisement_rules=updated_adv_rules,
tenant=tenant,
current_body=tier1_dict)
def build_advertisement_rule(self, name, action=None, prefix_operator=None,
route_advertisement_types=None, subnets=None):
return core_defs.RouteAdvertisementRule(
name=name, action=action, prefix_operator=prefix_operator,
route_advertisement_types=route_advertisement_types,
subnets=subnets)
def update_advertisement_rules(self, tier1_id, rules=None,
name_prefix=None,
tenant=constants.POLICY_INFRA_TENANT):
"""Update the router advertisement rules
If name_prefix is None, replace the entire list of NSX rules with the
new given 'rules'.
Else - delete the NSX rules with this name prefix, and add 'rules' to
the rest.
"""
tier1_dict = self.get(tier1_id, tenant)
current_rules = tier1_dict.get('route_advertisement_rules', [])
if name_prefix:
# delete rules with this prefix:
new_rules = []
for rule in current_rules:
if (not rule.get('name') or
not rule['name'].startswith(name_prefix)):
new_rules.append(rule)
# add new rules if provided
if rules:
new_rules.extend(rules)
else:
new_rules = rules
self.update(tier1_id,
route_advertisement_rules=new_rules,
tenant=tenant,
current_body=tier1_dict)
@staticmethod
def _locale_service_id(tier1_id):
# Supporting only a single locale-service per router for now
# with the same id as the router id with a constant suffix
return tier1_id + NsxPolicyTier1Api.LOCALE_SERVICE_SUFF
def create_locale_service(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
t1service_def = core_defs.Tier1LocaleServiceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
tenant=tenant)
self._create_or_store(t1service_def)
def delete_locale_service(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
t1service_def = core_defs.Tier1LocaleServiceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
tenant=tenant)
self._delete_with_retry(t1service_def)
def set_edge_cluster_path(self, tier1_id, edge_cluster_path,
tenant=constants.POLICY_INFRA_TENANT):
t1service_def = core_defs.Tier1LocaleServiceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
edge_cluster_path=edge_cluster_path,
tenant=tenant)
self._create_or_store(t1service_def)
def remove_edge_cluster(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Reset the path in the locale-service (deleting it is not allowed)"""
t1service_def = core_defs.Tier1LocaleServiceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
edge_cluster_path="",
tenant=tenant)
self.policy_api.create_or_update(t1service_def)
def get_edge_cluster_path(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
t1service_def = core_defs.Tier1LocaleServiceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
tenant=tenant)
try:
t1service = self.policy_api.get(t1service_def)
return t1service.get('edge_cluster_path')
except exceptions.ResourceNotFound:
return
def get_edge_cluster_path_by_searching(
self, tier1_id, tenant=constants.POLICY_INFRA_TENANT):
"""Get the edge_cluster path of a Tier1 router"""
services = self.get_locale_tier1_services(tier1_id, tenant=tenant)
for srv in services:
if 'edge_cluster_path' in srv:
return srv['edge_cluster_path']
def get_locale_tier1_services(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
t1service_def = core_defs.Tier1LocaleServiceDef(
tier1_id=tier1_id,
tenant=constants.POLICY_INFRA_TENANT)
return self.policy_api.list(t1service_def)['results']
def add_segment_interface(self, tier1_id, interface_id, segment_id,
subnets, ipv6_ndra_profile_id=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
args = {'tier1_id': tier1_id,
'service_id': self._locale_service_id(tier1_id),
'interface_id': interface_id,
'segment_id': segment_id,
'subnets': subnets,
'tenant': tenant}
if ipv6_ndra_profile_id != IGNORE:
args['ipv6_ndra_profile_id'] = ipv6_ndra_profile_id
t1interface_def = core_defs.Tier1InterfaceDef(**args)
self.policy_api.create_or_update(t1interface_def)
def remove_segment_interface(self, tier1_id, interface_id,
tenant=constants.POLICY_INFRA_TENANT):
t1interface_def = core_defs.Tier1InterfaceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
interface_id=interface_id,
tenant=tenant)
self._delete_with_retry(t1interface_def)
def list_segment_interface(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
t1interface_def = core_defs.Tier1InterfaceDef(
tier1_id=tier1_id,
service_id=self._locale_service_id(tier1_id),
tenant=tenant)
return self._list(t1interface_def)
def get_realized_state(self, tier1_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
return self._get_realized_state(tier1_def, entity_type=entity_type,
realization_info=realization_info)
def get_realized_id(self, tier1_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
if self.nsx_api:
# Use MP search api to find the LR ID as it is faster
return self._get_realized_id_using_search(
self.get_path(tier1_id, tenant=tenant),
self.nsx_api.logical_router.resource_type,
resource_def=tier1_def, entity_type=entity_type)
return self._get_realized_id(tier1_def, entity_type=entity_type,
realization_info=realization_info)
def get_realization_info(self, tier1_id, entity_type=None,
silent=False,
tenant=constants.POLICY_INFRA_TENANT):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
return self._get_realization_info(tier1_def, silent=silent)
def wait_until_realized(self, tier1_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
sleep=None, max_attempts=None):
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
return self._wait_until_realized(tier1_def, entity_type=entity_type,
sleep=sleep,
max_attempts=max_attempts)
@check_allowed_passthrough
def update_transport_zone(self, tier1_id, transport_zone_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Use the pass-through api to update the TZ zone on the NSX router"""
realization_info = self.wait_until_realized(
tier1_id, entity_type='RealizedLogicalRouter', tenant=tenant)
nsx_router_uuid = self.get_realized_id(
tier1_id, tenant=tenant, realization_info=realization_info)
self.nsx_api.logical_router.update(
nsx_router_uuid,
transport_zone_id=transport_zone_id)
@check_allowed_passthrough
def _get_realized_downlink_port(
self, tier1_id, segment_id,
tenant=constants.POLICY_INFRA_TENANT,
sleep=None, max_attempts=None):
"""Return the realized ID of a tier1 downlink port of a segment
If not found, wait until it has been realized
"""
if sleep is None:
sleep = self.nsxlib_config.realization_wait_sec
if max_attempts is None:
max_attempts = self.nsxlib_config.realization_max_attempts
tier1_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
path = tier1_def.get_resource_full_path()
test_num = 0
while test_num < max_attempts:
# get all the realized resources of the tier1
entities = self.policy_api.get_realized_entities(path)
for e in entities:
# Look for router ports
if (e['entity_type'] == 'RealizedLogicalRouterPort' and
e['state'] == constants.STATE_REALIZED):
# Get the NSX port to check if its the downlink port
port = self.nsx_api.logical_router_port.get(
e['realization_specific_identifier'])
# compare the segment ID to the port display name as this
# is the way policy sets it
port_type = port.get('resource_type')
if (port_type == nsx_constants.LROUTERPORT_DOWNLINK and
segment_id in port.get('display_name', '')):
return port['id']
eventlet.sleep(sleep)
test_num += 1
raise exceptions.DetailedRealizationTimeoutError(
resource_type='Tier1',
resource_id=tier1_id,
realized_type="downlink port",
related_type="segment",
related_id=segment_id,
attempts=max_attempts,
sleep=sleep)
@check_allowed_passthrough
def set_dhcp_relay(self, tier1_id, segment_id, relay_service_uuid,
tenant=constants.POLICY_INFRA_TENANT):
"""Set relay service on the nsx logical router port
Using passthrough api, as the policy api does not support this yet
"""
downlink_port_id = self._get_realized_downlink_port(
tier1_id, segment_id, tenant=tenant)
self.nsx_api.logical_router_port.update(
downlink_port_id, relay_service_uuid=relay_service_uuid)
def set_standby_relocation(self, tier1_id,
enable_standby_relocation=True,
tenant=constants.POLICY_INFRA_TENANT):
"""Set the flag for standby relocation on the Tier1 router
"""
return self.update(tier1_id,
enable_standby_relocation=enable_standby_relocation,
tenant=tenant)
class NsxPolicyTier0Api(NsxPolicyResourceBase):
"""NSX Tier0 API """
@property
def entry_def(self):
return core_defs.Tier0Def
def create_or_overwrite(self, name, tier0_id=None,
description=IGNORE,
ha_mode=constants.ACTIVE_ACTIVE,
failover_mode=constants.NON_PREEMPTIVE,
dhcp_config=IGNORE,
force_whitelisting=IGNORE,
default_rule_logging=IGNORE,
transit_subnets=IGNORE,
disable_firewall=IGNORE,
ipv6_ndra_profile_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
tier0_id = self._init_obj_uuid(tier0_id)
tier0_def = self._init_def(tier0_id=tier0_id,
name=name,
description=description,
ha_mode=ha_mode,
failover_mode=failover_mode,
dhcp_config=dhcp_config,
force_whitelisting=force_whitelisting,
default_rule_logging=default_rule_logging,
transit_subnets=transit_subnets,
disable_firewall=disable_firewall,
ipv6_ndra_profile_id=ipv6_ndra_profile_id,
tags=tags,
tenant=tenant)
self.policy_api.create_or_update(tier0_def)
return tier0_id
def delete(self, tier0_id, tenant=constants.POLICY_INFRA_TENANT):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
self._delete_with_retry(tier0_def)
def get(self, tier0_id, tenant=constants.POLICY_INFRA_TENANT,
silent=False):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
return self.policy_api.get(tier0_def, silent=silent)
def get_path(self, tier0_id, tenant=constants.POLICY_INFRA_TENANT):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
return tier0_def.get_resource_full_path()
def list(self, tenant=constants.POLICY_INFRA_TENANT):
tier0_def = self.entry_def(tenant=tenant)
return self._list(tier0_def)
def update(self, tier0_id, name=IGNORE, description=IGNORE,
failover_mode=IGNORE,
dhcp_config=IGNORE,
force_whitelisting=IGNORE,
default_rule_logging=IGNORE,
transit_subnets=IGNORE,
disable_firewall=IGNORE,
ipv6_ndra_profile_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(tier0_id=tier0_id,
name=name,
description=description,
failover_mode=failover_mode,
dhcp_config=dhcp_config,
force_whitelisting=force_whitelisting,
default_rule_logging=default_rule_logging,
transit_subnets=transit_subnets,
disable_firewall=disable_firewall,
ipv6_ndra_profile_id=ipv6_ndra_profile_id,
tags=tags,
tenant=tenant)
def get_locale_services(self, tier0_id,
tenant=constants.POLICY_INFRA_TENANT):
t0service_def = core_defs.Tier0LocaleServiceDef(
tier0_id=tier0_id,
tenant=constants.POLICY_INFRA_TENANT)
return self.policy_api.list(t0service_def)['results']
def get_edge_cluster_path(self, tier0_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Get the edge_cluster path of a Tier0 router"""
services = self.get_locale_services(tier0_id, tenant=tenant)
for srv in services:
if 'edge_cluster_path' in srv:
return srv['edge_cluster_path']
@check_allowed_passthrough
def get_overlay_transport_zone(
self, tier0_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Use the pass-through api to get the TZ zone of the NSX tier0"""
realization_info = self.wait_until_realized(
tier0_id, entity_type='RealizedLogicalRouter', tenant=tenant)
nsx_router_uuid = self.get_realized_id(
tier0_id, tenant=tenant,
realization_info=realization_info)
return self.nsx_api.router.get_tier0_router_overlay_tz(
nsx_router_uuid)
def get_realized_state(self, tier0_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
return self._get_realized_state(tier0_def, entity_type=entity_type,
realization_info=realization_info)
def get_realized_id(self, tier0_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
return self._get_realized_id(tier0_def, entity_type=entity_type,
realization_info=realization_info)
def get_realization_info(self, tier0_id, entity_type=None,
silent=False,
tenant=constants.POLICY_INFRA_TENANT):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
return self._get_realization_info(tier0_def, entity_type=entity_type,
silent=silent)
def wait_until_realized(self, tier0_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
sleep=None, max_attempts=None):
tier0_def = self.entry_def(tier0_id=tier0_id, tenant=tenant)
return self._wait_until_realized(tier0_def, entity_type=entity_type,
sleep=sleep,
max_attempts=max_attempts)
@check_allowed_passthrough
def get_transport_zones(self, tier0_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Return a list of the transport zones IDs connected to the tier0
Currently this is supported only with the passthrough api
"""
realization_info = self.wait_until_realized(
tier0_id, entity_type='RealizedLogicalRouter', tenant=tenant)
nsx_router_uuid = self.get_realized_id(
tier0_id, tenant=tenant,
realization_info=realization_info)
return self.nsx_api.router.get_tier0_router_tz(
nsx_router_uuid)
def _get_uplink_subnets(self, tier0_id,
tenant=constants.POLICY_INFRA_TENANT):
subnets = []
services = self.get_locale_services(tier0_id, tenant=tenant)
for srv in services:
# get the interfaces of this service
t0interface_def = core_defs.Tier0InterfaceDef(
tier0_id=tier0_id,
service_id=srv['id'],
tenant=constants.POLICY_INFRA_TENANT)
interfaces = self.policy_api.list(
t0interface_def).get('results', [])
for interface in interfaces:
if interface.get('type') == 'EXTERNAL':
subnets.extend(interface.get('subnets', []))
return subnets
def get_uplink_ips(self, tier0_id, tenant=constants.POLICY_INFRA_TENANT):
"""Return a link of all uplink ips of this tier0 router"""
subnets = self._get_uplink_subnets(tier0_id, tenant=tenant)
uplink_ips = []
for subnet in subnets:
uplink_ips.extend(subnet.get('ip_addresses', []))
return uplink_ips
def get_uplink_cidrs(self, tier0_id, tenant=constants.POLICY_INFRA_TENANT):
"""Return a link of all uplink cidrs of this tier0 router"""
subnets = self._get_uplink_subnets(tier0_id, tenant=tenant)
cidrs = []
for subnet in subnets:
for ip_address in subnet.get('ip_addresses'):
cidrs.append('%s/%s' % (ip_address,
subnet.get('prefix_len')))
return cidrs
def get_bgp_config(self, tier0_id, tenant=constants.POLICY_INFRA_TENANT):
services = self.get_locale_services(tier0_id, tenant=tenant)
for srv in services:
bgpconfig_def = core_defs.BgpRoutingConfigDef(
tier0_id=tier0_id,
service_id=srv['id'],
tenant=constants.POLICY_INFRA_TENANT)
try:
return self.policy_api.get(bgpconfig_def)
except exceptions.ResourceNotFound:
continue
def build_route_redistribution_rule(self, name=None, types=None,
route_map_path=None):
return core_defs.Tier0RouteRedistributionRule(
name, types, route_map_path)
def build_route_redistribution_config(self, enabled=None, rules=None):
return core_defs.Tier0RouteRedistributionConfig(enabled, rules)
def get_route_redistribution_config(self, tier0_id,
tenant=constants.POLICY_INFRA_TENANT):
services = self.get_locale_services(tier0_id, tenant=tenant)
for srv in services:
if srv.get('route_redistribution_config'):
return srv['route_redistribution_config']
def update_route_redistribution_config(
self, tier0_id, redistribution_config, service_id=None,
tenant=constants.POLICY_INFRA_TENANT):
if not service_id:
# Update on the first locale service
services = self.get_locale_services(tier0_id, tenant=tenant)
if len(services) > 0:
service_id = services[0]['id']
if not service_id:
err_msg = (_("Cannot update route redistribution config without "
"locale service on Tier0 router"))
raise exceptions.ManagerError(details=err_msg)
service_def = core_defs.Tier0LocaleServiceDef(
nsx_version=self.version,
tier0_id=tier0_id,
service_id=service_id,
route_redistribution_config=redistribution_config,
tenant=tenant)
self.policy_api.create_or_update(service_def)
class NsxPolicyTier0NatRuleApi(NsxPolicyResourceBase):
DEFAULT_NAT_ID = 'USER'
@property
def entry_def(self):
return core_defs.Tier0NatRule
def create_or_overwrite(self, name, tier0_id,
nat_id=DEFAULT_NAT_ID,
nat_rule_id=None,
description=IGNORE,
source_network=IGNORE,
destination_network=IGNORE,
translated_network=IGNORE,
firewall_match=constants.NAT_FIREWALL_MATCH_BYPASS,
action=IGNORE,
sequence_number=IGNORE,
log=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT,
enabled=IGNORE):
nat_rule_id = self._init_obj_uuid(nat_rule_id)
nat_rule_def = self._init_def(tier0_id=tier0_id,
nat_id=nat_id,
nat_rule_id=nat_rule_id,
name=name,
description=description,
source_network=source_network,
destination_network=destination_network,
translated_network=translated_network,
firewall_match=firewall_match,
action=action,
sequence_number=sequence_number,
log=log,
tags=tags,
tenant=tenant,
enabled=enabled)
self._create_or_store(nat_rule_def)
return nat_rule_id
def delete(self, tier0_id, nat_rule_id, nat_id=DEFAULT_NAT_ID,
tenant=constants.POLICY_INFRA_TENANT):
nat_rule_def = self.entry_def(tier0_id=tier0_id, nat_id=nat_id,
nat_rule_id=nat_rule_id, tenant=tenant)
self._delete_with_retry(nat_rule_def)
def get(self, tier0_id, nat_rule_id, nat_id=DEFAULT_NAT_ID,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
nat_rule_def = self.entry_def(tier0_id=tier0_id, nat_id=nat_id,
nat_rule_id=nat_rule_id, tenant=tenant)
return self.policy_api.get(nat_rule_def, silent=silent)
def list(self, tier0_id, nat_id=DEFAULT_NAT_ID,
tenant=constants.POLICY_INFRA_TENANT):
nat_rule_def = self.entry_def(tier0_id=tier0_id, nat_id=nat_id,
tenant=tenant)
return self._list(nat_rule_def)
def update(self, tier0_id, nat_rule_id,
nat_id=DEFAULT_NAT_ID,
name=IGNORE,
description=IGNORE,
source_network=IGNORE,
destination_network=IGNORE,
translated_network=IGNORE,
firewall_match=IGNORE,
action=IGNORE,
sequence_number=IGNORE,
log=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT,
enabled=IGNORE):
self._update(tier0_id=tier0_id,
nat_id=nat_id,
nat_rule_id=nat_rule_id,
name=name,
description=description,
source_network=source_network,
destination_network=destination_network,
translated_network=translated_network,
firewall_match=firewall_match,
action=action,
sequence_number=sequence_number,
log=log,
tags=tags,
tenant=tenant,
enabled=enabled)
class NsxPolicyTier1NatRuleApi(NsxPolicyResourceBase):
DEFAULT_NAT_ID = 'USER'
@property
def entry_def(self):
return core_defs.Tier1NatRule
def create_or_overwrite(self, name, tier1_id,
nat_id=DEFAULT_NAT_ID,
nat_rule_id=None,
description=IGNORE,
source_network=IGNORE,
destination_network=IGNORE,
translated_network=IGNORE,
firewall_match=constants.NAT_FIREWALL_MATCH_BYPASS,
action=IGNORE,
sequence_number=IGNORE,
log=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT,
enabled=IGNORE):
nat_rule_id = self._init_obj_uuid(nat_rule_id)
nat_rule_def = self._init_def(tier1_id=tier1_id,
nat_id=nat_id,
nat_rule_id=nat_rule_id,
name=name,
description=description,
source_network=source_network,
destination_network=destination_network,
translated_network=translated_network,
firewall_match=firewall_match,
action=action,
sequence_number=sequence_number,
log=log,
tags=tags,
tenant=tenant,
enabled=enabled)
self._create_or_store(nat_rule_def)
return nat_rule_id
def delete(self, tier1_id, nat_rule_id, nat_id=DEFAULT_NAT_ID,
tenant=constants.POLICY_INFRA_TENANT):
nat_rule_def = self.entry_def(tier1_id=tier1_id, nat_id=nat_id,
nat_rule_id=nat_rule_id, tenant=tenant)
self._delete_or_store(nat_rule_def)
def get(self, tier1_id, nat_rule_id, nat_id=DEFAULT_NAT_ID,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
nat_rule_def = self.entry_def(tier1_id=tier1_id, nat_id=nat_id,
nat_rule_id=nat_rule_id, tenant=tenant)
return self.policy_api.get(nat_rule_def, silent=silent)
def list(self, tier1_id, nat_id=DEFAULT_NAT_ID,
tenant=constants.POLICY_INFRA_TENANT):
nat_rule_def = self.entry_def(tier1_id=tier1_id, nat_id=nat_id,
tenant=tenant)
return self._list(nat_rule_def)
def update(self, tier1_id, nat_rule_id,
nat_id=DEFAULT_NAT_ID,
name=IGNORE,
description=IGNORE,
source_network=IGNORE,
destination_network=IGNORE,
translated_network=IGNORE,
firewall_match=IGNORE,
action=IGNORE,
sequence_number=IGNORE,
log=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT,
enabled=IGNORE):
self._update(tier1_id=tier1_id,
nat_id=nat_id,
nat_rule_id=nat_rule_id,
name=name,
description=description,
source_network=source_network,
destination_network=destination_network,
translated_network=translated_network,
firewall_match=firewall_match,
action=action,
sequence_number=sequence_number,
log=log,
tags=tags,
tenant=tenant,
enabled=enabled)
class NsxPolicyTier1StaticRouteApi(NsxPolicyResourceBase):
@property
def entry_def(self):
return core_defs.Tier1StaticRoute
def create_or_overwrite(self, name, tier1_id,
static_route_id=None,
description=IGNORE,
network=IGNORE,
next_hop=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
static_route_id = self._init_obj_uuid(static_route_id)
static_route_def = self._init_def(tier1_id=tier1_id,
static_route_id=static_route_id,
name=name,
description=description,
network=network,
next_hop=next_hop,
tags=tags,
tenant=tenant)
self._create_or_store(static_route_def)
return static_route_id
def delete(self, tier1_id, static_route_id,
tenant=constants.POLICY_INFRA_TENANT):
static_route_def = self.entry_def(tier1_id=tier1_id,
static_route_id=static_route_id,
tenant=tenant)
self._delete_with_retry(static_route_def)
def get(self, tier1_id, static_route_id,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
static_route_def = self.entry_def(tier1_id=tier1_id,
static_route_id=static_route_id,
tenant=tenant)
return self.policy_api.get(static_route_def, silent=silent)
def list(self, tier1_id,
tenant=constants.POLICY_INFRA_TENANT):
static_route_def = self.entry_def(tier1_id=tier1_id,
tenant=tenant)
return self._list(static_route_def)
def update(self, tier1_id, static_route_id,
name=IGNORE,
description=IGNORE,
network=IGNORE,
next_hop=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(tier1_id=tier1_id,
static_route_id=static_route_id,
name=name,
description=description,
network=network,
next_hop=next_hop,
tags=tags,
tenant=tenant)
class NsxPolicyTier1SegmentApi(NsxPolicyResourceBase):
"""NSX Tier1 Segment API """
@property
def entry_def(self):
return core_defs.Tier1SegmentDef
def build_subnet(self, gateway_address, dhcp_ranges=None,
dhcp_config=None):
return core_defs.Subnet(gateway_address, dhcp_ranges, dhcp_config)
def build_dhcp_config_v4(self, server_address, dns_servers=None,
lease_time=None, options=None):
return core_defs.SegmentDhcpConfigV4(server_address, dns_servers,
lease_time, options)
def build_dhcp_config_v6(self, server_address, dns_servers=None,
lease_time=None, domain_names=None):
return core_defs.SegmentDhcpConfigV6(server_address, dns_servers,
lease_time, domain_names)
def create_or_overwrite(self, name, tier1_id,
segment_id=None,
description=IGNORE,
subnets=IGNORE,
dhcp_config=IGNORE,
dns_domain_name=IGNORE,
vlan_ids=IGNORE,
default_rule_logging=IGNORE,
ip_pool_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
segment_id = self._init_obj_uuid(segment_id)
segment_def = self._init_def(tier1_id=tier1_id,
segment_id=segment_id,
name=name,
description=description,
subnets=subnets,
dhcp_config=dhcp_config,
dns_domain_name=dns_domain_name,
vlan_ids=vlan_ids,
default_rule_logging=default_rule_logging,
ip_pool_id=ip_pool_id,
tags=tags,
tenant=tenant)
self._create_or_store(segment_def)
return segment_id
def delete(self, tier1_id, segment_id,
tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(tier1_id=tier1_id,
segment_id=segment_id,
tenant=tenant)
self._delete_with_retry(segment_def)
def get(self, tier1_id, segment_id,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
segment_def = self.entry_def(tier1_id=tier1_id,
segment_id=segment_id,
tenant=tenant)
return self.policy_api.get(segment_def, silent=silent)
def list(self, tier1_id, tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(tier1_id=tier1_id, tenant=tenant)
return self._list(segment_def)
def update(self, tier1_id, segment_id,
name=IGNORE,
description=IGNORE,
subnets=IGNORE,
dhcp_config=IGNORE,
dns_domain_name=IGNORE,
vlan_ids=IGNORE,
default_rule_logging=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(tier1_id=tier1_id,
segment_id=segment_id,
name=name,
description=description,
subnets=subnets,
dhcp_config=dhcp_config,
dns_domain_name=dns_domain_name,
vlan_ids=vlan_ids,
default_rule_logging=default_rule_logging,
tags=tags,
tenant=tenant)
class NsxPolicySegmentApi(NsxPolicyResourceBase):
"""NSX Infra Segment API """
@property
def entry_def(self):
return core_defs.SegmentDef
def build_subnet(self, gateway_address, dhcp_ranges=None,
dhcp_config=None):
return core_defs.Subnet(gateway_address, dhcp_ranges, dhcp_config)
def build_dhcp_config_v4(self, server_address, dns_servers=None,
lease_time=None, options=None):
return core_defs.SegmentDhcpConfigV4(server_address, dns_servers,
lease_time, options)
def build_dhcp_config_v6(self, server_address, dns_servers=None,
lease_time=None, domain_names=None):
return core_defs.SegmentDhcpConfigV6(server_address, dns_servers,
lease_time, domain_names)
def create_or_overwrite(self, name,
segment_id=None,
tier1_id=IGNORE,
tier0_id=IGNORE,
description=IGNORE,
subnets=IGNORE,
dns_domain_name=IGNORE,
vlan_ids=IGNORE,
transport_zone_id=IGNORE,
ip_pool_id=IGNORE,
metadata_proxy_id=IGNORE,
dhcp_server_config_id=IGNORE,
admin_state=IGNORE,
ls_id=IGNORE,
unique_id=IGNORE,
ep_id=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
if tier0_id != IGNORE and tier1_id != IGNORE:
err_msg = (_("Cannot connect Segment to a Tier-0 and Tier-1 "
"Gateway simultaneously"))
raise exceptions.InvalidInput(details=err_msg)
segment_id = self._init_obj_uuid(segment_id)
segment_def = self._init_def(
segment_id=segment_id,
name=name,
description=description,
tier1_id=tier1_id,
tier0_id=tier0_id,
subnets=subnets,
dns_domain_name=dns_domain_name,
vlan_ids=vlan_ids,
transport_zone_id=transport_zone_id,
ip_pool_id=ip_pool_id,
metadata_proxy_id=metadata_proxy_id,
dhcp_server_config_id=dhcp_server_config_id,
admin_state=admin_state,
ls_id=ls_id,
unique_id=unique_id,
ep_id=ep_id,
tags=tags,
tenant=tenant)
self._create_or_store(segment_def)
return segment_id
def delete(self, segment_id,
tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
@utils.retry_upon_exception(
exceptions.NsxSegemntWithVM,
delay=self.nsxlib_config.realization_wait_sec,
max_attempts=self.nsxlib_config.realization_max_attempts)
def do_delete():
self._delete_with_retry(segment_def)
do_delete()
def get(self, segment_id,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return self.policy_api.get(segment_def, silent=silent)
def list(self, tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(tenant=tenant)
return self._list(segment_def)
def update(self, segment_id, name=IGNORE, description=IGNORE,
tier1_id=IGNORE, tier0_id=IGNORE, subnets=IGNORE,
dns_domain_name=IGNORE,
vlan_ids=IGNORE, metadata_proxy_id=IGNORE,
dhcp_server_config_id=IGNORE, admin_state=IGNORE,
tags=IGNORE, tenant=constants.POLICY_INFRA_TENANT):
self._update(segment_id=segment_id,
name=name,
description=description,
tier1_id=tier1_id,
tier0_id=tier0_id,
subnets=subnets,
dns_domain_name=dns_domain_name,
vlan_ids=vlan_ids,
metadata_proxy_id=metadata_proxy_id,
dhcp_server_config_id=dhcp_server_config_id,
admin_state=admin_state,
tags=tags,
tenant=tenant)
def remove_connectivity_and_subnets(
self, segment_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Disconnect a segment from a router and remove its subnets.
PATCH does not support this action so PUT is used for this
"""
# Get the current segment and update it
segment = self.get(segment_id)
segment['subnets'] = None
segment['connectivity_path'] = None
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
path = segment_def.get_resource_path()
self.policy_api.client.update(path, segment)
def remove_connectivity_path(self, segment_id,
tenant=constants.POLICY_INFRA_TENANT):
"""Disconnect a segment from a router.
PATCH does not support this action so PUT is used for this
"""
# Get the current segment and update it
segment = self.get(segment_id)
segment['connectivity_path'] = None
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
path = segment_def.get_resource_path()
self.policy_api.client.update(path, segment)
def get_realized_state(self, segment_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return self._get_realized_state(segment_def, entity_type=entity_type,
realization_info=realization_info)
def get_realized_id(self, segment_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return self._get_realized_id(segment_def, entity_type=entity_type,
realization_info=realization_info)
def get_path(self, segment_id, tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return segment_def.get_resource_full_path()
def get_realized_logical_switch_id(self, segment_id,
tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
entity_type = 'RealizedLogicalSwitch'
if self.nsx_api:
# Use MP search api to find the LS ID as it is faster
return self._get_realized_id_using_search(
self.get_path(segment_id, tenant=tenant),
self.nsx_api.logical_switch.resource_type,
resource_def=segment_def, entity_type=entity_type)
realization_info = self._wait_until_realized(
segment_def, entity_type=entity_type)
return self._get_realized_id(segment_def,
realization_info=realization_info)
def get_realization_info(self, segment_id, entity_type=None,
silent=False,
tenant=constants.POLICY_INFRA_TENANT):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return self._get_realization_info(segment_def,
entity_type=entity_type,
silent=silent)
def wait_until_realized(self, segment_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
sleep=None, max_attempts=None):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return self._wait_until_realized(segment_def, entity_type=entity_type,
sleep=sleep,
max_attempts=max_attempts)
def wait_until_state_successful(self, segment_id,
tenant=constants.POLICY_INFRA_TENANT,
sleep=None, max_attempts=None,
with_refresh=False):
segment_def = self.entry_def(segment_id=segment_id, tenant=tenant)
self._wait_until_state_successful(segment_def, sleep=sleep,
max_attempts=max_attempts,
with_refresh=with_refresh)
@check_allowed_passthrough
def set_admin_state(self, segment_id, admin_state,
tenant=constants.POLICY_INFRA_TENANT):
"""Set the segment admin state using the passthrough/policy api"""
if (version.LooseVersion(self.version) >=
version.LooseVersion(nsx_constants.NSX_VERSION_3_0_0)):
return self.update(segment_id, admin_state=admin_state,
tenant=tenant)
realization_info = self.wait_until_realized(
segment_id, entity_type='RealizedLogicalSwitch', tenant=tenant)
nsx_ls_uuid = self.get_realized_id(
segment_id, tenant=tenant, realization_info=realization_info)
self.nsx_api.logical_switch.update(
nsx_ls_uuid,
admin_state=admin_state)
def get_transport_zone_id(self, segment_id,
tenant=constants.POLICY_INFRA_TENANT):
segment = self.get(segment_id, tenant=tenant)
tz_path = segment.get('transport_zone_path')
if tz_path:
return p_utils.path_to_id(tz_path)
class NsxPolicySegmentPortApi(NsxPolicyResourceBase):
"""NSX Segment Port API """
@property
def entry_def(self):
return core_defs.SegmentPortDef
def build_address_binding(self, ip_address, mac_address,
vlan_id=None):
return core_defs.PortAddressBinding(ip_address,
mac_address,
vlan_id)
def create_or_overwrite(self, name,
segment_id,
port_id=None,
description=IGNORE,
address_bindings=IGNORE,
attachment_type=IGNORE,
vif_id=IGNORE,
app_id=IGNORE,
context_id=IGNORE,
traffic_tag=IGNORE,
allocate_addresses=IGNORE,
hyperbus_mode=IGNORE,
admin_state=IGNORE,
init_state=IGNORE,
extra_configs=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
port_id = self._init_obj_uuid(port_id)
port_def = self._init_def(segment_id=segment_id,
port_id=port_id,
name=name,
description=description,
address_bindings=address_bindings,
attachment_type=attachment_type,
vif_id=vif_id,
app_id=app_id,
context_id=context_id,
traffic_tag=traffic_tag,
allocate_addresses=allocate_addresses,
hyperbus_mode=hyperbus_mode,
admin_state=admin_state,
init_state=init_state,
extra_configs=extra_configs,
tags=tags,
tenant=tenant)
self._create_or_store(port_def)
return port_id
def delete(self, segment_id, port_id,
tenant=constants.POLICY_INFRA_TENANT):
port_def = self.entry_def(segment_id=segment_id,
port_id=port_id,
tenant=tenant)
self._delete_with_retry(port_def)
def get(self, segment_id, port_id,
tenant=constants.POLICY_INFRA_TENANT,
silent=False):
port_def = self.entry_def(segment_id=segment_id,
port_id=port_id,
tenant=tenant)
return self.policy_api.get(port_def, silent=silent)
def list(self, segment_id, tenant=constants.POLICY_INFRA_TENANT):
port_def = self.entry_def(segment_id=segment_id, tenant=tenant)
return self._list(port_def)
def update(self, segment_id, port_id,
name=IGNORE,
description=IGNORE,
address_bindings=IGNORE,
hyperbus_mode=IGNORE,
admin_state=IGNORE,
extra_configs=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
self._update(segment_id=segment_id,
port_id=port_id,
name=name,
description=description,
address_bindings=address_bindings,
hyperbus_mode=hyperbus_mode,
admin_state=admin_state,
extra_configs=extra_configs,
tags=tags,
tenant=tenant)
def detach(self, segment_id, port_id, vif_id=None, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
"""Reset the attachment with or without a vif_id"""