Simap Connector: (acf4d381) · Commits · TFS / controller

src/simap_connector/service/SimapConnectorService.py

+7 −2

Original line number	Diff line number	Diff line
		@@ -18,7 +18,9 @@ from common.Constants import ServiceNameEnum
		from common.Settings import get_service_port_grpc
		from common.proto.simap_connector_pb2 import DESCRIPTOR as SIMAP_CONNECTOR_DESCRIPTOR
		from common.proto.simap_connector_pb2_grpc import add_SimapConnectorServiceServicer_to_server
		from common.tools.rest_conf.client.RestConfClient import RestConfClient
		from common.tools.service.GenericGrpcService import GenericGrpcService
		from .telemetry.TelemetryPool import TelemetryPool
		from .SimapConnectorServiceServicerImpl import SimapConnectorServiceServicerImpl


		@@ -27,11 +29,14 @@ LOGGER = logging.getLogger(__name__)

		class SimapConnectorService(GenericGrpcService):
		def __init__(
		self, db_engine : sqlalchemy.engine.Engine, cls_name : str = __name__
		self, db_engine : sqlalchemy.engine.Engine, restconf_client : RestConfClient,
		telemetry_pool : TelemetryPool, cls_name : str = __name__
		) -> None:
		port = get_service_port_grpc(ServiceNameEnum.SIMAP_CONNECTOR)
		super().__init__(port, cls_name=cls_name)
		self.simap_connector_servicer = SimapConnectorServiceServicerImpl(db_engine)
		self.simap_connector_servicer = SimapConnectorServiceServicerImpl(
		db_engine, restconf_client, telemetry_pool
		)

		def install_servicers(self):
		add_SimapConnectorServiceServicer_to_server(self.simap_connector_servicer, self.server)

src/simap_connector/service/SimapConnectorServiceServicerImpl.py

+101 −7

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


		import grpc, logging, sqlalchemy
		from typing import List
		from common.proto.context_pb2 import Empty
		from common.proto.simap_connector_pb2 import 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 .database.Subscription import subscription_set, subscription_delete

		from device.client.DeviceClient import DeviceClient
		from .database.Subscription import subscription_get, subscription_set, subscription_delete
		from .database.SubSubscription import (
		sub_subscription_list, sub_subscription_set, sub_subscription_delete
		)
		from .telemetry.worker.data.AggregationCache import AggregationCache
		from .telemetry.TelemetryPool import TelemetryPool
		from .Tools import (
		SupportingLink, create_kafka_topic, delete_kafka_topic, delete_underlay_subscription,
		discover_supporting_links, establish_underlay_subscription, get_controller_id,
		)

		LOGGER = logging.getLogger(__name__)

		@@ -27,17 +38,100 @@ METRICS_POOL = MetricsPool('SimapConnector', 'RPC')


		class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):
		def __init__(self, db_engine : sqlalchemy.engine.Engine) -> None:
		def __init__(
		self, db_engine : sqlalchemy.engine.Engine, restconf_client : RestConfClient,
		telemetry_pool : TelemetryPool
		) -> None:
		LOGGER.debug('Creating Servicer...')
		self.db_engine = db_engine
		self._db_engine = db_engine
		self._restconf_client = restconf_client
		self._telemetry_pool = telemetry_pool
		LOGGER.debug('Servicer Created')


		def _get_metrics(self) -> MetricsPool: return METRICS_POOL


		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def EstablishSubscription(self, request : Subscription, context : grpc.ServicerContext) -> SubscriptionId:
		return subscription_set(self.db_engine, request)
		def EstablishSubscription(
		self, request : Subscription, context : grpc.ServicerContext
		) -> SubscriptionId:
		datastore = request.datastore
		xpath_filter = request.xpath_filter
		period = request.period
		supporting_links : List[SupportingLink] = discover_supporting_links(
		self._restconf_client, xpath_filter
		)

		parent_subscription_uuid, parent_subscription_id = subscription_set(
		self._db_engine, datastore, xpath_filter, period
		)

		aggregation_cache = AggregationCache()

		device_client = DeviceClient()
		sup_link_xpath_filters : List[str] = list()
		for supporting_link in supporting_links:
		controller_id = get_controller_id(supporting_link.network_id)
		sup_link_xpath_filter = supporting_link.get_xpath_filter()
		sup_link_xpath_filters.append(sup_link_xpath_filter)

		if controller_id is not None:
		underlay_sub_id = establish_underlay_subscription(
		device_client, controller_id, sup_link_xpath_filter, period
		)

		collector_name = '{:s}:{:s}'.format(
		controller_id, str(underlay_sub_id.subscription_id)
		)
		self._telemetry_pool.start_collector(
		collector_name, underlay_sub_id.subscription_id, controller_id,
		underlay_sub_id.subscription_uri, aggregation_cache, period
		)

		sub_request = Subscription()
		sub_request.datastore = datastore
		sub_request.xpath_filter = sup_link_xpath_filter
		sub_request.period = period
		sub_subscription_set(
		self._db_engine, parent_subscription_uuid, controller_id, datastore,
		sup_link_xpath_filter, period, underlay_sub_id.subscription_id,
		underlay_sub_id.subscription_uri
		)

		topic = 'subscription.{:d}'.format(parent_subscription_id)
		create_kafka_topic(topic)

		aggregator_name = '{:s}:{:s}'.format(
		controller_id, str(parent_subscription_id)
		)
		self._telemetry_pool.start_aggregator(
		aggregator_name, parent_subscription_id, aggregation_cache, topic, period
		)

		return SubscriptionId(subscription_id=parent_subscription_id)


		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def DeleteSubscription(self, request : SubscriptionId, context : grpc.ServicerContext) -> Empty:
		return subscription_delete(self.db_engine, request)
		parent_subscription_id = request.subscription_id
		subscription = subscription_get(self._db_engine, parent_subscription_id)
		if subscription is None: return Empty()

		# TODO: desactivate subscription aggregator and collectors

		topic = 'subscription.{:d}'.format(parent_subscription_id)
		delete_kafka_topic(topic)

		parent_subscription_uuid = subscription['subscription_uuid']

		device_client = DeviceClient()
		sub_subscriptions = sub_subscription_list(self._db_engine, parent_subscription_uuid)
		for sub_subscription in sub_subscriptions:
		sub_subscription_id = sub_subscription['sub_subscription_id']
		controller_id = sub_subscription['controller_uuid' ]
		delete_underlay_subscription(device_client, controller_id, sub_subscription_id)
		sub_subscription_delete(self._db_engine, parent_subscription_uuid, sub_subscription_id)

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

