VNT Manager component:

- Updated code to use Kafka topics for recommendations to NBI

src/vnt_manager/service/VNTManagerServiceServicerImpl.py

@@ -12,149 +12,83 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import grpc
-import json
-import logging
-import threading
-import time
-from websockets.sync.client import connect
-from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME
+import grpc, json, logging, uuid
+from confluent_kafka import Consumer as KafkaConsumer
+from confluent_kafka import Producer as KafkaProducer
+from confluent_kafka import KafkaError
 from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
-from common.proto.context_pb2 import ContextId, Empty, Link, LinkId, LinkList, TopologyId
-from common.proto.vnt_manager_pb2 import VNTSubscriptionRequest, VNTSubscriptionReply
+from common.proto.context_pb2 import Empty, Link, LinkId, LinkList
 from common.proto.vnt_manager_pb2_grpc import VNTManagerServiceServicer
-from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string
-from common.tools.object_factory.Context import json_context_id
-from common.tools.object_factory.Topology import json_topology_id
+from common.tools.grpc.Tools import grpc_message_to_json_string
+from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
 from context.client.ContextClient import ContextClient
-from context.client.EventsCollector import EventsCollector
 from .vntm_config_device import configure, deconfigure
 
+
 LOGGER = logging.getLogger(__name__)
 
 METRICS_POOL = MetricsPool("VNTManager", "RPC")
 
-context_client: ContextClient = ContextClient()
-
-JSON_ADMIN_CONTEXT_ID = json_context_id(DEFAULT_CONTEXT_NAME)
-ADMIN_CONTEXT_ID = ContextId(**JSON_ADMIN_CONTEXT_ID)
-ADMIN_TOPOLOGY_ID = TopologyId(**json_topology_id(DEFAULT_TOPOLOGY_NAME, context_id=JSON_ADMIN_CONTEXT_ID))
-
-GET_EVENT_TIMEOUT = 0.5
-
-
-class VNTMEventDispatcher(threading.Thread):
-    def __init__(self, host, port) -> None:
-        LOGGER.debug('Creating VNTM connector...')
-        self.host = host
-        self.port = port
-        super().__init__(name='VNTMEventDispatcher', daemon=True)
-        self._terminate = threading.Event()
-        LOGGER.debug('VNTM connector created')
-
-    def start(self) -> None:
-        self._terminate.clear()
-        return super().start()
-
-    def stop(self):
-        self._terminate.set()
-
-    def send_msg(self, msg):
-        try:
-            self.websocket.send(msg)
-        except Exception as e:
-            LOGGER.exception('Unable to send message')
-
-    def recv_msg(self):
-        message = self.websocket.recv()
-        return message
-
-    def run(self) -> None:
-        events_collector = EventsCollector(
-            context_client,
-            log_events_received            = True,
-            activate_context_collector     = True,
-            activate_topology_collector    = True,
-            activate_device_collector      = True,
-            activate_link_collector        = True,
-            activate_service_collector     = False,
-            activate_slice_collector       = False,
-            activate_connection_collector  = False,
-        )
-        events_collector.start()
-
-        try:
-            url = "ws://" + str(self.host) + ":" + str(self.port)
-            LOGGER.info("Connecting to events server...: {:s}".format(url))
-            self.websocket = connect(url)
-        except Exception as ex:
-            MSG = 'Error connecting to {:s}'
-            LOGGER.exception(MSG.format(str(url)))
-        else:
-            LOGGER.info('Connected to {:s}'.format(url))
-            context_id = json_context_id(DEFAULT_CONTEXT_NAME)
-            topology_id = json_topology_id(DEFAULT_TOPOLOGY_NAME, context_id)
-            
-            try:
-                topology_details = context_client.GetTopologyDetails(TopologyId(**topology_id))
-            except Exception as ex:
-                LOGGER.warning('No topology found')
-            else:
-                self.send_msg(grpc_message_to_json_string(topology_details))
-
-            while not self._terminate.is_set():
-                event = events_collector.get_event(block=True, timeout=GET_EVENT_TIMEOUT)
-                if event is None: continue
-                LOGGER.debug('Event type: {}'.format(event))
-                topology_details = context_client.GetTopologyDetails(TopologyId(**topology_id))
-                to_send = grpc_message_to_json_string(topology_details)
-                self.send_msg(to_send)
-        
-            LOGGER.info('Exiting')
-            events_collector.stop()
-
 
 class VNTManagerServiceServicerImpl(VNTManagerServiceServicer):
     def __init__(self):
         LOGGER.debug("Creating Servicer...")
