feat: Updated Version of SIMAP Connector (1784774c) · Commits · TFS / controller

src/simap_connector/service/SimapConnectorServiceServicerImpl.py

+6 −1

Original line number	Diff line number	Diff line
		@@ -165,7 +165,12 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):
		link_id = request.link_id
		bandwidth_factor = request.bandwidth_factor
		latency_factor = request.latency_factor
		# connection_count = request.connection_count

		# TODO: Remove bandwidth_factor and latency_factor from the request, as they are not used in the current implementation.
		# Add connection_count to the request.

		connection_count = 0
		synthesizer_name = '{:s}:{:s}'.format(network_id, link_id)
		synthesizer : Optional[_Worker] = self._telemetry_pool.get_worker(
		WorkerTypeEnum.SYNTHESIZER, synthesizer_name
		@@ -175,5 +180,5 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):
		raise Exception(MSG.format(synthesizer_name))
		assert isinstance(synthesizer, SynthesizerWorker), \
		'Expected SynthesizerWorker, got {:s}'.format(type(synthesizer).__name__)
		synthesizer.change_resources(bandwidth_factor, latency_factor)
		synthesizer.change_resources(connection_count)
		return Empty()

src/simap_connector/service/simap_updater/AllowedLinks.py

+18 −0

Original line number	Diff line number	Diff line
		@@ -18,3 +18,21 @@ ALLOWED_LINKS_PER_CONTROLLER = {
		'L11ba', 'L12ab', 'L12ba', 'L13', 'L14' },
		'trans-pkt': { 'L5', 'L6', 'L9', 'L10' },
		}
		# NOTE: Ranges should be less than 100 because the schema does not allow
		# bandwidth-utilization to exceed 100%
		# As per schema below: (percentage of link capacity)
		# /* --- Local typedefs --- */
		# typedef percent {
		# type decimal64 {
		# fraction-digits 2;
		# range "0 .. 100";
		# }
		# units "percent";
		# description "0–100 percent value.";
		# }
		LINKS_CAPACITY = {
		'L1' : 30, 'L2' : 30, 'L3' : 70, 'L4' : 70,
		'L5' : 90, 'L6' : 90, 'L9' : 90, 'L10' : 90,
		'L7ab' : 50, 'L7ba' : 50, 'L8ab' : 50, 'L8ba' : 50, 'L11ab' : 50,
		'L11ba' : 50, 'L12ab': 50, 'L12ba': 50, 'L13' : 30, 'L14' : 30,
		}

