Commit 8617abfb authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

SIMAP Connector: 

- Added show logs script
- Corrected manifest file
- Added requirements.in
- Added Update methods in SimapClient
- Added ENDPOINT to ObjectCache
- Improved object key and instance handining in ObjectCache
- Implemented Create/Update/Remove events for Topology/Device/Link in SimapUpdater
parent e52cacb5

manifests/simap_connectorservice.yaml

+8 −2
Original line number Diff line number Diff line
@@ -40,9 +40,15 @@ spec: 
            - name: SIMAP_SERVER_SCHEME

              value: "http"

            - name: SIMAP_SERVER_ADDRESS

              value: "10.254.0.9"

              # Assuming SIMAP Server is deployed in a local Docker container, as per:

              # - ./src/tests/tools/simap_server/build.sh

              # - ./src/tests/tools/simap_server/deploy.sh

              value: "172.17.0.1"

            - name: SIMAP_SERVER_PORT

              value: "80"

              # Assuming SIMAP Server is deployed in a local Docker container, as per:

              # - ./src/tests/tools/simap_server/build.sh

              # - ./src/tests/tools/simap_server/deploy.sh

              value: "8080"

            - name: SIMAP_SERVER_USERNAME

              value: "admin"

            - name: SIMAP_SERVER_PASSWORD

scripts/show_logs_simap_connector.sh

0 → 100755
+27 −0
Original line number Diff line number Diff line 
#!/bin/bash

# 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.



########################################################################################################################

# Define your deployment settings here

########################################################################################################################



# If not already set, set the name of the Kubernetes namespace to deploy to.

export TFS_K8S_NAMESPACE=${TFS_K8S_NAMESPACE:-"tfs"}



########################################################################################################################

# Automated steps start here

########################################################################################################################



kubectl --namespace $TFS_K8S_NAMESPACE logs deployment/simap-connectorservice -c server

src/simap_connector/requirements.in

+1 −0
Original line number Diff line number Diff line
@@ -12,3 +12,4 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



requests==2.27.*

src/simap_connector/service/simap_updater/ObjectCache.py

+76 −23
Original line number Diff line number Diff line
@@ -28,47 +28,100 @@ LOGGER = logging.getLogger(__name__) 
class CachedEntities(Enum):

    TOPOLOGY = 'topology'

    DEVICE   = 'device'

    ENDPOINT = 'endpoint'

    LINK     = 'link'





KEY_LENGTHS = {

    CachedEntities.TOPOLOGY : 1,

    CachedEntities.DEVICE   : 1,

    CachedEntities.ENDPOINT : 2,

    CachedEntities.LINK     : 1,

}





def compose_object_key(entity : CachedEntities, *object_uuids : str) -> Tuple[str, ...]:

    expected_length = KEY_LENGTHS.get(entity)

    entity_name = str(entity.value)

    if expected_length is None:

        MSG = 'Unsupported ({:s}, {:s})'

        raise Exception(MSG.format(entity_name.title(), str(object_uuids)))



    if len(object_uuids) == expected_length:

        return (entity_name, *object_uuids)



    MSG = 'Invalid Key ({:s}, {:s})'

    raise Exception(MSG.format(entity_name.title(), str(object_uuids)))





