Commit 46c1cd4e authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Simap Connector: 

- Corrected empty controller_uuid conditions
- Fixed keys in AggregationCache
- Enhanced Collector worker to support underlay controller streams and direct SIMAP polling
parent 006ff10d

src/simap_connector/service/SimapConnectorServiceServicerImpl.py

+23 −16
Original line number Diff line number Diff line
@@ -76,17 +76,25 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer): 
            sup_link_xpath_filter = supporting_link.get_xpath_filter()

            sup_link_xpath_filters.append(sup_link_xpath_filter)



            if controller_id is not None:

            if controller_id is None:

                collector_name = 'SIMAP:{:s}:{:s}'.format(

                    str(supporting_link.network_id), str(supporting_link.link_id)

                )

                target_uri = sup_link_xpath_filter

                underlay_subscription_id = 0

            else:

                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)

                )

                target_uri = underlay_sub_id.subscription_uri

                underlay_subscription_id = 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

                collector_name, controller_id, supporting_link.network_id,

                supporting_link.link_id, target_uri, aggregation_cache, period

            )



            sub_request = Subscription()
@@ -95,8 +103,7 @@ class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer): 
            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

                sup_link_xpath_filter, period, underlay_subscription_id, target_uri

            )



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

src/simap_connector/service/database/SubSubscription.py

