feat: Update SIMAP connection remove and created handling (5e07d785) · Commits · TFS / controller

src/common/tools/context_queries/Connection.py

+1 −9

Original line number	Diff line number	Diff line
		@@ -15,7 +15,7 @@
		import grpc, logging
		from typing import List, Optional
		from common.Constants import DEFAULT_CONTEXT_NAME
		from common.proto.context_pb2 import Connection, ConnectionId, ContextId
		from common.proto.context_pb2 import Connection, ConnectionId
		from context.client.ContextClient import ContextClient

		LOGGER = logging.getLogger(__name__)
		@@ -42,11 +42,3 @@ def get_connection_by_uuid(
		connection_id = ConnectionId()
		connection_id.connection_uuid.uuid = connection_uuid
		return get_connection_by_id(context_client, connection_id, rw_copy=rw_copy)

		def get_connections(
		context_client : ContextClient, context_uuid : str = DEFAULT_CONTEXT_NAME
		) -> List[Connection]:
		context_id = ContextId()
		context_id.context_uuid.uuid = context_uuid
		connections = context_client.ListConnections(context_id)
		return [c for c in connections.connections]

src/simap_connector/service/simap_updater/ObjectCache.py

+18 −7

Original line number	Diff line number	Diff line
		@@ -20,7 +20,7 @@ from common.tools.context_queries.Device import get_device, get_devices
		from common.tools.context_queries.Link import get_link, get_links
		from common.tools.context_queries.Topology import get_topology, get_topologies
		from common.tools.context_queries.Service import get_service_by_uuid, get_services
		from common.tools.context_queries.Connection import get_connection_by_uuid, get_connections
		from common.tools.context_queries.Connection import get_connection_by_uuid
		from context.client.ContextClient import ContextClient


		@@ -64,6 +64,7 @@ class ObjectCache:
		def __init__(self, context_client : ContextClient):
		self._context_client = context_client
		self._object_cache : Dict[Tuple[str, str], Any] = dict()
		# self.populate_all_cache() # Added for testing purposes; can be removed.

		def get(
		self, entity : CachedEntities, *object_uuids : str,
		@@ -180,12 +181,6 @@ class ObjectCache:
		(s.service_id.service_uuid.uuid, s.name) : s
		for s in objects
		}
		elif entity == CachedEntities.CONNECTION:
		objects = get_connections(self._context_client)
		objects = {
		(c.connection_id.connection_uuid.uuid, c.connection_id.connection_uuid.uuid) : c
		for c in objects
		}
		else:
		MSG = 'Not Supported ({:s})'
		LOGGER.warning(MSG.format(str(entity.value).title()))
		@@ -214,3 +209,19 @@ class ObjectCache:
		def delete(self, entity : CachedEntities, *object_uuids : str) -> None:
		object_key = compose_object_key(entity, *object_uuids)
		self._object_cache.pop(object_key, None)

		def populate_all_cache(self) -> None:
		"""Populate cache with all entities for testing purposes."""
		LOGGER.info('Populating cache with all entities for testing...')
		for entity in CachedEntities:
		if entity in (CachedEntities.ENDPOINT, CachedEntities.CONNECTION):
		# Endpoints are populated when devices are updated
		# Connections are service-scoped; cached on-demand during events
		continue
		try:
		self._update_all(entity)
		# LOGGER.info('Populated cache for entity: {:s}'.format(entity.value))
		except Exception as e:
		LOGGER.warning('Failed to populate cache for entity {:s}: {:s}'.format(
		entity.value, str(e)))
		LOGGER.info('Cache population completed')

src/simap_connector/service/simap_updater/SimapUpdater.py

+111 −51

Original line number	Diff line number	Diff line
		@@ -51,7 +51,7 @@ class EventDispatcher(BaseEventDispatcher):
		# Telemetry scaling configuration
		BASE_BANDWIDTH_OFFSET = 5.0 # Minimum bandwidth utilization % (no services)
		MAX_BANDWIDTH_OFFSET = 90.0 # Maximum bandwidth utilization % (at capacity)
		MAX_EXPECTED_SERVICES = 10 # Expected maximum concurrent services
		MAX_EXPECTED_CONNECTIONS = 10 # Expected maximum concurrent connections per link for scaling purposes
		DEFAULT_LINK_OFFSET = 25.0 # Default offset used in _dispatch_link_set

		def __init__(
		@@ -671,6 +671,12 @@ class EventDispatcher(BaseEventDispatcher):
		# Extract connection UUID from event
		connection_uuid = connection_event.connection_id.connection_uuid.uuid

		# Clean up any stale mapping for this connection (e.g., if connection is being re-created)
		old_mapping = self._object_cache.get(CachedEntities.CONNECTION, connection_uuid, auto_retrieve=False)
		if old_mapping is not None and isinstance(old_mapping, dict) and 'domain' in old_mapping:
		self._object_cache.delete(CachedEntities.CONNECTION, connection_uuid)
		LOGGER.debug('Removed stale mapping for connection {:s} before processing'.format(connection_uuid))

		try:
		# Use common helper to prepare connection data
		result = self._prepare_connection_processing(connection_uuid)
		@@ -679,10 +685,10 @@ class EventDispatcher(BaseEventDispatcher):
		(topology_name, processed_links) = result

		# Update telemetry for each link involved in this connection
		bandwidth_factor = self._calculate_bandwidth_factor()

		for _, link_name in processed_links:
		LOGGER.info('Connection {:s} uses allowed link: {:s}'.format(connection_uuid, link_name))
		for link_uuid, link_name in processed_links:
		# Calculate bandwidth factor specific to this link
		bandwidth_factor = self._calculate_bandwidth_factor(link_uuid, topology_name)
		LOGGER.info('Connection {:s} uses allowed link: {:s} (uuid: {:s})'.format(connection_uuid, link_name, link_uuid))
		worker_name = '{:s}:{:s}'.format(topology_name, link_name)

		# Worker should already exist from _dispatch_link_set (link creation event)
		@@ -731,6 +737,7 @@ class EventDispatcher(BaseEventDispatcher):

		return True


		def _prepare_connection_processing(self, connection_uuid: str):
		"""
		Extract common logic for processing connection events.
		@@ -770,65 +777,81 @@ class EventDispatcher(BaseEventDispatcher):
		connection_uuid, domain_name))
		return None

		# Cache the connection-to-links mapping for later retrieval (e.g., during REMOVE events)
		mapping = {
		'domain': domain_name,
		'links': {link_uuid: {'name': link_name} for link_uuid, link_name in processed_links}
		}
		self._object_cache.set(CachedEntities.CONNECTION, mapping, connection_uuid)
		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

		def _calculate_bandwidth_factor(self) -> float:

		def _calculate_bandwidth_factor(self, link_uuid: str, domain_name: str,
		active_connection_count: Optional[int] = None) -> float:
		"""
		Calculate bandwidth scaling factor based on active service count.
		Calculate bandwidth scaling factor based on active connection count for a specific link.

		Args:
		link_uuid: UUID of the link to calculate factor for
		domain_name: Domain name to filter connections
		active_connection_count: Pre-calculated connection count (if None, will be calculated)

		Returns:
		float: Bandwidth factor to multiply with existing worker offset
		"""
		try:
		# Query all services from Context
		all_services = self._object_cache.get_all(CachedEntities.SERVICE, fresh=False)

		# Count active services (SERVICESTATUS_ACTIVE or SERVICESTATUS_UPDATING)
		active_service_count = 0
		for service in all_services:
		service_status = service.service_status.service_status
		if service_status in (ServiceStatusEnum.SERVICESTATUS_ACTIVE,
		ServiceStatusEnum.SERVICESTATUS_UPDATING):
		# Skip sub-services (UUID-based names)
		try:
		uuid.UUID(hex=service.name)
		continue # Skip sub-services
		except: # pylint: disable=bare-except
		active_service_count += 1
		# If connection count not provided, calculate it from cache
		if active_connection_count is None:
		all_cached_connections = self._object_cache.get_all(CachedEntities.CONNECTION, fresh=False)
		active_connection_count = 0
		for cached_obj in all_cached_connections:
		if not isinstance(cached_obj, dict) or 'domain' not in cached_obj or 'links' not in cached_obj:
		continue

		if cached_obj['domain'] != domain_name:
		continue

		active_service_count = int(active_service_count / 2) # Each service appears as two connections (uuid and name)
		LOGGER.info('Active service count: {:d}'.format(int(active_service_count)))
		if link_uuid in cached_obj['links']:
		active_connection_count += 1

		LOGGER.info('Active connection count on link {:s} in domain {:s}: {:d}'.format(
		link_uuid, domain_name, active_connection_count))

		# Calculate bandwidth offset using linear scaling
		service_ratio = min(active_service_count / self.MAX_EXPECTED_SERVICES, 1.0)
		target_bandwidth_offset = (self.BASE_BANDWIDTH_OFFSET + (service_ratio *
		connection_ratio = min(active_connection_count / self.MAX_EXPECTED_CONNECTIONS, 1.0)
		target_bandwidth_offset = (self.BASE_BANDWIDTH_OFFSET + (connection_ratio *
		(self.MAX_BANDWIDTH_OFFSET - self.BASE_BANDWIDTH_OFFSET)))

		# Calculate adjustment factor relative to default offset
		bandwidth_factor = target_bandwidth_offset / self.DEFAULT_LINK_OFFSET

		LOGGER.info('Calculated bandwidth_factor={:.2f} (service_count={:d}, target_offset={:.2f})'.format(
		bandwidth_factor, active_service_count, target_bandwidth_offset))
		LOGGER.info('Calculated bandwidth_factor={:.2f} (connection_count={:d}, target_offset={:.2f})'.format(
		bandwidth_factor, active_connection_count, target_bandwidth_offset))

		return bandwidth_factor

		except Exception as e:
		LOGGER.exception('Failed to calculate bandwidth factor: {:s}'.format(str(e)))
		# Return default factor (1.0 = no change)
		return 1.0


		def dispatch_connection_create(self, connection_event : ConnectionEvent) -> None:
		if not self.dispatch_connection_set(connection_event): return

		MSG = 'Skipping Connection Create Event: {:s}'
		LOGGER.debug(MSG.format(grpc_message_to_json_string(connection_event)))


		def dispatch_connection_update(self, connection_event : ConnectionEvent) -> None:
		if not self.dispatch_connection_set(connection_event): return

		MSG = 'Skipping Connection Update Event: {:s}'
		LOGGER.debug(MSG.format(grpc_message_to_json_string(connection_event)))


		def dispatch_connection_remove(self, connection_event : ConnectionEvent) -> None:
		MSG = 'Processing Connection Remove Event: {:s}'
		LOGGER.info(MSG.format(grpc_message_to_json_string(connection_event)))
		@@ -836,28 +859,65 @@ class EventDispatcher(BaseEventDispatcher):
		connection_uuid = connection_event.connection_id.connection_uuid.uuid

		try:
		result = self._prepare_connection_processing(connection_uuid)
		if result is None:
		return
		(topology_name, processed_links) = result
		mapping = self._object_cache.get(CachedEntities.CONNECTION, connection_uuid, auto_retrieve=False)

		if mapping is None:
		MSG = 'Connection {:s} not found in cache, cannot process removal (service may have restarted)'
		raise Exception(MSG.format(connection_uuid))

		# Defensive: distinguish mapping dicts from potential protobuf Connection objects
		if not isinstance(mapping, dict) or 'domain' not in mapping or 'links' not in mapping:
		MSG = 'Invalid mapping structure for connection {:s}: expected dict with domain/links keys'
		raise Exception(MSG.format(connection_uuid))

		# Extract domain and links from cached mapping
		topology_name = mapping['domain']
		link_uuids_dict = mapping['links']
		processed_links = [(link_uuid, link_data['name']) for link_uuid, link_data in link_uuids_dict.items()]

		# Update telemetry factor for remaining services
		bandwidth_factor = self._calculate_bandwidth_factor()
		LOGGER.info('Retrieved cached mapping for connection {:s}: domain={:s}, links={:d}'.format(
		connection_uuid, topology_name, len(processed_links)))

		for _, link_name in processed_links:
		# Process each link: count remaining connections and stop/update worker accordingly
		for link_uuid, link_name in processed_links:
		worker_name = '{:s}:{:s}'.format(topology_name, link_name)

		if not self._telemetry_pool.has_worker(WorkerTypeEnum.SYNTHESIZER, worker_name):
		LOGGER.warning('Worker not found for link {:s}, skipping telemetry update for connection removal'.format(link_name))
		continue

		# Count how many OTHER connections (excluding current one being removed) use this link
		remaining_connections_count = 0
		for cache_key, cached_obj in self._object_cache._object_cache.items():
		# Filter for CONNECTION entity type
		if cache_key[0] != 'connection':
		continue
		cached_conn_uuid = cache_key[1]
		if cached_conn_uuid == connection_uuid:
		continue

		if isinstance(cached_obj, dict) and 'links' in cached_obj:
		if link_uuid in cached_obj['links']:
		remaining_connections_count += 1

		if remaining_connections_count == 0:
		# 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))
		else:
		# Other connections still use this link, recalculate bandwidth factor
		bandwidth_factor = self._calculate_bandwidth_factor(link_uuid, topology_name, remaining_connections_count)
		worker = self._telemetry_pool.get_worker(WorkerTypeEnum.SYNTHESIZER, worker_name)
		assert isinstance(worker, SynthesizerWorker), \
		'Expected SynthesizerWorker, got {:s}'.format(type(worker).__name__)

		# Update bandwidth scaling
		worker.change_resources(bandwidth_factor, latency_factor=1.0)
		LOGGER.info('Updated telemetry for link {:s} after connection removal'.format(link_name))
		LOGGER.info('Updated telemetry for link {:s} after connection removal, {:d} connections remain'.format(
		link_name, remaining_connections_count))

		# Clean up the mapping for this connection
		self._object_cache.delete(CachedEntities.CONNECTION, connection_uuid)
		LOGGER.debug('Deleted cached mapping for connection {:s}'.format(connection_uuid))

		except Exception as e:
		LOGGER.exception('Failed to process connection removal {:s}: {:s}'.format(