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

Events: 

- Add new AnyEvent message and GetAllEvents() RPC method.
- Add methods in Context client and server
- Adapt Simap Updater
parent 5a4356d4

proto/context.proto

+14 −1
Original line number Diff line number Diff line
@@ -78,6 +78,7 @@ service ContextService { 
  rpc RemoveConnection   (ConnectionId  ) returns (       Empty           ) {}

  rpc GetConnectionEvents(Empty         ) returns (stream ConnectionEvent ) {}



  rpc GetAllEvents       (Empty         ) returns (stream AnyEvent        ) {}



  // ------------------------------ Experimental -----------------------------

  rpc GetOpticalConfig       (Empty            ) returns (OpticalConfigList) {}
@@ -118,6 +119,18 @@ message Event { 
  EventTypeEnum event_type = 2;

}



message AnyEvent {

  oneof event {

    ContextEvent    context    = 1;

    TopologyEvent   topology   = 2;

    DeviceEvent     device     = 3;

    LinkEvent       link       = 4;

    ServiceEvent    service    = 5;

    SliceEvent      slice      = 6;

    ConnectionEvent connection = 7;

  }

}



// ----- Context -------------------------------------------------------------------------------------------------------

message ContextId {

  Uuid context_uuid = 1;

src/context/client/ContextClient.py

+8 −1
Original line number Diff line number Diff line
@@ -19,7 +19,7 @@ from common.Settings import get_service_host, get_service_port_grpc 
from common.tools.client.RetryDecorator import retry, delay_exponential

from common.tools.grpc.Tools import grpc_message_to_json_string

from common.proto.context_pb2 import (

    Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList,

    AnyEvent, Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList,

    Context, ContextEvent, ContextId, ContextIdList, ContextList,

    Device, DeviceEvent, DeviceFilter, DeviceId, DeviceIdList, DeviceList,

    Empty, EndPointIdList, EndPointNameList,
@@ -61,6 +61,13 @@ class ContextClient: 
        self.stub = None

        self.policy_stub = None



    @RETRY_DECORATOR

    def GetAllEvents(self, request: Empty) -> Iterator[AnyEvent]:

        LOGGER.debug('GetAllEvents request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.GetAllEvents(request)

        LOGGER.debug('GetAllEvents result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def ListContextIds(self, request: Empty) -> ContextIdList:

        LOGGER.debug('ListContextIds request: {:s}'.format(grpc_message_to_json_string(request)))

src/context/service/ContextServiceServicerImpl.py

+7 −2
Original line number Diff line number Diff line
@@ -16,7 +16,7 @@ import grpc, logging, sqlalchemy 
from typing import Iterator

from common.message_broker.MessageBroker import MessageBroker

from common.proto.context_pb2 import (

    Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList,

    AnyEvent, Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList,

    Context, ContextEvent, ContextId, ContextIdList, ContextList,

    Device, DeviceEvent, DeviceFilter, DeviceId, DeviceIdList, DeviceList,

    Empty, EndPointIdList, EndPointNameList,
@@ -41,7 +41,7 @@ from .database.Device import ( 
    device_delete, device_get, device_list_ids, device_list_objs, device_select, device_set

)

from .database.EndPoint import endpoint_list_names

from .database.Events import EventTopicEnum, consume_events

from .database.Events import EventTopicEnum, consume_all_events, consume_events

from .database.Link import (

    link_delete, link_get, link_list_ids, link_list_objs, link_set

)
@@ -83,6 +83,11 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 
    def _get_metrics(self) -> MetricsPool: return METRICS_POOL





    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def GetAllEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[AnyEvent]:

        for message in consume_all_events(self.messagebroker): yield message





    # ----- Context ----------------------------------------------------------------------------------------------------



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

src/context/service/database/Events.py

+28 −1
Original line number Diff line number Diff line
@@ -17,7 +17,7 @@ from typing import Dict, Iterator, Set 
from common.message_broker.Message import Message

from common.message_broker.MessageBroker import MessageBroker

from common.proto.context_pb2 import (

    ConnectionEvent, ContextEvent, DeviceEvent, EventTypeEnum, LinkEvent,

    AnyEvent, ConnectionEvent, ContextEvent, DeviceEvent, EventTypeEnum, LinkEvent,

    ServiceEvent, SliceEvent, TopologyEvent, OpticalConfigEvent

)
@@ -130,3 +130,30 @@ def consume_events( 
            MSG = 'Unable to identify EventClass for Message({:s}). Ignoring...'

            LOGGER.warning(MSG.format(str(message)))

            continue



def consume_all_events(

    messagebroker : MessageBroker, consume_timeout : float = CONSUME_TIMEOUT

) -> Iterator[AnyEvent]:

    topic_names = [EventTopicEnum.ALL.value]

    for message in messagebroker.consume(topic_names, consume_timeout=consume_timeout):

        data = json.loads(message.content)

        if 'context_id' in data:

            yield AnyEvent(context=data)

        elif 'topology_id' in data:

            yield AnyEvent(topology=data)

        elif 'device_id' in data:

            yield AnyEvent(device=data)

        elif 'opticalconfig_id' in data:

            yield AnyEvent(device=data)

        elif 'link_id' in data:

            yield AnyEvent(link=data)

        elif 'service_id' in data:

            yield AnyEvent(service=data)

        elif 'slice_id' in data:

            yield AnyEvent(slice=data)

        elif 'connection_id' in data:

            yield AnyEvent(connection=data)

        else:

            MSG = 'Unable to identify EventClass for Message({:s}). Ignoring...'

            LOGGER.warning(MSG.format(str(message)))

            continue

src/simap_connector/service/simap_updater/SimapUpdater.py

+34 −9
Original line number Diff line number Diff line
@@ -14,7 +14,7 @@ 




import logging, queue, threading

from typing import Any, Optional, Set

from typing import Any, Callable, Optional, Set

from common.DeviceTypes import DeviceTypeEnum

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

from common.tools.grpc.BaseEventCollector import BaseEventCollector
@@ -28,7 +28,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 EventTypeEnum, get_device_endpoint, get_link_endpoint





LOGGER = logging.getLogger(__name__)
@@ -68,6 +68,28 @@ class EventDispatcher(BaseEventDispatcher): 
        MSG = 'Unexpected Event: {:s}'

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



    # overwrite default _get_dispatcher() method

    def _get_dispatcher(self, event : Any) -> Optional[Callable]:

        object_name = str(event.__class__.__name__).lower().replace('event', '')

        if object_name == 'any':

            object_name = event.WhichOneof('event')

            event = getattr(event, object_name)



        event_type  = EventTypeEnum.Name(event.event.event_type).lower().replace('eventtype_', '')



        method_name = 'dispatch_{:s}_{:s}'.format(object_name, event_type)

        dispatcher  = getattr(self, method_name, None)

        if dispatcher is not None: return dispatcher



        method_name = 'dispatch_{:s}'.format(object_name)

        dispatcher  = getattr(self, method_name, None)

        if dispatcher is not None: return dispatcher



        method_name = 'dispatch'

        dispatcher  = getattr(self, method_name, None)

        if dispatcher is not None: return dispatcher



        return None



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

        MSG = 'Processing Topology Event: {:s}'
@@ -506,14 +528,17 @@ class SimapUpdater: 


        self._event_collector = BaseEventCollector(terminate=terminate)

        self._event_collector.install_collector(

            self._context_client.GetTopologyEvents, Empty(), log_events_received=True

        )

        self._event_collector.install_collector(

            self._context_client.GetDeviceEvents, Empty(), log_events_received=True

        )

        self._event_collector.install_collector(

            self._context_client.GetLinkEvents, Empty(), log_events_received=True

            self._context_client.GetAllEvents, Empty(), log_events_received=True

        )

        #self._event_collector.install_collector(

        #    self._context_client.GetTopologyEvents, Empty(), log_events_received=True

        #)

        #self._event_collector.install_collector(

        #    self._context_client.GetDeviceEvents, Empty(), log_events_received=True

        #)

        #self._event_collector.install_collector(

        #    self._context_client.GetLinkEvents, Empty(), log_events_received=True

        #)



        self._event_dispatcher = EventDispatcher(

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