Simap Connector and NBI component: (cac5074d) · Commits · TFS / controller

proto/simap_connector.proto

+10 −2

Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ import "context.proto";
		service SimapConnectorService {
		rpc EstablishSubscription (Subscription ) returns (SubscriptionId) {}
		rpc DeleteSubscription (SubscriptionId) returns (context.Empty ) {}
		rpc AffectSampleSynthesizer(Affectation ) returns (context.Empty ) {}
		}

		message SubscriptionId {
		@@ -33,3 +34,10 @@ message Subscription {
		string xpath_filter = 2;
		float period = 3;
		}

		message Affectation {
		string network_id = 1;
		string link_id = 2;
		float bandwidth_factor = 3;
		float latency_factor = 4;
		}

src/nbi/service/sse_telemetry/AffectSampleSynthesizer.py

0 → 100644

+60 −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 logging
		from flask import jsonify, request
		from flask_restful import Resource
		from werkzeug.exceptions import BadRequest, UnsupportedMediaType
		from common.proto.simap_connector_pb2 import Affectation
		from simap_connector.client.SimapConnectorClient import SimapConnectorClient


		LOGGER = logging.getLogger(__name__)


		class AffectSampleSynthesizer(Resource):
		# @HTTP_AUTH.login_required
		def post(self):
		if not request.is_json:
		raise UnsupportedMediaType('JSON payload is required')

		request_data = request.json
		LOGGER.debug('[post] Affectation request: {:s}'.format(str(request_data)))

		if 'network_id' not in request_data:
		raise BadRequest('Missing field(network_id)')
		network_id = str(request_data['network_id'])

		if 'link_id' not in request_data:
		raise BadRequest('Missing field(link_id)')
		link_id = str(request_data['link_id'])

		if 'bandwidth_factor' not in request_data:
		raise BadRequest('Missing field(bandwidth_factor)')
		bandwidth_factor = float(request_data['bandwidth_factor'])

		if 'latency_factor' not in request_data:
		raise BadRequest('Missing field(latency_factor)')
		latency_factor = float(request_data['latency_factor'])

		affectation = Affectation()
		affectation.network_id = network_id
		affectation.link_id = link_id
		affectation.bandwidth_factor = bandwidth_factor
		affectation.latency_factor = latency_factor

		simap_connector_client = SimapConnectorClient()
		simap_connector_client.AffectSampleSynthesizer(affectation)
		return jsonify({})

src/nbi/service/sse_telemetry/init.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -21,11 +21,11 @@


		from nbi.service.NbiApplication import NbiApplication
		from .AffectSampleSynthesizer import AffectSampleSynthesizer
		from .EstablishSubscription import EstablishSubscription
		from .DeleteSubscription import DeleteSubscription
		from .StreamSubscription import StreamSubscription


		def register_telemetry_subscription(nbi_app: NbiApplication):
		nbi_app.add_rest_api_resource(
		EstablishSubscription,
		@@ -45,3 +45,9 @@ def register_telemetry_subscription(nbi_app: NbiApplication):
		'/restconf/stream/<int:subscription_id>/',
		endpoint='sse.stream',
		)
		nbi_app.add_rest_api_resource(
		AffectSampleSynthesizer,
		'/affect_sample_synthesizer',
		'/affect_sample_synthesizer/',
		endpoint='sse.affect_sample_synthesizer',
		)

src/simap_connector/client/SimapConnectorClient.py

+8 −1

Original line number	Diff line number	Diff line
		@@ -16,7 +16,7 @@ import grpc, logging
		from common.Constants import ServiceNameEnum
		from common.Settings import get_service_host, get_service_port_grpc
		from common.proto.context_pb2 import Empty
		from common.proto.simap_connector_pb2 import Subscription, SubscriptionId
		from common.proto.simap_connector_pb2 import Affectation, Subscription, SubscriptionId
		from common.proto.simap_connector_pb2_grpc import SimapConnectorServiceStub
		from common.tools.client.RetryDecorator import delay_exponential, retry
		from common.tools.grpc.Tools import grpc_message_to_json_string
		@@ -62,3 +62,10 @@ class SimapConnectorClient:
		response = self.stub.DeleteSubscription(request)
		LOGGER.debug('DeleteSubscription result: {:s}'.format(grpc_message_to_json_string(response)))
		return response

		@RETRY_DECORATOR
		def AffectSampleSynthesizer(self, request : Affectation) -> Empty:
		LOGGER.debug('AffectSampleSynthesizer request: {:s}'.format(grpc_message_to_json_string(request)))
		response = self.stub.AffectSampleSynthesizer(request)
		LOGGER.debug('AffectSampleSynthesizer result: {:s}'.format(grpc_message_to_json_string(response)))
		return response

src/simap_connector/service/SimapConnectorServiceServicerImpl.py

+23 −1

Original line number	Diff line number	Diff line
		@@ -14,12 +14,14 @@


		import grpc, logging, sqlalchemy
		from typing import Optional
		from common.proto.context_pb2 import Empty
		from common.proto.simap_connector_pb2 import Subscription, SubscriptionId
		from common.proto.simap_connector_pb2 import Affectation, Subscription, SubscriptionId
		from common.proto.simap_connector_pb2_grpc import SimapConnectorServiceServicer
		from common.tools.rest_conf.client.RestConfClient import RestConfClient
		from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
		from device.client.DeviceClient import DeviceClient
		from simap_connector.service.telemetry.worker.SynthesizerWorker import SynthesizerWorker
		from simap_connector.service.telemetry.worker._Worker import WorkerTypeEnum
		from .database.Subscription import subscription_get, subscription_set, subscription_delete
		from .database.SubSubscription import (
		@@ -153,3 +155,23 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):

		subscription_delete(self._db_engine, parent_subscription_id)
		return Empty()


		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def AffectSampleSynthesizer(
		self, request : Affectation, context : grpc.ServicerContext
		) -> Empty:
		network_id = request.network_id
		link_id = request.link_id
		bandwidth_factor = request.bandwidth_factor
		latency_factor = request.latency_factor

		synthesizer_name = '{:s}:{:s}'.format(network_id, link_id)
		synthesizer : Optional[SynthesizerWorker] = self._telemetry_pool.get_worker(
		WorkerTypeEnum.SYNTHESIZER, synthesizer_name
		)
		if synthesizer is None:
		MSG = 'Synthesizer({:s}) not found'
		raise Exception(MSG.format(synthesizer_name))
		synthesizer.change_resources(bandwidth_factor, latency_factor)
		return Empty()