Device component - NCE Driver: (60d48b21) · Commits · TFS / controller

src/common/tools/client/RestConfClient.py

+16 −7

Original line number	Diff line number	Diff line
		@@ -37,7 +37,7 @@ class RestConfClient(RestApiClient):

		def _discover_base_url(self) -> None:
		host_meta_url = HOST_META_URL.format(self._scheme, self._address, self._port)
		host_meta : Dict = self.get(host_meta_url, expected_status_codes={requests.codes['OK']})
		host_meta : Dict = super().get(host_meta_url, expected_status_codes={requests.codes['OK']})

		links = host_meta.get('links')
		if links is None: raise AttributeError('Missing attribute "links" in host-meta reply')
		@@ -56,14 +56,14 @@ class RestConfClient(RestApiClient):
		if href is None: raise AttributeError('Missing attribute "links[0]" in host-meta reply')
		if not isinstance(href, str): raise AttributeError('Attribute "links[0].href" must be a str')

		self._base_url = str(href + '/data').replace('//', '/')
		self._base_url = str(href).replace('//', '/')

		def get(
		self, endpoint : str,
		expected_status_codes : Set[int] = {requests.codes['OK']}
		) -> Optional[Any]:
		return super().get(
		endpoint,
		('/data/{:s}'.format(endpoint)).replace('//', '/')
		expected_status_codes=expected_status_codes
		)

		@@ -72,7 +72,7 @@ class RestConfClient(RestApiClient):
		expected_status_codes : Set[int] = {requests.codes['CREATED']}
		) -> Optional[Any]:
		return super().post(
		endpoint, body=body,
		('/data/{:s}'.format(endpoint)).replace('//', '/'), body=body,
		expected_status_codes=expected_status_codes
		)

		@@ -81,7 +81,7 @@ class RestConfClient(RestApiClient):
		expected_status_codes : Set[int] = {requests.codes['CREATED'], requests.codes['NO_CONTENT']}
		) -> Optional[Any]:
		return super().put(
		endpoint, body=body,
		('/data/{:s}'.format(endpoint)).replace('//', '/'), body=body,
		expected_status_codes=expected_status_codes
		)

		@@ -90,7 +90,7 @@ class RestConfClient(RestApiClient):
		expected_status_codes : Set[int] = {requests.codes['NO_CONTENT']}
		) -> Optional[Any]:
		return super().patch(
		endpoint, body=body,
		('/data/{:s}'.format(endpoint)).replace('//', '/'), body=body,
		expected_status_codes=expected_status_codes
		)

		@@ -99,6 +99,15 @@ class RestConfClient(RestApiClient):
		expected_status_codes : Set[int] = {requests.codes['NO_CONTENT']}
		) -> Optional[Any]:
		return super().delete(
		endpoint, body=body,
		('/data/{:s}'.format(endpoint)).replace('//', '/'), body=body,
		expected_status_codes=expected_status_codes
		)

		def rpc(
		self, endpoint : str, body : Optional[Any] = None,
		expected_status_codes : Set[int] = {requests.codes['CREATED']}
		) -> Optional[Any]:
		return super().post(
		('/operations/{:s}'.format(endpoint)).replace('//', '/'), body=body,
		expected_status_codes=expected_status_codes
		)

src/device/service/drivers/nce/NCEDriver.py

+28 −23