-        LOGGER.debug("Servicer Created")
+        self.context_client = ContextClient()
         self.links = []
-
-    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def VNTSubscript(self, request: VNTSubscriptionRequest, context: grpc.ServicerContext) -> VNTSubscriptionReply:
-        LOGGER.info("Subscript request: {:s}".format(str(grpc_message_to_json(request))))
-        reply = VNTSubscriptionReply()
-        reply.subscription = "OK"
-
-        self.event_dispatcher = VNTMEventDispatcher(request.host, int(request.port))
-        self.host = request.host
-        self.port = request.port
-        LOGGER.info('sleeping 5...')
-        time.sleep(5)
-        self.event_dispatcher.start()
-        return reply
+        LOGGER.debug("Servicer Created")
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def ListVirtualLinks(self, request : Empty, context : grpc.ServicerContext) -> LinkList:
-        return [link for link in context_client.ListLinks(Empty()).links if link.virtual]
+        links = self.context_client.ListLinks(Empty()).links
+        return [link for link in links if link.virtual]
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def GetVirtualLink(self, request : LinkId, context : grpc.ServicerContext) -> Link:
-        link = context_client.GetLink(request)
+        link = self.context_client.GetLink(request)
         return link if link.virtual else Empty()
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def SetVirtualLink(self, request : Link, context : grpc.ServicerContext) -> LinkId:
         try:
-            LOGGER.info('SETTING virtual link')
-            self.event_dispatcher.send_msg(grpc_message_to_json_string(request))
+            LOGGER.info('[SetVirtualLink] request={:s}'.format(grpc_message_to_json_string(request)))
+            request_key = str(uuid.uuid4())
+            kafka_producer = KafkaProducer({
+                'bootstrap.servers' : KafkaConfig.get_kafka_address()
+            })
+
+            vntm_request = json.dumps({
+                'event': 'vlink_create', 'data': grpc_message_to_json_string(request)
+            }).encode('utf-8')
+            LOGGER.info('[SetVirtualLink] vntm_request={:s}'.format(str(vntm_request)))
+            kafka_producer.produce(
+                KafkaTopic.VNTMANAGER_REQUEST.value, key=request_key, value=vntm_request
+            )
+            kafka_producer.flush()
+
+            kafka_consumer = KafkaConsumer({
+                'bootstrap.servers' : KafkaConfig.get_kafka_address(),
+                'group.id'          : str(uuid.uuid4()),
+                'auto.offset.reset' : 'latest'
+            })
+            kafka_consumer.subscribe([KafkaTopic.VNTMANAGER_RESPONSE.value])
+            while True:
+                receive_msg = kafka_consumer.poll(2.0)
+                if receive_msg is None: continue
+                LOGGER.info('[SetVirtualLink] receive_msg={:s}'.format(str(receive_msg)))
+                if receive_msg.error():
+                    if receive_msg.error().code() == KafkaError._PARTITION_EOF: continue
+                    LOGGER.error('Consumer error: {:s}'.format(str(receive_msg.error())))
+                    break
+                reply_key = receive_msg.key().decode('utf-8')
+                if reply_key == request_key: break
+
+            link = Link(**json.loads(receive_msg.value().decode('utf-8')))
+            # at this point, we know the request was accepted and an optical connection was created
+
             # configure('CSGW1', 'xe5', 'CSGW2', 'xe5', 'ecoc2024-1')
-            response = self.event_dispatcher.recv_msg()
-            message_json = json.loads(response)
-            link = Link(**message_json)
-            context_client.SetLink(link)
-        except Exception as e:
-            MSG = 'Exception setting virtual link={:s}')
+            self.context_client.SetLink(link)
+        except: # pylint: disable=bare-except
+            MSG = 'Exception setting virtual link={:s}'
             LOGGER.exception(MSG.format(str(request.link_id.link_uuid.uuid)))
         return request.link_id
 
@@ -162,20 +96,45 @@ class VNTManagerServiceServicerImpl(VNTManagerServiceServicer):
     def RemoveVirtualLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:
         try:
             LOGGER.debug('Removing virtual link')
-            self.event_dispatcher.send_msg(grpc_message_to_json_string(request))
-            # deconfigure('CSGW1', 'xe5', 'CSGW2', 'xe5', 'ecoc2024-1')
-            response = self.event_dispatcher.recv_msg()
-            message_json = json.loads(response)
-            link_id = LinkId(**message_json)
-            context_client.RemoveLink(link_id)
+            request_key = str(uuid.uuid4())
+
+            kafka_producer = KafkaProducer({
+                'bootstrap.servers' : KafkaConfig.get_kafka_address()
+            })
+
+            vntm_request = json.dumps({
+                'event': 'vlink_remove', 'data': grpc_message_to_json_string(request)
+            }).encode('utf-8')
+            LOGGER.info('[RemoveVirtualLink] vntm_request={:s}'.format(str(vntm_request)))
+            kafka_producer.produce(
+                KafkaTopic.VNTMANAGER_REQUEST.value, key=request_key, value=vntm_request
+            )
+            kafka_producer.flush()
+
+            kafka_consumer = KafkaConsumer({
+                'bootstrap.servers' : KafkaConfig.get_kafka_address(),
+                'group.id'          : str(uuid.uuid4()),
+                'auto.offset.reset' : 'latest'
+            })
+            kafka_consumer.subscribe([KafkaTopic.VNTMANAGER_RESPONSE.value])
+            while True:
+                receive_msg = kafka_consumer.poll(2.0)
+                if receive_msg is None: continue
+                if receive_msg.error():
+                    if receive_msg.error().code() == KafkaError._PARTITION_EOF: continue
+                    LOGGER.error('Consumer error: {:s}'.format(str(receive_msg.error())))
+                    break
+                reply_key = receive_msg.key().decode('utf-8')
+                if reply_key == request_key: break
+
+            link_id = LinkId(**json.loads(receive_msg.value().decode('utf-8')))
+            # at this point, we know the request was accepted and an optical connection was deleted
 
+            # deconfigure('CSGW1', 'xe5', 'CSGW2', 'xe5', 'ecoc2024-1')
+            self.context_client.RemoveLink(link_id)
             LOGGER.info('Removed')
-        except Exception as e:
+        except: # pylint: disable=bare-except
             MSG = 'Exception removing virtual link={:s}'
             LOGGER.exception(MSG.format(str(request.link_uuid.uuid)))
-            return msg_error
-        else:
-            context_client.RemoveLink(request)
-            LOGGER.info('Removed')
 
         return Empty()

src/vnt_manager/service/old_code.py