class ObjectCache:

    def __init__(self, context_client : ContextClient):

        self._context_client = context_client

        self._object_cache : Dict[Tuple[str, str], Any] = dict()



    def get(self, entity : CachedEntities, object_uuid : str) -> Optional[Any]:

        object_key = (entity.value, object_uuid)

    def get(

        self, entity : CachedEntities, *object_uuids : str, auto_retrieve : bool = True

    ) -> Optional[Any]:

        object_key = compose_object_key(entity, *object_uuids)

        if object_key in self._object_cache:

            return self._object_cache[object_key]

        return self._update(entity, object_uuid)



    def _retrieve(

        self, entity : CachedEntities, entity_uuid : str

    ) -> Optional[Any]:

        if not auto_retrieve: return None

        return self._update(entity, *object_uuids)



    def set(self, entity : CachedEntities, object_inst : Any, *object_uuids : str) -> None:

        object_key = compose_object_key(entity, *object_uuids)

        self._object_cache[object_key] = object_inst



    def _update(self, entity : CachedEntities, *object_uuids : str) -> Optional[Any]:

        if entity == CachedEntities.TOPOLOGY:

            return get_topology(self._context_client, entity_uuid, rw_copy=False)

        if entity == CachedEntities.DEVICE:

            return get_device(

                self._context_client, entity_uuid, rw_copy=False, include_endpoints=True,

            object_inst = get_topology(

                self._context_client, object_uuids[0], rw_copy=False

            )

        elif entity == CachedEntities.DEVICE:

            object_inst = get_device(

                self._context_client, object_uuids[0], rw_copy=False, include_endpoints=True,

                include_components=False, include_config_rules=False,

            )

        if entity == CachedEntities.LINK:

            return get_link(self._context_client, entity_uuid, rw_copy=False)

        elif entity == CachedEntities.ENDPOINT:

            # Endpoints are only updated when updating a Device

            return None

        elif entity == CachedEntities.LINK:

            object_inst = get_link(

                self._context_client, object_uuids[0], rw_copy=False

            )

        else:

            MSG = 'Not Supported ({:s}, {:s})'

        LOGGER.warning(MSG.format(str(entity.value).title(), str(entity_uuid)))

            LOGGER.warning(MSG.format(str(entity.value).title(), str(object_uuids)))

            return None



    def _update(self, entity : CachedEntities, object_uuid : str) -> Optional[Any]:

        object_inst = self._retrieve(entity, object_uuid)

        if object_inst is None:

            MSG = 'Not Found ({:s}, {:s})'

            LOGGER.warning(MSG.format(str(entity).title(), str(object_uuid)))

            LOGGER.warning(MSG.format(str(entity.value).title(), str(object_uuids)))

            return None



        object_key = (entity.value, object_uuid)

        self._object_cache[object_key] = object_inst

        self.set(entity, object_inst, object_uuids[0])

        self.set(entity, object_inst, object_inst.name)



        if entity == CachedEntities.DEVICE:

            device_uuid = object_inst.device_id.device_uuid.uuid

            device_name = object_inst.name



            for endpoint in object_inst.device_endpoints:

                endpoint_device_uuid = endpoint.endpoint_id.device_id.device_uuid.uuid

                if device_uuid != endpoint_device_uuid:

                    MSG = 'DeviceUUID({:s}) != Endpoint.DeviceUUID({:s})'

                    raise Exception(str(device_uuid), str(endpoint_device_uuid))



                endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid

                endpoint_name = endpoint.name

                self.set(CachedEntities.ENDPOINT, endpoint, device_uuid, endpoint_uuid)

                self.set(CachedEntities.ENDPOINT, endpoint, device_uuid, endpoint_name)

                self.set(CachedEntities.ENDPOINT, endpoint, device_name, endpoint_uuid)

                self.set(CachedEntities.ENDPOINT, endpoint, device_name, endpoint_name)



        return object_inst



    def delete(self, entity : CachedEntities, object_uuid : str) -> None:

        object_key = (entity.value, object_uuid)

    def delete(self, entity : CachedEntities, *object_uuids : str) -> None:

        object_key = compose_object_key(entity, *object_uuids)

        self._object_cache.pop(object_key, None)

src/simap_connector/service/simap_updater/SimapUpdater.py

+163 −45
Original line number Diff line number Diff line
@@ -15,24 +15,19 @@ 


import logging, queue, threading

from typing import Any, Optional

from common.proto.context_pb2 import DeviceEvent, Empty, TopologyEvent

from common.tools.context_queries.Device import get_device

from common.tools.context_queries.Link import get_link



from common.proto.context_pb2 import DeviceEvent, Empty, LinkEvent, TopologyEvent

from common.tools.grpc.BaseEventCollector import BaseEventCollector

from common.tools.grpc.BaseEventDispatcher import BaseEventDispatcher

from common.tools.grpc.Tools import grpc_message_to_json_string

from common.DeviceTypes import DeviceTypeEnum

from context.client.ContextClient import ContextClient

from simap_connector.service.simap_updater.ObjectCache import CachedEntities

from .simap_client.RestConfClient import RestConfClient

from .simap_client.SimapClient import SimapClient

from .ObjectCache import ObjectCache



from simap_connector.Config import (

    SIMAP_SERVER_SCHEME, SIMAP_SERVER_ADDRESS, SIMAP_SERVER_PORT,

    SIMAP_SERVER_USERNAME, SIMAP_SERVER_PASSWORD,

)

from .simap_client.RestConfClient import RestConfClient

from .simap_client.SimapClient import SimapClient

from .ObjectCache import CachedEntities, ObjectCache

from .Tools import get_device_endpoint, get_link_endpoint





LOGGER = logging.getLogger(__name__)
@@ -55,6 +50,12 @@ class EventDispatcher(BaseEventDispatcher): 
        )

        self._simap_client = SimapClient(self._restconf_client)





    def dispatch(self, event : Any) -> None:

        MSG = 'Unexpected Event: {:s}'

        LOGGER.warning(MSG.format(grpc_message_to_json_string(event)))





    def dispatch_topology_create(self, topology_event : TopologyEvent) -> None:

        MSG = 'Processing Topology Event: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -67,6 +68,7 @@ class EventDispatcher(BaseEventDispatcher): 
        MSG = 'Topology Created: {:s}'

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





    def dispatch_topology_update(self, topology_event : TopologyEvent) -> None:

        MSG = 'Processing Topology Event: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -74,11 +76,12 @@ class EventDispatcher(BaseEventDispatcher): 
        topology_uuid = topology_event.topology_id.topology_uuid.uuid

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name

        self._simap_client.network(topology_name).create()

        self._simap_client.network(topology_name).update()



        MSG = 'Topology Updated: {:s}'

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





    def dispatch_topology_remove(self, topology_event : TopologyEvent) -> None:

        MSG = 'Processing Topology Event: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -89,52 +92,167 @@ class EventDispatcher(BaseEventDispatcher): 
        self._simap_client.network(topology_name).delete()



        self._object_cache.delete(CachedEntities.TOPOLOGY, topology_uuid)

        self._object_cache.delete(CachedEntities.TOPOLOGY, topology_name)



        MSG = 'Topology Remove: {:s}'

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





    def dispatch_device_create(self, device_event : DeviceEvent) -> None:

        MSG = 'Processing Device Event: {:s}'

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



        device_uuid = device_event.device_id.device_uuid.uuid

        device = self._object_cache.get(CachedEntities.DEVICE, device_uuid)

        device_name = device.name



        topology_uuid, endpoint_names = get_device_endpoint(device)

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        te_topo.node(device_name).create(termination_point_ids=endpoint_names)



        MSG = 'Device Created: {:s}'

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



    #def dispatch_device_create(self, device_event : DeviceEvent) -> None:

    #    MSG = 'Processing Device Create: {:s}'

    #    LOGGER.info(MSG.format(grpc_message_to_json_string(device_event)))

        topology_uuid = topology_event.topology_id.topology_uuid.uuid

        topology = get_topology(

            self._context_client, topology_uuid, rw_copy=False,

            include_endpoints=False, include_config_rules=True,

            include_components=False

        )

        device_type = device.device_type



    #    tfs_ctrl_settings = get_tfs_controller_settings(

    #        self._context_client, device_event

    #    )

    #    if tfs_ctrl_settings is None: return

    #    self._subscriptions.add_subscription(tfs_ctrl_settings)



    #def dispatch_device_update(self, device_event : DeviceEvent) -> None:

    #    MSG = 'Processing Device Update: {:s}'

    #    LOGGER.info(MSG.format(grpc_message_to_json_string(device_event)))

    #    tfs_ctrl_settings = get_tfs_controller_settings(

    #        self._context_client, device_event

    #    )

    #    if tfs_ctrl_settings is None: return

    #    self._subscriptions.add_subscription(tfs_ctrl_settings)



    #def dispatch_device_remove(self, device_event : DeviceEvent) -> None:

    #    MSG = 'Processing Device Remove: {:s}'

    #    LOGGER.info(MSG.format(grpc_message_to_json_string(device_event)))

    #    device_uuid = device_event.device_id.device_uuid.uuid

    #    self._subscriptions.remove_subscription(device_uuid)



    def dispatch(self, event : Any) -> None:

        MSG = 'Unexpected Event: {:s}'

        LOGGER.warning(MSG.format(grpc_message_to_json_string(event)))

    def dispatch_device_update(self, device_event : DeviceEvent) -> None:

        MSG = 'Processing Device Event: {:s}'

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



        device_uuid = device_event.device_id.device_uuid.uuid

        device = self._object_cache.get(CachedEntities.DEVICE, device_uuid)

        device_name = device.name



        topology_uuid, endpoint_names = get_device_endpoint(device)

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        te_device = te_topo.node(device_name)

        te_device.update()



        for endpoint_name in endpoint_names:

            te_device.termination_point(endpoint_name).update()



        MSG = 'Device Updated: {:s}'

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





    def dispatch_device_remove(self, device_event : DeviceEvent) -> None:

        MSG = 'Processing Device Event: {:s}'

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



        device_uuid = device_event.device_id.device_uuid.uuid

        device = self._object_cache.get(CachedEntities.DEVICE, device_uuid)

        device_name = device.name



        topology_uuid, endpoint_names = get_device_endpoint(device)

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        te_device = te_topo.node(device_name)

        for endpoint_name in endpoint_names:

            te_device.termination_point(endpoint_name).delete()



            endpoint = self._object_cache.get(CachedEntities.ENDPOINT, device_uuid, endpoint_name)

            endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid

            self._object_cache.delete(CachedEntities.DEVICE, device_uuid, endpoint_uuid)

            self._object_cache.delete(CachedEntities.DEVICE, device_uuid, endpoint_name)

            self._object_cache.delete(CachedEntities.DEVICE, device_name, endpoint_uuid)

            self._object_cache.delete(CachedEntities.DEVICE, device_name, endpoint_name)



        te_device.delete()



        self._object_cache.delete(CachedEntities.DEVICE, device_uuid)

        self._object_cache.delete(CachedEntities.DEVICE, device_name)



        MSG = 'Device Remove: {:s}'

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





    def dispatch_link_create(self, link_event : LinkEvent) -> None:

        MSG = 'Processing Link Event: {:s}'

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



        link_uuid = link_event.link_id.link_uuid.uuid

        link = self._object_cache.get(CachedEntities.LINK, link_uuid)

        link_name = link.name



        topology_uuid, endpoint_uuids = get_link_endpoint(link)

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        src_device_name   = self._object_cache.get(CachedEntities.DEVICE,   endpoint_uuids[0][0], auto_retrieve=False)

        src_endpoint_name = self._object_cache.get(CachedEntities.ENDPOINT, *(endpoint_uuids[0]), auto_retrieve=False)

        dst_device_name   = self._object_cache.get(CachedEntities.DEVICE,   endpoint_uuids[1][0], auto_retrieve=False)

        dst_endpoint_name = self._object_cache.get(CachedEntities.ENDPOINT, *(endpoint_uuids[1]), auto_retrieve=False)



        te_topo.link(link_name).create(src_device_name, src_endpoint_name, dst_device_name, dst_endpoint_name)



        MSG = 'Link Created: {:s}'

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



    def dispatch_link_update(self, link_event : LinkEvent) -> None:

        MSG = 'Processing Link Event: {:s}'

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



        link_uuid = link_event.link_id.link_uuid.uuid

        link = self._object_cache.get(CachedEntities.LINK, link_uuid)

        link_name = link.name



        topology_uuid, endpoint_uuids = get_link_endpoint(link)

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        src_device_name   = self._object_cache.get(CachedEntities.DEVICE,   endpoint_uuids[0][0], auto_retrieve=False)

        src_endpoint_name = self._object_cache.get(CachedEntities.ENDPOINT, *(endpoint_uuids[0]), auto_retrieve=False)

        dst_device_name   = self._object_cache.get(CachedEntities.DEVICE,   endpoint_uuids[1][0], auto_retrieve=False)

        dst_endpoint_name = self._object_cache.get(CachedEntities.ENDPOINT, *(endpoint_uuids[1]), auto_retrieve=False)



        te_link = te_topo.link(link_name)

        te_link.update(src_device_name, src_endpoint_name, dst_device_name, dst_endpoint_name)



        MSG = 'Link Updated: {:s}'

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



    def dispatch_link_remove(self, link_event : LinkEvent) -> None:

        MSG = 'Processing Link Event: {:s}'

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



        link_uuid = link_event.link_id.link_uuid.uuid

        link = self._object_cache.get(CachedEntities.LINK, link_uuid)

        link_name = link.name



        topology_uuid, _ = get_link_endpoint(link)

        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        te_link = te_topo.link(link_name)

        te_link.delete()



        self._object_cache.delete(CachedEntities.LINK, link_uuid)

        self._object_cache.delete(CachedEntities.LINK, link_name)



        MSG = 'Link Remove: {:s}'

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





class SimapUpdater: