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

VNT Manager component: 

- Fixed race condition while subscribing to Kafka
parent 096d8137

src/vnt_manager/service/VNTManagerServiceServicerImpl.py

+53 −16
Original line number Diff line number Diff line
@@ -13,7 +13,7 @@ 
# limitations under the License.



from typing import Dict, Optional

import grpc, json, logging, uuid

import grpc, json, logging, time, uuid

from confluent_kafka import Consumer as KafkaConsumer

from confluent_kafka import Producer as KafkaProducer

from confluent_kafka import KafkaError
@@ -68,6 +68,30 @@ class VNTManagerServiceServicerImpl(VNTManagerServiceServicer): 
        self.kafka_producer.flush()

        return request_key



    def create_reply_consumer(self) -> KafkaConsumer:

        LOGGER.info('[create_reply_consumer] begin')

        kafka_consumer = KafkaConsumer({

            'bootstrap.servers'   : KafkaConfig.get_kafka_address(),

            'group.id'            : str(uuid.uuid4()),

            'auto.offset.reset'   : 'latest',

            'enable.auto.commit'  : False,

            'max.poll.interval.ms': 600000,

            'session.timeout.ms'  : 60000,

        })

        kafka_consumer.subscribe([KafkaTopic.VNTMANAGER_RESPONSE.value])



        deadline = time.monotonic() + 15.0

        while time.monotonic() < deadline:

            kafka_consumer.poll(0.2)

            assignment = kafka_consumer.assignment()

            if len(assignment) > 0:

                LOGGER.info('[create_reply_consumer] assigned=%s', str(assignment))

                return kafka_consumer



        LOGGER.error('[create_reply_consumer] timed out waiting for topic assignment')

        kafka_consumer.close()

        raise Exception('Kafka consumer subscription to VNT Manager reply topic was not assigned')



    def send_vlink_create(self, request : Link) -> str:

        return self.send_recommendation({

            'event': 'vlink_create', 'data': grpc_message_to_json_string(request)
@@ -78,21 +102,24 @@ class VNTManagerServiceServicerImpl(VNTManagerServiceServicer): 
            'event': 'vlink_remove', 'data': grpc_message_to_json_string(request)

        })



    def wait_for_reply(self, request_key : str) -> Optional[Dict]:

    def wait_for_reply(self, request_key : str, kafka_consumer : KafkaConsumer) -> Optional[Dict]:

        LOGGER.info('[wait_for_reply] request_key={:s}'.format(str(request_key)))



        self.kafka_consumer = KafkaConsumer({

            'bootstrap.servers'   : KafkaConfig.get_kafka_address(),

            'group.id'            : str(uuid.uuid4()),

            'auto.offset.reset'   : 'latest',

            'max.poll.interval.ms': 600000,

            'session.timeout.ms'  : 60000,

        })

        self.kafka_consumer.subscribe([KafkaTopic.VNTMANAGER_RESPONSE.value])



        deadline = time.monotonic() + 120.0

        polls_without_message = 0

        while True:

            receive_msg = self.kafka_consumer.poll(2.0)

            if receive_msg is None: continue

            receive_msg = kafka_consumer.poll(2.0)

            if receive_msg is None:

                polls_without_message += 1

                if polls_without_message % 5 == 0:

                    LOGGER.info(

                        '[wait_for_reply] request_key=%s still waiting... assignment=%s',

                        str(request_key), str(kafka_consumer.assignment())

                    )

                if time.monotonic() >= deadline:

                    raise TimeoutError('Timed out waiting for VNT Manager reply for request_key={:s}'.format(

                        str(request_key)

                    ))

                continue

            LOGGER.info('[wait_for_reply] receive_msg={:s}'.format(str(receive_msg)))

            if receive_msg.error():

                if receive_msg.error().code() == KafkaError._PARTITION_EOF: continue
@@ -127,8 +154,13 @@ class VNTManagerServiceServicerImpl(VNTManagerServiceServicer): 
    def SetVirtualLink(self, request : Link, context : grpc.ServicerContext) -> LinkId:

        try:

            LOGGER.info('[SetVirtualLink] request={:s}'.format(grpc_message_to_json_string(request)))

            kafka_consumer = self.create_reply_consumer()

            request_key = self.send_vlink_create(request)

            reply = self.wait_for_reply(request_key)

            try:

                reply = self.wait_for_reply(request_key, kafka_consumer)

            finally:

                LOGGER.info('[SetVirtualLink] closing reply consumer')

                kafka_consumer.close()

            LOGGER.info('[SetVirtualLink] reply={:s}'.format(str(reply)))



            # At this point, we know the request is processed and an optical connection was created
@@ -193,8 +225,13 @@ class VNTManagerServiceServicerImpl(VNTManagerServiceServicer): 
    def RemoveVirtualLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:

        try:

            LOGGER.info('[RemoveVirtualLink] request={:s}'.format(grpc_message_to_json_string(request)))

            kafka_consumer = self.create_reply_consumer()

            request_key = self.send_vlink_remove(request)

            reply = self.wait_for_reply(request_key)

            try:

                reply = self.wait_for_reply(request_key, kafka_consumer)

            finally:

                LOGGER.info('[RemoveVirtualLink] closing reply consumer')

                kafka_consumer.close()

            LOGGER.info('[RemoveVirtualLink] reply={:s}'.format(str(reply)))



            # At this point, we know the request is processed and an optical connection was removed