8932612069

Committed 03 May 2024 01:02AM UTC coverage: 51.117% (+0.07%) from 51.046%

Build # 8932612069

Build Type

Pull #264

github

Committed by

web-flow

Commit Message

Merge 971db41d7 into 2b273a5c3

Pull Request Pull Request #264: Add end point to list all connections

Run Details

293 of 541 branches covered (54.16%)

Branch coverage included in aggregate %.

19 of 32 new or added lines in 5 files covered. (59.38%)

22 existing lines in 4 files now uncovered.

714 of 1429 relevant lines covered (49.97%)

2.0 hits per line

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

52.43

/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_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 = os.getenv("SUB_QUEUE")

logger = logging.getLogger(__name__)


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

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

        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)

    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):
        MESSAGE_ID = 0
        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 = []
        num_domain_topos = 0

        # This part reads from DB when SDX controller initially starts.
        # It looks for domain_list, and num_domain_topos, if they are already in DB,
        # Then use the existing ones from DB.
        domain_list_from_db = db_instance.read_from_db("domains", "domain_list")
        latest_topo_from_db = db_instance.read_from_db("topologies", "latest_topo")
        num_domain_topos_from_db = db_instance.read_from_db("topologies", "num_domain_topos")

        if domain_list_from_db:
            domain_list = domain_list_from_db["domain_list"]
            logger.debug("Read domain_list from db: ")
            logger.debug(domain_list)

        if latest_topo_from_db:
            latest_topo = latest_topo_from_db["latest_topo"]
            logger.debug("Read latest_topo from db: ")
            logger.debug(latest_topo)

        if num_domain_topos_from_db:
            num_domain_topos = num_domain_topos_from_db["num_domain_topos"]
            logger.debug("Read num_domain_topos from db: ")
            logger.debug(num_domain_topos)
            for topo in range(1, num_domain_topos + 2):
                db_key = f"LC-{topo}"
                topology = db_instance.read_from_db("topologies", db_key)

                if topology:
                    # Get the actual thing minus the Mongo ObjectID.
                    topology = topology[db_key]
                    topo_json = json.loads(topology)
                    self.te_manager.add_topology(topo_json)
                    logger.debug(f"Read {db_key}: {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))
            else:
                logger.info("Saving to database.")
                if parse_helper.is_json(msg):
                    if "version" in str(msg):
                        lc_message_handler.process_lc_json_msg(
                            msg,
                            latest_topo,
                            domain_list,
                            num_domain_topos,
                        )
                    else:
                        logger.info("got message from MQ: " + str(msg))
                

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

atlanticwave-sdx / sdx-controller / 8932612069

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