+"""
+a1s rmr functionality
+"""
# ==================================================================================
-# Copyright (c) 2019 Nokia
-# Copyright (c) 2018-2019 AT&T Intellectual Property.
+# Copyright (c) 2019-2020 Nokia
+# Copyright (c) 2018-2020 AT&T Intellectual Property.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# limitations under the License.
# ==================================================================================
import os
-import gevent
-from rmr import rmr
-from a1 import get_module_logger
-from a1.exceptions import MessageSendFailure, ExpectedAckNotReceived
+import queue
+import time
+import json
+from threading import Thread
+from ricxappframe.rmr import rmr, helpers
+from mdclogpy import Logger
+from a1 import data, messages
+from a1.exceptions import PolicyTypeNotFound, PolicyInstanceNotFound
-logger = get_module_logger(__name__)
+mdc_logger = Logger(name=__name__)
-RMR_RCV_RETRY_INTERVAL = int(os.environ.get("RMR_RCV_RETRY_INTERVAL", 1000))
-RETRY_TIMES = int(os.environ.get("RMR_RETRY_TIMES", 4))
-MRC = None
+RETRY_TIMES = int(os.environ.get("A1_RMR_RETRY_TIMES", 4))
+A1_POLICY_REQUEST = 20010
+A1_POLICY_RESPONSE = 20011
+A1_POLICY_QUERY = 20012
-RECEIVED_MESSAGES = [] # used to store messages we need but havent been procedded yet
-WAITING_TRANSIDS = {} # used to store transactionids we are waiting for, so we can filter other stuff out
+# Note; yes, globals are bad, but this is a private (to this module) global
+# No other module can import/access this (well, python doesn't enforce this, but all linters will complain)
+__RMR_LOOP__ = None
-def _dequeue_all_waiting_messages():
+class _RmrLoop:
"""
- dequeue all waiting rmr messages from rmr, put them into RECEIVED_MESSAGES
+ class represents an rmr loop that constantly reads from rmr and performs operations based on waiting messages
+ this launches a thread, it should probably only be called once; the public facing method to access these ensures this
+
+ TODO: the xapp frame has a version of this looping structure. See if A1 can switch to that.
"""
- new_messages = []
- sbuf = rmr.rmr_alloc_msg(MRC, 4096)
- while True:
- sbuf = rmr.rmr_torcv_msg(MRC, sbuf, 0) # set the timeout to 0 so this doesn't block!!
- summary = rmr.message_summary(sbuf)
- if summary["message state"] == 12 and summary["message status"] == "RMR_ERR_TIMEOUT":
- break
- elif summary["transaction id"] in WAITING_TRANSIDS: # message is relevent
- new_messages.append(summary)
- else:
- logger.debug("A message was received by a1, but a1 was not expecting it! It's being dropped: %s", summary)
- # do nothing with message, effectively dropped
- return new_messages
+ def __init__(self, init_func_override=None, rcv_func_override=None):
+ self.keep_going = True
+ self.rcv_func = None
+ self.last_ran = time.time()
-def _check_if_ack_received(target_transid, target_type):
- """
- Try to recieve the latest messages, then search the current queue for the target ACK
- TODO: probably a slightly more efficient data structure than list. Maybe a dict by message type
- However, in the near term, where there are not many xapps under A1, this is fine. Revisit later.
- TODO: do we need to deal with duplicate ACKs for the same transaction id?
- Is it possible if the downstream xapp uses rmr_rts? Might be harmless to sit in queue.. might slow things
+ # see docs/overview#resiliency for a discussion of this
+ self.instance_send_queue = queue.Queue() # thread safe queue https://docs.python.org/3/library/queue.html
+ # intialize rmr context
+ if init_func_override:
+ self.mrc = init_func_override()
+ else:
+ mdc_logger.debug("Waiting for rmr to initialize..")
+ # rmr.RMRFL_MTCALL puts RMR into a multithreaded mode, where a receiving thread populates an
+ # internal ring of messages, and receive calls read from that
+ # currently the size is 2048 messages, so this is fine for the foreseeable future
+ self.mrc = rmr.rmr_init(b"4562", rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
+ while rmr.rmr_ready(self.mrc) == 0:
+ time.sleep(0.5)
+
+ # set the receive function
+ self.rcv_func = (
+ rcv_func_override
+ if rcv_func_override
+ else lambda: helpers.rmr_rcvall_msgs_raw(self.mrc, [A1_POLICY_RESPONSE, A1_POLICY_QUERY])
+ )
+
+ # start the work loop
+ self.thread = Thread(target=self.loop)
+ self.thread.start()
+
+ def _assert_good_send(self, sbuf, pre_send_summary):
+ """
+ common helper function for _send_msg and _rts_msg
+ """
+ post_send_summary = rmr.message_summary(sbuf)
+ if post_send_summary["message state"] == 0 and post_send_summary["message status"] == "RMR_OK":
+ return True
+ mdc_logger.debug("Message NOT sent!")
+ mdc_logger.debug("Pre-send summary: {0}, Post-send summary: {1}".format(pre_send_summary, post_send_summary))
+ return False
+
+ def _send_msg(self, pay, mtype, subid):
+ """
+ sends a msg
+ """
+ for _ in range(0, RETRY_TIMES):
+ sbuf = rmr.rmr_alloc_msg(self.mrc, len(pay), payload=pay, gen_transaction_id=True, mtype=mtype, sub_id=subid)
+ sbuf.contents.sub_id = subid
+ pre_send_summary = rmr.message_summary(sbuf)
+ mdc_logger.debug("Trying to send message: {}".format(pre_send_summary))
+ sbuf = rmr.rmr_send_msg(self.mrc, sbuf) # send
+ if self._assert_good_send(sbuf, pre_send_summary):
+ rmr.rmr_free_msg(sbuf) # free
+ return
+
+ mdc_logger.debug("A1 did NOT send the message successfully after {} retries!".format(RETRY_TIMES))
+
+ def _rts_msg(self, pay, sbuf_rts, mtype):
+ """
+ sends a message using rts
+ we do not call free here because we may rts many times; it is called after the rts loop
+ """
+ for _ in range(0, RETRY_TIMES):
+ pre_send_summary = rmr.message_summary(sbuf_rts)
+ sbuf_rts = rmr.rmr_rts_msg(self.mrc, sbuf_rts, payload=pay, mtype=mtype)
+ if self._assert_good_send(sbuf_rts, pre_send_summary):
+ break
+ return sbuf_rts # in some cases rts may return a new sbuf
+
+ def _handle_sends(self):
+ # send out all messages waiting for us
+ while not self.instance_send_queue.empty():
+ work_item = self.instance_send_queue.get(block=False, timeout=None)
+ payload = json.dumps(messages.a1_to_handler(*work_item)).encode("utf-8")
+ self._send_msg(payload, A1_POLICY_REQUEST, work_item[1])
+
+ def loop(self):
+ """
+ This loop runs forever, and has 3 jobs:
+ - send out any messages that have to go out (create instance, delete instance)
+ - read a1s mailbox and update the status of all instances based on acks from downstream policy handlers
+ - clean up the database (eg delete the instance) under certain conditions based on those statuses (NOT DONE YET)
+ """
+ # loop forever
+ mdc_logger.debug("Work loop starting")
+ while self.keep_going:
+
+ # Update 3/20/2020
+ # We now handle our sends in a thread (that will just exit when it's done) because there is a difference between how send works in SI95 vs NNG.
+ # Send_msg via NNG formerly never blocked.
+ # However under SI95 this send may block for some arbitrary period of time on the first send to an endpoint for which a connection is not established
+ # If this send takes too long, this loop blocks, and the healthcheck will fail, which will cause A1s healthcheck to fail, which will cause Kubernetes to whack A1 and all kinds of horrible things happen.
+ # Therefore, now under SI95, we thread this.
+ Thread(target=self._handle_sends).start()
+
+ # read our mailbox
+ for (msg, sbuf) in self.rcv_func():
+ # TODO: in the future we may also have to catch SDL errors
+ try:
+ mtype = msg["message type"]
+ except (KeyError, TypeError, json.decoder.JSONDecodeError):
+ mdc_logger.debug("Dropping malformed policy ack/query message: {0}".format(msg))
+
+ if mtype == A1_POLICY_RESPONSE:
+ try:
+ # got a policy response, update status
+ pay = json.loads(msg["payload"])
+ data.set_policy_instance_status(
+ pay["policy_type_id"], pay["policy_instance_id"], pay["handler_id"], pay["status"]
+ )
+ mdc_logger.debug("Successfully received status update: {0}".format(pay))
+ except (PolicyTypeNotFound, PolicyInstanceNotFound):
+ mdc_logger.debug("Received a response for a non-existent instance")
+ except (KeyError, TypeError, json.decoder.JSONDecodeError):
+ mdc_logger.debug("Dropping malformed policy ack message: {0}".format(msg))
+
+ elif mtype == A1_POLICY_QUERY:
+ try:
+ # got a query, do a lookup and send out all instances
+ pti = json.loads(msg["payload"])["policy_type_id"]
+ instance_list = data.get_instance_list(pti) # will raise if a bad type
+ mdc_logger.debug("Received a query for a good type: {0}".format(msg))
+ for pii in instance_list:
+ instance = data.get_policy_instance(pti, pii)
+ payload = json.dumps(messages.a1_to_handler("CREATE", pti, pii, instance)).encode("utf-8")
+ sbuf = self._rts_msg(payload, sbuf, A1_POLICY_REQUEST)
+ except (PolicyTypeNotFound):
+ mdc_logger.debug("Received a query for a non-existent type: {0}".format(msg))
+ except (KeyError, TypeError, json.decoder.JSONDecodeError):
+ mdc_logger.debug("Dropping malformed policy query message: {0}".format(msg))
+
+ else:
+ mdc_logger.debug("Received message type {0} but A1 does not handle this".format(mtype))
+
+ # we must free each sbuf
+ rmr.rmr_free_msg(sbuf)
+
+ self.last_ran = time.time()
+ time.sleep(1)
+
+ mdc_logger.debug("RMR Thread Ending!")
+
+
+# Public
+
+
+def start_rmr_thread(init_func_override=None, rcv_func_override=None):
"""
- new_messages = _dequeue_all_waiting_messages() # dequeue all waiting messages
- global RECEIVED_MESSAGES # this is ugly, but fine.. we just need an in memory list across the async calls
- RECEIVED_MESSAGES += new_messages
- for index, summary in enumerate(RECEIVED_MESSAGES): # Search the queue for the target message
- if (
- summary["message state"] == 0
- and summary["message status"] == "RMR_OK"
- and summary["message type"] == target_type
- and summary["transaction id"] == target_transid
- ): # Found; delete it from queue
- del RECEIVED_MESSAGES[index]
- return summary
- return None
-
-
-def init_rmr():
- """
- called from run; not called for unit tests
+ Start a1s rmr thread
"""
- global MRC
- MRC = rmr.rmr_init(b"4562", rmr.RMR_MAX_RCV_BYTES, 0x00)
-
- while rmr.rmr_ready(MRC) == 0:
- gevent.sleep(1)
- logger.debug("not yet ready")
+ global __RMR_LOOP__
+ if __RMR_LOOP__ is None:
+ __RMR_LOOP__ = _RmrLoop(init_func_override, rcv_func_override)
-def send(payload, message_type=0):
+def stop_rmr_thread():
"""
- sends a message up to RETRY_TIMES
- If the message is sent successfully, it returns the transactionid
- Raises an exception (MessageSendFailure) otherwise
+ stops the rmr thread
"""
- # we may be called many times in asyncronous loops, so for now, it is safer not to share buffers. We can investifgate later whether this is really a problem.
- sbuf = rmr.rmr_alloc_msg(MRC, 4096)
- payload = payload if isinstance(payload, bytes) else payload.encode("utf-8")
-
- # retry RETRY_TIMES to send the message
- tried = 0
- while True:
- # setup the send message
- rmr.set_payload_and_length(payload, sbuf)
- rmr.generate_and_set_transaction_id(sbuf)
- sbuf.contents.state = 0
- sbuf.contents.mtype = message_type
- pre_send_summary = rmr.message_summary(sbuf)
- logger.debug("Pre message send summary: %s", pre_send_summary)
- transaction_id = pre_send_summary["transaction id"] # save the transactionid because we need it later
-
- # send
- sbuf = rmr.rmr_send_msg(MRC, sbuf)
- post_send_summary = rmr.message_summary(sbuf)
- logger.debug("Post message send summary: %s", rmr.message_summary(sbuf))
-
- # check success or failure
- if post_send_summary["message state"] == 0 and post_send_summary["message status"] == "RMR_OK":
- return transaction_id # we are good
- if post_send_summary["message state"] == 10 and post_send_summary["message status"] == "RMR_ERR_RETRY":
- # in this state, we should retry
- if tried == RETRY_TIMES:
- # we have tried RETRY_TIMES and we are still not getting a good state, raise an exception and let the caller deal with it
- raise MessageSendFailure(str(post_send_summary))
- else:
- tried += 1
- else:
- # we hit a state where we should not even retry
- raise MessageSendFailure(str(post_send_summary))
+ __RMR_LOOP__.keep_going = False
-def send_ack_retry(payload, expected_ack_message_type, message_type=0):
+def queue_instance_send(item):
"""
- send a message and check for an ACK.
- If no ACK is recieved, defer execution for RMR_RCV_RETRY_INTERVAL ms, then check again.
- If no ack is received before the timeout (set by _rmr_init), send again and try again up to RETRY_TIMES
+ push an item into the work queue
+ currently the only type of work is to send out messages
+ """
+ __RMR_LOOP__.instance_send_queue.put(item)
- It is critical here to set the RMR_TIMEOUT to 0 in the rmr_rcv_to function, which causes that function NOT to block.
- Instead, if the message isn't there, we give up execution for the interval, which allows the gevent server to process other requests in the meantime.
- Amazing props to https://sdiehl.github.io/gevent-tutorial/
- (which also runs this whole server)
+def healthcheck_rmr_thread(seconds=30):
+ """
+ returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
+ 1. is it running?,
+ 2. is it stuck in a long (> seconds) loop?
"""
+ return __RMR_LOOP__.thread.is_alive() and ((time.time() - __RMR_LOOP__.last_ran) < seconds)
- # try to send the msg to the downstream policy handler
- expected_transaction_id = send(payload, message_type)
- WAITING_TRANSIDS[expected_transaction_id] = 1
-
- gevent.sleep(0.01) # wait 10ms before we try the first recieve
- for _ in range(0, RETRY_TIMES):
- logger.debug("Seeing if return message is fufilled")
- summary = _check_if_ack_received(expected_transaction_id, expected_ack_message_type)
- if summary:
- logger.debug("Target ack Message received!: %s", summary)
- logger.debug("current queue size is %d", len(RECEIVED_MESSAGES))
- del WAITING_TRANSIDS[expected_transaction_id]
- return summary["payload"]
- else:
- logger.debug("Deffering execution for %s seconds", str(RMR_RCV_RETRY_INTERVAL / 1000))
- gevent.sleep(RMR_RCV_RETRY_INTERVAL / 1000)
- # we still didn't get the ACK we want
- raise ExpectedAckNotReceived()
+def replace_rcv_func(rcv_func):
+ """purely for the ease of unit testing to test different rcv scenarios"""
+ __RMR_LOOP__.rcv_func = rcv_func
Code Review / ric-plt / a1.git / blobdiff