Simap Connector: (02517bc4) · Commits · TFS / controller

src/simap_connector/service/SimapConnectorServiceServicerImpl.py

+16 −8

Original line number	Diff line number	Diff line
		@@ -29,8 +29,8 @@ from .database.SubSubscription import (
		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,
		LinkDetails, create_kafka_topic, delete_kafka_topic, delete_underlay_subscription,
		discover_link_details, establish_underlay_subscription, get_controller_id
		)

		LOGGER = logging.getLogger(__name__)
		@@ -60,9 +60,10 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):
		datastore = request.datastore
		xpath_filter = request.xpath_filter
		period = request.period
		supporting_links : List[SupportingLink] = discover_supporting_links(
		link_details : LinkDetails = discover_link_details(
		self._restconf_client, xpath_filter
		)
		xpath_filter = link_details.link.get_xpath_filter()

		parent_subscription_uuid, parent_subscription_id = subscription_set(
		self._db_engine, datastore, xpath_filter, period
		@@ -72,14 +73,15 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):

		device_client = DeviceClient()
		sup_link_xpath_filters : List[str] = list()
		for supporting_link in supporting_links:
		for supporting_link in link_details.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 None:
		collector_name = '{:d}:SIMAP:{:s}:{:s}'.format(
		parent_subscription_id, str(supporting_link.network_id), str(supporting_link.link_id)
		parent_subscription_id, str(supporting_link.network_id),
		str(supporting_link.link_id)
		)
		target_uri = sup_link_xpath_filter
		underlay_subscription_id = 0
		@@ -88,7 +90,8 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):
		device_client, controller_id, sup_link_xpath_filter, period
		)
		collector_name = '{:d}:{:s}:{:s}'.format(
		parent_subscription_id, controller_id, str(underlay_sub_id.subscription_id)
		parent_subscription_id, controller_id,
		str(underlay_sub_id.subscription_id)
		)
		target_uri = underlay_sub_id.subscription_uri
		underlay_subscription_id = underlay_sub_id.subscription_id
		@@ -112,15 +115,20 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):
		create_kafka_topic(topic)

		aggregator_name = str(parent_subscription_id)
		network_id = link_details.link.network_id
		link_id = link_details.link.link_id
		self._telemetry_pool.start_aggregator(
		aggregator_name, parent_subscription_id, aggregation_cache, topic, period
		aggregator_name, network_id, link_id, 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:
		def DeleteSubscription(
		self, request : SubscriptionId, context : grpc.ServicerContext
		) -> Empty:
		parent_subscription_id = request.subscription_id
		subscription = subscription_get(self._db_engine, parent_subscription_id)
		if subscription is None: return Empty()

src/simap_connector/service/Tools.py

+35 −16

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


		import logging
		from dataclasses import dataclass
		import logging, re
		from dataclasses import dataclass, field
		from kafka.admin import KafkaAdminClient, NewTopic
		from kafka.errors import BrokerResponseError
		from typing import List, Optional
		@@ -34,8 +34,13 @@ XPATH_LINK_TEMPLATE = (
		'/ietf-network-topology:link={:s}/simap-telemetry:simap-telemetry'
		)

		RE_XPATH_LINK = re.compile(
		r'^/ietf-network:networks/network=([^/]+)/ietf-network-topology:link=([^/]+)/?.*$'
		)


		@dataclass
		class SupportingLink:
		class Link:
		network_id : str
		link_id : str

		@@ -43,30 +48,44 @@ class SupportingLink:
		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_2.replace('/simap-telemetry', '')
		xpath_data = restconf_client.get(xpath_filter_2)
		@dataclass
		class LinkDetails:
		link : Link
		supporting_links : List[Link] = field(default_factory=list)


		def discover_link_details(restconf_client : RestConfClient, xpath_filter : str) -> LinkDetails:
		link_xpath_match = RE_XPATH_LINK.match(xpath_filter)
		if link_xpath_match is None:
		raise Exception('Unsupported xpath_filter({:s})'.format(str(xpath_filter)))

		network_id, link_id = link_xpath_match.groups()
		link_details = LinkDetails(Link(network_id, link_id))

		xpath_filter = link_details.link.get_xpath_filter()
		xpath_data = restconf_client.get(xpath_filter)
		if not xpath_data:
		MSG = 'Resource({:s} => {:s}) not found in SIMAP Server'
		raise Exception(MSG.format(str(xpath_filter), str(xpath_filter_2)))
		raise Exception('Resource({:s}) not found in SIMAP Server'.format(str(xpath_filter)))

		links = xpath_data.get('ietf-network-topology:link', list())
		if len(links) == 0:
		raise Exception('Link({:s}) not found'.format(str(xpath_filter_2)))
		raise Exception('Link({:s}) not found'.format(str(xpath_filter)))
		if len(links) > 1:
		raise Exception('Multiple occurrences for Link({:s})'.format(str(xpath_filter_2)))
		raise Exception('Multiple occurrences for Link({:s})'.format(str(xpath_filter)))
		link = links[0]
		if link['link-id'] != link_id:
		MSG = 'Retieved Link({:s}) does not match xpath_filter({:s})'
		raise Exception(MSG.format(str(link), str(xpath_filter)))
		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
		for sup_link in supporting_links:
		link_details.supporting_links.append(Link(
		sup_link['network-ref'], sup_link['link-ref']
		))
		return link_details


		#def compose_establish_subscription(datastore : str, xpath_filter : str, period : float) -> Dict:

src/simap_connector/service/telemetry/TelemetryPool.py

+4 −4

Original line number	Diff line number	Diff line
		@@ -55,12 +55,12 @@ class TelemetryPool:


		def start_aggregator(
		self, worker_name : str, parent_subscription_id : int, aggregation_cache : AggregationCache,
		topic : str, sampling_interval : float
		self, worker_name : str, network_id : str, link_id : str, parent_subscription_id : int,
		aggregation_cache : AggregationCache, topic : str, sampling_interval : float
		) -> None:
		self._start_worker(
		WorkerTypeEnum.AGGREGATOR, worker_name, parent_subscription_id, aggregation_cache,
		topic, sampling_interval
		WorkerTypeEnum.AGGREGATOR, worker_name, self._simap_client, network_id, link_id,
		parent_subscription_id, aggregation_cache, topic, sampling_interval
		)

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

+13 −2

Original line number	Diff line number	Diff line
		@@ -18,6 +18,7 @@ from dataclasses import dataclass
		from kafka import KafkaProducer
		from typing import Dict, Optional, Union
		from common.tools.kafka.Variables import KafkaConfig
		from simap_connector.service.simap_updater.SimapClient import SimapClient
		from .data.AggregationCache import AggregationCache
		from ._Worker import _Worker, WorkerTypeEnum

		@@ -55,10 +56,14 @@ class ServerSentEvent:

		class AggregatorWorker(_Worker):
		def __init__(
		self, worker_name : str, parent_subscription_id : int, aggregation_cache : AggregationCache,
		topic : str, sampling_interval : float, terminate : Optional[threading.Event] = None
		self, worker_name : str, simap_client : SimapClient, network_id : str, link_id : str,
		parent_subscription_id : int, aggregation_cache : AggregationCache, topic : str,
		sampling_interval : float, terminate : Optional[threading.Event] = None
		) -> None:
		super().__init__(WorkerTypeEnum.AGGREGATOR, worker_name, terminate=terminate)
		self._simap_client = simap_client
		self._network_id = network_id
		self._link_id = link_id
		self._parent_subscription_id = parent_subscription_id
		self._aggregation_cache = aggregation_cache
		self._topic = topic
		@@ -100,6 +105,12 @@ class AggregatorWorker(_Worker):
		)
		kafka_producer.flush()

		simap_link = self._simap_client.network(self._network_id).link(self._link_id)
		simap_link.telemetry.update(
		link_sample.bandwidth_utilization, link_sample.latency,
		related_service_ids=list(link_sample.related_service_ids)
		)

		update_counter += 1

		# Make wait responsible to terminations