NBI Component - SSE Telemetry: (05ab026c) · Commits · TFS / controller

src/nbi/service/sse_telemetry/SimapClient.py

0 → 100644

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


		from typing import Dict, List, Tuple
		from common.tools.rest_conf.client.RestConfClient import RestConfClient


		class TerminationPoint:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}/node={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:termination-point={:s}'

		def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str, tp_id : str):
		self._restconf_client = restconf_client
		self._network_id = network_id
		self._node_id = node_id
		self._tp_id = tp_id

		def create(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None:
		endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)
		tp = {'tp-id': self._tp_id}
		stps = [
		{'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id}
		for snet_id,snode_id,stp_id in supporting_termination_point_ids
		]
		if len(stps) > 0: tp['supporting-termination-point'] = stps
		node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]}
		network = {'network-id': self._network_id, 'node': [node]}
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.post(endpoint, payload)

		def get(self) -> Dict:
		endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)
		node : Dict = self._restconf_client.get(endpoint)
		return node['ietf-network-topology:termination-point'][0]

		def update(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None:
		endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)
		tp = {'tp-id': self._tp_id}
		stps = [
		{'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id}
		for snet_id,snode_id,stp_id in supporting_termination_point_ids
		]
		if len(stps) > 0: tp['supporting-termination-point'] = stps
		node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]}
		network = {'network-id': self._network_id, 'node': [node]}
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.patch(endpoint, payload)

		def delete(self) -> None:
		endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)
		self._restconf_client.delete(endpoint)

		class Node:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/node={:s}'

		def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str):
		self._restconf_client = restconf_client
		self._network_id = network_id
		self._node_id = node_id
		self._tps : Dict[str, TerminationPoint] = dict()

		def termination_points(self) -> List[Dict]:
		tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID)
		return tps['ietf-network-topology:termination-point'].get('termination-point', list())

		def termination_point(self, tp_id : str) -> TerminationPoint:
		_tp = self._tps.get(tp_id)
		if _tp is not None: return _tp
		_tp = TerminationPoint(self._restconf_client, self._network_id, self._node_id, tp_id)
		return self._tps.setdefault(tp_id, _tp)

		def create(
		self, termination_point_ids : List[str] = [],
		supporting_node_ids : List[Tuple[str, str]] = []
		) -> None:
		endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)
		node = {'node-id': self._node_id}
		tps = [{'tp-id': tp_id} for tp_id in termination_point_ids]
		if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps
		sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids]
		if len(sns) > 0: node['supporting-node'] = sns
		network = {'network-id': self._network_id, 'node': [node]}
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.post(endpoint, payload)

		def get(self) -> Dict:
		endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)
		node : Dict = self._restconf_client.get(endpoint)
		return node['ietf-network:node'][0]

		def update(
		self, termination_point_ids : List[str] = [],
		supporting_node_ids : List[Tuple[str, str]] = []
		) -> None:
		endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)
		node = {'node-id': self._node_id}
		tps = [{'tp-id': tp_id} for tp_id in termination_point_ids]
		if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps
		sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids]
		if len(sns) > 0: node['supporting-node'] = sns
		network = {'network-id': self._network_id, 'node': [node]}
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.patch(endpoint, payload)

		def delete(self) -> None:
		endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)
		self._restconf_client.delete(endpoint)

		class Link:
		ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link={:s}'

		def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str):
		self._restconf_client = restconf_client
		self._network_id = network_id
		self._link_id = link_id

		def create(
		self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str,
		supporting_link_ids : List[Tuple[str, str]] = []
		) -> None:
		endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id)
		link = {
		'link-id' : self._link_id,
		'source' : {'source-node': src_node_id, 'source-tp': src_tp_id},
		'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id},
		}
		sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids]
		if len(sls) > 0: link['supporting-link'] = sls
		network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]}
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.post(endpoint, payload)

		def get(self) -> Dict:
		endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id)
		link : Dict = self._restconf_client.get(endpoint)
		return link['ietf-network-topology:link'][0]

		def update(
		self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str,
		supporting_link_ids : List[Tuple[str, str]] = []
		) -> None:
		endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id)
		link = {
		'link-id' : self._link_id,
		'source' : {'source-node': src_node_id, 'source-tp': src_tp_id},
		'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id},
		}
		sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids]
		if len(sls) > 0: link['supporting-link'] = sls
		network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]}
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.patch(endpoint, payload)

		def delete(self) -> None:
		endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id)
		self._restconf_client.delete(endpoint)


		class Network:
		ENDPOINT_NO_ID = '/ietf-network:networks'
		ENDPOINT_ID = ENDPOINT_NO_ID + '/network={:s}'

		def __init__(self, restconf_client : RestConfClient, network_id : str):
		self._restconf_client = restconf_client
		self._network_id = network_id
		self._nodes : Dict[str, Node] = dict()
		self._links : Dict[str, Link] = dict()

		def nodes(self) -> List[Dict]:
		reply : Dict = self._restconf_client.get(Node.ENDPOINT_NO_ID.format(self._network_id))
		return reply['ietf-network:network'][0].get('node', list())

		def links(self) -> List[Dict]:
		reply : Dict = self._restconf_client.get(Link.ENDPOINT_NO_ID.format(self._network_id))
		return reply['ietf-network:network'][0].get('ietf-network-topology:link', list())

		def node(self, node_id : str) -> Node:
		_node = self._nodes.get(node_id)
		if _node is not None: return _node
		_node = Node(self._restconf_client, self._network_id, node_id)
		return self._nodes.setdefault(node_id, _node)

		def link(self, link_id : str) -> Link:
		_link = self._links.get(link_id)
		if _link is not None: return _link
		_link = Link(self._restconf_client, self._network_id, link_id)
		return self._links.setdefault(link_id, _link)

		def create(self, supporting_network_ids : List[str] = []) -> None:
		endpoint = Network.ENDPOINT_ID.format(self._network_id)
		network = {'network-id': self._network_id}
		sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids]
		if len(sns) > 0: network['supporting-network'] = sns
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.post(endpoint, payload)

		def get(self) -> Dict:
		endpoint = Network.ENDPOINT_ID.format(self._network_id)
		networks : Dict = self._restconf_client.get(endpoint)
		return networks['ietf-network:network'][0]

		def update(self, supporting_network_ids : List[str] = []) -> None:
		endpoint = Network.ENDPOINT_ID.format(self._network_id)
		network = {'network-id': self._network_id}
		sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids]
		if len(sns) > 0: network['supporting-network'] = sns
		payload = {'ietf-network:networks': {'network': [network]}}
		self._restconf_client.patch(endpoint, payload)

		def delete(self) -> None:
		endpoint = Network.ENDPOINT_ID.format(self._network_id)
		self._restconf_client.delete(endpoint)


		class SimapClient:
		def __init__(self, restconf_client : RestConfClient) -> None:
		self._restconf_client = restconf_client
		self._networks : Dict[str, Network] = dict()

		def networks(self) -> List[Dict]:
		reply : Dict = self._restconf_client.get(Network.ENDPOINT_NO_ID)
		return reply['ietf-network:networks'].get('network', list())

		def network(self, network_id : str) -> Network:
		_network = self._networks.get(network_id)
		if _network is not None: return _network
		_network = Network(self._restconf_client, network_id)
		return self._networks.setdefault(network_id, _network)

