SIMAP Connector: (7633957c) · Commits · TFS / controller

src/simap_connector/service/simap_updater/SimapUpdater.py

+5 −1

Original line number	Diff line number	Diff line
		@@ -30,6 +30,7 @@ from simap_connector.Config import (
		SIMAP_SERVER_USERNAME, SIMAP_SERVER_PASSWORD,
		)
		from simap_connector.service.simap_updater.MockSimaps import delete_mock_simap, set_mock_simap
		from simap_connector.service.telemetry.Resources import Resources
		from simap_connector.service.telemetry.TelemetryPool import TelemetryPool
		from .SimapClient import SimapClient
		from .ObjectCache import CachedEntities, ObjectCache
		@@ -511,7 +512,10 @@ class EventDispatcher(BaseEventDispatcher):
		#dom_link = domain_topo.link(link_name)
		#dom_link.update(src_dev_name, src_ep_name, dst_dev_name, dst_ep_name)

		self._telemetry_pool.start_worker(domain_name)

		resources = Resources()
		sampling_interval = 1.0
		self._telemetry_pool.start_worker(domain_name, resources, sampling_interval)
		return True

src/simap_connector/service/telemetry/Resources.py

0 → 100644

+62 −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 dataclasses import dataclass, field
		from typing import List
		from simap_connector.service.simap_updater.SimapClient import SimapClient
		from simap_connector.service.telemetry.SyntheticSamplers import SyntheticSampler


		@dataclass
		class ResourceNode:
		node_name : str
		cpu_utilization_sampler : SyntheticSampler
		related_service_ids : List[str] = field(default_factory=list)

		def generate_samples(self, simap_client : SimapClient, domain_name : str) -> None:
		cpu_utilization = self.cpu_utilization_sampler.get_sample()
		simap_node = simap_client.network(domain_name).node(self.node_name)
		simap_node.telemetry.update(
		cpu_utilization.value, related_service_ids=self.related_service_ids
		)

		@dataclass
		class ResourceLink:
		link_name : str
		bandwidth_utilization_sampler : SyntheticSampler
		latency_sampler : SyntheticSampler
		related_service_ids : List[str] = field(default_factory=list)

		def generate_samples(self, simap_client : SimapClient, domain_name : str) -> None:
		bandwidth_utilization = self.bandwidth_utilization_sampler.get_sample()
		latency = self.latency_sampler.get_sample()
		simap_link = simap_client.network(domain_name).link(self.link_name)
		simap_link.telemetry.update(
		bandwidth_utilization.value, latency.value,
		related_service_ids=self.related_service_ids
		)


		@dataclass
		class Resources:
		nodes : List[ResourceNode] = field(default_factory=list)
		links : List[ResourceLink] = field(default_factory=list)

		def generate_samples(self, simap_client : SimapClient, domain_name : str) -> None:
		for resource in self.nodes:
		resource.generate_samples(simap_client, domain_name)

		for resource in self.links:
		resource.generate_samples(simap_client, domain_name)

src/simap_connector/service/telemetry/SyntheticSamplers.py

0 → 100644

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


		import math, random, threading
		from dataclasses import dataclass
		from datetime import datetime
		from typing import Dict


		@dataclass
		class Sample:
		timestamp : float
		value : float


		@dataclass
		class SyntheticSampler:
		amplitude : float = 0.0
		phase : float = 0.0
		period : float = 1.0
		offset : float = 0.0
		noise_ratio : float = 0.0

		@classmethod
		def create_random(
		cls, amplitude_scale : float, phase_scale : float, period_scale : float,
		offset_scale : float, noise_ratio : float
		) -> 'SyntheticSampler':
		amplitude = amplitude_scale * random.random()
		phase = phase_scale * random.random()
		period = period_scale * random.random()
		offset = offset_scale * random.random() + amplitude
		return cls(amplitude, phase, period, offset, noise_ratio)

		def get_sample(self) -> Sample:
		timestamp = datetime.timestamp(datetime.utcnow())

		waveform = math.sin(2 * math.pi * timestamp / self.period + self.phase)
		waveform *= self.amplitude
		waveform += self.offset

		noise = self.amplitude * random.random()
		value = abs((1.0 - self.noise_ratio) * waveform + self.noise_ratio * noise)

		return Sample(timestamp, value)


		class SyntheticSamplers:
		def __init__(self) -> None:
		self._lock = threading.Lock()
		self._samplers : Dict[str, SyntheticSampler] = dict()

		def add_sampler(
		self, sampler_name : str, amplitude_scale : float, phase_scale : float,
		period_scale : float, offset_scale : float, noise_ratio : float
		) -> None:
		with self._lock:
		if sampler_name in self._samplers:
		MSG = 'SyntheticSampler({:s}) already exists'
		raise Exception(MSG.format(sampler_name))
		self._samplers[sampler_name] = SyntheticSampler.create_random(
		amplitude_scale, phase_scale, period_scale, offset_scale, noise_ratio
		)

		def remove_sampler(self, sampler_name : str) -> None:
		with self._lock:
		self._samplers.pop(sampler_name, None)

		def get_sample(self, sampler_name : str) -> Sample:
		with self._lock:
		sampler = self._samplers.get(sampler_name)
		if sampler_name not in self._samplers:
		MSG = 'SyntheticSampler({:s}) does not exist'
		raise Exception(MSG.format(sampler_name))
		return sampler.get_sample()