src/simap_connector/service/Tools.py

0 → 100644

+156 −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 dataclasses import dataclass
		from kafka.admin import KafkaAdminClient, NewTopic
		from kafka.errors import BrokerResponseError
		from typing import List, Optional
		from common.proto.monitoring_pb2 import (
		SSEMonitoringSubscriptionConfig, SSEMonitoringSubscriptionResponse
		)
		from common.tools.kafka.Variables import KafkaConfig
		from common.tools.rest_conf.client.RestConfClient import RestConfClient
		from device.client.DeviceClient import DeviceClient


		LOGGER = logging.getLogger(__name__)


		XPATH_LINK_TEMPLATE = (
		'/ietf-network:networks/network={:s}'
		'/ietf-network-topology:link={:s}/simap-telemetry:simap-telemetry'
		)

		@dataclass
		class SupportingLink:
		network_id : str
		link_id : str

		def get_xpath_filter(self) -> str:
		return XPATH_LINK_TEMPLATE.format(self.network_id, self.link_id)


		def discover_supporting_links(restconf_client : RestConfClient, xpath_filter : str) -> List[SupportingLink]:
		xpath_filter_2 = xpath_filter.replace('/simap-telemetry:simap-telemetry', '')
		xpath_filter_2 = xpath_filter.replace('/simap-telemetry', '')
		xpath_data = restconf_client.get(xpath_filter_2)
		if not xpath_data:
		MSG = 'Resource({:s} => {:s}) not found in SIMAP Server'
		raise Exception(MSG.format(str(xpath_filter), str(xpath_filter_2)))

		links = xpath_data.get('ietf-network-topology:link', list())
		if len(links) == 0:
		raise Exception('Link({:s}) not found'.format(str(xpath_filter_2)))
		if len(links) > 1:
		raise Exception('Multiple occurrences for Link({:s})'.format(str(xpath_filter_2)))
		link = links[0]
		supporting_links = link.get('supporting-link', list())
		if len(supporting_links) == 0:
		MSG = 'No supporting links found for Resource({:s}, {:s})'
		raise Exception(MSG.format(str(xpath_filter), str(xpath_data)))

		supporting_link_xpaths : List[SupportingLink] = [
		SupportingLink(sup_link['network-ref'], sup_link['link-ref'])
		for sup_link in supporting_links
		]
		return supporting_link_xpaths


		#def compose_establish_subscription(datastore : str, xpath_filter : str, period : float) -> Dict:
		# return {
		# 'ietf-subscribed-notifications:input': {
		# 'datastore': datastore,
		# 'ietf-yang-push:datastore-xpath-filter': xpath_filter,
		# 'ietf-yang-push:periodic': {
		# 'ietf-yang-push:period': period,
		# }
		# }
		# }


		CONTROLLER_MAP = {
		'e2e' : 'TFS-E2E',
		'agg' : 'TFS-AGG',
		'trans-pkt': 'TFS-IP',
		'trans-opt': 'NCE-T',
		'access' : 'NCE-FAN',
		'admin' : None, # controller-less
		}

		def get_controller_id(network_id : str) -> Optional[str]:
		# TODO: Future improvement: infer controller based on topology data
		if network_id not in CONTROLLER_MAP:
		MSG = 'Unable to identify controller for SimapNetwork({:s})'
		raise Exception(MSG.format(str(network_id)))
		return CONTROLLER_MAP[network_id]


		@dataclass
		class UnderlaySubscriptionId:
		subscription_id : int
		subscription_uri : str

		@classmethod
		def from_reply(cls, sse_sub_rep : SSEMonitoringSubscriptionResponse) -> 'UnderlaySubscriptionId':
		return cls(
		subscription_id = sse_sub_rep.identifier,
		subscription_uri = sse_sub_rep.uri,
		)

		def establish_underlay_subscription(
		device_client : DeviceClient, controller_uuid : str, xpath_filter : str,
		sampling_interval : float
		) -> UnderlaySubscriptionId:
		sse_sub_req = SSEMonitoringSubscriptionConfig()
		sse_sub_req.device_id.device_uuid.uuid = controller_uuid
		sse_sub_req.config_type = SSEMonitoringSubscriptionConfig.Subscribe
		sse_sub_req.uri = xpath_filter
		sse_sub_req.sampling_interval = str(sampling_interval)
		sse_sub_rep = device_client.SSETelemetrySubscribe(sse_sub_req)
		return UnderlaySubscriptionId.from_reply(sse_sub_rep)

		def delete_underlay_subscription(
		device_client : DeviceClient, controller_uuid : str, subscription_id : int
		) -> None:
		sse_unsub_req = SSEMonitoringSubscriptionConfig()
		sse_unsub_req.device_id.device_uuid.uuid = controller_uuid
		sse_unsub_req.config_type = SSEMonitoringSubscriptionConfig.Unsubscribe
		sse_unsub_req.identifier = subscription_id
		device_client.SSETelemetrySubscribe(sse_unsub_req)


		KAFKA_BOOT_SERVERS = KafkaConfig.get_kafka_address()

		def create_kafka_topic(topic : str) -> None:
		try:
		kafka_admin = KafkaAdminClient(bootstrap_servers=KAFKA_BOOT_SERVERS)
		existing_topics = set(kafka_admin.list_topics())
		if topic in existing_topics: return
		to_create = [NewTopic(topic, num_partitions=3, replication_factor=1)]
		kafka_admin.create_topics(to_create, validate_only=False)
		except BrokerResponseError:
		MSG = 'Error creating Topic({:s})'
		LOGGER.exception(MSG.format(str(topic)))

		def delete_kafka_topic(topic : str) -> None:
		try:
		kafka_admin = KafkaAdminClient(bootstrap_servers=KAFKA_BOOT_SERVERS)
		existing_topics = set(kafka_admin.list_topics())
		if topic not in existing_topics: return
		kafka_admin.delete_topics([topic])
		except BrokerResponseError:
		MSG = 'Error deleting Topic({:s})'
		LOGGER.exception(MSG.format(str(topic)))

