Device component - IETF L2VPN Driver: (b1997768) · Commits · TFS / controller

src/device/service/drivers/init.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -74,6 +74,15 @@ DRIVERS.append(
		#}
		]))

		from .ietf_l2vpn.IetfL2VpnDriver import IetfL2VpnDriver # pylint: disable=wrong-import-position
		DRIVERS.append(
		(IetfL2VpnDriver, [
		{
		FilterFieldEnum.DEVICE_TYPE: DeviceTypeEnum.TERAFLOWSDN_CONTROLLER,
		FilterFieldEnum.DRIVER: DeviceDriverEnum.DEVICEDRIVER_IETF_L2VPN,
		}
		]))

		if LOAD_ALL_DEVICE_DRIVERS:
		from .openconfig.OpenConfigDriver import OpenConfigDriver # pylint: disable=wrong-import-position
		DRIVERS.append(

src/device/service/drivers/ietf_l2vpn/IetfL2VpnDriver.py

+115 −3

Original line number	Diff line number	Diff line
		@@ -12,6 +12,118 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		class IetfL2VpnDriver:
		def __init__(self) -> None:
		pass
		import logging, requests, threading
		from requests.auth import HTTPBasicAuth
		from typing import Any, Iterator, List, Optional, Tuple, Union
		from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
		from common.type_checkers.Checkers import chk_string, chk_type
		from device.service.driver_api._Driver import _Driver
		from . import ALL_RESOURCE_KEYS
		from .Tools import create_connectivity_service, find_key, config_getter, delete_connectivity_service

		LOGGER = logging.getLogger(__name__)

		METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': 'ietf_l2vpn'})

		class IetfL2VpnDriver(_Driver):
		def __init__(self, address: str, port: int, **settings) -> None: # pylint: disable=super-init-not-called
		self.__lock = threading.Lock()
		self.__started = threading.Event()
		self.__terminate = threading.Event()
		username = settings.get('username')
		password = settings.get('password')
		self.__auth = HTTPBasicAuth(username, password) if username is not None and password is not None else None
		scheme = settings.get('scheme', 'http')
		self.__tapi_root = '{:s}://{:s}:{:d}'.format(scheme, address, int(port))
		self.__timeout = int(settings.get('timeout', 120))

		def Connect(self) -> bool:
		url = self.__tapi_root + '/restconf/data/tapi-common:context'
		with self.__lock:
		if self.__started.is_set(): return True
		try:
		requests.get(url, timeout=self.__timeout, verify=False, auth=self.__auth)
		except requests.exceptions.Timeout:
		LOGGER.exception('Timeout connecting {:s}'.format(str(self.__tapi_root)))
		return False
		except Exception: # pylint: disable=broad-except
		LOGGER.exception('Exception connecting {:s}'.format(str(self.__tapi_root)))
		return False
		else:
		self.__started.set()
		return True

		def Disconnect(self) -> bool:
		with self.__lock:
		self.__terminate.set()
		return True

		@metered_subclass_method(METRICS_POOL)
		def GetInitialConfig(self) -> List[Tuple[str, Any]]:
		with self.__lock:
		return []

		@metered_subclass_method(METRICS_POOL)
		def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
		chk_type('resources', resource_keys, list)
		results = []
		with self.__lock:
		if len(resource_keys) == 0: resource_keys = ALL_RESOURCE_KEYS
		for i, resource_key in enumerate(resource_keys):
		str_resource_name = 'resource_key[#{:d}]'.format(i)
		chk_string(str_resource_name, resource_key, allow_empty=False)
		results.extend(config_getter(
		self.__tapi_root, resource_key, timeout=self.__timeout, auth=self.__auth))
		return results

		@metered_subclass_method(METRICS_POOL)
		def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
		results = []
		if len(resources) == 0:
		return results
		with self.__lock:
		for resource in resources:
		LOGGER.info('resource = {:s}'.format(str(resource)))

		input_sip = find_key(resource, 'input_sip')
		output_sip = find_key(resource, 'output_sip')
		uuid = find_key(resource, 'uuid')
		capacity_value = find_key(resource, 'capacity_value')
		capacity_unit = find_key(resource, 'capacity_unit')
		layer_protocol_name = find_key(resource, 'layer_protocol_name')
		layer_protocol_qualifier = find_key(resource, 'layer_protocol_qualifier')
		direction = find_key(resource, 'direction')

		data = create_connectivity_service(
		self.__tapi_root, uuid, input_sip, output_sip, direction, capacity_value, capacity_unit,
		layer_protocol_name, layer_protocol_qualifier, timeout=self.__timeout, auth=self.__auth)
		results.extend(data)
		return results

		@metered_subclass_method(METRICS_POOL)
		def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
		results = []
		if len(resources) == 0: return results
		with self.__lock:
		for resource in resources:
		LOGGER.info('resource = {:s}'.format(str(resource)))
		uuid = find_key(resource, 'uuid')
		results.extend(delete_connectivity_service(
		self.__tapi_root, uuid, timeout=self.__timeout, auth=self.__auth))
		return results

		@metered_subclass_method(METRICS_POOL)
		def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
		# TODO: TAPI does not support monitoring by now
		return [False for _ in subscriptions]

		@metered_subclass_method(METRICS_POOL)
		def UnsubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
		# TODO: TAPI does not support monitoring by now
		return [False for _ in subscriptions]

		def GetState(
		self, blocking=False, terminate : Optional[threading.Event] = None
		) -> Iterator[Tuple[float, str, Any]]:
		# TODO: TAPI does not support monitoring by now
		return []

src/device/service/drivers/ietf_l2vpn/MockOSM.py

0 → 100644

+62 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
		#
		# 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 logging
		from .WimconnectorIETFL2VPN import WimconnectorIETFL2VPN

		LOGGER = logging.getLogger(__name__)

		class MockOSM:
		def __init__(self, url, mapping, username, password):
		wim = {'wim_url': url}
		wim_account = {'user': username, 'password': password}
		config = {'mapping_not_needed': False, 'service_endpoint_mapping': mapping}
		self.wim = WimconnectorIETFL2VPN(wim, wim_account, config=config)
		self.conn_info = {} # internal database emulating OSM storage provided to WIM Connectors

		def create_connectivity_service(self, service_type, connection_points):
		LOGGER.info('[create_connectivity_service] service_type={:s}'.format(str(service_type)))
		LOGGER.info('[create_connectivity_service] connection_points={:s}'.format(str(connection_points)))
		self.wim.check_credentials()
		result = self.wim.create_connectivity_service(service_type, connection_points)
		LOGGER.info('[create_connectivity_service] result={:s}'.format(str(result)))
		service_uuid, conn_info = result
		self.conn_info[service_uuid] = conn_info
		return service_uuid

		def get_connectivity_service_status(self, service_uuid):
		LOGGER.info('[get_connectivity_service] service_uuid={:s}'.format(str(service_uuid)))
		conn_info = self.conn_info.get(service_uuid)
		if conn_info is None: raise Exception('ServiceId({:s}) not found'.format(str(service_uuid)))
		LOGGER.info('[get_connectivity_service] conn_info={:s}'.format(str(conn_info)))
		self.wim.check_credentials()
		result = self.wim.get_connectivity_service_status(service_uuid, conn_info=conn_info)
		LOGGER.info('[get_connectivity_service] result={:s}'.format(str(result)))
		return result

		def edit_connectivity_service(self, service_uuid, connection_points):
		LOGGER.info('[edit_connectivity_service] service_uuid={:s}'.format(str(service_uuid)))
		LOGGER.info('[edit_connectivity_service] connection_points={:s}'.format(str(connection_points)))
		conn_info = self.conn_info.get(service_uuid)
		if conn_info is None: raise Exception('ServiceId({:s}) not found'.format(str(service_uuid)))
		LOGGER.info('[edit_connectivity_service] conn_info={:s}'.format(str(conn_info)))
		self.wim.edit_connectivity_service(service_uuid, conn_info=conn_info, connection_points=connection_points)

		def delete_connectivity_service(self, service_uuid):
		LOGGER.info('[delete_connectivity_service] service_uuid={:s}'.format(str(service_uuid)))
		conn_info = self.conn_info.get(service_uuid)
		if conn_info is None: raise Exception('ServiceId({:s}) not found'.format(str(service_uuid)))
		LOGGER.info('[delete_connectivity_service] conn_info={:s}'.format(str(conn_info)))
		self.wim.check_credentials()
		self.wim.delete_connectivity_service(service_uuid, conn_info=conn_info)

src/device/service/drivers/ietf_l2vpn/Tools.py

0 → 100644

+184 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
		#
		# 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 json, logging, operator, requests
		from requests.auth import HTTPBasicAuth
		from typing import Dict, Optional
		from device.service.driver_api._Driver import RESOURCE_ENDPOINTS

		LOGGER = logging.getLogger(__name__)

		HTTP_OK_CODES = {
		200, # OK
		201, # Created
		202, # Accepted
		204, # No Content
		}

		def find_key(resource, key):
		return json.loads(resource[1])[key]


		def config_getter(
		root_url : str, resource_key : str, auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None
		):
		url = '{:s}/restconf/data/tapi-common:context'.format(root_url)
		result = []
		try:
		response = requests.get(url, timeout=timeout, verify=False, auth=auth)
		except requests.exceptions.Timeout:
		LOGGER.exception('Timeout connecting {:s}'.format(url))
		return result
		except Exception as e: # pylint: disable=broad-except
		LOGGER.exception('Exception retrieving {:s}'.format(resource_key))
		result.append((resource_key, e))
		return result

		try:
		context = json.loads(response.content)
		except Exception as e: # pylint: disable=broad-except
		LOGGER.warning('Unable to decode reply: {:s}'.format(str(response.content)))
		result.append((resource_key, e))
		return result

		if resource_key != RESOURCE_ENDPOINTS: return result

		if 'tapi-common:context' in context:
		context = context['tapi-common:context']
		elif 'context' in context:
		context = context['context']

		for sip in context['service-interface-point']:
		layer_protocol_name = sip.get('layer-protocol-name', '?')
		supportable_spectrum = sip.get('tapi-photonic-media:media-channel-service-interface-point-spec', {})
		supportable_spectrum = supportable_spectrum.get('mc-pool', {})
		supportable_spectrum = supportable_spectrum.get('supportable-spectrum', [])
		supportable_spectrum = supportable_spectrum[0] if len(supportable_spectrum) == 1 else {}
		grid_type = supportable_spectrum.get('frequency-constraint', {}).get('grid-type')
		granularity = supportable_spectrum.get('frequency-constraint', {}).get('adjustment-granularity')
		direction = sip.get('direction', '?')
		endpoint_type = [layer_protocol_name, grid_type, granularity, direction]
		str_endpoint_type = ':'.join(filter(lambda i: operator.is_not(i, None), endpoint_type))
		endpoint_url = '/endpoints/endpoint[{:s}]'.format(sip['uuid'])
		endpoint_data = {'uuid': sip['uuid'], 'type': str_endpoint_type}
		result.append((endpoint_url, endpoint_data))

		return result

		def create_connectivity_service(
		root_url, uuid, input_sip, output_sip, direction, capacity_value, capacity_unit, layer_protocol_name,
		layer_protocol_qualifier,
		auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None
		):

		url = '{:s}/restconf/data/tapi-common:context/tapi-connectivity:connectivity-context'.format(root_url)
		headers = {'content-type': 'application/json'}
		data = {
		'tapi-connectivity:connectivity-service': [
		{
		'uuid': uuid,
		'connectivity-constraint': {
		'requested-capacity': {
		'total-size': {
		'value': capacity_value,
		'unit': capacity_unit
		}
		},
		'connectivity-direction': direction
		},
		'end-point': [
		{
		'service-interface-point': {
		'service-interface-point-uuid': input_sip
		},
		'layer-protocol-name': layer_protocol_name,
		'layer-protocol-qualifier': layer_protocol_qualifier,
		'local-id': input_sip
		},
		{
		'service-interface-point': {
		'service-interface-point-uuid': output_sip
		},
		'layer-protocol-name': layer_protocol_name,
		'layer-protocol-qualifier': layer_protocol_qualifier,
		'local-id': output_sip
		}
		]
		}
		]
		}
		results = []
		try:
		LOGGER.info('Connectivity service {:s}: {:s}'.format(str(uuid), str(data)))
		response = requests.post(
		url=url, data=json.dumps(data), timeout=timeout, headers=headers, verify=False, auth=auth)
		LOGGER.info('TAPI response: {:s}'.format(str(response)))
		except Exception as e: # pylint: disable=broad-except
		LOGGER.exception('Exception creating ConnectivityService(uuid={:s}, data={:s})'.format(str(uuid), str(data)))
		results.append(e)
		else:
		if response.status_code not in HTTP_OK_CODES:
		msg = 'Could not create ConnectivityService(uuid={:s}, data={:s}). status_code={:s} reply={:s}'
		LOGGER.error(msg.format(str(uuid), str(data), str(response.status_code), str(response)))
		results.append(response.status_code in HTTP_OK_CODES)
		return results

		def delete_connectivity_service(root_url, uuid, auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None):
		url = '{:s}/restconf/data/tapi-common:context/tapi-connectivity:connectivity-context/connectivity-service={:s}'
		url = url.format(root_url, uuid)
		results = []
		try:
		response = requests.delete(url=url, timeout=timeout, verify=False, auth=auth)
		except Exception as e: # pylint: disable=broad-except
		LOGGER.exception('Exception deleting ConnectivityService(uuid={:s})'.format(str(uuid)))
		results.append(e)
		else:
		if response.status_code not in HTTP_OK_CODES:
		msg = 'Could not delete ConnectivityService(uuid={:s}). status_code={:s} reply={:s}'
		LOGGER.error(msg.format(str(uuid), str(response.status_code), str(response)))
		results.append(response.status_code in HTTP_OK_CODES)
		return results

		def compose_service_endpoint_id(site_id : str, endpoint_id : Dict):
		device_uuid = endpoint_id['device_id']['device_uuid']['uuid']
		endpoint_uuid = endpoint_id['endpoint_uuid']['uuid']
		return ':'.join([site_id, device_uuid, endpoint_uuid])

		def wim_mapping(site_id, ce_endpoint_id, pe_device_id : Optional[Dict] = None, priority=None, redundant=[]):
		ce_device_uuid = ce_endpoint_id['device_id']['device_uuid']['uuid']
		ce_endpoint_uuid = ce_endpoint_id['endpoint_uuid']['uuid']
		service_endpoint_id = compose_service_endpoint_id(site_id, ce_endpoint_id)
		if pe_device_id is None:
		bearer = '{:s}:{:s}'.format(ce_device_uuid, ce_endpoint_uuid)
		else:
		pe_device_uuid = pe_device_id['device_uuid']['uuid']
		bearer = '{:s}:{:s}'.format(ce_device_uuid, pe_device_uuid)
		mapping = {
		'service_endpoint_id': service_endpoint_id,
		'datacenter_id': site_id, 'device_id': ce_device_uuid, 'device_interface_id': ce_endpoint_uuid,
		'service_mapping_info': {
		'site-id': site_id,
		'bearer': {'bearer-reference': bearer},
		}
		}
		if priority is not None: mapping['service_mapping_info']['priority'] = priority
		if len(redundant) > 0: mapping['service_mapping_info']['redundant'] = redundant
		return service_endpoint_id, mapping

		def connection_point(service_endpoint_id : str, encapsulation_type : str, vlan_id : int):
		return {
		'service_endpoint_id': service_endpoint_id,
		'service_endpoint_encapsulation_type': encapsulation_type,
		'service_endpoint_encapsulation_info': {'vlan': vlan_id}
		}

src/device/service/drivers/ietf_l2vpn/WimconnectorIETFL2VPN.py

0 → 100644

+543 −0

File added.

Preview size limit exceeded, changes collapsed.