13395484668

Committed 18 Feb 2025 04:32PM UTC coverage: 55.457% (-0.3%) from 55.763%

Build # 13395484668

Build Type

Pull #417

github

Committed by

web-flow

Commit Message

Merge ebab391fa into 4870b19a5

Pull Request Pull Request #417: Use constants from datamodel

Run Details

9 of 12 new or added lines in 8 files covered. (75.0%)

7 existing lines in 1 file now uncovered.

1123 of 2025 relevant lines covered (55.46%)

1.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

57.65

/sdx_controller/messaging/rpc_queue_consumer.py

#!/usr/bin/env python
import json
import logging
import os
import threading
from queue import Queue

import pika
from sdx_datamodel.constants import Constants, MessageQueueNames, MongoCollections

from sdx_controller.handlers.lc_message_handler import LcMessageHandler
from sdx_controller.utils.parse_helper import ParseHelper

MQ_HOST = os.getenv("MQ_HOST")
MQ_PORT = os.getenv("MQ_PORT") or 5672
MQ_USER = os.getenv("MQ_USER") or "guest"
MQ_PASS = os.getenv("MQ_PASS") or "guest"

# subscribe to the corresponding queue
SUB_QUEUE = MessageQueueNames.OXP_UPDATE

logger = logging.getLogger(__name__)


class RpcConsumer(object):
    def __init__(self, thread_queue, exchange_name, te_manager):
        self.logger = logging.getLogger(__name__)

        self.logger.info(f"[MQ] Using amqp://{MQ_USER}@{MQ_HOST}:{MQ_PORT}")

        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(
                host=MQ_HOST,
                port=MQ_PORT,
                credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
            )
        )

        self.channel = self.connection.channel()
        self.exchange_name = exchange_name

        self.channel.queue_declare(queue=SUB_QUEUE)
        self._thread_queue = thread_queue

        self.te_manager = te_manager

        self._exit_event = threading.Event()

    def on_request(self, ch, method, props, message_body):
        response = message_body
        self._thread_queue.put(message_body)

        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(
                host=MQ_HOST,
                port=MQ_PORT,
                credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
            )
        )
        self.channel = self.connection.channel()

        try:
            ch.basic_publish(
                exchange=self.exchange_name,
                routing_key=props.reply_to,
                properties=pika.BasicProperties(correlation_id=props.correlation_id),
                body=str(response),
            )
            ch.basic_ack(delivery_tag=method.delivery_tag)
        except Exception as err:
            self.logger.info(f"[MQ] encountered error when publishing: {err}")

    def start_consumer(self):
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(queue=SUB_QUEUE, on_message_callback=self.on_request)

        self.logger.info(" [MQ] Awaiting requests from queue: " + SUB_QUEUE)
        self.channel.start_consuming()

    def start_sdx_consumer(self, thread_queue, db_instance):
        HEARTBEAT_ID = 0

        rpc = RpcConsumer(thread_queue, "", self.te_manager)
        t1 = threading.Thread(target=rpc.start_consumer, args=(), daemon=True)
        t1.start()

        lc_message_handler = LcMessageHandler(db_instance, self.te_manager)
        parse_helper = ParseHelper()

        latest_topo = {}
        domain_list = []

        # This part reads from DB when SDX controller initially starts.
        # It looks for domain_list, if already in DB,
        # Then use the existing ones from DB.
        domain_list_from_db = db_instance.read_from_db(
            MongoCollections.DOMAINS, Constants.DOMAIN_LIST
        )
        latest_topo_from_db = db_instance.read_from_db(
            MongoCollections.TOPOLOGIES, Constants.LATEST_TOPO
        )

        if domain_list_from_db:
            domain_list = domain_list_from_db[Constants.DOMAIN_LIST]
            logger.debug("Domain list already exists in db: ")
            logger.debug(domain_list)

        if latest_topo_from_db:
            latest_topo = latest_topo_from_db[Constants.LATEST_TOPO]
            logger.debug("Topology already exists in db: ")
            logger.debug(latest_topo)

        # If topologies already saved in db, use them to initialize te_manager
        if domain_list:
            for domain in domain_list:
                topology = db_instance.read_from_db(MongoCollections.TOPOLOGIES, domain)

                if not topology:
                    continue

                # Get the actual thing minus the Mongo ObjectID.
                topology = topology[domain]
                topo_json = json.loads(topology)
                self.te_manager.add_topology(topo_json)
                logger.debug(f"Read {domain}: {topology}")

        while not self._exit_event.is_set():
            # Queue.get() will block until there's an item in the queue.
            msg = thread_queue.get()
            logger.debug("MQ received message:" + str(msg))

            if "Heart Beat" in str(msg):
                HEARTBEAT_ID += 1
                logger.debug("Heart beat received. ID: " + str(HEARTBEAT_ID))
                continue

            if not parse_helper.is_json(msg):
                continue

            if "version" not in str(msg):
                logger.info("Got message (NO VERSION) from MQ: " + str(msg))

            lc_message_handler.process_lc_json_msg(
                msg,
                latest_topo,
                domain_list,
            )

    def stop_threads(self):
        """
        Signal threads that we're ready to stop.
        """
        logger.info("[MQ] Stopping threads.")
        self.channel.stop_consuming()
        self._exit_event.set()

1	#!/usr/bin/env python
2	import json	2✔
3	import logging	2✔
4	import os	2✔
5	import threading	2✔
6	from queue import Queue	2✔
7
8	import pika	2✔
9	from sdx_datamodel.constants import Constants, MessageQueueNames, MongoCollections	2✔
10
11	from sdx_controller.handlers.lc_message_handler import LcMessageHandler	2✔
12	from sdx_controller.utils.parse_helper import ParseHelper	2✔
13
14	MQ_HOST = os.getenv("MQ_HOST")	2✔
15	MQ_PORT = os.getenv("MQ_PORT") or 5672	2✔
16	MQ_USER = os.getenv("MQ_USER") or "guest"	2✔
17	MQ_PASS = os.getenv("MQ_PASS") or "guest"	2✔
18
19	# subscribe to the corresponding queue
20	SUB_QUEUE = MessageQueueNames.OXP_UPDATE	2✔
21
22	logger = logging.getLogger(__name__)	2✔
23
24
25	class RpcConsumer(object):	2✔
26	def __init__(self, thread_queue, exchange_name, te_manager):	2✔
27	self.logger = logging.getLogger(__name__)	2✔
28
29	self.logger.info(f"[MQ] Using amqp://{MQ_USER}@{MQ_HOST}:{MQ_PORT}")	2✔
30
31	self.connection = pika.BlockingConnection(	2✔
32	pika.ConnectionParameters(
33	host=MQ_HOST,
34	port=MQ_PORT,
35	credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
36	)
37	)
38
39	self.channel = self.connection.channel()	2✔
40	self.exchange_name = exchange_name	2✔
41
42	self.channel.queue_declare(queue=SUB_QUEUE)	2✔
43	self._thread_queue = thread_queue	2✔
44
45	self.te_manager = te_manager	2✔
46
47	self._exit_event = threading.Event()	2✔
48
49	def on_request(self, ch, method, props, message_body):	2✔
50	response = message_body	×
51	self._thread_queue.put(message_body)	×
52
53	self.connection = pika.BlockingConnection(	×
54	pika.ConnectionParameters(
55	host=MQ_HOST,
56	port=MQ_PORT,
57	credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
58	)
59	)
60	self.channel = self.connection.channel()	×
61
62	try:	×
63	ch.basic_publish(	×
64	exchange=self.exchange_name,
65	routing_key=props.reply_to,
66	properties=pika.BasicProperties(correlation_id=props.correlation_id),
67	body=str(response),
68	)
69	ch.basic_ack(delivery_tag=method.delivery_tag)	×
70	except Exception as err:	×
71	self.logger.info(f"[MQ] encountered error when publishing: {err}")	×
72
73	def start_consumer(self):	2✔
74	self.channel.basic_qos(prefetch_count=1)	2✔
75	self.channel.basic_consume(queue=SUB_QUEUE, on_message_callback=self.on_request)	2✔
76
77	self.logger.info(" [MQ] Awaiting requests from queue: " + SUB_QUEUE)	2✔
78	self.channel.start_consuming()	2✔
79
80	def start_sdx_consumer(self, thread_queue, db_instance):	2✔
81	HEARTBEAT_ID = 0	2✔
82
83	rpc = RpcConsumer(thread_queue, "", self.te_manager)	2✔
84	t1 = threading.Thread(target=rpc.start_consumer, args=(), daemon=True)	2✔
85	t1.start()	2✔
86
87	lc_message_handler = LcMessageHandler(db_instance, self.te_manager)	2✔
88	parse_helper = ParseHelper()	2✔
89
90	latest_topo = {}	2✔
91	domain_list = []	2✔
92
93	# This part reads from DB when SDX controller initially starts.
94	# It looks for domain_list, if already in DB,
95	# Then use the existing ones from DB.
96	domain_list_from_db = db_instance.read_from_db(	2✔
97	MongoCollections.DOMAINS, Constants.DOMAIN_LIST
98	)
99	latest_topo_from_db = db_instance.read_from_db(	2✔
100	MongoCollections.TOPOLOGIES, Constants.LATEST_TOPO
101	)
102
103	if domain_list_from_db:	2✔
104	domain_list = domain_list_from_db[Constants.DOMAIN_LIST]	×
105	logger.debug("Domain list already exists in db: ")	×
106	logger.debug(domain_list)	×
107
108	if latest_topo_from_db:	2✔
NEW 109	latest_topo = latest_topo_from_db[Constants.LATEST_TOPO]	×
110	logger.debug("Topology already exists in db: ")	×
111	logger.debug(latest_topo)	×
112
113	# If topologies already saved in db, use them to initialize te_manager
114	if domain_list:	2✔
115	for domain in domain_list:	×
116	topology = db_instance.read_from_db(MongoCollections.TOPOLOGIES, domain)	×
117
118	if not topology:	×
119	continue	×
120
121	# Get the actual thing minus the Mongo ObjectID.
122	topology = topology[domain]	×
123	topo_json = json.loads(topology)	×
124	self.te_manager.add_topology(topo_json)	×
125	logger.debug(f"Read {domain}: {topology}")	×
126
127	while not self._exit_event.is_set():	2✔
128	# Queue.get() will block until there's an item in the queue.
129	msg = thread_queue.get()	2✔
130	logger.debug("MQ received message:" + str(msg))	×
131
132	if "Heart Beat" in str(msg):	×
133	HEARTBEAT_ID += 1	×
134	logger.debug("Heart beat received. ID: " + str(HEARTBEAT_ID))	×
135	continue	×
136
137	if not parse_helper.is_json(msg):	×
138	continue	×
139
140	if "version" not in str(msg):	×
141	logger.info("Got message (NO VERSION) from MQ: " + str(msg))	×
142
143	lc_message_handler.process_lc_json_msg(	×
144	msg,
145	latest_topo,
146	domain_list,
147	)
148
149	def stop_threads(self):	2✔
150	"""
151	Signal threads that we're ready to stop.
152	"""
153	logger.info("[MQ] Stopping threads.")	×
154	self.channel.stop_consuming()	×
155	self._exit_event.set()	×

atlanticwave-sdx / sdx-controller / 13395484668

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous