Loading src/tests/tools/simap_server/run_client.sh 0 → 100755 +19 −0 Original line number Diff line number Diff line #!/bin/bash # 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. # Make folder containing the script the root folder for its execution cd $(dirname $0)/../../../ python -m tests.tools.simap_server.simap_client src/tests/tools/simap_server/simap_client/SimapClient.py 0 → 100644 +352 −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 typing import Dict, List, Optional, Tuple from common.tools.rest_conf.client.RestConfClient import RestConfClient class TerminationPoint: ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}/node={:s}' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:termination-point={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str, tp_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tp_id = tp_id def create(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network-topology:termination-point'][0] def update(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) self._restconf_client.delete(endpoint) class NodeTelemetry: ENDPOINT = '/ietf-network:networks/network={:s}/node={:s}' # ENDPOINT = '/ietf-network:networks/network={:s}/node={:s}/simap-telemetry:simap-telemetry' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id def create( self, cpu_utilization : float, related_service_ids : List[str] = [] ) -> None: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) telemetry = { 'cpu-utilization': '{:.2f}'.format(cpu_utilization), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) telemetry : Dict = self._restconf_client.get(endpoint) return telemetry def update( self, cpu_utilization : float, related_service_ids : List[str] = [] ) -> None: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) telemetry = { 'cpu-utilization': '{:.2f}'.format(cpu_utilization), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) self._restconf_client.delete(endpoint) class Node: ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}' ENDPOINT_ID = ENDPOINT_NO_ID + '/node={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tps : Dict[str, TerminationPoint] = dict() self._telemetry : Optional[NodeTelemetry] = None @property def telemetry(self) -> NodeTelemetry: if self._telemetry is None: self._telemetry = NodeTelemetry(self._restconf_client, self._network_id, self._node_id) return self._telemetry def termination_points(self) -> List[Dict]: tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID) return tps['ietf-network-topology:termination-point'].get('termination-point', list()) def termination_point(self, tp_id : str) -> TerminationPoint: _tp = self._tps.get(tp_id) if _tp is not None: return _tp _tp = TerminationPoint(self._restconf_client, self._network_id, self._node_id, tp_id) return self._tps.setdefault(tp_id, _tp) def create( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network:node'][0] def update( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) self._restconf_client.delete(endpoint) class LinkTelemetry: ENDPOINT = '/ietf-network:networks/network={:s}/ietf-network-topology:link={:s}' # ENDPOINT = '/ietf-network:networks/network={:s}/ietf-network-topology:link={:s}/simap-telemetry:simap-telemetry' def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str): self._restconf_client = restconf_client self._network_id = network_id self._link_id = link_id def create( self, bandwidth_utilization : float, latency : float, related_service_ids : List[str] = [] ) -> None: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) telemetry = { 'bandwidth-utilization': '{:.2f}'.format(bandwidth_utilization), 'latency' : '{:.3f}'.format(latency), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) telemetry : Dict = self._restconf_client.get(endpoint) return telemetry def update( self, bandwidth_utilization : float, latency : float, related_service_ids : List[str] = [] ) -> None: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) telemetry = { 'bandwidth-utilization': '{:.2f}'.format(bandwidth_utilization), 'latency' : '{:.3f}'.format(latency), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) self._restconf_client.delete(endpoint) class Link: ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str): self._restconf_client = restconf_client self._network_id = network_id self._link_id = link_id self._telemetry : Optional[LinkTelemetry] = None @property def telemetry(self) -> LinkTelemetry: if self._telemetry is None: self._telemetry = LinkTelemetry(self._restconf_client, self._network_id, self._link_id) return self._telemetry def create( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link : Dict = self._restconf_client.get(endpoint) return link['ietf-network-topology:link'][0] def update( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) self._restconf_client.delete(endpoint) class Network: ENDPOINT_NO_ID = '/ietf-network:networks' ENDPOINT_ID = ENDPOINT_NO_ID + '/network={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str): self._restconf_client = restconf_client self._network_id = network_id self._nodes : Dict[str, Node] = dict() self._links : Dict[str, Link] = dict() def nodes(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Node.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('node', list()) def links(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Link.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('ietf-network-topology:link', list()) def node(self, node_id : str) -> Node: _node = self._nodes.get(node_id) if _node is not None: return _node _node = Node(self._restconf_client, self._network_id, node_id) return self._nodes.setdefault(node_id, _node) def link(self, link_id : str) -> Link: _link = self._links.get(link_id) if _link is not None: return _link _link = Link(self._restconf_client, self._network_id, link_id) return self._links.setdefault(link_id, _link) def create(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Network.ENDPOINT_ID.format(self._network_id) networks : Dict = self._restconf_client.get(endpoint) return networks['ietf-network:network'][0] def update(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) self._restconf_client.delete(endpoint) class SimapClient: def __init__(self, restconf_client : RestConfClient) -> None: self._restconf_client = restconf_client self._networks : Dict[str, Network] = dict() def networks(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Network.ENDPOINT_NO_ID) return reply['ietf-network:networks'].get('network', list()) def network(self, network_id : str) -> Network: _network = self._networks.get(network_id) if _network is not None: return _network _network = Network(self._restconf_client, network_id) return self._networks.setdefault(network_id, _network) src/tests/tools/simap_server/simap_client/SimapMetricsGenerator.py 0 → 100644 +174 −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 random import math import logging from typing import Dict, List, Tuple LOGGER = logging.getLogger(__name__) # Congestion curve types CURVE_LINEAR = 'linear' # x - steady increase CURVE_EXPONENTIAL = 'exponential' # exp(x)-1 - slow start, rapid end CURVE_LOGARITHMIC = 'logarithmic' # log(1+x) - fast start, plateau # Link profiles: (base_bw%, base_latency_ms, sensitivity, curve_type) # - sensitivity: 1.0 = highly affected by load, 0.3 = minimally affected # - curve_type: how congestion scales with load LINK_PROFILES = { 'L1' : (15.0, 1.0, 1.0, CURVE_EXPONENTIAL), 'L3' : (10.0, 0.8, 0.7, CURVE_EXPONENTIAL), 'L5' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L9' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L13': (12.0, 0.5, 0.5, CURVE_LOGARITHMIC), } MAX_SERVICES = 5 class SimapMetricsGenerator: """ Generates realistic SIMAP telemetry metrics based on service count. Higher service counts cause non-linear congestion effects. Access links are more sensitive to load than core links. """ def __init__(self, service_count: int = 0, seed: int = None): LOGGER.info("Initiating SimapMetricsGenerator") self._random = random.Random(seed) self._service_count = 0 self._service_ids: Dict[str, List[str]] = { 'te' : [], 'trans' : [], 'agg' : [], 'e2e' : [], } self.set_service_count(service_count) @property def service_count(self) -> int: return self._service_count def set_service_count(self, count: int) -> None: """Update service count and regenerate domain-specific service IDs.""" if count < 0 or count > MAX_SERVICES: raise ValueError(f"Service count must be 0-{MAX_SERVICES}, got {count}") self._service_count = count # Each domain has its own service IDs self._service_ids = { 'te' : [f'te-svc-{i+1}' for i in range(count)], 'trans' : [f'trans-svc-{i+1}' for i in range(count)], 'agg' : [f'agg-svc-{i+1}' for i in range(count)], 'e2e' : [f'e2e-svc-{i+1}' for i in range(count)], } LOGGER.info(f"Service count set to {count}, IDs per domain: {self._service_ids}") def get_service_ids(self, domain: str = 'e2e') -> List[str]: """Return current list of active service IDs for a specific domain.""" if domain not in self._service_ids: raise ValueError(f"Unknown domain: {domain}. Valid: {list(self._service_ids.keys())}") return self._service_ids[domain].copy() def get_all_service_ids(self) -> Dict[str, List[str]]: """Return all domain service IDs.""" return {k: v.copy() for k, v in self._service_ids.items()} def _compute_congestion_factor(self, curve_type: str, load_ratio: float) -> float: """ Compute congestion factor based on curve type and load ratio (0-1). """ if curve_type == CURVE_LINEAR: return load_ratio elif curve_type == CURVE_EXPONENTIAL: # Exponential: slow start, rapid increase at high load return (math.exp(load_ratio * 2) - 1) / (math.e ** 2 - 1) elif curve_type == CURVE_LOGARITHMIC: # Logarithmic: fast initial increase, then plateau return math.log1p(load_ratio * 2.7) / math.log1p(2.7) else: return load_ratio # Default to linear def generate_link_metrics(self, link_id: str) -> Tuple[float, float]: """ Generate BW and latency for a specific TE link using distinct congestion patterns. Returns: Tuple of (bandwidth_utilization%, latency_ms) """ if link_id not in LINK_PROFILES: raise ValueError(f"Unknown link ID: {link_id}") base_bw, base_latency, sensitivity, curve_type = LINK_PROFILES[link_id] # Load ratio (0 to 1) load_ratio = self._service_count / MAX_SERVICES # Compute congestion factor using link-specific curve congestion_factor = self._compute_congestion_factor(curve_type, load_ratio) # Calculate base metrics with congestion bw_utilization = base_bw + (congestion_factor * sensitivity * 60.0) latency = base_latency * (1.0 + congestion_factor * sensitivity * 4.0) # Add uniform noise (5%) bw_noise = self._random.uniform(-0.05, 0.05) * bw_utilization lat_noise = self._random.uniform(-0.05, 0.05) * latency bw_utilization = max(0.0, min(100.0, bw_utilization + bw_noise)) latency = max(0.1, latency + lat_noise) return (bw_utilization, latency) def generate_all_te_metrics(self) -> Dict[str, Tuple[float, float]]: """ Generate metrics for all TE links in the path. Returns: Dict mapping link_id to (bandwidth%, latency_ms) """ return {link_id: self.generate_link_metrics(link_id) for link_id in LINK_PROFILES} def aggregate_abstract_metrics( self, te_metrics: Dict[str, Tuple[float, float]] ) -> Dict[str, Tuple[float, float]]: """ Aggregate TE metrics into abstract layer metrics. BW: average, Latency: sum Returns: Dict with 'Trans-L1', 'AggNet-L1', 'E2E-L1' metrics """ bw_L1, lat_L1 = te_metrics['L1'] bw_L3, lat_L3 = te_metrics['L3'] bw_L5, lat_L5 = te_metrics['L5'] bw_L9, lat_L9 = te_metrics['L9'] bw_L13, lat_L13 = te_metrics['L13'] # Trans-L1: L5 + L9 bw_trans = (bw_L5 + bw_L9) / 2 lat_trans = lat_L5 + lat_L9 # AggNet-L1: L3 + Trans-L1 + L13 bw_aggnet = (bw_L3 + bw_trans + bw_L13) / 3 lat_aggnet = lat_L3 + lat_trans + lat_L13 # E2E-L1: L1 + AggNet-L1 bw_e2e = (bw_L1 + bw_aggnet) / 2 lat_e2e = lat_L1 + lat_aggnet return { 'Trans-L1' : (bw_trans, lat_trans), 'AggNet-L1': (bw_aggnet, lat_aggnet), 'E2E-L1' : (bw_e2e, lat_e2e), } src/tests/tools/simap_server/simap_client/Tools.py 0 → 100644 +109 −0 File added.Preview size limit exceeded, changes collapsed. Show changes src/tests/tools/simap_server/simap_client/__init__.py 0 → 100644 +14 −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. Loading
src/tests/tools/simap_server/run_client.sh 0 → 100755 +19 −0 Original line number Diff line number Diff line #!/bin/bash # 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. # Make folder containing the script the root folder for its execution cd $(dirname $0)/../../../ python -m tests.tools.simap_server.simap_client
src/tests/tools/simap_server/simap_client/SimapClient.py 0 → 100644 +352 −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 typing import Dict, List, Optional, Tuple from common.tools.rest_conf.client.RestConfClient import RestConfClient class TerminationPoint: ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}/node={:s}' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:termination-point={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str, tp_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tp_id = tp_id def create(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network-topology:termination-point'][0] def update(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) tp = {'tp-id': self._tp_id} stps = [ {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id} for snet_id,snode_id,stp_id in supporting_termination_point_ids ] if len(stps) > 0: tp['supporting-termination-point'] = stps node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id) self._restconf_client.delete(endpoint) class NodeTelemetry: ENDPOINT = '/ietf-network:networks/network={:s}/node={:s}' # ENDPOINT = '/ietf-network:networks/network={:s}/node={:s}/simap-telemetry:simap-telemetry' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id def create( self, cpu_utilization : float, related_service_ids : List[str] = [] ) -> None: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) telemetry = { 'cpu-utilization': '{:.2f}'.format(cpu_utilization), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) telemetry : Dict = self._restconf_client.get(endpoint) return telemetry def update( self, cpu_utilization : float, related_service_ids : List[str] = [] ) -> None: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) telemetry = { 'cpu-utilization': '{:.2f}'.format(cpu_utilization), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids node = {'node-id': self._node_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = NodeTelemetry.ENDPOINT.format(self._network_id, self._node_id) self._restconf_client.delete(endpoint) class Node: ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}' ENDPOINT_ID = ENDPOINT_NO_ID + '/node={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str, node_id : str): self._restconf_client = restconf_client self._network_id = network_id self._node_id = node_id self._tps : Dict[str, TerminationPoint] = dict() self._telemetry : Optional[NodeTelemetry] = None @property def telemetry(self) -> NodeTelemetry: if self._telemetry is None: self._telemetry = NodeTelemetry(self._restconf_client, self._network_id, self._node_id) return self._telemetry def termination_points(self) -> List[Dict]: tps : Dict = self._restconf_client.get(TerminationPoint.ENDPOINT_NO_ID) return tps['ietf-network-topology:termination-point'].get('termination-point', list()) def termination_point(self, tp_id : str) -> TerminationPoint: _tp = self._tps.get(tp_id) if _tp is not None: return _tp _tp = TerminationPoint(self._restconf_client, self._network_id, self._node_id, tp_id) return self._tps.setdefault(tp_id, _tp) def create( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node : Dict = self._restconf_client.get(endpoint) return node['ietf-network:node'][0] def update( self, termination_point_ids : List[str] = [], supporting_node_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) node = {'node-id': self._node_id} tps = [{'tp-id': tp_id} for tp_id in termination_point_ids] if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids] if len(sns) > 0: node['supporting-node'] = sns network = {'network-id': self._network_id, 'node': [node]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id) self._restconf_client.delete(endpoint) class LinkTelemetry: ENDPOINT = '/ietf-network:networks/network={:s}/ietf-network-topology:link={:s}' # ENDPOINT = '/ietf-network:networks/network={:s}/ietf-network-topology:link={:s}/simap-telemetry:simap-telemetry' def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str): self._restconf_client = restconf_client self._network_id = network_id self._link_id = link_id def create( self, bandwidth_utilization : float, latency : float, related_service_ids : List[str] = [] ) -> None: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) telemetry = { 'bandwidth-utilization': '{:.2f}'.format(bandwidth_utilization), 'latency' : '{:.3f}'.format(latency), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) telemetry : Dict = self._restconf_client.get(endpoint) return telemetry def update( self, bandwidth_utilization : float, latency : float, related_service_ids : List[str] = [] ) -> None: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) telemetry = { 'bandwidth-utilization': '{:.2f}'.format(bandwidth_utilization), 'latency' : '{:.3f}'.format(latency), } if len(related_service_ids) > 0: telemetry['related-service-ids'] = related_service_ids link = {'link-id': self._link_id, 'simap-telemetry:simap-telemetry': telemetry} network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = LinkTelemetry.ENDPOINT.format(self._network_id, self._link_id) self._restconf_client.delete(endpoint) class Link: ENDPOINT_NO_ID = '/ietf-network:networks/network={:s}' ENDPOINT_ID = ENDPOINT_NO_ID + '/ietf-network-topology:link={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str, link_id : str): self._restconf_client = restconf_client self._network_id = network_id self._link_id = link_id self._telemetry : Optional[LinkTelemetry] = None @property def telemetry(self) -> LinkTelemetry: if self._telemetry is None: self._telemetry = LinkTelemetry(self._restconf_client, self._network_id, self._link_id) return self._telemetry def create( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link : Dict = self._restconf_client.get(endpoint) return link['ietf-network-topology:link'][0] def update( self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str, supporting_link_ids : List[Tuple[str, str]] = [] ) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) link = { 'link-id' : self._link_id, 'source' : {'source-node': src_node_id, 'source-tp': src_tp_id}, 'destination': {'dest-node' : dst_node_id, 'dest-tp' : dst_tp_id}, } sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids] if len(sls) > 0: link['supporting-link'] = sls network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]} payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id) self._restconf_client.delete(endpoint) class Network: ENDPOINT_NO_ID = '/ietf-network:networks' ENDPOINT_ID = ENDPOINT_NO_ID + '/network={:s}' def __init__(self, restconf_client : RestConfClient, network_id : str): self._restconf_client = restconf_client self._network_id = network_id self._nodes : Dict[str, Node] = dict() self._links : Dict[str, Link] = dict() def nodes(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Node.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('node', list()) def links(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Link.ENDPOINT_NO_ID.format(self._network_id)) return reply['ietf-network:network'][0].get('ietf-network-topology:link', list()) def node(self, node_id : str) -> Node: _node = self._nodes.get(node_id) if _node is not None: return _node _node = Node(self._restconf_client, self._network_id, node_id) return self._nodes.setdefault(node_id, _node) def link(self, link_id : str) -> Link: _link = self._links.get(link_id) if _link is not None: return _link _link = Link(self._restconf_client, self._network_id, link_id) return self._links.setdefault(link_id, _link) def create(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.post(endpoint, payload) def get(self) -> Dict: endpoint = Network.ENDPOINT_ID.format(self._network_id) networks : Dict = self._restconf_client.get(endpoint) return networks['ietf-network:network'][0] def update(self, supporting_network_ids : List[str] = []) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) network = {'network-id': self._network_id} sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids] if len(sns) > 0: network['supporting-network'] = sns payload = {'ietf-network:networks': {'network': [network]}} self._restconf_client.patch(endpoint, payload) def delete(self) -> None: endpoint = Network.ENDPOINT_ID.format(self._network_id) self._restconf_client.delete(endpoint) class SimapClient: def __init__(self, restconf_client : RestConfClient) -> None: self._restconf_client = restconf_client self._networks : Dict[str, Network] = dict() def networks(self) -> List[Dict]: reply : Dict = self._restconf_client.get(Network.ENDPOINT_NO_ID) return reply['ietf-network:networks'].get('network', list()) def network(self, network_id : str) -> Network: _network = self._networks.get(network_id) if _network is not None: return _network _network = Network(self._restconf_client, network_id) return self._networks.setdefault(network_id, _network)
src/tests/tools/simap_server/simap_client/SimapMetricsGenerator.py 0 → 100644 +174 −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 random import math import logging from typing import Dict, List, Tuple LOGGER = logging.getLogger(__name__) # Congestion curve types CURVE_LINEAR = 'linear' # x - steady increase CURVE_EXPONENTIAL = 'exponential' # exp(x)-1 - slow start, rapid end CURVE_LOGARITHMIC = 'logarithmic' # log(1+x) - fast start, plateau # Link profiles: (base_bw%, base_latency_ms, sensitivity, curve_type) # - sensitivity: 1.0 = highly affected by load, 0.3 = minimally affected # - curve_type: how congestion scales with load LINK_PROFILES = { 'L1' : (15.0, 1.0, 1.0, CURVE_EXPONENTIAL), 'L3' : (10.0, 0.8, 0.7, CURVE_EXPONENTIAL), 'L5' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L9' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L13': (12.0, 0.5, 0.5, CURVE_LOGARITHMIC), } MAX_SERVICES = 5 class SimapMetricsGenerator: """ Generates realistic SIMAP telemetry metrics based on service count. Higher service counts cause non-linear congestion effects. Access links are more sensitive to load than core links. """ def __init__(self, service_count: int = 0, seed: int = None): LOGGER.info("Initiating SimapMetricsGenerator") self._random = random.Random(seed) self._service_count = 0 self._service_ids: Dict[str, List[str]] = { 'te' : [], 'trans' : [], 'agg' : [], 'e2e' : [], } self.set_service_count(service_count) @property def service_count(self) -> int: return self._service_count def set_service_count(self, count: int) -> None: """Update service count and regenerate domain-specific service IDs.""" if count < 0 or count > MAX_SERVICES: raise ValueError(f"Service count must be 0-{MAX_SERVICES}, got {count}") self._service_count = count # Each domain has its own service IDs self._service_ids = { 'te' : [f'te-svc-{i+1}' for i in range(count)], 'trans' : [f'trans-svc-{i+1}' for i in range(count)], 'agg' : [f'agg-svc-{i+1}' for i in range(count)], 'e2e' : [f'e2e-svc-{i+1}' for i in range(count)], } LOGGER.info(f"Service count set to {count}, IDs per domain: {self._service_ids}") def get_service_ids(self, domain: str = 'e2e') -> List[str]: """Return current list of active service IDs for a specific domain.""" if domain not in self._service_ids: raise ValueError(f"Unknown domain: {domain}. Valid: {list(self._service_ids.keys())}") return self._service_ids[domain].copy() def get_all_service_ids(self) -> Dict[str, List[str]]: """Return all domain service IDs.""" return {k: v.copy() for k, v in self._service_ids.items()} def _compute_congestion_factor(self, curve_type: str, load_ratio: float) -> float: """ Compute congestion factor based on curve type and load ratio (0-1). """ if curve_type == CURVE_LINEAR: return load_ratio elif curve_type == CURVE_EXPONENTIAL: # Exponential: slow start, rapid increase at high load return (math.exp(load_ratio * 2) - 1) / (math.e ** 2 - 1) elif curve_type == CURVE_LOGARITHMIC: # Logarithmic: fast initial increase, then plateau return math.log1p(load_ratio * 2.7) / math.log1p(2.7) else: return load_ratio # Default to linear def generate_link_metrics(self, link_id: str) -> Tuple[float, float]: """ Generate BW and latency for a specific TE link using distinct congestion patterns. Returns: Tuple of (bandwidth_utilization%, latency_ms) """ if link_id not in LINK_PROFILES: raise ValueError(f"Unknown link ID: {link_id}") base_bw, base_latency, sensitivity, curve_type = LINK_PROFILES[link_id] # Load ratio (0 to 1) load_ratio = self._service_count / MAX_SERVICES # Compute congestion factor using link-specific curve congestion_factor = self._compute_congestion_factor(curve_type, load_ratio) # Calculate base metrics with congestion bw_utilization = base_bw + (congestion_factor * sensitivity * 60.0) latency = base_latency * (1.0 + congestion_factor * sensitivity * 4.0) # Add uniform noise (5%) bw_noise = self._random.uniform(-0.05, 0.05) * bw_utilization lat_noise = self._random.uniform(-0.05, 0.05) * latency bw_utilization = max(0.0, min(100.0, bw_utilization + bw_noise)) latency = max(0.1, latency + lat_noise) return (bw_utilization, latency) def generate_all_te_metrics(self) -> Dict[str, Tuple[float, float]]: """ Generate metrics for all TE links in the path. Returns: Dict mapping link_id to (bandwidth%, latency_ms) """ return {link_id: self.generate_link_metrics(link_id) for link_id in LINK_PROFILES} def aggregate_abstract_metrics( self, te_metrics: Dict[str, Tuple[float, float]] ) -> Dict[str, Tuple[float, float]]: """ Aggregate TE metrics into abstract layer metrics. BW: average, Latency: sum Returns: Dict with 'Trans-L1', 'AggNet-L1', 'E2E-L1' metrics """ bw_L1, lat_L1 = te_metrics['L1'] bw_L3, lat_L3 = te_metrics['L3'] bw_L5, lat_L5 = te_metrics['L5'] bw_L9, lat_L9 = te_metrics['L9'] bw_L13, lat_L13 = te_metrics['L13'] # Trans-L1: L5 + L9 bw_trans = (bw_L5 + bw_L9) / 2 lat_trans = lat_L5 + lat_L9 # AggNet-L1: L3 + Trans-L1 + L13 bw_aggnet = (bw_L3 + bw_trans + bw_L13) / 3 lat_aggnet = lat_L3 + lat_trans + lat_L13 # E2E-L1: L1 + AggNet-L1 bw_e2e = (bw_L1 + bw_aggnet) / 2 lat_e2e = lat_L1 + lat_aggnet return { 'Trans-L1' : (bw_trans, lat_trans), 'AggNet-L1': (bw_aggnet, lat_aggnet), 'E2E-L1' : (bw_e2e, lat_e2e), }
src/tests/tools/simap_server/simap_client/Tools.py 0 → 100644 +109 −0 File added.Preview size limit exceeded, changes collapsed. Show changes
src/tests/tools/simap_server/simap_client/__init__.py 0 → 100644 +14 −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.