src/simap_connector/service/telemetry/TelemetryPool.py

+13 −4

Original line number	Diff line number	Diff line
		@@ -15,6 +15,8 @@

		import logging, threading
		from typing import Dict, Optional
		from simap_connector.service.simap_updater.SimapClient import SimapClient
		from .Resources import Resources
		from .TelemetryWorker import TelemetryWorker


		@@ -23,27 +25,34 @@ LOGGER = logging.getLogger(__name__)

		class TelemetryPool:
		def __init__(
		self, terminate : Optional[threading.Event] = None
		self, simap_client : SimapClient, terminate : Optional[threading.Event] = None
		) -> None:
		self._simap_client = simap_client
		self._workers : Dict[str, TelemetryWorker] = dict()
		self._lock = threading.Lock()
		self._terminate = threading.Event() if terminate is None else terminate


		def start_worker(self, domain_name : str) -> None:
		def start_worker(
		self, domain_name : str, resources : Resources, sampling_interval : float
		) -> None:
		with self._lock:
		if domain_name in self._workers:
		MSG = '[start_worker] Worker already running for Domain({:s})'
		LOGGER.debug(MSG.format(str(domain_name)))
		return

		worker = TelemetryWorker(domain_name, terminate=self._terminate)
		self._workers[domain_name] = worker
		worker = TelemetryWorker(
		domain_name, self._simap_client, resources, sampling_interval,
		terminate=self._terminate
		)
		worker.start()

		MSG = '[start_worker] Started worker for Domain({:s})'
		LOGGER.info(MSG.format(str(domain_name)))

		self._workers[domain_name] = worker


		def stop_worker(self, domain_name : str) -> None:
		with self._lock:

src/simap_connector/service/telemetry/TelemetryWorker.py

+19 −10

Original line number	Diff line number	Diff line
		@@ -15,6 +15,8 @@

		import logging, threading, time
		from typing import Optional
		from simap_connector.service.simap_updater.SimapClient import SimapClient
		from .Resources import Resources


		LOGGER = logging.getLogger(__name__)
		@@ -22,38 +24,45 @@ LOGGER = logging.getLogger(__name__)

		class TelemetryWorker(threading.Thread):
		def __init__(
		self, domain_name : str, terminate : Optional[threading.Event] = None
		self, domain_name : str, simap_client : SimapClient, resources : Resources,
		sampling_interval : float, terminate : Optional[threading.Event] = None
		) -> None:
		name = 'TelemetryWorker-{:s}'.format(str(domain_name))
		name = 'TelemetryWorker({:s})'.format(str(domain_name))
		super().__init__(name=name, daemon=True)
		self.domain_name = domain_name
		self._domain_name = domain_name
		self._simap_client = simap_client
		self._resources = resources
		self._sampling_interval = sampling_interval
		self._stop_event = threading.Event()
		self._terminate = threading.Event() if terminate is None else terminate

		def stop(self) -> None:
		MSG = '[stop][{:s}] Stopping...'
		LOGGER.info(MSG.format(str(self.domain_name)))
		LOGGER.info(MSG.format(str(self._domain_name)))
		self._stop_event.set()
		self.join()

		def run(self) -> None:
		MSG = '[run][{:s}] Starting...'
		LOGGER.info(MSG.format(str(self.domain_name)))
		LOGGER.info(MSG.format(str(self._domain_name)))

		try:
		while not self._stop_event.is_set() and not self._terminate.is_set():
		MSG = '[run][{:s}] Sampling...'
		LOGGER.info(MSG.format(str(self._domain_name)))

		MSG = '[run][{:s}] Heartbeat'
		LOGGER.info(MSG.format(str(self.domain_name)))
		self._resources.generate_samples(self._simap_client, self._domain_name)

		for _ in range(10):
		# Make wait responsible to terminations
		iterations = self._sampling_interval / 0.1
		for _ in range(iterations):
		if self._stop_event.is_set(): break
		if self._terminate.is_set() : break
		time.sleep(0.1)

		except Exception:
		MSG = '[run][{:s}] Unhandled Exception'
		LOGGER.info(MSG.format(str(self.domain_name)))
		LOGGER.info(MSG.format(str(self._domain_name)))
		finally:
		MSG = '[run][{:s}] Terminated'
		LOGGER.info(MSG.format(str(self.domain_name)))
		LOGGER.info(MSG.format(str(self._domain_name)))