src/simap_connector/service/main.py

+6 −6

Original line number	Diff line number	Diff line
		@@ -35,7 +35,7 @@ from .SimapConnectorService import SimapConnectorService
		TERMINATE = threading.Event()

		LOG_LEVEL = get_log_level()
		logging.basicConfig(level=LOG_LEVEL, format="[%(asctime)s] %(levelname)s:%(name)s:%(message)s")
		logging.basicConfig(level=LOG_LEVEL, format='[%(asctime)s] %(levelname)s:%(name)s:%(message)s')
		logging.getLogger('RestConfClient').setLevel(logging.WARN)

		LOGGER = logging.getLogger(__name__)
		@@ -75,16 +75,16 @@ def main():

		rebuild_database(db_engine)

		# Starting service
		grpc_service = SimapConnectorService(db_engine)
		grpc_service.start()


		restconf_client = RestConfClient(
		scheme=SIMAP_SERVER_SCHEME, address=SIMAP_SERVER_ADDRESS,
		port=SIMAP_SERVER_PORT, username=SIMAP_SERVER_USERNAME,
		password=SIMAP_SERVER_PASSWORD,
		)

		# Starting service
		grpc_service = SimapConnectorService(db_engine, restconf_client)
		grpc_service.start()

		simap_client = SimapClient(restconf_client)
		telemetry_pool = TelemetryPool(simap_client, terminate=TERMINATE)
		simap_updater = SimapUpdater(simap_client, telemetry_pool, TERMINATE)