Original line number	Diff line number	Diff line
		@@ -21,13 +21,12 @@ from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS, RESOU
		from device.service.driver_api.AnyTreeTools import (
		TreeNode, dump_subtree, get_subnode, set_subnode_value,
		)
		from .handlers.AppFlowHandler import AppFlowHandler
		from .handlers.NetworkTopologyHandler import NetworkTopologyHandler
		from .Constants import SPECIAL_RESOURCE_MAPPINGS
		from .nce_fan_client import (
		NCEClient,
		SubscribedNotificationsSchema,
		UnsubscribedNotificationsSchema,
		from .handlers.SubscriptionHandler import (
		SubscribedNotificationsSchema, SubscriptionHandler, UnsubscribedNotificationsSchema
		)
		from .Constants import SPECIAL_RESOURCE_MAPPINGS
		from .Tools import compose_resource_endpoint


		@@ -60,14 +59,11 @@ class NCEDriver(_Driver):
		restconf_settings['logger'] = logging.getLogger(__name__ + '.RestConfClient')
		self._rest_conf_client = RestConfClient(address, port=port, **restconf_settings)
		self._handler_net_topology = NetworkTopologyHandler(self._rest_conf_client, **settings)
		self._handler_app_flow = AppFlowHandler(self._rest_conf_client)
		self._handler_subscription = SubscriptionHandler(self._rest_conf_client)

		self.__running = TreeNode('.')
		scheme = self.settings.get('scheme', 'http')
		username = self.settings.get('username')
		password = self.settings.get('password')
		self.nce = NCEClient(
		self.address, self.port, scheme=scheme, username=username, password=password,
		)

		endpoints = self.settings.get('endpoints', [])
		endpoint_resources = []
		for endpoint in endpoints:
		@@ -92,7 +88,7 @@ class NCEDriver(_Driver):
		resource_key, resource_value = resource
		chk_string(str_resource_name, resource_key, allow_empty=False)
		resource_path = resource_key.split('/')
		except Exception as e: # pylint: disable=broad-except
		except Exception as e:
		LOGGER.exception(
		'Exception validating {:s}: {:s}'.format(
		str_resource_name, str(resource_key)
		@@ -119,7 +115,7 @@ class NCEDriver(_Driver):
		except requests.exceptions.Timeout:
		LOGGER.exception('Timeout exception checking connectivity')
		return False
		except Exception: # pylint: disable=broad-except
		except Exception:
		LOGGER.exception('Unhandled exception checking connectivity')
		return False
		else:
		@@ -151,6 +147,17 @@ class NCEDriver(_Driver):
		chk_string(str_resource_name, resource_key, allow_empty=False)
		if resource_key == RESOURCE_ENDPOINTS:
		results.extend(self._handler_net_topology.get())
		elif resource_key == RESOURCE_SERVICES:
		app_flows = self._handler_app_flow.retrieve()
		app_flow_names = [
		app_flow['name']
		for app_flow in app_flows['huawei-nce-app-flow:app-flows']['app-flow']
		]
		if len(app_flow_names) == 1:
		resource_key = '/service[{:s}]/AppFlow'.format(app_flow_names[0])
		results.append((resource_key, app_flows))
		elif len(app_flow_names) > 1:
		raise Exception('Support for multiple app-flow retrieval not properly managed')
		else:
		resource_key = SPECIAL_RESOURCE_MAPPINGS.get(resource_key, resource_key)
		resource_path = resource_key.split('/')
		@@ -158,7 +165,7 @@ class NCEDriver(_Driver):
		# if not found, resource_node is None
		if resource_node is None: continue
		results.extend(dump_subtree(resource_node))
		except Exception as e: # pylint: disable=broad-except
		except Exception as e:
		MSG = 'Error processing resource_key({:s}, {:s})'
		LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))
		results.append((resource_key, e)) # if processing fails, store the exception
		@@ -178,7 +185,7 @@ class NCEDriver(_Driver):
		continue
		try:
		resource_value = json.loads(resource_value)
		self.nce.create_app_flow(resource_value)
		self._handler_app_flow.create(resource_value)
		results.append((resource_key, True))
		except Exception as e: # pylint: disable=broad-except
		MSG = 'Unhandled error processing SET resource_key({:s})'
		@@ -198,9 +205,7 @@ class NCEDriver(_Driver):
		continue
		try:
		resource_value = json.loads(resource_value)
		app_flows = resource_value['huawei-nce-app-flow:app-flows']
		app_flow_name = app_flows['app-flow'][0]['app-name']
		self.nce.delete_app_flow(app_flow_name)
		self._handler_app_flow.delete(resource_value)
		results.append((resource_key, True))
		except Exception as e:
		MSG = 'Unhandled error processing DELETE resource_key({:s})'
		@@ -216,7 +221,7 @@ class NCEDriver(_Driver):
		raise ValueError('NCE driver supports only one subscription at a time')
		s = subscriptions[0]
		uri = s[0]
		_ = s[1] # sampling duration
		#sampling_duration = s[1]
		sampling_interval = s[2]
		s_data : SubscribedNotificationsSchema = {
		'ietf-subscribed-notifications:input': {
		@@ -225,7 +230,7 @@ class NCEDriver(_Driver):
		'ietf-yang-push:periodic': {'ietf-yang-push:period': str(sampling_interval)},
		}
		}
		s_id = self.nce.subscribe_telemetry(s_data)
		s_id = self._handler_subscription.subscribe(s_data)
		return [s_id]

		@metered_subclass_method(METRICS_POOL)
		@@ -241,7 +246,7 @@ class NCEDriver(_Driver):
		'identifier': identifier,
		}
		}
		self.nce.unsubscribe_telemetry(s_data)
		self._handler_subscription.unsubscribe(s_data)
		return [True]

		def GetState(

src/device/service/drivers/nce/handlers/AppFlowHandler.py

0 → 100644

+160 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2025 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.
		# 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, requests
		from typing import Dict
		from common.tools.client.RestConfClient import RestConfClient


		LOGGER = logging.getLogger(__name__)


		class AppFlowHandler:
		def __init__(self, rest_conf_client : RestConfClient) -> None:
		self._rest_conf_client = rest_conf_client

		self._url_qos_profile = '/huawei-nce-app-flow:qos-profiles'
		self._url_qos_profile_item = self._url_qos_profile + '/qos-profile={:s}'

		self._url_application = '/huawei-nce-app-flow:applications'
		self._url_application_item = self._url_application + '/application={:s}'

		self._url_app_flow = '/huawei-nce-app-flow:app-flows'
		self._url_app_flow_item = self._url_app_flow + '/app-flow={:s}'


		def create(self, data : Dict) -> None:
		MSG = '[create] data={:s}'
		LOGGER.debug(MSG.format(str(data)))

		try:
		qos_profiles = (
		data
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('qos-profiles', dict())
		.get('qos-profile', list())
		)
		for qos_profile in qos_profiles:
		request = {'huawei-nce-app-flow:qos-profiles': {'qos-profile': qos_profile}}
		LOGGER.info('Creating QoS Profile: {:s}'.format(str(request)))
		self._rest_conf_client.post(self._url_qos_profile, json=request)

		applications = (
		data
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('applications', dict())
		.get('application', list())
		)
		for application in applications:
		request = {'huawei-nce-app-flow:applications': {'application': application}}
		LOGGER.info('Creating Application: {:s}'.format(str(request)))
		self._rest_conf_client.post(self._url_application, json=request)

		app_flows = (
		data
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('app-flow', list())
		)
		for app_flow in app_flows:
		request = {'huawei-nce-app-flow:app-flows': {'app-flow': app_flow}}
		LOGGER.info('Creating App Flow: {:s}'.format(str(request)))
		self._rest_conf_client.post(self._url_app_flow, json=request)

		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send POST requests to NCE FAN NBI'
		raise Exception(MSG) from e


		def retrieve(self) -> Dict:
		try:
		LOGGER.info('Retrieving QoS Profiles')
		qos_profiles = self._rest_conf_client.get(self._url_qos_profile)

		LOGGER.info('Retrieving Applications')
		applications = self._rest_conf_client.get(self._url_application)

		LOGGER.info('Retrieving App Flows')
		app_flows = self._rest_conf_client.get(self._url_app_flow)
		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send GET requests to NCE FAN NBI'
		raise Exception(MSG) from e

		qos_profiles = (
		qos_profiles
		.get('huawei-nce-app-flow:qos-profiles', dict())
		.get('qos-profile', list())
		)

		applications = (
		applications
		.get('huawei-nce-app-flow:applications', dict())
		.get('application', list())
		)

		app_flows = (
		app_flows
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('app-flow', list())
		)

		return {'huawei-nce-app-flow:app-flows': {
		'qos-profiles': {'qos-profile': qos_profiles},
		'applications': {'application': applications},
		'app-flow': app_flows,
		}}


		def delete(self, data : Dict) -> None:
		MSG = '[delete] data={:s}'
		LOGGER.debug(MSG.format(str(data)))

		try:
		app_flows = (
		data
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('app-flow', list())
		)
		for app_flow in app_flows:
		app_flow_name = app_flow['name']
		LOGGER.info('Deleting App Flow: {:s}'.format(str(app_flow_name)))
		app_flow_url = self._url_app_flow_item.format(app_flow_name)
		self._rest_conf_client.delete(app_flow_url)

		applications = (
		data
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('applications', dict())
		.get('application', list())
		)
		for application in applications:
		application_name = application['name']
		LOGGER.info('Deleting Application: {:s}'.format(str(application_name)))
		application_url = self._url_application_item.format(application_name)
		self._rest_conf_client.delete(application_url)

		qos_profiles = (
		data
		.get('huawei-nce-app-flow:app-flows', dict())
		.get('qos-profiles', dict())
		.get('qos-profile', list())
		)
		for qos_profile in qos_profiles:
		qos_profile_name = qos_profile['name']
		LOGGER.info('Deleting QoS Profile: {:s}'.format(str(qos_profile_name)))
		qos_profile_url = self._url_qos_profile_item.format(qos_profile_name)
		self._rest_conf_client.delete(qos_profile_url)

		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send POST requests to NCE FAN NBI'
		raise Exception(MSG) from e

src/device/service/drivers/nce/handlers/SubscriptionHandler.py

0 → 100644

+86 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2025 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.
		# 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, requests
		from typing_extensions import TypedDict
		from common.tools.client.RestConfClient import RestConfClient


		LOGGER = logging.getLogger(__name__)


		Periodic = TypedDict('Periodic', {'ietf-yang-push:period': str})

		Input = TypedDict(
		'Input',
		{
		'datastore': str,
		'ietf-yang-push:datastore-xpath-filter': str,
		'ietf-yang-push:periodic': Periodic,
		},
		)

		SubscribedNotificationsSchema = TypedDict(
		'SubscribedNotificationsSchema', {'ietf-subscribed-notifications:input': Input}
		)

		SubscriptionSchema = TypedDict('SubscriptionSchema', {'identifier': str})

		UnsubscribedNotificationsSchema = TypedDict(
		'UnsubscribedNotificationsSchema', {'delete-subscription': SubscriptionSchema}
		)


		class SubscriptionId(TypedDict):
		identifier: str
		uri: str


		class SubscriptionHandler:
		def __init__(self, rest_conf_client : RestConfClient) -> None:
		self._rest_conf_client = rest_conf_client

		self._url_qos_profile = '/huawei-nce-app-flow:qos-profiles'
		self._url_qos_profile_item = self._url_qos_profile + '/qos-profile={:s}'

		self._url_app_flow = '/huawei-nce-app-flow:app-flows'
		self._url_app_flow_item = self._url_app_flow + '/app-flow={:s}'


		def subscribe(
		self, subscription_data : SubscribedNotificationsSchema
		) -> SubscriptionId:
		MSG = '[subscribe] subscription_data={:s}'
		LOGGER.debug(MSG.format(str(subscription_data)))
		try:
		url = '/subscriptions:establish-subscription'
		LOGGER.debug('Subscribing to telemetry: {:s}'.format(str(subscription_data)))
		return self._rest_conf_client.rpc(url, json=subscription_data)
		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send RPC request'
		raise Exception(MSG) from e


		def unsubscribe(
		self, unsubscription_data : UnsubscribedNotificationsSchema
		) -> SubscriptionId:
		MSG = '[unsubscribe] unsubscription_data={:s}'
		LOGGER.debug(MSG.format(str(unsubscription_data)))
		try:
		url = '/subscriptions:delete-subscription'
		LOGGER.debug('Unsubscribing from telemetry: {:s}'.format(str(unsubscription_data)))
		return self._rest_conf_client.rpc(url, json=unsubscription_data)
		except requests.exceptions.ConnectionError as e:
		MSG = 'Failed to send RPC request'
		raise Exception(MSG) from e

src/device/service/drivers/nce/nce_fan_client.py

deleted100644 → 0

+0 −143

Original line number	Diff line number	Diff line
		# Copyright 2022-2025 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.
		# 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 typing import Optional
		from typing_extensions import List, TypedDict

		import requests
		from requests.auth import HTTPBasicAuth


		Periodic = TypedDict('Periodic', {'ietf-yang-push:period': str})

		Input = TypedDict(
		'Input',
		{
		'datastore': str,
		'ietf-yang-push:datastore-xpath-filter': str,
		'ietf-yang-push:periodic': Periodic,
		},
		)

		SubscribedNotificationsSchema = TypedDict(
		'SubscribedNotificationsSchema', {'ietf-subscribed-notifications:input': Input}
		)

		SubscriptionSchema = TypedDict('SubscriptionSchema', {'identifier': str})

		UnsubscribedNotificationsSchema = TypedDict(
		'UnsubscribedNotificationsSchema', {'delete-subscription': SubscriptionSchema}
		)


		class SubscriptionId(TypedDict):
		identifier: str
		uri: str


		LOGGER = logging.getLogger(__name__)

		NCE_FAN_URL = '{:s}://{:s}:{:d}'
		TIMEOUT = 30

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

		MAPPING_STATUS = {
		'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,
		'DEVICEOPERATIONALSTATUS_DISABLED': 1,
		'DEVICEOPERATIONALSTATUS_ENABLED': 2,
		}

		MAPPING_DRIVER = {
		'DEVICEDRIVER_UNDEFINED': 0,
		'DEVICEDRIVER_OPENCONFIG': 1,
		'DEVICEDRIVER_TRANSPORT_API': 2,
		'DEVICEDRIVER_P4': 3,
		'DEVICEDRIVER_IETF_NETWORK_TOPOLOGY': 4,
		'DEVICEDRIVER_ONF_TR_532': 5,
		'DEVICEDRIVER_XR': 6,
		'DEVICEDRIVER_IETF_L2VPN': 7,
		'DEVICEDRIVER_GNMI_OPENCONFIG': 8,
		'DEVICEDRIVER_OPTICAL_TFS': 9,
		'DEVICEDRIVER_IETF_ACTN': 10,
		'DEVICEDRIVER_OC': 11,
		}

		HEADERS = {'Content-Type': 'application/json'}


		class NCEClient:
		def __init__(
		self,
		address: str,
		port: str,
		scheme: str = 'http',
		username: Optional[str] = None,
		password: Optional[str] = None,
		) -> None:
		self._nce_fan_url = NCE_FAN_URL.format(scheme, address, int(port))
		self._auth = None

		def create_app_flow(self, app_flow_data: dict) -> None:
		try:
		app_data = app_flow_data['huawei-nce-app-flow:app-flows']['applications']
		app_url = self._nce_fan_url + '/restconf/v1/data' + '/app-flows/apps'
		LOGGER.info(f'Creating app: {app_data} URL: {app_url}')
		requests.post(app_url, json=app_data, headers=HEADERS)

		app_flow_data = {'app-flow': app_flow_data['huawei-nce-app-flow:app-flows']['app-flow']}
		app_flow_url = self._nce_fan_url + '/restconf/v1/data' + '/app-flows'
		LOGGER.info(f'Creating app flow: {app_flow_data} URL: {app_flow_url}')
		requests.post(app_flow_url, json=app_flow_data, headers=HEADERS)
		except requests.exceptions.ConnectionError:
		raise Exception('faild to send post requests to NCE FAN')

		def delete_app_flow(self, app_flow_name: str) -> None:
		try:
		app_url = (
		self._nce_fan_url
		+ '/restconf/v1/data'
		+ f'/app-flows/apps/application={app_flow_name}'
		)
		LOGGER.info(f'Deleting app: {app_flow_name} URL: {app_url}')
		requests.delete(app_url)

		app_flow_url = (
		self._nce_fan_url + '/restconf/v1/data' + f'/app-flows/app-flow={app_flow_name}'
		)
		LOGGER.info(f'Deleting app flow: {app_flow_name} URL: {app_flow_url}')
		requests.delete(app_flow_url)
		except requests.exceptions.ConnectionError:
		raise Exception('faild to send delete request to NCE FAN')

		def subscribe_telemetry(
		self, subscription_data: SubscribedNotificationsSchema
		) -> SubscriptionId:
		url = self._nce_fan_url + '/restconf/operations/subscriptions:establish-subscription'
		LOGGER.debug(f'Subscribing to telemetry with data: {subscription_data} URL: {url}')
		r = requests.post(url, json=subscription_data, headers=HEADERS)
		r.raise_for_status()
		return r.json()

		def unsubscribe_telemetry(self, unsubscription_data: UnsubscribedNotificationsSchema) -> None:
		url = self._nce_fan_url + '/restconf/operations/subscriptions:delete-subscription'
		LOGGER.debug(f'Unsubscribing to telemetry with data: {unsubscription_data} URL: {url}')
		r = requests.post(url, json=unsubscription_data, headers=HEADERS)
		r.raise_for_status()