Merge branch 'feat/68-cttc-add-sbi-driver-for-flex-scale-optical-sdn-controller' into 'develop' (8d72cd43) · Commits · TFS / controller

src/context/service/database/models/enums/DeviceDriver.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -31,6 +31,7 @@ class ORM_DeviceDriverEnum(enum.Enum):
		XR = DeviceDriverEnum.DEVICEDRIVER_XR
		IETF_L2VPN = DeviceDriverEnum.DEVICEDRIVER_IETF_L2VPN
		GNMI_OPENCONFIG = DeviceDriverEnum.DEVICEDRIVER_GNMI_OPENCONFIG
		FLEXSCALE = DeviceDriverEnum.DEVICEDRIVER_FLEXSCALE

		grpc_to_enum__device_driver = functools.partial(
		grpc_to_enum, DeviceDriverEnum, ORM_DeviceDriverEnum)

src/device/service/drivers/init.py

+10 −0

Original line number	Diff line number	Diff line
		@@ -148,3 +148,13 @@ if LOAD_ALL_DEVICE_DRIVERS:
		FilterFieldEnum.DRIVER : DeviceDriverEnum.DEVICEDRIVER_XR,
		}
		]))

		if LOAD_ALL_DEVICE_DRIVERS:
		from .flexscale.FlexScaleDriver import FlexScaleDriver # pylint: disable=wrong-import-position
		DRIVERS.append(
		(FlexScaleDriver, [
		{
		FilterFieldEnum.DEVICE_TYPE: DeviceTypeEnum.OPEN_LINE_SYSTEM,
		FilterFieldEnum.DRIVER: DeviceDriverEnum.DEVICEDRIVER_FLEXSCALE,
		}
		]))

src/device/service/drivers/flexscale/FlexScaleDriver.py

0 → 100644

+151 −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, 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 find_key, add_lightpath, del_lightpath, get_lightpaths
		from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS
		from device.service.drivers.ietf_l2vpn.TfsDebugApiClient import TfsDebugApiClient
		from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum, get_import_topology

		LOGGER = logging.getLogger(__name__)

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


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

		# 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.TOPOLOGY)


		def Connect(self) -> bool:
		url = self.__flexscale_root + '/OpticalTFS/GetLightpaths'
		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)

		if resource_key == RESOURCE_ENDPOINTS:
		# return endpoints through debug-api and list-devices method
		results.extend(self.dac.get_devices_endpoints(self.__import_topology))

		# results.extend(get_lightpaths(
		# self.__flexscale_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)))

		src_node = find_key(resource, 'src_node')
		dst_node = find_key(resource, 'dst_node')
		bitrate = find_key(resource, 'bitrate')

		response = add_lightpath(self.__flexscale_root, src_node, dst_node, bitrate,
		auth=self.__auth, timeout=self.__timeout)

		results.extend(response)
		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)))
		flow_id = find_key(resource, 'flow_id')
		src_node = find_key(resource, 'src_node')
		dst_node = find_key(resource, 'dst_node')
		bitrate = find_key(resource, 'bitrate')

		response = del_lightpath(self.__flexscale_root, flow_id, src_node, dst_node, bitrate)
		results.extend(response)

		return results

		@metered_subclass_method(METRICS_POOL)
		def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
		# FlexScale 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]]:
		# FlexScale 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]]:
		# FlexScale does not support monitoring by now
		return []

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

0 → 100644

+143 −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, requests
		from requests.auth import HTTPBasicAuth
		from typing import Optional

		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 get_lightpaths(root_url : str, resource_key : str,auth : Optional[HTTPBasicAuth] = None,
		timeout : Optional[int] = None):
		headers = {'accept': 'application/json'}
		url = '{:s}/OpticalTFS/GetLightpaths'.format(root_url)

		result = []
		try:
		response = requests.get(url, timeout=timeout, headers=headers, 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:
		flows = 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

		for flow in flows:
		flow_id = flow.get('flow_id')
		source = flow.get('src')
		destination = flow.get('dst')
		bitrate = flow.get('bitrate')

		endpoint_url = '/flows/flow[{:s}]'.format(flow_id)
		endpoint_data = {'flow_id': flow_id, 'src': source, 'dst': destination, 'bitrate': bitrate}
		result.append((endpoint_url, endpoint_data))

		return result


		def add_lightpath(root_url, src_node, dst_node, bitrate,
		auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None):

		headers = {'accept': 'application/json'}
		url = '{:s}/OpticalTFS/AddLightpath/{:s}/{:s}/{:s}'.format(
		root_url, src_node, dst_node, bitrate)

		results = []
		try:
		LOGGER.info('Lightpath request: {:s} <-> {:s} with {:s} bitrate'.format(
		str(src_node), str(dst_node), str(bitrate)))
		response = requests.put(url=url, timeout=timeout, headers=headers, verify=False, auth=auth)
		results.append(response.json())
		LOGGER.info('Response: {:s}'.format(str(response)))

		except Exception as e: # pylint: disable=broad-except
		LOGGER.exception('Exception requesting Lightpath: {:s} <-> {:s} with {:s} bitrate'.format(
		str(src_node), str(dst_node), str(bitrate)))
		results.append(e)
		else:
		if response.status_code not in HTTP_OK_CODES:
		msg = 'Could not create Lightpath(status_code={:s} reply={:s}'
		LOGGER.error(msg.format(str(response.status_code), str(response)))
		results.append(response.status_code in HTTP_OK_CODES)

		return results



		def del_lightpath(root_url, flow_id, src_node, dst_node, bitrate,
		auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None):
		url = '{:s}/OpticalTFS/DelLightpath/{:s}/{:s}/{:s}/{:s}'.format(
		root_url, flow_id, src_node, dst_node, bitrate)
		headers = {'accept': 'application/json'}

		results = []

		try:
		response = requests.delete(
		url=url, timeout=timeout, headers=headers, verify=False, auth=auth)
		except Exception as e: # pylint: disable=broad-except
		LOGGER.exception('Exception deleting Lightpath(uuid={:s})'.format(str(flow_id)))
		results.append(e)
		else:
		if response.status_code not in HTTP_OK_CODES:
		msg = 'Could not delete Lightpath(flow_id={:s}). status_code={:s} reply={:s}'
		LOGGER.error(msg.format(str(flow_id), str(response.status_code), str(response)))
		results.append(response.status_code in HTTP_OK_CODES)

		return results


		def get_topology(root_url : str, resource_key : str,auth : Optional[HTTPBasicAuth] = None,
		timeout : Optional[int] = None):
		headers = {'accept': 'application/json'}
		url = '{:s}/OpticalTFS/GetLinks'.format(root_url)

		result = []
		try:
		response = requests.get(url, timeout=timeout, headers=headers, 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:
		response = 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

		result.append(response)
		return result

src/device/service/drivers/flexscale/init.py

0 → 100644

+20 −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.

		from device.service.driver_api._Driver import RESOURCE_ENDPOINTS, RESOURCE_SERVICES

		ALL_RESOURCE_KEYS = [
		RESOURCE_ENDPOINTS,
		RESOURCE_SERVICES,
		]