Commit c1ff4a57 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

SIMAP Connector: 

- Implemented TelemetryPool skeleton
parent cb25c717

src/simap_connector/service/__main__.py

+4 −1
Original line number Diff line number Diff line
@@ -20,6 +20,7 @@ from common.Settings import ( 
    get_log_level, get_metrics_port, wait_for_environment_variables

)

from .simap_updater.SimapUpdater import SimapUpdater

from .telemetry.TelemetryPool import TelemetryPool

from .SimapConnectorService import SimapConnectorService



TERMINATE = threading.Event()
@@ -55,7 +56,8 @@ def main(): 
    grpc_service = SimapConnectorService()

    grpc_service.start()



    simap_updater = SimapUpdater(TERMINATE)

    telemetry_pool = TelemetryPool(terminate=TERMINATE)

    simap_updater = SimapUpdater(TERMINATE, telemetry_pool)

    simap_updater.start()



    LOGGER.info('Running...')
@@ -64,6 +66,7 @@ def main(): 


    LOGGER.info('Terminating...')

    simap_updater.stop()

    telemetry_pool.stop_all()

    grpc_service.stop()



    LOGGER.info('Bye')

src/simap_connector/service/simap_updater/SimapUpdater.py

+10 −2
Original line number Diff line number Diff line
@@ -30,6 +30,7 @@ from simap_connector.Config import ( 
    SIMAP_SERVER_USERNAME, SIMAP_SERVER_PASSWORD,

)

from simap_connector.service.simap_updater.MockSimaps import delete_mock_simap, set_mock_simap

from simap_connector.service.telemetry.TelemetryPool import TelemetryPool

from .SimapClient import SimapClient

from .ObjectCache import CachedEntities, ObjectCache

from .Tools import get_device_endpoint, get_link_endpoint, get_service_endpoint
@@ -51,10 +52,12 @@ class EventDispatcher(BaseEventDispatcher): 
    def __init__(

        self, events_queue : queue.PriorityQueue,

        context_client : ContextClient,

        telemetry_pool : TelemetryPool,

        terminate : Optional[threading.Event] = None

    ) -> None:

        super().__init__(events_queue, terminate)

        self._context_client = context_client

        self._telemetry_pool = telemetry_pool

        self._object_cache = ObjectCache(self._context_client)

        self._restconf_client = RestConfClient(

            scheme=SIMAP_SERVER_SCHEME, address=SIMAP_SERVER_ADDRESS,
@@ -507,6 +510,8 @@ class EventDispatcher(BaseEventDispatcher): 
        #)

        #dom_link = domain_topo.link(link_name)

        #dom_link.update(src_dev_name, src_ep_name, dst_dev_name, dst_ep_name)



        self._telemetry_pool.start_worker(domain_name)

        return True
@@ -602,12 +607,15 @@ class EventDispatcher(BaseEventDispatcher): 
        #self._object_cache.delete(CachedEntities.SERVICE, service_uuid)

        #self._object_cache.delete(CachedEntities.SERVICE, service_name)



        self._telemetry_pool.stop_worker(domain_name)



        MSG = 'Logical Link Removed for Service: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(service_event)))





class SimapUpdater:

    def __init__(self, terminate : threading.Event) -> None:

    def __init__(self, terminate : threading.Event, telemetry_pool : TelemetryPool) -> None:

        self._telemetry_pool = telemetry_pool

        self._context_client = ContextClient()



        self._event_collector = BaseEventCollector(terminate=terminate)
@@ -617,7 +625,7 @@ class SimapUpdater: 


        self._event_dispatcher = EventDispatcher(

            self._event_collector.get_events_queue(), self._context_client,

            terminate=terminate

            self._telemetry_pool, terminate=terminate

        )



    def start(self) -> None:

src/simap_connector/service/telemetry/TelemetryPool.py

0 → 100644
+74 −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, threading

from typing import Dict, Optional

from .TelemetryWorker import TelemetryWorker





LOGGER = logging.getLogger(__name__)





class TelemetryPool:

    def __init__(

        self, terminate : Optional[threading.Event] = None

    ) -> None:

        self._workers : Dict[str, TelemetryWorker] = dict()

        self._lock = threading.Lock()

        self._terminate = threading.Event() if terminate is None else terminate





    def start_worker(self, domain_name : str) -> None:

        with self._lock:

            if domain_name in self._workers:

                MSG = '[start_worker] Worker already running for Domain({:s})'

                LOGGER.debug(MSG.format(str(domain_name)))

                return



            worker = TelemetryWorker(domain_name, terminate=self._terminate)

            self._workers[domain_name] = worker

            worker.start()



            MSG = '[start_worker] Started worker for Domain({:s})'

            LOGGER.info(MSG.format(str(domain_name)))





    def stop_worker(self, domain_name : str) -> None:

        with self._lock:

            worker = self._workers.pop(domain_name, None)



        if worker is None:

            MSG = '[stop_worker] No worker found for Domain({:s})'

            LOGGER.debug(MSG.format(str(domain_name)))

            return



        worker.stop()



        MSG = '[stop_worker] Stopped worker for Domain({:s})'

        LOGGER.info(MSG.format(str(domain_name)))





    def stop_all(self) -> None:

        LOGGER.info('[stop_all] Stopping all worker')



        with self._lock:

            names = list(self._workers.keys())



        for name in names:

            try:

                self.stop_worker(name)

            except Exception:

                MSG = '[stop_all] Unhandled Exception stopping Worker({:s})'

                LOGGER.exception(MSG.format(str(name)))

src/simap_connector/service/telemetry/TelemetryWorker.py

0 → 100644
+59 −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, threading, time

from typing import Optional





LOGGER = logging.getLogger(__name__)





class TelemetryWorker(threading.Thread):

    def __init__(

        self, domain_name : str, terminate : Optional[threading.Event] = None

    ) -> None:

        name = 'TelemetryWorker-{:s}'.format(str(domain_name))

        super().__init__(name=name, daemon=True)

        self.domain_name = domain_name

        self._stop_event = threading.Event()

        self._terminate = threading.Event() if terminate is None else terminate



    def stop(self) -> None:

        MSG = '[stop][{:s}] Stopping...'

        LOGGER.info(MSG.format(str(self.domain_name)))

        self._stop_event.set()

        self.join()



    def run(self) -> None:

        MSG = '[run][{:s}] Starting...'

        LOGGER.info(MSG.format(str(self.domain_name)))



        try:

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



                MSG = '[run][{:s}] Heartbeat'

                LOGGER.info(MSG.format(str(self.domain_name)))



                for _ in range(10):

                    if self._stop_event.is_set(): break

                    if self._terminate.is_set() : break

                    time.sleep(0.1)



        except Exception:

            MSG = '[run][{:s}] Unhandled Exception'

            LOGGER.info(MSG.format(str(self.domain_name)))

        finally:

            MSG = '[run][{:s}] Terminated'

            LOGGER.info(MSG.format(str(self.domain_name)))

src/simap_connector/service/telemetry/__init__.py

0 → 100644
+13 −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.