fix: Enhance logging in AggregatorWorker and AggregationCache for better traceability (9aa27391) · Commits · TFS / controller

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

+11 −1

Original line number	Diff line number	Diff line
		@@ -72,13 +72,17 @@ class AggregatorWorker(_Worker):

		def run(self) -> None:
		self._logger.info('[run] Starting...')
		MSG = '[run] Aggregating link ({:s}, {:s}) every {:.1f}s'
		self._logger.info(MSG.format(
		self._network_id, self._link_id, self._sampling_interval
		))

		kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_BOOT_SERVERS)
		update_counter = 1

		try:
		while not self._stop_event.is_set() and not self._terminate.is_set():
		#self._logger.debug('[run] Aggregating...')
		self._logger.debug('[run] Aggregation cycle #{:d}...'.format(update_counter))

		link_sample = self._aggregation_cache.aggregate()

		@@ -111,6 +115,12 @@ class AggregatorWorker(_Worker):
		related_service_ids=list(link_sample.related_service_ids)
		)

		MSG = '[run] Updated SIMAP link ({:s}, {:s}): BW={:.2f}%, Latency={:.3f}ms'
		self._logger.debug(MSG.format(
		self._network_id, self._link_id,
		link_sample.bandwidth_utilization, link_sample.latency
		))

		update_counter += 1

		# Make wait responsible to terminations

src/simap_connector/service/telemetry/worker/data/AggregationCache.py

+34 −2

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


		import threading
		import logging, threading
		from dataclasses import dataclass, field
		from datetime import datetime
		from typing import Dict, Set, Tuple


		LOGGER = logging.getLogger(__name__)


		@dataclass
		class LinkSample:
		network_id : str
		@@ -47,11 +50,32 @@ class AggregationCache:
		with self._lock:
		self._samples[link_key] = link_sample

		MSG = '[update] Received sample for link ({:s}, {:s}): BW={:.2f}%, Latency={:.3f}ms, Services={:s}'
		LOGGER.debug(MSG.format(
		link_sample.network_id, link_sample.link_id,
		link_sample.bandwidth_utilization, link_sample.latency,
		str(link_sample.related_service_ids)
		))


		def aggregate(self) -> AggregatedLinkSample:
		with self._lock:
		num_samples = len(self._samples)
		if num_samples > 0:
		MSG = '[aggregate] Aggregating {:d} supporting link(s)'
		LOGGER.info(MSG.format(num_samples))

		agg = AggregatedLinkSample(timestamp=datetime.utcnow())
		for sample in self._samples.values():
		for link_key, sample in self._samples.items():
		network_id, link_id = link_key

		MSG = '[aggregate] - Link ({:s}, {:s}): BW={:.2f}%, Latency={:.3f}ms, Services={:s}'
		LOGGER.debug(MSG.format(
		network_id, link_id,
		sample.bandwidth_utilization, sample.latency,
		str(sample.related_service_ids)
		))

		agg.bandwidth_utilization = max(
		agg.bandwidth_utilization, sample.bandwidth_utilization
		)
		@@ -59,4 +83,12 @@ class AggregationCache:
		agg.related_service_ids = agg.related_service_ids.union(
		sample.related_service_ids
		)

		if num_samples > 0:
		MSG = '[aggregate] Result: BW={:.2f}% (max), Latency={:.3f}ms (sum), Services={:s}'
		LOGGER.info(MSG.format(
		agg.bandwidth_utilization, agg.latency,
		str(agg.related_service_ids)
		))

		return agg