Commit 495f649b authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

SIMAP Connector: 

- Factorized create-update code into set
- Fixed creation of simap virtual links, using now update instead
- Fixed identification of service endpoints
parent 8a6d8fe9

src/simap_connector/service/simap_updater/SimapUpdater.py

+52 −152
Original line number Diff line number Diff line
@@ -83,7 +83,7 @@ class EventDispatcher(BaseEventDispatcher): 
        LOGGER.debug(MSG.format(grpc_message_to_json_string(context_event)))





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

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

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

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -101,27 +101,16 @@ class EventDispatcher(BaseEventDispatcher): 
            supporting_network_ids=supporting_network_ids

        )



        MSG = 'Topology Created: {:s}'

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



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

        self._dispatch_topology_set(topology_event)



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

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

        MSG = 'Topology Create: {:s}'

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



        topology_uuid = topology_event.topology_id.topology_uuid.uuid

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

        topology_name = topology.name



        supporting_network_ids = list()

        if topology_name != DEFAULT_TOPOLOGY_NAME:

            supporting_network_ids.append(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

        )

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

        self._dispatch_topology_set(topology_event)



        MSG = 'Topology Updated: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))
@@ -143,7 +132,7 @@ class EventDispatcher(BaseEventDispatcher): 
        LOGGER.info(MSG.format(grpc_message_to_json_string(topology_event)))





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

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

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

        LOGGER.info(MSG.format(grpc_message_to_json_string(device_event)))
@@ -198,64 +187,16 @@ class EventDispatcher(BaseEventDispatcher): 


        #self._remove_skipped_device(device)



        MSG = 'Device Created: {:s}'

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



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

        self._dispatch_device_set(device_event)



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

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

        MSG = 'Device Created: {: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_type = device.device_type

        if device_type in SKIPPED_DEVICE_TYPES:

            self._add_skipped_device(device)

            MSG = (

                'DeviceEvent({:s}) skipped, is of a skipped device type. '

                'SIMAP should be updated by him: {:s}'

            )

            str_device_event = grpc_message_to_json_string(device_event)

            str_device = grpc_message_to_json_string(device)

            LOGGER.warning(MSG.format(str_device_event, str_device))

            return



        #device_controller_uuid = device.controller_id.device_uuid.uuid

        #if len(device_controller_uuid) > 0:

        #    self._add_skipped_device(device)

        #    MSG = (

        #        'DeviceEvent({:s}) skipped, is a remotely-managed device. '

        #        'SIMAP should be updated by remote controller: {:s}'

        #    )

        #    str_device_event = grpc_message_to_json_string(device_event)

        #    str_device = grpc_message_to_json_string(device)

        #    LOGGER.warning(MSG.format(str_device_event, str_device))

        #    return



        topology_uuid, endpoint_names = get_device_endpoint(device)

        if topology_uuid is None:

            #self._add_skipped_device(device)

            MSG = 'DeviceEvent({:s}) skipped, no endpoints to identify topology: {:s}'

            str_device_event = grpc_message_to_json_string(device_event)

            str_device = grpc_message_to_json_string(device)

            LOGGER.warning(MSG.format(str_device_event, str_device))

            return



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

        topology_name = topology.name



        te_topo = self._simap_client.network(topology_name)

        te_topo.update()



        device_name = device.name

        te_device = te_topo.node(device_name)

        te_device.update()



        for endpoint_name in endpoint_names:

            te_device.termination_point(endpoint_name).update()



        #self._remove_skipped_device(device)

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

        self._dispatch_device_set(device_event)



        MSG = 'Device Updated: {:s}'

        LOGGER.info(MSG.format(grpc_message_to_json_string(device_event)))
@@ -323,7 +264,7 @@ class EventDispatcher(BaseEventDispatcher): 
        LOGGER.info(MSG.format(grpc_message_to_json_string(device_event)))





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

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

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

        LOGGER.info(MSG.format(grpc_message_to_json_string(link_event)))
