Device component: (0de242d5) · Commits · TFS / controller

src/device/service/drivers/init.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -82,7 +82,7 @@ DRIVERS.append(
		]))


		from .ietf_l3vpn.driver import IetfL3VpnDriver # pylint: disable=wrong-import-position
		from .ietf_l3vpn.IetfL3VpnDriver import IetfL3VpnDriver # pylint: disable=wrong-import-position
		DRIVERS.append(
		(IetfL3VpnDriver, [
		{

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

+19 −39

Original line number	Diff line number	Diff line
		@@ -13,21 +13,13 @@
		# limitations under the License.

		import logging, requests
		from requests.auth import HTTPBasicAuth
		from typing import Dict, List, Optional
		from common.tools.client.RestClient import RestClient
		from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum

		GET_DEVICES_URL = '{:s}://{:s}:{:d}/tfs-api/devices'
		GET_LINKS_URL = '{:s}://{:s}:{:d}/tfs-api/links'

		TIMEOUT = 30

		HTTP_OK_CODES = {
		200, # OK
		201, # Created
		202, # Accepted
		204, # No Content
		}
		GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'
		GET_DEVICES_URL = '/tfs-api/devices'
		GET_LINKS_URL = '/tfs-api/links'

		MAPPING_STATUS = {
		'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,
		@@ -54,23 +46,23 @@ MAPPING_DRIVER = {
		'DEVICEDRIVER_NCE' : 15,
		}

		MSG_ERROR = 'Could not retrieve devices in remote TeraFlowSDN instance({:s}). status_code={:s} reply={:s}'

		LOGGER = logging.getLogger(__name__)

		class TfsApiClient:
		class TfsApiClient(RestClient):
		def __init__(
		self, address : str, port : int, scheme : str = 'http',
		username : Optional[str] = None, password : Optional[str] = None
		username : Optional[str] = None, password : Optional[str] = None,
		timeout : Optional[int] = 30
		) -> None:
		self._devices_url = GET_DEVICES_URL.format(scheme, address, port)
		self._links_url = GET_LINKS_URL.format(scheme, address, port)
		self._auth = (
		HTTPBasicAuth(username, password)
		if username is not None and password is not None
		else None
		super().__init__(
		address, port, scheme=scheme, username=username, password=password,
		timeout=timeout, verify_certs=False, allow_redirects=True, logger=LOGGER
		)

		def check_credentials(self) -> None:
		self.get(GET_CONTEXT_IDS_URL, expected_status_codes={requests.codes['OK']})
		LOGGER.info('Credentials checked')

		def get_devices_endpoints(
		self, import_topology : ImportTopologyEnum = ImportTopologyEnum.DEVICES
		) -> List[Dict]:
		@@ -78,20 +70,14 @@ class TfsApiClient:
		MSG = '[get_devices_endpoints] import_topology={:s}'
		LOGGER.debug(MSG.format(str(import_topology)))

		reply = requests.get(self._devices_url, timeout=TIMEOUT, verify=False, auth=self._auth)
		if reply.status_code not in HTTP_OK_CODES:
		msg = MSG_ERROR.format(
		str(self._devices_url), str(reply.status_code), str(reply)
		)
		LOGGER.error(msg)
		raise Exception(msg)

		if import_topology == ImportTopologyEnum.DISABLED:
		MSG = 'Unsupported import_topology mode: {:s}'
		raise Exception(MSG.format(str(import_topology)))

		devices = self.get(GET_DEVICES_URL, expected_status_codes={requests.codes['OK']})

		result = list()
		for json_device in reply.json()['devices']:
		for json_device in devices['devices']:
		device_uuid : str = json_device['device_id']['device_uuid']['uuid']
		device_type : str = json_device['device_type']
		#if not device_type.startswith('emu-'): device_type = 'emu-' + device_type
		@@ -124,15 +110,9 @@ class TfsApiClient:
		LOGGER.debug('[get_devices_endpoints] devices only; returning')
		return result

		reply = requests.get(self._links_url, timeout=TIMEOUT, verify=False, auth=self._auth)
		if reply.status_code not in HTTP_OK_CODES:
		msg = MSG_ERROR.format(
		str(self._links_url), str(reply.status_code), str(reply)
		)
		LOGGER.error(msg)
		raise Exception(msg)
		links = self.get(GET_LINKS_URL, expected_status_codes={requests.codes['OK']})

		for json_link in reply.json()['links']:
		for json_link in links['links']:
		link_uuid : str = json_link['link_id']['link_uuid']['uuid']
		link_url = '/links/link[{:s}]'.format(link_uuid)
		link_endpoint_ids = [

src/device/service/drivers/ietf_l3vpn/driver.py→src/device/service/drivers/ietf_l3vpn/IetfL3VpnDriver.py

+52 −91

Original line number	Diff line number	Diff line
		# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
		# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (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.
		@@ -12,41 +12,20 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import json
		import logging
		import re
		import threading
		from typing import Any, Iterator, List, Optional, Tuple, Union

		import anytree
		import requests
		from requests.auth import HTTPBasicAuth

		import anytree, json, logging, re, requests, threading
		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_length, chk_string, chk_type
		from device.service.driver_api._Driver import (
		RESOURCE_ENDPOINTS,
		RESOURCE_SERVICES,
		_Driver,
		)
		from device.service.driver_api.AnyTreeTools import (
		TreeNode,
		dump_subtree,
		get_subnode,
		set_subnode_value,
		)
		from device.service.driver_api.ImportTopologyEnum import (
		ImportTopologyEnum,
		get_import_topology,
		)

		from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS, RESOURCE_SERVICES
		from device.service.driver_api.AnyTreeTools import TreeNode, dump_subtree, get_subnode, set_subnode_value
		from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum, get_import_topology
		from .Constants import SPECIAL_RESOURCE_MAPPINGS
		from .TfsApiClient import TfsApiClient
		from .Tools import compose_resource_endpoint

		LOGGER = logging.getLogger(__name__)


		ALL_RESOURCE_KEYS = [
		RESOURCE_ENDPOINTS,
		RESOURCE_SERVICES,
		@@ -57,9 +36,8 @@ RE_GET_ENDPOINT_FROM_INTERFACE = re.compile(r"^\/interface\[([^\]]+)\].*")
		RE_IETF_L3VPN_DATA = re.compile(r"^\/service\[[^\]]+\]\/IETFL3VPN$")
		RE_IETF_L3VPN_OPERATION = re.compile(r"^\/service\[[^\]]+\]\/IETFL3VPN\/operation$")

		DRIVER_NAME = "ietf_l3vpn"
		METRICS_POOL = MetricsPool("Device", "Driver", labels={"driver": DRIVER_NAME})

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

		class IetfL3VpnDriver(_Driver):
		def __init__(self, address : str, port : str, **settings) -> None:
		@@ -67,30 +45,25 @@ class IetfL3VpnDriver(_Driver):
		self.__lock = threading.Lock()
		self.__started = threading.Event()
		self.__terminate = threading.Event()
		self.__running = TreeNode(".")
		scheme = self.settings.get("scheme", "http")
		username = self.settings.get("username")
		password = self.settings.get("password")
		self.__running = TreeNode('.')
		username = self.settings.get('username')
		password = self.settings.get('password')
		scheme = self.settings.get('scheme', 'http')
		timeout = int(self.settings.get('timeout', 60))
		self.tac = TfsApiClient(
		self.address,
		self.port,
		scheme=scheme,
		username=username,
		password=password,
		)
		self.__auth = None
		# (
		# HTTPBasicAuth(username, password)
		# if username is not None and password is not None
		# else None
		# )
		self.__tfs_nbi_root = "{:s}://{:s}:{:d}".format(
		scheme, self.address, int(self.port)
		)
		self.__timeout = int(self.settings.get("timeout", 120))
		self.__import_topology = get_import_topology(
		self.settings, default=ImportTopologyEnum.DEVICES
		self.address, self.port, scheme=scheme, username=username,
		password=password, timeout=timeout
		)
		#self.__tfs_nbi_root = "{:s}://{:s}:{:d}".format(scheme, self.address, int(self.port))

		# Options are:
		# disabled --> just import endpoints as usual
		# devices --> imports sub-devices but not links connecting them.
		# (a remotely-controlled transport domain might exist between them)
		# topology --> imports sub-devices and links connecting them.
		# (not supported by XR driver)
		self.__import_topology = get_import_topology(self.settings, default=ImportTopologyEnum.DEVICES)

		endpoints = self.settings.get("endpoints", [])
		endpoint_resources = []
		for endpoint in endpoints:
		@@ -139,20 +112,12 @@ class IetfL3VpnDriver(_Driver):
		return results

		def Connect(self) -> bool:
		url = (
		self.__tfs_nbi_root + "/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services"
		)
		with self.__lock:
		if self.__started.is_set():
		return True
		if self.__started.is_set(): return True
		try:
		# requests.get(url, timeout=self.__timeout, auth=self.__auth)
		...
		except requests.exceptions.Timeout:
		LOGGER.exception("Timeout connecting {:s}".format(url))
		return False
		except Exception: # pylint: disable=broad-except
		LOGGER.exception("Exception connecting {:s}".format(url))
		self.tac.check_credentials()
		except: # pylint: disable=bare-except
		LOGGER.exception('Exception checking credentials')
		return False
		else:
		self.__started.set()
		@@ -172,39 +137,36 @@ class IetfL3VpnDriver(_Driver):
		def GetConfig(
		self, resource_keys : List[str] = []
		) -> List[Tuple[str, Union[Any, None, Exception]]]:
		chk_type("resources", resource_keys, list)
		with self.__lock:
		if len(resource_keys) == 0:
		return dump_subtree(self.__running)
		chk_type('resources', resource_keys, list)
		results = []
		resolver = anytree.Resolver(pathattr="name")
		with self.__lock:
		self.tac.check_credentials()
		if len(resource_keys) == 0: resource_keys = ALL_RESOURCE_KEYS
		#if len(resource_keys) == 0:
		# return dump_subtree(self.__running)
		resolver = anytree.Resolver(pathattr='name')
		for i, resource_key in enumerate(resource_keys):
		str_resource_name = "resource_key[#{:d}]".format(i)
		str_resource_name = 'resource_key[#{:d}]'.format(i)
		try:
		chk_string(str_resource_name, resource_key, allow_empty=False)
		if resource_key == RESOURCE_ENDPOINTS:
		# return endpoints through TFS NBI API and list-devices method
		results.extend(self.tac.get_devices_endpoints(self.__import_topology))
		else:
		resource_key = SPECIAL_RESOURCE_MAPPINGS.get(
		resource_key, resource_key
		)
		resource_path = resource_key.split("/")
		except Exception as e: # pylint: disable=broad-except
		LOGGER.exception(
		"Exception validating {:s}: {:s}".format(
		str_resource_name, str(resource_key)
		)
		)
		results.append(
		(resource_key, e)
		) # if validation fails, store the exception
		continue

		resource_path = resource_key.split('/')
		resource_node = get_subnode(
		resolver, self.__running, resource_path, default=None
		)
		# if not found, resource_node is None
		if resource_node is None:
		continue
		if resource_node is None: continue
		results.extend(dump_subtree(resource_node))
		return results
		except Exception as e:
		MSG = 'Unhandled error processing {:s}: resource_key({:s})'
		LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))
		results.append((resource_key, e))
		return results

		@metered_subclass_method(METRICS_POOL)
		@@ -212,8 +174,7 @@ class IetfL3VpnDriver(_Driver):
		self, resources : List[Tuple[str, Any]]
		) -> List[Union[bool, Exception]]:
		results = []
		if len(resources) == 0:
		return results
		if len(resources) == 0: return results
		with self.__lock:
		for resource in resources:
		resource_key, resource_value = resource
		@@ -224,7 +185,7 @@ class IetfL3VpnDriver(_Driver):
		else:
		raise Exception("operation type not found in resources")
		for resource in resources:
		LOGGER.info("resource = {:s}".format(str(resource)))
		LOGGER.info('resource = {:s}'.format(str(resource)))
		resource_key, resource_value = resource
		if not RE_IETF_L3VPN_DATA.match(resource_key):
		continue
		@@ -292,18 +253,18 @@ class IetfL3VpnDriver(_Driver):
		def SubscribeState(
		self, subscriptions : List[Tuple[str, float, float]]
		) -> List[Union[bool, Exception]]:
		# TODO: IETF L3VPN does not support monitoring by now
		# TODO: 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: IETF L3VPN does not support monitoring by now
		# TODO: 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: IETF L3VPN does not support monitoring by now
		# TODO: does not support monitoring by now
		return []

src/device/service/drivers/ietf_l3vpn/TfsApiClient.py

+44 −63

Original line number	Diff line number	Diff line
		@@ -13,22 +13,14 @@
		# limitations under the License.

		import logging, requests
		from requests.auth import HTTPBasicAuth
		from typing import Dict, List, Optional
		from common.tools.client.RestClient import RestClient
		from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum

		GET_DEVICES_URL = '{:s}://{:s}:{:d}/tfs-api/devices'
		GET_LINKS_URL = '{:s}://{:s}:{:d}/tfs-api/links'
		L3VPN_URL = '{:s}://{:s}:{:d}/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services'

		TIMEOUT = 30

		HTTP_OK_CODES = {
		200, # OK
		201, # Created
		202, # Accepted
		204, # No Content
		}
		GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'
		GET_DEVICES_URL = '/tfs-api/devices'
		GET_LINKS_URL = '/tfs-api/links'
		L3VPN_URL = '/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services'

		MAPPING_STATUS = {
		'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,
		@@ -55,24 +47,23 @@ MAPPING_DRIVER = {
		'DEVICEDRIVER_NCE' : 15,
		}

		MSG_ERROR = 'Could not retrieve devices in remote TeraFlowSDN instance({:s}). status_code={:s} reply={:s}'

		LOGGER = logging.getLogger(__name__)

		class TfsApiClient:
		class TfsApiClient(RestClient):
		def __init__(
		self, address : str, port : int, scheme : str = 'http',
		username : Optional[str] = None, password : Optional[str] = None
		username : Optional[str] = None, password : Optional[str] = None,
		timeout : Optional[int] = 30
		) -> None:
		self._devices_url = GET_DEVICES_URL.format(scheme, address, port)
		self._links_url = GET_LINKS_URL.format(scheme, address, port)
		self._l3vpn_url = L3VPN_URL.format(scheme, address, port)
		self._auth = (
		HTTPBasicAuth(username, password)
		if username is not None and password is not None
		else None
		super().__init__(
		address, port, scheme=scheme, username=username, password=password,
		timeout=timeout, verify_certs=False, allow_redirects=True, logger=LOGGER
		)

		def check_credentials(self) -> None:
		self.get(GET_CONTEXT_IDS_URL, expected_status_codes={requests.codes['OK']})
		LOGGER.info('Credentials checked')

		def get_devices_endpoints(
		self, import_topology : ImportTopologyEnum = ImportTopologyEnum.DEVICES
		) -> List[Dict]:
		@@ -80,20 +71,14 @@ class TfsApiClient:
		MSG = '[get_devices_endpoints] import_topology={:s}'
		LOGGER.debug(MSG.format(str(import_topology)))

		reply = requests.get(self._devices_url, timeout=TIMEOUT, verify=False, auth=self._auth)
		if reply.status_code not in HTTP_OK_CODES:
		msg = MSG_ERROR.format(
		str(self._devices_url), str(reply.status_code), str(reply)
		)
		LOGGER.error(msg)
		raise Exception(msg)

		if import_topology == ImportTopologyEnum.DISABLED:
		MSG = 'Unsupported import_topology mode: {:s}'
		raise Exception(MSG.format(str(import_topology)))

		devices = self.get(GET_DEVICES_URL, expected_status_codes={requests.codes['OK']})

		result = list()
		for json_device in reply.json()['devices']:
		for json_device in devices['devices']:
		device_uuid : str = json_device['device_id']['device_uuid']['uuid']
		device_type : str = json_device['device_type']
		#if not device_type.startswith('emu-'): device_type = 'emu-' + device_type
		@@ -126,15 +111,9 @@ class TfsApiClient:
		LOGGER.debug('[get_devices_endpoints] devices only; returning')
		return result

		reply = requests.get(self._links_url, timeout=TIMEOUT, verify=False, auth=self._auth)
		if reply.status_code not in HTTP_OK_CODES:
		msg = MSG_ERROR.format(
		str(self._links_url), str(reply.status_code), str(reply)
		)
		LOGGER.error(msg)
		raise Exception(msg)
		links = self.get(GET_LINKS_URL, expected_status_codes={requests.codes['OK']})

		for json_link in reply.json()['links']:
		for json_link in links['links']:
		link_uuid : str = json_link['link_id']['link_uuid']['uuid']
		link_url = '/links/link[{:s}]'.format(link_uuid)
		link_endpoint_ids = [
		@@ -155,28 +134,30 @@ class TfsApiClient:
		return result

		def create_connectivity_service(self, l3vpn_data : dict) -> None:
		try:
		requests.post(self._l3vpn_url, json=l3vpn_data)
		MSG = '[create_connectivity_service] l3vpn_data={:s}'
		LOGGER.debug(MSG.format(str(l3vpn_data)))
		except requests.exceptions.ConnectionError:
		raise Exception('Failed to send POST request to TFS L3VPN NBI')
		try:
		self.post(L3VPN_URL, body=l3vpn_data)
		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send POST request to TFS L3VPN NBI'
		raise Exception(MSG) from e

		def update_connectivity_service(self, l3vpn_data : dict) -> None:
		vpn_id = l3vpn_data['ietf-l3vpn-svc:l3vpn-svc']['vpn-services']['vpn-service'][0]['vpn-id']
		url = self._l3vpn_url + f'/vpn-service={vpn_id}'
		try:
		requests.put(url, json=l3vpn_data)
		MSG = '[update_connectivity_service] l3vpn_data={:s}'
		LOGGER.debug(MSG.format(str(l3vpn_data)))
		except requests.exceptions.ConnectionError:
		raise Exception('Failed to send PUT request to TFS L3VPN NBI')
		vpn_id = l3vpn_data['ietf-l3vpn-svc:l3vpn-svc']['vpn-services']['vpn-service'][0]['vpn-id']
		try:
		self.put(L3VPN_URL + f'/vpn-service={vpn_id}', body=l3vpn_data)
		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send PUT request to TFS L3VPN NBI'
		raise Exception(MSG) from e

		def delete_connectivity_service(self, service_uuid : str) -> None:
		url = self._l3vpn_url + f'/vpn-service={service_uuid}'
		try:
		requests.delete(url)
		url = L3VPN_URL + f'/vpn-service={service_uuid}'
		MSG = '[delete_connectivity_service] url={:s}'
		LOGGER.debug(MSG.format(str(url)))
		except requests.exceptions.ConnectionError:
		raise Exception('Failed to send DELETE request to TFS L3VPN NBI')
		try:
		self.delete(url)
		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send DELETE request to TFS L3VPN NBI'
		raise Exception(MSG) from e

src/device/service/drivers/optical_tfs/OpticalTfsDriver.py

+31 −18

File changed.

Preview size limit exceeded, changes collapsed.