src/simap_connector/service/simap_updater/ObjectCache.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -64,7 +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() # NOTE: Added for testing purposes; can be removed.
		# self.populate_all_cache() # NOTE: Added for testing purposes; should be removed/commented.

		def get(
		self, entity : CachedEntities, *object_uuids : str,

src/simap_connector/service/simap_updater/SimapUpdater.py

+54 −109

Original line number	Diff line number	Diff line
		@@ -30,7 +30,7 @@ from simap_connector.service.telemetry.worker.data.Resources import (
		)
		from simap_connector.service.telemetry.worker._Worker import WorkerTypeEnum
		from simap_connector.service.telemetry.TelemetryPool import SynthesizerWorker, TelemetryPool
		from .AllowedLinks import ALLOWED_LINKS_PER_CONTROLLER
		from .AllowedLinks import ALLOWED_LINKS_PER_CONTROLLER, LINKS_CAPACITY
		from .MockSimaps import delete_mock_simap, set_mock_simap
		from .ObjectCache import CachedEntities, ObjectCache
		from .SimapClient import SimapClient
		@@ -48,12 +48,6 @@ SKIPPED_DEVICE_TYPES = {


		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_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__(
		self, events_queue : queue.PriorityQueue,
		simap_client : SimapClient,
		@@ -365,23 +359,15 @@ class EventDispatcher(BaseEventDispatcher):
		worker_name = '{:s}:{:s}'.format(topology_name, link_name)
		resources = Resources()
		resources.links.append(ResourceLink(
		domain_name=topology_name, link_name=link_name,
		bandwidth_utilization_sampler=SyntheticSampler.create_random(
		amplitude_scale = 25.0,
		phase_scale = 1e-7,
		period_scale = 86_400,
		offset_scale = 25,
		noise_ratio = 0.05,
		min_value = 0.0,
		max_value = 100.0,
		),
		latency_sampler=SyntheticSampler.create_random(
		amplitude_scale = 0.5,
		phase_scale = 1e-7,
		period_scale = 60.0,
		offset_scale = 10.0,
		noise_ratio = 0.05,
		min_value = 0.0,
		domain_name = topology_name,
		link_name = link_name,
		metrics_sampler = SyntheticSampler.create_random(
		base_bw_range = (5.0, 25.0),
		base_latency_range = (0.3, 2.0),
		sensitivity_range = (0.3, 1.0),
		curve_type = None, # Default is LINEAR
		connection_count = 0,
		link_capacity = LINKS_CAPACITY.get(link_name, 100.0)
		),
		related_service_ids=[],
		))
		@@ -686,8 +672,8 @@ class EventDispatcher(BaseEventDispatcher):

		# Update telemetry for each link involved in this connection
		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)
		# Count active connections on this link
		active_conn_count = self._count_active_connections(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)

		@@ -698,23 +684,15 @@ class EventDispatcher(BaseEventDispatcher):
		# Create worker with same parameters as in _dispatch_link_set
		resources = Resources()
		resources.links.append(ResourceLink(
		domain_name=topology_name, link_name=link_name,
		bandwidth_utilization_sampler=SyntheticSampler.create_random(
		amplitude_scale = 25.0,
		phase_scale = 1e-7,
		period_scale = 86_400,
		offset_scale = 25,
		noise_ratio = 0.05,
		min_value = 0.0,
		max_value = 100.0,
		),
		latency_sampler=SyntheticSampler.create_random(
		amplitude_scale = 0.5,
		phase_scale = 1e-7,
		period_scale = 60.0,
		offset_scale = 10.0,
		noise_ratio = 0.05,
		min_value = 0.0,
		domain_name = topology_name,
		link_name = link_name,
		metrics_sampler = SyntheticSampler.create_random(
		base_bw_range = (5.0, 25.0),
		base_latency_range = (0.3, 2.0),
		sensitivity_range = (0.3, 1.0),
		curve_type = None, # Random curve type
		connection_count = active_conn_count,
		link_capacity = LINKS_CAPACITY.get(link_name, 100.0)
		),
		related_service_ids=[],
		))
		@@ -722,14 +700,14 @@ class EventDispatcher(BaseEventDispatcher):
		self._telemetry_pool.start_synthesizer(worker_name, resources, sampling_interval)
		LOGGER.info('Started new synthesizer worker: {:s}'.format(worker_name))
		else:
		# Worker exists, update bandwidth scaling factor
		# Worker exists, update connection count for congestion simulation
		worker = self._telemetry_pool.get_worker(WorkerTypeEnum.SYNTHESIZER, worker_name)
		assert isinstance(worker, SynthesizerWorker), \
		'Expected SynthesizerWorker, got {:s}'.format(type(worker).__name__)

		worker.change_resources(bandwidth_factor, latency_factor=1.0)
		LOGGER.info('Updated telemetry of already running worker: link {:s}, and bandwidth_factor={:.2f}'.format(
		link_name, bandwidth_factor))
		worker.change_resources(active_conn_count)
		LOGGER.info('Updated telemetry of already running worker: link {:s}, connection_count={:d}'.format(
		link_name, active_conn_count))

		except Exception as e:
		LOGGER.exception('Failed to process connection event {:s}: {:s}'.format(connection_uuid, str(e)))
		@@ -789,24 +767,19 @@ class EventDispatcher(BaseEventDispatcher):
		return domain_name, processed_links # NOTE: Domain name = topology name


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

		Args:
		link_uuid: UUID of the link to calculate factor for
		link_uuid: UUID of the link to count connections 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
		int: Number of active connections using this link
		"""
		try:
		# 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
		active_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
		@@ -815,27 +788,11 @@ class EventDispatcher(BaseEventDispatcher):
		continue

		if link_uuid in cached_obj['links']:
		active_connection_count += 1
		active_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
		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} (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 1.0
		link_uuid, domain_name, active_count))
		return active_count


		def dispatch_connection_create(self, connection_event : ConnectionEvent) -> None:
		@@ -879,6 +836,10 @@ class EventDispatcher(BaseEventDispatcher):
		LOGGER.info('Retrieved cached mapping for connection {:s}: domain={:s}, links={:d}'.format(
		connection_uuid, topology_name, len(processed_links)))

		# Delete the connection from cache first (we already have the mapping)
		self._object_cache.delete(CachedEntities.CONNECTION, connection_uuid)
		LOGGER.debug('Deleted cached mapping for connection {:s}'.format(connection_uuid))

		# 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)
		@@ -887,38 +848,22 @@ class EventDispatcher(BaseEventDispatcher):
		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
		# Count remaining connections on this link (now excluding the deleted one)
		remaining_conn_count = self._count_active_connections(link_uuid, topology_name)

		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:
		if remaining_conn_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)
		# Other connections still use this link, update worker with new count
		worker = self._telemetry_pool.get_worker(WorkerTypeEnum.SYNTHESIZER, worker_name)
		assert isinstance(worker, SynthesizerWorker), \
		'Expected SynthesizerWorker, got {:s}'.format(type(worker).__name__)

		worker.change_resources(bandwidth_factor, latency_factor=1.0)
		worker.change_resources(remaining_conn_count)
		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))
		link_name, remaining_conn_count))

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

src/simap_connector/service/telemetry/worker/SynthesizerWorker.py

+2 −3

Original line number	Diff line number	Diff line
		@@ -34,11 +34,10 @@ class SynthesizerWorker(_Worker):
		self._resources = resources
		self._sampling_interval = sampling_interval

		def change_resources(self, bandwidth_factor : float, latency_factor : float) -> None:
		def change_resources(self, connection_count: int) -> None:
		with self._lock:
		for link in self._resources.links:
		link.bandwidth_utilization_sampler.offset *= bandwidth_factor
		link.latency_sampler.offset *= latency_factor
		link.metrics_sampler.connection_count = connection_count

		def run(self) -> None:
		self._logger.info('[run] Starting...')