@@ -389,83 +330,24 @@ class EventDispatcher(BaseEventDispatcher): 
            MSG = '{:s} in Link({:s})'

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



        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)

        if topology_uuid is None:

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

            str_link_event = grpc_message_to_json_string(link_event)

            str_link = grpc_message_to_json_string(link)

            LOGGER.warning(MSG.format(str_link_event, str_link))

            return



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

        topology_name = topology.name

        te_link = te_topo.link(link_name)

        te_link.update(src_device.name, src_endpoint.name, dst_device.name, dst_endpoint.name)



        te_topo = self._simap_client.network(topology_name)

        te_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)



        # Skip links that connect two management endpoints

        if src_endpoint is not None and dst_endpoint is not None:

            if str(src_endpoint.name).lower() == 'mgmt' and str(dst_endpoint.name).lower() == 'mgmt':

                MSG = 'LinkEvent({:s}) skipped, connects two management endpoints: {:s}'

                str_link_event = grpc_message_to_json_string(link_event)

                str_link = grpc_message_to_json_string(link)

                LOGGER.warning(MSG.format(str_link_event, str_link))

                return

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

        self._dispatch_link_set(link_event)



        # Skip links that connect to devices previously marked as skipped

        src_uuid = src_device.device_id.device_uuid.uuid

        dst_uuid = dst_device.device_id.device_uuid.uuid

        src_name = src_device.name

        dst_name = dst_device.name

        if (src_uuid in self._skipped_devices or src_name in self._skipped_devices

            or dst_uuid in self._skipped_devices or dst_name in self._skipped_devices):

            MSG = 'LinkEvent({:s}) skipped, connects to skipped device(s): {:s}'

            str_link_event = grpc_message_to_json_string(link_event)

            str_link = grpc_message_to_json_string(link)

            LOGGER.warning(MSG.format(str_link_event, str_link))

            return

        MSG = 'Link Created: {:s}'

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



        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 Link({:s})'

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



        te_link = te_topo.link(link_name)

        te_link.update(src_device.name, src_endpoint.name, dst_device.name, dst_endpoint.name)

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

        self._dispatch_link_set(link_event)



        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)))
@@ -517,7 +399,8 @@ 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:



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

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

        LOGGER.info(MSG.format(grpc_message_to_json_string(service_event)))
@@ -533,6 +416,7 @@ class EventDispatcher(BaseEventDispatcher): 
            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)
@@ -540,6 +424,13 @@ class EventDispatcher(BaseEventDispatcher): 
            LOGGER.warning(MSG.format(str_service_event, str_service))

            return



        if len(endpoint_uuids) < 2:

            MSG = 'ServiceEvent({:s}) skipped, not enough endpoint_ids to compose link: {: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)
@@ -584,28 +475,36 @@ class EventDispatcher(BaseEventDispatcher): 


        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_1.update(supporting_node_ids=[(parent_domain_name, src_dev_name)])

        site_1_tp = site_1.termination_point(src_ep_name)

        site_1_tp.update(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)]

        )

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

        site_2_tp = site_2.termination_point(dst_ep_name)

        site_2_tp.update(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)

        dom_link = domain_topo.link(link_name)

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





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

        self._dispatch_service_set(service_event)



        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)

        self._dispatch_service_set(service_event)



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

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





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

        MSG = 'Processing Service Event: {:s}'
@@ -675,6 +574,7 @@ class EventDispatcher(BaseEventDispatcher): 
        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:

        self._context_client = ContextClient()

src/simap_connector/service/simap_updater/Tools.py

+6 −4
Original line number Diff line number Diff line
@@ -125,12 +125,12 @@ def get_link_endpoint(link : Link) -> Tuple[Optional[str], List[Tuple[str, str]] 




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

    if len(service.service_endpoint_ids) == 0:

        return None, list()



    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)
@@ -144,15 +144,17 @@ def get_service_endpoint(service : Service) -> Tuple[Optional[str], List[Tuple[s 
        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:

        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