src/simap_connector/service/database/SubSubscription.py

0 → 100644

+112 −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 datetime, logging
		from sqlalchemy.dialects.postgresql import insert
		from sqlalchemy.engine import Engine
		from sqlalchemy.orm import Session, sessionmaker
		from sqlalchemy_cockroachdb import run_transaction
		from typing import Dict, List, Optional, Tuple
		from common.method_wrappers.ServiceExceptions import NotFoundException
		from .models.SubSubscriptionModel import SubSubscriptionModel


		LOGGER = logging.getLogger(__name__)


		def sub_subscription_list(db_engine : Engine, parent_subscription_uuid : str) -> List[Dict]:
		def callback(session : Session) -> List[Dict]:
		obj_list : List[SubSubscriptionModel] = (
		session
		.query(SubSubscriptionModel)
		.filter_by(parent=parent_subscription_uuid)
		.all()
		)
		return [obj.dump() for obj in obj_list]
		return run_transaction(sessionmaker(bind=db_engine), callback)


		def sub_subscription_get(
		db_engine : Engine, parent_subscription_uuid : str, sub_subscription_id : int
		) -> Dict:
		def callback(session : Session) -> Optional[Dict]:
		obj : Optional[SubSubscriptionModel] = (
		session
		.query(SubSubscriptionModel)
		.filter_by(parent=parent_subscription_uuid, sub_subscription_id=sub_subscription_id)
		.one_or_none()
		)
		return None if obj is None else obj.dump()
		obj = run_transaction(sessionmaker(bind=db_engine), callback)
		if obj is None:
		sub_sub_key = '{:s}/{:s}'.format(str(parent_subscription_uuid), str(sub_subscription_id))
		raise NotFoundException('SubSubscription', sub_sub_key)
		return obj


		def sub_subscription_set(
		db_engine : Engine, parent_subscription_uuid : str, controller_uuid : str, datastore : str,
		xpath_filter : str, period : float, sub_subscription_id : int, sub_subscription_uri : str
		) -> str:
		now = datetime.datetime.now(datetime.timezone.utc)
		sub_subscription_data = {
		'parent' : parent_subscription_uuid,
		'controller_uuid' : controller_uuid,
		'datastore' : datastore,
		'xpath_filter' : xpath_filter,
		'period' : period,
		'sub_subscription_id' : sub_subscription_id,
		'sub_subscription_uri': sub_subscription_uri,
		'created_at' : now,
		'updated_at' : now,
		}

		def callback(session : Session) -> Tuple[bool, str]:
		stmt = insert(SubSubscriptionModel).values([sub_subscription_data])
		stmt = stmt.on_conflict_do_update(
		index_elements=[SubSubscriptionModel.subscription_uuid],
		set_=dict(
		controller_uuid = stmt.excluded.controller_uuid,
		datastore = stmt.excluded.datastore,
		xpath_filter = stmt.excluded.xpath_filter,
		period = stmt.excluded.period,
		sub_subscription_id = stmt.excluded.sub_subscription_id,
		sub_subscription_uri = stmt.excluded.sub_subscription_uri,
		updated_at = stmt.excluded.updated_at,
		)
		)
		stmt = stmt.returning(
		SubSubscriptionModel.created_at, SubSubscriptionModel.updated_at,
		SubSubscriptionModel.sub_subscription_uuid
		)
		return_values = session.execute(stmt).fetchone()
		created_at,updated_at,subscription_uuid = return_values
		return updated_at > created_at, subscription_uuid

		_, subscription_uuid = run_transaction(sessionmaker(bind=db_engine), callback)
		return subscription_uuid


		def sub_subscription_delete(
		db_engine : Engine, parent_subscription_uuid : str, sub_subscription_id : int
		) -> bool:
		def callback(session : Session) -> bool:
		num_deleted = (
		session
		.query(SubSubscriptionModel)
		.filter_by(parent=parent_subscription_uuid, sub_subscription_id=sub_subscription_id)
		.delete()
		)
		return num_deleted > 0
		return run_transaction(sessionmaker(bind=db_engine), callback)