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

SIMAP Connector: 

- Added support for Services in ObjectCache
- Added support for list-all in ObjectCache
- Added handling of service events in SimapUpdater
parent 7c7ccd6c

src/simap_connector/service/simap_updater/ObjectCache.py

+83 −12
Original line number Diff line number Diff line
@@ -15,10 +15,11 @@ 


import logging

from enum import Enum

from typing import Any, Dict, Optional, Tuple

from common.tools.context_queries.Device import get_device

from common.tools.context_queries.Link import get_link

from common.tools.context_queries.Topology import get_topology

from typing import Any, Dict, List, Optional, Tuple

from common.tools.context_queries.Device import get_device, get_devices

from common.tools.context_queries.Link import get_link, get_links

from common.tools.context_queries.Topology import get_topology, get_topologies

from common.tools.context_queries.Service import get_service_by_uuid, get_services

from context.client.ContextClient import ContextClient
@@ -30,6 +31,8 @@ class CachedEntities(Enum): 
    DEVICE     = 'device'

    ENDPOINT   = 'endpoint'

    LINK       = 'link'

    SERVICE    = 'service'

    CONNECTION = 'connection'





KEY_LENGTHS = {
@@ -37,6 +40,8 @@ KEY_LENGTHS = { 
    CachedEntities.DEVICE     : 1,

    CachedEntities.ENDPOINT   : 2,

    CachedEntities.LINK       : 1,

    CachedEntities.SERVICE    : 2,

    CachedEntities.CONNECTION : 3,

}
@@ -69,6 +74,17 @@ class ObjectCache: 
        if not auto_retrieve: return None

        return self._update(entity, *object_uuids)



    def get_all(

        self, entity : CachedEntities, fresh : bool = False

    ) -> List[Any]:

        if fresh: self._update_all(entity)

        entity_name = str(entity.value)

        return [

            obj

            for obj_key, obj in self._object_cache.items()

            if obj_key[0] == entity_name

        ]



    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
@@ -90,6 +106,10 @@ class ObjectCache: 
            object_inst = get_link(

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

            )

        elif entity == CachedEntities.SERVICE:

            object_inst = get_service_by_uuid(

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

            )

        else:

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

            LOGGER.warning(MSG.format(str(entity.value).title(), str(object_uuids)))
@@ -122,6 +142,57 @@ class ObjectCache: 


        return object_inst



    def _update_all(self, entity : CachedEntities) -> None:

        if entity == CachedEntities.TOPOLOGY:

            objects = get_topologies(self._context_client)

            objects = {

                (t.topology_id.topology_uuid.uuid, t.name) : t

                for t in objects

            }

        elif entity == CachedEntities.DEVICE:

            objects = get_devices(self._context_client)

            objects = {

                (d.device_id.device_uuid.uuid, d.name) : d

                for d in objects

            }

        elif entity == CachedEntities.ENDPOINT:

            # Endpoints are only updated when updating a Device

            return None

        elif entity == CachedEntities.LINK:

            objects = get_links(self._context_client)

            objects = {

                (l.link_id.link_uuid.uuid, l.name) : l

                for l in objects

            }

        elif entity == CachedEntities.SERVICE:

            objects = get_services(self._context_client)

            objects = {

                (s.service_id.service_uuid.uuid, s.name) : s

                for s in objects

            }

        else:

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

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

            return None



        for (object_uuid, object_name), object_inst in objects.items():

            self.set(entity, object_inst, object_uuid)

            self.set(entity, object_inst, object_name)



            if entity == CachedEntities.DEVICE:

                for endpoint in object_inst.device_endpoints:

                    endpoint_device_uuid = endpoint.endpoint_id.device_id.device_uuid.uuid

                    if object_uuid != endpoint_device_uuid:

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

                        raise Exception(str(object_uuid), str(endpoint_device_uuid))



                    endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid

                    endpoint_name = endpoint.name

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

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

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

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



    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

+176 −5
Original line number Diff line number Diff line
@@ -13,10 +13,11 @@ 
# limitations under the License.





import logging, queue, threading

import logging, queue, threading, uuid

from typing import Any, Optional, Set

from common.Constants import DEFAULT_TOPOLOGY_NAME

from common.DeviceTypes import DeviceTypeEnum

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

from common.proto.context_pb2 import ContextEvent, DeviceEvent, Empty, LinkEvent, ServiceEvent, 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
@@ -28,7 +29,7 @@ from simap_connector.Config import ( 
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

from .Tools import get_device_endpoint, get_link_endpoint, get_service_endpoint





LOGGER = logging.getLogger(__name__)
@@ -90,9 +91,14 @@ class EventDispatcher(BaseEventDispatcher): 
        topology = self._object_cache.get(CachedEntities.TOPOLOGY, topology_uuid)

        topology_name = topology.name

        

        if topology_name != DEFAULT_TOPOLOGY_NAME:

            supporting_network_ids = [DEFAULT_TOPOLOGY_NAME]



        # Theoretically it should be create(), but given we have multiple clients

        # updating same SIMAP server, use update to skip tricks on get-check-create-or-update.

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

        self._simap_client.network(topology_name).update(

            supporting_network_ids=supporting_network_ids

        )



        MSG = 'Topology Created: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -105,7 +111,15 @@ 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).update()



        if topology_name != DEFAULT_TOPOLOGY_NAME:

            supporting_network_ids = [DEFAULT_TOPOLOGY_NAME]



        # Theoretically it should be create(), but given we have multiple clients

        # updating same SIMAP server, use update to skip tricks on get-check-create-or-update.

        self._simap_client.network(topology_name).update(

            supporting_network_ids=supporting_network_ids

        )



        MSG = 'Topology Updated: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -501,6 +515,163 @@ class EventDispatcher(BaseEventDispatcher): 
        MSG = 'Link Remove: {:s}'

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



    def dispatch_service_create(self, service_event : ServiceEvent) -> None:

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

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



        service_uuid = service_event.service_id.service_uuid.uuid

        service = self._object_cache.get(CachedEntities.SERVICE, service_uuid)

        service_name = service.name



        try:

            uuid.UUID(hex=service_name)

            # skip it if properly parsed, means it is a service with a UUID-based name, i.e., a sub-service

            return

        except: # pylint: disable=bare-except

            pass



        topology_uuid, endpoint_uuids = get_service_endpoint(service)

        if topology_uuid is None:

            MSG = 'ServiceEvent({:s}) skipped, no endpoint_ids to identify topology: {:s}'

            str_service_event = grpc_message_to_json_string(service_event)

            str_service = grpc_message_to_json_string(service)

            LOGGER.warning(MSG.format(str_service_event, str_service))

            return



        topologies = self._object_cache.get_all(CachedEntities.TOPOLOGY, fresh=False)

        topology_names = {t.name for t in topologies}

        topology_names.discard(DEFAULT_TOPOLOGY_NAME)

        if len(topology_names) != 1:

            MSG = 'ServiceEvent({:s}) skipped, unable to identify on which topology to insert it'

            str_service_event = grpc_message_to_json_string(service_event)

            LOGGER.warning(MSG.format(str_service_event))

            return

        domain_name = topology_names.pop()  # trans-pkt/agg-net/e2e-net



        domain_topo = self._simap_client.network(domain_name)

        domain_topo.update()



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

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

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

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



        try:

            if src_device is None:

                MSG = 'Device({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[0][0])))

            if src_endpoint is None:

                MSG = 'Endpoint({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[0])))

            if dst_device is None:

                MSG = 'Device({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[1][0])))

            if dst_endpoint is None:

                MSG = 'Endpoint({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[1])))

        except Exception as e:

            MSG = '{:s} in Service({:s})'

            raise Exception(MSG.format(str(e), grpc_message_to_json_string(service))) from e



        src_dev_name = src_device.name

        src_ep_name  = src_endpoint.name

        dst_dev_name = dst_device.name

        dst_ep_name  = dst_endpoint.name



        parent_domain_name = DEFAULT_TOPOLOGY_NAME      # TODO: compute from service settings



        site_1_name = 'site1'                           # TODO: compute from service settings

        site_1 = domain_topo.node(site_1_name)

        site_1.create(supporting_node_ids=[(parent_domain_name, src_dev_name)])

        site_1.termination_point(src_ep_name).create(

            supporting_termination_point_ids=[(parent_domain_name, src_dev_name, src_ep_name)]

        )



        site_2_name = 'site2'                           # TODO: compute from service settings

        site_2 = domain_topo.node(site_2_name)

        site_2.create(supporting_node_ids=[(parent_domain_name, dst_dev_name)])

        site_2.termination_point(dst_ep_name).create(

            supporting_termination_point_ids=[(parent_domain_name, dst_dev_name, dst_ep_name)]

        )



        link_name = '{:s}:{:s}-{:s}=={:s}-{:s}'.format(

            service_name, src_dev_name, src_ep_name, dst_dev_name, dst_ep_name

        )

        domain_topo.link(link_name).create(src_dev_name, src_ep_name, dst_dev_name, dst_ep_name)



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

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



    def dispatch_service_update(self, service_event : ServiceEvent) -> None:

        self.dispatch_service_create(service_event)



    def dispatch_service_remove(self, service_event : ServiceEvent) -> None:

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

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



        service_uuid = service_event.service_id.service_uuid.uuid

        service = self._object_cache.get(CachedEntities.SERVICE, service_uuid)

        service_name = service.name



        topology_uuid, endpoint_uuids = get_service_endpoint(service)

        if topology_uuid is None:

            MSG = 'ServiceEvent({:s}) skipped, no endpoint_ids to identify topology: {:s}'

            str_service_event = grpc_message_to_json_string(service_event)

            str_service = grpc_message_to_json_string(service)

            LOGGER.warning(MSG.format(str_service_event, str_service))

            return



        topologies = self._object_cache.get_all(CachedEntities.TOPOLOGY, fresh=False)

        topology_names = {t.name for t in topologies}

        topology_names.discard(DEFAULT_TOPOLOGY_NAME)

        if len(topology_names) != 1:

            MSG = 'ServiceEvent({:s}) skipped, unable to identify on which topology to insert it'

            str_service_event = grpc_message_to_json_string(service_event)

            LOGGER.warning(MSG.format(str_service_event))

            return

        domain_name = topology_names.pop()  # trans-pkt/agg-net/e2e-net



        domain_topo = self._simap_client.network(domain_name)

        domain_topo.update()



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

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

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

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



        try:

            if src_device is None:

                MSG = 'Device({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[0][0])))

            if src_endpoint is None:

                MSG = 'Endpoint({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[0])))

            if dst_device is None:

                MSG = 'Device({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[1][0])))

            if dst_endpoint is None:

                MSG = 'Endpoint({:s}) not found in cache'

                raise Exception(MSG.format(str(endpoint_uuids[1])))

        except Exception as e:

            MSG = '{:s} in Service({:s})'

            raise Exception(MSG.format(str(e), grpc_message_to_json_string(service))) from e



        src_dev_name = src_device.name

        src_ep_name  = src_endpoint.name

        dst_dev_name = dst_device.name

        dst_ep_name  = dst_endpoint.name



        link_name = '{:s}:{:s}-{:s}=={:s}-{:s}'.format(

            service_name, src_dev_name, src_ep_name, dst_dev_name, dst_ep_name

        )

        te_link = domain_topo.link(link_name)

        te_link.delete()



        self._object_cache.delete(CachedEntities.SERVICE, service_uuid)

        self._object_cache.delete(CachedEntities.SERVICE, service_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:

src/simap_connector/service/simap_updater/Tools.py

+37 −1
Original line number Diff line number Diff line
@@ -17,7 +17,7 @@ import enum 
from typing import List, Optional, Set, Tuple, Union

from common.proto.context_pb2 import (

    EVENTTYPE_CREATE, EVENTTYPE_REMOVE, EVENTTYPE_UPDATE, Device,

    DeviceEvent, Link, LinkEvent, ServiceEvent, SliceEvent, TopologyEvent

    DeviceEvent, Link, LinkEvent, Service, ServiceEvent, SliceEvent, TopologyEvent

)

from common.tools.grpc.Tools import grpc_message_to_json_string
@@ -122,3 +122,39 @@ def get_link_endpoint(link : Link) -> Tuple[Optional[str], List[Tuple[str, str]] 
        raise Exception(MSG.format(str(e), grpc_message_to_json_string(link))) from e



    return topology_uuid, endpoint_uuids





def get_service_endpoint(service : Service) -> Tuple[Optional[str], List[Tuple[str, str]]]:

    topology_uuids : Set[str] = set()

    endpoint_uuids : List[Tuple[str, str]] = list()



    if len(service.service_endpoint_ids) == 0:

        return None, endpoint_uuids



    for endpoint_id in service.service_endpoint_ids:

        topology_uuid = endpoint_id.topology_id.topology_uuid.uuid

        topology_uuids.add(topology_uuid)



        device_uuid = endpoint_id.device_id.device_uuid.uuid

        endpoint_uuid = endpoint_id.endpoint_uuid.uuid

        endpoint_uuids.append((device_uuid, endpoint_uuid))



    try:

        # Check topology UUIDs

        if len(topology_uuids) != 1:

            MSG = 'Unsupported: no/multiple Topologies({:s}) referenced'

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

        topology_uuid = list(topology_uuids)[0]

        if len(topology_uuid) == 0:

            MSG = 'Unsupported: empty TopologyUUID({:s}) referenced'

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



        # Check Count Endpoints

        if len(endpoint_uuids) != 2:

            MSG = 'Unsupported: non-p2p service ServiceUUIDs({:s})'

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

    except Exception as e:

        MSG = '{:s} in Service({:s})'

        raise Exception(MSG.format(str(e), grpc_message_to_json_string(service))) from e



    return topology_uuid, endpoint_uuids