+1 −0
Original line number Diff line number Diff line
@@ -60,6 +60,7 @@ def sub_subscription_set( 
    xpath_filter : str, period : float, sub_subscription_id : int, sub_subscription_uri : str

) -> str:

    now = datetime.datetime.now(datetime.timezone.utc)

    if controller_uuid is None: controller_uuid = ''

    sub_subscription_data = {

        'parent'              : parent_subscription_uuid,

        'controller_uuid'     : controller_uuid,

src/simap_connector/service/telemetry/TelemetryPool.py

+4 −4
Original line number Diff line number Diff line
@@ -65,12 +65,12 @@ class TelemetryPool: 




    def start_collector(

        self, worker_name : str, subscription_id : int, controller_uuid : Optional[str],

        subscription_uri : str, aggregation_cache : AggregationCache, sampling_interval : float

        self, worker_name : str, controller_uuid : Optional[str], network_id : str, link_id : str,

        target_uri : str, aggregation_cache : AggregationCache, sampling_interval : float

    ) -> None:

        self._start_worker(

            WorkerTypeEnum.COLLECTOR, worker_name, subscription_id, controller_uuid,

            subscription_uri, aggregation_cache, sampling_interval

            WorkerTypeEnum.COLLECTOR, worker_name, controller_uuid, network_id, link_id,

            target_uri, aggregation_cache, sampling_interval

        )

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

+121 −72
Original line number Diff line number Diff line
@@ -13,7 +13,7 @@ 
# limitations under the License.





import json, requests, threading

import json, math, requests, threading, time

from requests.exceptions import ReadTimeout

from typing import Optional

from .data.AggregationCache import AggregationCache, LinkSample
@@ -29,17 +29,21 @@ CONTROLLER_TO_ADDRESS_PORT = { 
    'SIMAP'   : ('10.254.0.9',  80),

}



WAIT_LOOP_GRANULARITY = 0.5





class CollectorWorker(_Worker):

    def __init__(

        self, worker_name : str, subscription_id : int, controller_uuid : Optional[str],

        subscription_uri : str, aggregation_cache : AggregationCache, sampling_interval : float,

        self, worker_name : str, controller_uuid : Optional[str],

        network_id : str, link_id : str, target_uri : str,

        aggregation_cache : AggregationCache, sampling_interval : float,

        terminate : Optional[threading.Event] = None

    ) -> None:

        super().__init__(WorkerTypeEnum.COLLECTOR, worker_name, terminate=terminate)

        self._subscription_id = subscription_id

        self._controller_uuid = controller_uuid

        self._subscription_uri = subscription_uri

        self._network_id = network_id

        self._link_id = link_id

        self._target_uri = target_uri

        self._aggregation_cache = aggregation_cache

        self._sampling_interval = sampling_interval
@@ -48,10 +52,21 @@ class CollectorWorker(_Worker): 


        try:

            address_port = CONTROLLER_TO_ADDRESS_PORT.get(self._controller_uuid)

            if address_port is None: address_port = CONTROLLER_TO_ADDRESS_PORT['SIMAP']

            if address_port is None:

                address, port = CONTROLLER_TO_ADDRESS_PORT['SIMAP']

                self.direct_simap_polling(address, port)

            else:

                address, port = address_port

            stream_url = 'http://{:s}:{:d}{:s}'.format(address, port, self._subscription_uri)

            self._logger.info('[run] Opening stream "{:s}"...'.format(str(stream_url)))

                self.underlay_subscription_stream(address, port)

        except Exception:

            self._logger.exception('[run] Unhandled Exception')

        finally:

            self._logger.info('[run] Terminated')



    def underlay_subscription_stream(self, address : str, port : int) -> None:

        stream_url = 'http://{:s}:{:d}{:s}'.format(address, port, self._target_uri)

        MSG = '[underlay_subscription_stream] Opening stream "{:s}"...'

        self._logger.info(MSG.format(str(stream_url)))



        session = requests.Session()

        try:
@@ -74,7 +89,8 @@ class CollectorWorker(_Worker): 
                    if line is None: continue

                    if len(line) == 0: continue



                        self._logger.info('[run] ==> {:s}'.format(str(line)))

                    MSG = '[underlay_subscription_stream] ==> {:s}'

                    self._logger.info(MSG.format(str(line)))

                    if not line.startswith('data:'): continue

                    data = json.loads(line[5:])
@@ -106,7 +122,8 @@ class CollectorWorker(_Worker): 
                    related_service_ids   = simap_telemetry['related-service-ids']



                    link_sample = LinkSample(

                            subscription_id       = self._subscription_id,

                        network_id            = self._network_id,

                        link_id               = self._link_id,

                        bandwidth_utilization = bandwidth_utilization,

                        latency               = latency,

                        related_service_ids   = related_service_ids,
@@ -115,7 +132,39 @@ class CollectorWorker(_Worker): 
        finally:

            if session is not None:

                session.close()

        except Exception:

            self._logger.exception('[run] Unhandled Exception')

        finally:

            self._logger.info('[run] Terminated')
 


    def direct_simap_polling(self, address : str, port : int) -> None:

        simap_url = 'http://{:s}:{:d}{:s}'.format(address, port, self._target_uri)



        while not self._stop_event.is_set() and not self._terminate.is_set():

            MSG = '[direct_simap_polling] Requesting "{:s}"...'

            self._logger.info(MSG.format(str(simap_url)))



            with requests.get(simap_url, timeout=10) as reply:

                reply.raise_for_status()

                data = reply.json()



            if 'simap-telemetry:simap-telemetry' not in data:

                MSG = 'Field(simap-telemetry:simap-telemetry) missing: {:s}'

                raise Exception(MSG.format(str(data)))

            simap_telemetry = data['simap-telemetry:simap-telemetry']



            bandwidth_utilization = float(simap_telemetry['bandwidth-utilization'])

            latency               = float(simap_telemetry['latency'])

            related_service_ids   = simap_telemetry.get('related-service-ids', list())



            link_sample = LinkSample(

                network_id            = self._network_id,

                link_id               = self._link_id,

                bandwidth_utilization = bandwidth_utilization,

                latency               = latency,

                related_service_ids   = related_service_ids,

            )

            self._aggregation_cache.update(link_sample)



            # Make wait responsible to terminations

            iterations = int(math.ceil(self._sampling_interval / WAIT_LOOP_GRANULARITY))

            for _ in range(iterations):

                if self._stop_event.is_set(): break

                if self._terminate.is_set() : break

                time.sleep(WAIT_LOOP_GRANULARITY)

src/simap_connector/service/telemetry/worker/data/AggregationCache.py

+3 −2
Original line number Diff line number Diff line
@@ -21,7 +21,8 @@ from typing import Dict, Set, Tuple 


@dataclass

class LinkSample:

    subscription_id       : int

    network_id            : str

    link_id               : str

    bandwidth_utilization : float

    latency               : float

    related_service_ids   : Set[str] = field(default_factory=set)
@@ -42,7 +43,7 @@ class AggregationCache: 




    def update(self, link_sample : LinkSample) -> None:

        link_key = (link_sample.domain_name, link_sample.link_name)

        link_key = (link_sample.network_id, link_sample.link_id)

        with self._lock:

            self._samples[link_key] = link_sample