feat: Implement network data extraction and SIMAP network management in RealSimaps (3c579977) · Commits · TFS / controller

src/simap_connector/service/simap_updater/RealSimaps.py

+172 −9

Original line number	Diff line number	Diff line
		@@ -14,29 +14,155 @@


		import logging
		from typing import Dict, List, Tuple
		from context.client.ContextClient import ContextClient
		from common.proto.context_pb2 import DeviceId, DeviceIdList
		from .SimapClient import SimapClient


		LOGGER = logging.getLogger(__name__)

		# NOTE: for e2e --> network_data = [
		# ('ONT1', {'termination_points': ['200', '500']}),
		# ('POP2', {'termination_points': ['200', '201', '500']})
		# ]

		def set_simap_network(simap_client: SimapClient, network_id: str,
		network_data: list[dict]
		) -> None:


		# Use connection event log at IP: [2026-02-20 21:20:21,224] INFO:simap_connector.service.simap_updater.SimapUpdater:Processing Connection: {"connection_id": {"connection_uuid": {"uuid": "fdb61970-1a08-4f0e-acec-95d82a4b0d32"}}, "path_hops_endpoint_ids": [{"device_id": {"device_uuid": {"uuid": "c4b22f0f-d958-5895-a452-cac82e11ef90"}}, "endpoint_uuid": {"uuid": "97f60155-b852-5607-9ba5-1b41e228f04d"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "topology_uuid": {"uuid": "c76135e3-24a8-5e92-9bed-c3c9139359c8"}}}, {"device_id": {"device_uuid": {"uuid": "c4b22f0f-d958-5895-a452-cac82e11ef90"}}, "endpoint_uuid": {"uuid": "f9cea78e-0de3-5c8a-93f7-b99d207ae709"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "topology_uuid": {"uuid": "c76135e3-24a8-5e92-9bed-c3c9139359c8"}}}, {"device_id": {"device_uuid": {"uuid": "706e20a6-1f43-522d-9500-0bafd8131899"}}, "endpoint_uuid": {"uuid": "7dfc3453-a6df-584d-9e74-8ad4bfa301da"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "topology_uuid": {"uuid": "c76135e3-24a8-5e92-9bed-c3c9139359c8"}}}, {"device_id": {"device_uuid": {"uuid": "706e20a6-1f43-522d-9500-0bafd8131899"}}, "endpoint_uuid": {"uuid": "559070fb-857e-56b4-8453-dd427c489f59"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "topology_uuid": {"uuid": "c76135e3-24a8-5e92-9bed-c3c9139359c8"}}}, {"device_id": {"device_uuid": {"uuid": "7ae7e7bc-c4db-50a2-ad59-b5b7d9bcdc47"}}, "endpoint_uuid": {"uuid": "842aa058-7ac6-57f4-bca7-496070518b11"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "topology_uuid": {"uuid": "c76135e3-24a8-5e92-9bed-c3c9139359c8"}}}, {"device_id": {"device_uuid": {"uuid": "7ae7e7bc-c4db-50a2-ad59-b5b7d9bcdc47"}}, "endpoint_uuid": {"uuid": "dc6ac139-e4c9-5b27-a693-17e4a06aaf37"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "topology_uuid": {"uuid": "c76135e3-24a8-5e92-9bed-c3c9139359c8"}}}], "service_id": {"context_id": {"context_uuid": {"uuid": "43813baf-195e-5da6-af20-b3d0922e71a7"}}, "service_uuid": {"uuid": "e1ed09d2-dc76-5183-a245-855e5a596af2"}}, "settings": {}, "sub_service_ids: []}
		# We need to extract the connection's path hops and identify which links are involved, then determine the domain (topology) to which this connection belongs, and finally filter links based on allowed links per controller.
		def extract_network_data(context_client: ContextClient, network_id: str, network_connection: dict) -> list[tuple[str, dict]]:
		"""
		Extract network data from a connection object for SIMAP hierarchical network configuration.
		Extracts only the first and last devices (Service Demarcation Points) with their service-facing endpoints.

		Args:
		network_id: The network identifier (e.g., 'e2e', 'agg', 'trans-pkt')
		network_connection: Dictionary representation of a Connection protobuf message containing path_hops_endpoint_ids

		Returns:
		List of exactly 2 tuples: [(first_device_name, {'termination_points': [endpoint]}),
		(last_device_name, {'termination_points': [endpoint]})]
		For example: [('P-PE1', {'termination_points': ['200']}), ('P-PE2', {'termination_points': ['200']})]
		"""
		try:
		# Extract path_hops_endpoint_ids from network_connection dict
		path_hops = network_connection.get('path_hops_endpoint_ids', [])

		if not path_hops:
		LOGGER.warning(f"No path_hops_endpoint_ids found in network_connection for network {network_id}")
		return []

		if len(path_hops) < 2:
		LOGGER.warning(f"Connection path too short (less than 2 hops) for network {network_id}")
		return []

		# Extract first and last hops (SDPs - Service Demarcation Points)
		first_hop = path_hops[0]
		last_hop = path_hops[-1]

		# Extract device and endpoint UUIDs for SDPs
		first_device_uuid = first_hop.get('device_id', {}).get('device_uuid', {}).get('uuid', '')
		first_endpoint_uuid = first_hop.get('endpoint_uuid', {}).get('uuid', '')

		last_device_uuid = last_hop.get('device_id', {}).get('device_uuid', {}).get('uuid', '')
		last_endpoint_uuid = last_hop.get('endpoint_uuid', {}).get('uuid', '')

		if not all([first_device_uuid, first_endpoint_uuid, last_device_uuid, last_endpoint_uuid]):
		LOGGER.warning(f"Invalid first or last hop in path_hops_endpoint_ids for network {network_id}")
		return []

		# Prepare results for exactly 2 SDPs
		network_data: List[Tuple[str, Dict[str, List[str]]]] = []

		# Process first device (sdp1)
		first_device_id = DeviceId()
		first_device_id.device_uuid.uuid = first_device_uuid

		try:
		device_list = context_client.SelectDevice(
		DeviceIdList(device_ids=[first_device_id]),
		include_endpoints=True,
		include_config_rules=False,
		include_components=False
		)
		if not device_list.devices:
		LOGGER.warning(f"First device with UUID {first_device_uuid} not found in context")
		return []

		first_device = device_list.devices[0]
		first_device_name = first_device.name

		# Find the service-facing endpoint name
		first_endpoint_name = None
		for endpoint in first_device.device_endpoints:
		if endpoint.endpoint_id.endpoint_uuid.uuid == first_endpoint_uuid:
		first_endpoint_name = endpoint.name
		break

		if not first_endpoint_name:
		LOGGER.warning(f"First endpoint {first_endpoint_uuid} not found in device {first_device_name}")
		return []

		network_data.append((first_device_name, {'termination_points': [first_endpoint_name]}))

		except Exception as e:
		LOGGER.error(f"Error retrieving first device {first_device_uuid} from context: {e}")
		return []

		# Process last device (sdp2)
		last_device_id = DeviceId()
		last_device_id.device_uuid.uuid = last_device_uuid

		try:
		device_list = context_client.SelectDevice(
		DeviceIdList(device_ids=[last_device_id]),
		include_endpoints=True,
		include_config_rules=False,
		include_components=False
		)
		if not device_list.devices:
		LOGGER.warning(f"Last device with UUID {last_device_uuid} not found in context")
		return []

		last_device = device_list.devices[0]
		last_device_name = last_device.name

		# Find the service-facing endpoint name
		last_endpoint_name = None
		for endpoint in last_device.device_endpoints:
		if endpoint.endpoint_id.endpoint_uuid.uuid == last_endpoint_uuid:
		last_endpoint_name = endpoint.name
		break

		if not last_endpoint_name:
		LOGGER.warning(f"Last endpoint {last_endpoint_uuid} not found in device {last_device_name}")
		return []

		network_data.append((last_device_name, {'termination_points': [last_endpoint_name]}))

		except Exception as e:
		LOGGER.error(f"Error retrieving last device {last_device_uuid} from context: {e}")
		return []

		LOGGER.info(f"Extracted network data for {network_id}: {network_data}")
		return network_data

		except Exception as e:
		LOGGER.error(f"Error extracting network data from connection for network {network_id}: {e}")
		return []


		def set_simap_network(context_client: ContextClient, simap_client: SimapClient, network_id: str, network_connection: dict) -> None:
		"""
		Configure a SIMAP network with preset configurations.

		Args:
		context_client: ContextClient instance
		simap_client: SimapClient instance
		network_id: Network identifier ('e2e', 'agg', or 'trans-pkt')
		supp_net_ids: Tuple of supporting network IDs
		term_point_ids: List of termination point IDs
		network_connection: Dictionary representation of Connection protobuf with path_hops_endpoint_ids
		"""

		LOGGER.info(f"Setting SIMAP network: {network_id} for connection with {len(network_connection.get('path_hops_endpoint_ids', []))} hops")
		network_data : list[tuple[str, dict]] = extract_network_data(context_client, network_id, network_connection)

		if network_id == 'e2e':
		try:
		# E2E Network Configuration
		@@ -147,3 +273,40 @@ def set_simap_network(simap_client: SimapClient, network_id: str,
		return

		LOGGER.info(f'Successfully configured SIMAP network: {network_id}')


		def delete_simap_network(simap_client: SimapClient, network_id: str) -> None:
		"""
		Delete a SIMAP network configuration.

		Args:
		simap_client: SimapClient instance
		network_id: Network identifier ('e2e', 'agg', or 'trans-pkt')
		"""
		if network_id == 'e2e':
		simap = simap_client.network('e2e')
		simap.update(supporting_network_ids=['admin', 'agg'])

		link = simap.link('E2E-L1')
		link.delete()

		elif network_id == 'agg':
		simap = simap_client.network('agg')
		simap.update(supporting_network_ids=['admin', 'trans-pkt'])

		link = simap.link('AggNet-L1')
		link.delete()

		elif network_id == 'trans-pkt':
		simap = simap_client.network('trans-pkt')
		simap.update(supporting_network_ids=['admin'])

		link = simap.link('Trans-L1')
		link.delete()

		else:
		MSG = 'Unsupported network_id({:s}) to delete SIMAP'
		LOGGER.warning(MSG.format(str(network_id)))
		return

		LOGGER.info(f'Successfully deleted SIMAP network: {network_id}')

src/simap_connector/service/simap_updater/SimapUpdater.py

+22 −15

Original line number	Diff line number	Diff line
		@@ -23,13 +23,14 @@ from common.proto.context_pb2 import (
		)
		from common.tools.grpc.BaseEventCollector import BaseEventCollector
		from common.tools.grpc.BaseEventDispatcher import BaseEventDispatcher
		from common.tools.grpc.Tools import grpc_message_to_json_string
		from common.tools.grpc.Tools import grpc_message_to_json_string, grpc_message_to_json
		from context.client.ContextClient import ContextClient
		from simap_connector.service.telemetry.worker.data.Resources import (
		ResourceLink, Resources, SyntheticSampler
		)
		from simap_connector.service.telemetry.worker._Worker import WorkerTypeEnum
		from simap_connector.service.telemetry.TelemetryPool import SynthesizerWorker, TelemetryPool
		from src.simap_connector.service.simap_updater.RealSimaps import set_simap_network, delete_simap_network
		from .AllowedLinks import ALLOWED_LINKS_PER_CONTROLLER, LINKS_CAPACITY
		# from .MockSimaps import delete_mock_simap, set_mock_simap
		from .ObjectCache import CachedEntities, ObjectCache
		@@ -478,16 +479,16 @@ class EventDispatcher(BaseEventDispatcher):
		# LOGGER.warning(MSG.format(str_service_event, str_service))
		# return False

		topologies = self._object_cache.get_all(CachedEntities.TOPOLOGY, fresh=False)
		topology_names = {t.name for t in topologies}
		topology_names.discard(DEFAULT_TOPOLOGY_NAME)
		if len(topology_names) != 1:
		MSG = 'ServiceEvent({:s}) skipped, unable to identify on which topology to insert it'
		str_service_event = grpc_message_to_json_string(service_event)
		LOGGER.warning(MSG.format(str_service_event))
		return False
		# topologies = self._object_cache.get_all(CachedEntities.TOPOLOGY, fresh=False)
		# topology_names = {t.name for t in topologies}
		# topology_names.discard(DEFAULT_TOPOLOGY_NAME)
		# if len(topology_names) != 1:
		# MSG = 'ServiceEvent({:s}) skipped, unable to identify on which topology to insert it'
		# str_service_event = grpc_message_to_json_string(service_event)
		# LOGGER.warning(MSG.format(str_service_event))
		# return False

		domain_name = topology_names.pop() # trans-pkt/agg-net/e2e-net
		# domain_name = topology_names.pop() # trans-pkt/agg-net/e2e-net
		# set_mock_simap(self._simap_client, domain_name)

		#domain_topo = self._simap_client.network(domain_name)
		@@ -724,11 +725,9 @@ class EventDispatcher(BaseEventDispatcher):
		MSG = 'Processing Connection: {:s}'
		LOGGER.info(MSG.format(grpc_message_to_json_string(connection)))

		# NOTE: Actual Connection event object does not include service_id.

		_, link_uuids = get_connection_endpoints_and_links(connection_uuid)

		# Determine the controller's domain name
		# Determine the controller's domain name (network_id)
		topologies = self._object_cache.get_all(CachedEntities.TOPOLOGY, fresh=False)
		topology_names = {t.name for t in topologies}
		topology_names.discard(DEFAULT_TOPOLOGY_NAME)
		@@ -737,6 +736,11 @@ class EventDispatcher(BaseEventDispatcher):
		return None
		domain_name = topology_names.pop()

		# Call set_simap_network with proper parameters
		network_connection = grpc_message_to_json(connection)
		set_simap_network(self._context_client, self._simap_client, domain_name, network_connection)
		LOGGER.info('Set SIMAP network for connection {:s} in domain {:s}'.format(connection_uuid, domain_name))

		# Filter links based on ALLOWED_LINKS_PER_CONTROLLER
		allowed_link_names = ALLOWED_LINKS_PER_CONTROLLER.get(domain_name, set())
		LOGGER.debug('Allowed links for domain {:s}: {:s}'.format(domain_name, str(allowed_link_names)))
		@@ -764,7 +768,7 @@ class EventDispatcher(BaseEventDispatcher):
		LOGGER.debug('Cached connection {:s} mapping with {:d} links for domain {:s}'.format(
		connection_uuid, len(processed_links), domain_name))

		return domain_name, processed_links # NOTE: Domain name = topology name
		return domain_name, processed_links


		def _count_active_connections(self, link_uuid: str, domain_name: str, ) -> int:
		@@ -855,6 +859,9 @@ class EventDispatcher(BaseEventDispatcher):
		# No other connections use this link, stop the worker
		self._telemetry_pool.stop_worker(WorkerTypeEnum.SYNTHESIZER, worker_name)
		LOGGER.info('Stopped telemetry worker for link {:s}, no connections remain'.format(link_name))

		delete_simap_network(self._context_client, self._simap_client, domain_name)
		LOGGER.info('Deleted SIMAP network for domain {:s} after connection removal'.format(domain_name))
		else:
		# Other connections still use this link, update worker with new count
		worker = self._telemetry_pool.get_worker(WorkerTypeEnum.SYNTHESIZER, worker_name)