src/nbi/service/sse_telemetry/create_subscription.py

+4 −3

Original line number	Diff line number	Diff line
		@@ -45,7 +45,6 @@ from .topology import (
		get_controller_name,
		)

		from .database_tmp import SERVICE_ID


		class SubscriptionId(TypedDict):
		@@ -73,10 +72,11 @@ class CreateSubscription(Resource):
		LOGGER.debug('Received subscription request data: {:s}'.format(str(request_data)))

		# break the request into its abstract components for telemetry subscription
		list_db_ids = list_identifiers(db)
		request_identifier = str(
		choice([x for x in range(1000, 10000) if x not in list_identifiers(db)])
		choice([x for x in range(1000, 10000) if x not in list_db_ids])
		)
		sub_subs = decompose_subscription(request_data, SERVICE_ID)
		sub_subs = decompose_subscription(request_data)

		# subscribe to each component
		device_client = DeviceClient()
		@@ -86,6 +86,7 @@ class CreateSubscription(Resource):
		'ietf-yang-push:datastore-xpath-filter'
		]

		SERVICE_ID = ''
		device_controller = get_controller_name(xpath, SERVICE_ID, context_client)
		if device_controller == Controllers.CONTROLLERLESS:
		LOGGER.warning(

src/nbi/service/sse_telemetry/database/Subscription.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -18,7 +18,7 @@ from sqlalchemy.dialects.postgresql import insert
		from sqlalchemy.engine import Engine
		from sqlalchemy.orm import Session, sessionmaker
		from sqlalchemy_cockroachdb import run_transaction
		from typing import Any, List, Optional, TypedDict
		from typing import Any, List, Optional, Set, TypedDict

		from .models.Subscription import SSESubscriptionModel

		@@ -140,7 +140,7 @@ def get_subscriptions(db_engine: Engine) -> List[SSESubsciprionDict]:
		return run_transaction(sessionmaker(bind=db_engine), callback)


		def list_identifiers(db_engine: Engine) -> List[str]:
		def list_identifiers(db_engine: Engine) -> Set[str]:
		def callback(session: Session) -> set[str]:
		obj_list: List[SSESubscriptionModel] = session.query(SSESubscriptionModel).all()
		return {obj.identifier for obj in obj_list}

src/nbi/service/sse_telemetry/database_tmp.py

deleted100644 → 0

+0 −16

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.


		SERVICE_ID = 'simap1'

src/nbi/service/sse_telemetry/topology.py

+72 −64

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


		import json
		import logging
		import os
		import json, logging, os, re
		from enum import Enum
		from string import octdigits
		from typing_extensions import List, TypedDict, Optional
		import re

		from context.client.ContextClient import ContextClient
		from device.client.DeviceClient import DeviceClient
		from typing_extensions import List, TypedDict
		from common.proto.context_pb2 import Device, DeviceId, Empty
		from common.tools.object_factory.Device import json_device_id
		from common.tools.rest_conf.client.RestConfClient import RestConfClient
		from common.DeviceTypes import DeviceTypeEnum
		from context.client.ContextClient import ContextClient
		from device.client.DeviceClient import DeviceClient

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

		@@ -64,26 +60,26 @@ phy_network = re.compile(r'providerId-\d+-clientId-\d+-topologyId-\d+')

		LOGGER = logging.getLogger(__name__)

		dir_path = os.path.dirname(__file__)
		#dir_path = os.path.dirname(__file__)

		with open(os.path.join(dir_path, 'Full-Te-Topology-simap1.json'), 'r') as f:
		NETWORK_DATA_SIMAP1 = json.load(f)
		#with open(os.path.join(dir_path, 'Full-Te-Topology-simap1.json'), 'r') as f:
		# NETWORK_DATA_SIMAP1 = json.load(f)

		with open(os.path.join(dir_path, 'Full-Te-Topology-simap2.json'), 'r') as f:
		NETWORK_DATA_SIMAP2 = json.load(f)
		#with open(os.path.join(dir_path, 'Full-Te-Topology-simap2.json'), 'r') as f:
		# NETWORK_DATA_SIMAP2 = json.load(f)


		def get_network_data(service_id: str) -> dict:
		if service_id == 'simap1':
		return NETWORK_DATA_SIMAP1
		elif service_id == 'simap2':
		return NETWORK_DATA_SIMAP2
		else:
		raise ValueError(f'Unsupported service_id: {service_id}. Expected "simap1" or "simap2".')
		#def get_network_data(service_id: str) -> dict:
		# if service_id == 'simap1':
		# return NETWORK_DATA_SIMAP1
		# elif service_id == 'simap2':
		# return NETWORK_DATA_SIMAP2
		# else:
		# raise ValueError(f'Unsupported service_id: {service_id}. Expected "simap1" or "simap2".')


		def decompose_subscription(
		s: SubscribedNotificationsSchema, service_id: str
		s : SubscribedNotificationsSchema
		) -> List[SubscribedNotificationsSchema]:
		"""
		Decomposes a subscription into its components by finding supporting links
		@@ -92,52 +88,64 @@ def decompose_subscription(
		input_data = s['ietf-subscribed-notifications:input']
		xpath_filter = input_data['ietf-yang-push:datastore-xpath-filter']

		# Parse the XPath to extract network and link information
		# Format: /ietf-network:networks/network=<network-id>/ietf-network-topology:link=<link-id>/simap-telemetry
		parts = xpath_filter.split('/')
		network_part = None
		link_part = None

		for part in parts:
		if part.startswith('network='):
		network_part = part[8:] # Remove 'network=' prefix
		elif part.startswith('ietf-network-topology:link='):
		link_part = part[27:] # Remove 'ietf-network-topology:link=' prefix

		if not network_part or not link_part:
		raise ValueError('Invalid XPath filter format')

		# Find the network in the topology data
		networks = get_network_data(service_id)['ietf-network:networks']['network']
		target_network = None

		for network in networks:
		if network['network-id'] == network_part:
		target_network = network
		break

		if not target_network:
		raise ValueError(f'Network {network_part} not found in topology data')

		# Find the link in the network
		links = target_network.get('ietf-network-topology:link', [])
		target_link = None

		for link in links:
		if link['link-id'] == link_part:
		target_link = link
		break

		if not target_link:
		raise ValueError(f'Link {link_part} not found in network {network_part}')
		rest_conf_client = RestConfClient(
		'10.254.0.9', port=8080, scheme='http', username='admin', password='admin',
		logger=logging.getLogger('RestConfClient')
		)
		xpath_data = rest_conf_client.get(xpath_filter)
		if not xpath_data:
		MSG = 'Resource({:s}) not found in SIMAP Server'
		raise Exception(MSG.format(str(xpath_filter)))

		# # Parse the XPath to extract network and link information
		# # Format: /ietf-network:networks/network=<network-id>/ietf-network-topology:link=<link-id>/simap-telemetry
		# parts = xpath_filter.split('/')
		# network_part = None
		# link_part = None

		# for part in parts:
		# if part.startswith('network='):
		# network_part = part[8:] # Remove 'network=' prefix
		# elif part.startswith('ietf-network-topology:link='):
		# link_part = part[27:] # Remove 'ietf-network-topology:link=' prefix

		# if not network_part or not link_part:
		# raise ValueError('Invalid XPath filter format')

		# # Find the network in the topology data
		# networks = get_network_data(service_id)['ietf-network:networks']['network']
		# target_network = None

		# for network in networks:
		# if network['network-id'] == network_part:
		# target_network = network
		# break

		# if not target_network:
		# raise ValueError(f'Network {network_part} not found in topology data')

		# # Find the link in the network
		# links = target_network.get('ietf-network-topology:link', [])
		# target_link = None

		# for link in links:
		# if link['link-id'] == link_part:
		# target_link = link
		# break

		# if not target_link:
		# raise ValueError(f'Link {link_part} not found in network {network_part}')

		# Get supporting links
		supporting_links = target_link.get('ietf-network-topology:supporting-link', [])
		#supporting_links = target_link.get('ietf-network-topology:supporting-link', [])
		supporting_links = xpath_data.get('ietf-network-topology:supporting-link', list())

		if not supporting_links:
		raise ValueError(
		f'No supporting links found for link {link_part} in network {network_part}'
		)
		#raise ValueError(
		# f'No supporting links found for link {link_part} in network {network_part}'
		#)
		MSG = 'No supporting links found for Resource({:s}, {:s})'
		raise Exception(MSG.format(str(xpath_filter), str(xpath_data)))

		# Create decomposed subscriptions
		decomposed = []