+# Copyright 2022-2024 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 grpc
+import json
+import logging
+import threading
+import time
+from websockets.sync.client import connect
+from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME
+from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
+from common.proto.context_pb2 import ContextId, Empty, Link, LinkId, LinkList, TopologyId
+from common.proto.vnt_manager_pb2 import VNTSubscriptionRequest, VNTSubscriptionReply
+from common.proto.vnt_manager_pb2_grpc import VNTManagerServiceServicer
+from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string
+from common.tools.object_factory.Context import json_context_id
+from common.tools.object_factory.Topology import json_topology_id
+from context.client.ContextClient import ContextClient
+from context.client.EventsCollector import EventsCollector
+from .vntm_config_device import configure, deconfigure
+
+LOGGER = logging.getLogger(__name__)
+
+METRICS_POOL = MetricsPool("VNTManager", "RPC")
+
+context_client: ContextClient = ContextClient()
+
+JSON_ADMIN_CONTEXT_ID = json_context_id(DEFAULT_CONTEXT_NAME)
+ADMIN_CONTEXT_ID = ContextId(**JSON_ADMIN_CONTEXT_ID)
+ADMIN_TOPOLOGY_ID = TopologyId(**json_topology_id(DEFAULT_TOPOLOGY_NAME, context_id=JSON_ADMIN_CONTEXT_ID))
+
+GET_EVENT_TIMEOUT = 0.5
+
+
+class VNTMEventDispatcher(threading.Thread):
+    def __init__(self, host, port) -> None:
+        LOGGER.debug('Creating VNTM connector...')
+        self.host = host
+        self.port = port
+        super().__init__(name='VNTMEventDispatcher', daemon=True)
+        self._terminate = threading.Event()
+        LOGGER.debug('VNTM connector created')
+
+    def start(self) -> None:
+        self._terminate.clear()
+        return super().start()
+
+    def stop(self):
+        self._terminate.set()
+
+    def send_msg(self, msg):
+        try:
+            self.websocket.send(msg)
+        except Exception as e:
+            LOGGER.exception('Unable to send message')
+
+    def recv_msg(self):
+        message = self.websocket.recv()
+        return message
+
+    def run(self) -> None:
+        events_collector = EventsCollector(
+            context_client,
+            log_events_received            = True,
+            activate_context_collector     = True,
+            activate_topology_collector    = True,
+            activate_device_collector      = True,
+            activate_link_collector        = True,
+            activate_service_collector     = False,
+            activate_slice_collector       = False,
+            activate_connection_collector  = False,
+        )
+        events_collector.start()
+
+        try:
+            url = "ws://" + str(self.host) + ":" + str(self.port)
+            LOGGER.info("Connecting to events server...: {:s}".format(url))
+            self.websocket = connect(url)
+        except Exception as ex:
+            MSG = 'Error connecting to {:s}'
+            LOGGER.exception(MSG.format(str(url)))
+        else:
+            LOGGER.info('Connected to {:s}'.format(url))
+            context_id = json_context_id(DEFAULT_CONTEXT_NAME)
+            topology_id = json_topology_id(DEFAULT_TOPOLOGY_NAME, context_id)
+            
+            try:
+                topology_details = context_client.GetTopologyDetails(TopologyId(**topology_id))
+            except Exception as ex:
+                LOGGER.warning('No topology found')
+            else:
+                self.send_msg(grpc_message_to_json_string(topology_details))
+
+            while not self._terminate.is_set():
+                event = events_collector.get_event(block=True, timeout=GET_EVENT_TIMEOUT)
+                if event is None: continue
+                LOGGER.debug('Event type: {}'.format(event))
+                topology_details = context_client.GetTopologyDetails(TopologyId(**topology_id))
+                to_send = grpc_message_to_json_string(topology_details)
+                self.send_msg(to_send)
+        
+            LOGGER.info('Exiting')
+            events_collector.stop()
+
+
+class VNTManagerServiceServicerImpl(VNTManagerServiceServicer):
+    def __init__(self):
+        LOGGER.debug("Creating Servicer...")
+        LOGGER.debug("Servicer Created")
+        self.links = []
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def ListVirtualLinks(self, request : Empty, context : grpc.ServicerContext) -> LinkList:
+        links = context_client.ListLinks(Empty()).links
+        return [link for link in links if link.virtual]
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def GetVirtualLink(self, request : LinkId, context : grpc.ServicerContext) -> Link:
+        link = context_client.GetLink(request)
+        return link if link.virtual else Empty()
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def SetVirtualLink(self, request : Link, context : grpc.ServicerContext) -> LinkId:
+        try:
+            LOGGER.info('SETTING virtual link')
+            self.event_dispatcher.send_msg(grpc_message_to_json_string(request))
+            # configure('CSGW1', 'xe5', 'CSGW2', 'xe5', 'ecoc2024-1')
+            response = self.event_dispatcher.recv_msg()
+            message_json = json.loads(response)
+            link = Link(**message_json)
+            context_client.SetLink(link)
+        except Exception as e:
+            MSG = 'Exception setting virtual link={:s}')
+            LOGGER.exception(MSG.format(str(request.link_id.link_uuid.uuid)))
+        return request.link_id
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def RemoveVirtualLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:
+        try:
+            LOGGER.debug('Removing virtual link')
+            self.event_dispatcher.send_msg(grpc_message_to_json_string(request))
+            # deconfigure('CSGW1', 'xe5', 'CSGW2', 'xe5', 'ecoc2024-1')
+            response = self.event_dispatcher.recv_msg()
+            message_json = json.loads(response)
+            link_id = LinkId(**message_json)
+            context_client.RemoveLink(link_id)
+
+            LOGGER.info('Removed')
+        except Exception as e:
+            MSG = 'Exception removing virtual link={:s}'
+            LOGGER.exception(MSG.format(str(request.link_uuid.uuid)))
+            return msg_error
+        else:
+            context_client.RemoveLink(request)
+            LOGGER.info('Removed')
+
+        return Empty()