+# misc cruft
*.log
+integration_tests/log.txt
NOTES.txt
rmr/*
docs_and_diagrams/
.tox
docs/_build/
+# standard python ignore template
.pytest_cache/
xunit-results.xml
.DS_Store
+"""
+contains the app; broken out here for ease of unit testing
+"""
# ==================================================================================
# Copyright (c) 2019 Nokia
# Copyright (c) 2018-2019 AT&T Intellectual Property.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================
-import logging
import connexion
-def get_module_logger(mod_name):
- """
- To use this, do logger = get_module_logger(__name__)
- """
- logger = logging.getLogger(mod_name)
- handler = logging.StreamHandler()
- formatter = logging.Formatter(
- '%(asctime)s [%(name)-12s] %(levelname)-8s %(message)s')
- handler.setFormatter(formatter)
- logger.addHandler(handler)
- logger.setLevel(logging.DEBUG)
- return logger
-
-
-app = connexion.App(__name__, specification_dir='.')
-app.add_api('openapi.yaml', arguments={'title': 'My Title'})
+app = connexion.App(__name__, specification_dir=".")
+app.add_api("openapi.yaml", arguments={"title": "My Title"})
import json
from threading import Thread
from rmr import rmr, helpers
-from a1 import get_module_logger
+from mdclogpy import Logger
from a1 import data
from a1.exceptions import PolicyTypeNotFound, PolicyInstanceNotFound
-logger = get_module_logger(__name__)
+mdc_logger = Logger(name=__name__)
RETRY_TIMES = int(os.environ.get("A1_RMR_RETRY_TIMES", 4))
if init_func_override:
self.mrc = init_func_override()
else:
- logger.debug("Waiting for rmr to initialize..")
+ 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
- clean up the database (eg delete the instance) under certain conditions based on those statuses (NOT DONE YET)
"""
# loop forever
- logger.debug("Work loop starting")
+ mdc_logger.debug("Work loop starting")
while self.keep_going:
# send out all messages waiting for us
pre_send_summary = rmr.message_summary(sbuf)
sbuf = rmr.rmr_send_msg(self.mrc, sbuf) # send
post_send_summary = rmr.message_summary(sbuf)
- logger.debug("Pre-send summary: %s, Post-send summary: %s", pre_send_summary, post_send_summary)
+ mdc_logger.debug(
+ "Pre-send summary: {0}, Post-send summary: {1}".format(pre_send_summary, post_send_summary)
+ )
rmr.rmr_free_msg(sbuf) # free
if post_send_summary["message state"] == 0 and post_send_summary["message status"] == "RMR_OK":
- logger.debug("Message sent successfully!")
+ mdc_logger.debug("Message sent successfully!")
break
# read our mailbox and update statuses
pti = pay["policy_type_id"]
pii = pay["policy_instance_id"]
data.set_policy_instance_status(pti, pii, pay["handler_id"], pay["status"])
- except (PolicyTypeNotFound, PolicyInstanceNotFound, KeyError, json.decoder.JSONDecodeError):
+ except (PolicyTypeNotFound, PolicyInstanceNotFound, KeyError, TypeError, json.decoder.JSONDecodeError):
# TODO: in the future we may also have to catch SDL errors
- logger.debug(("Dropping malformed or non applicable message", msg))
+ mdc_logger.debug("Dropping malformed or non applicable message: {0}".format(msg))
# TODO: what's a reasonable sleep time? we don't want to hammer redis too much, and a1 isn't a real time component
self.last_ran = time.time()
from jsonschema import validate
from jsonschema.exceptions import ValidationError
import connexion
-from a1 import get_module_logger
+from mdclogpy import Logger
from a1 import a1rmr, exceptions, data
-logger = get_module_logger(__name__)
+mdc_logger = Logger(name=__name__)
def _try_func_return(func):
return "", 404
except BaseException as exc:
# catch all, should never happen...
- logger.exception(exc)
+ mdc_logger.exception(exc)
return Response(status=500)
+"""
+Represents A1s database and database access functions.
+In the future, this may change to use a different backend, possibly dramatically.
+Hopefully, the access functions are a good api so nothing else has to change when this happens
+
+For now, the database is in memory.
+We use dict data structures (KV) with the expectation of having to move this into Redis
+"""
# ==================================================================================
# Copyright (c) 2019 Nokia
# Copyright (c) 2018-2019 AT&T Intellectual Property.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================
-
-"""
-Represents A1s database and database access functions.
-In the future, this may change to use a different backend, possibly dramatically.
-Hopefully, the access functions are a good api so nothing else has to change when this happens
-
-For now, the database is in memory.
-We use dict data structures (KV) with the expectation of having to move this into Redis
-"""
import os
import time
from threading import Thread
import msgpack
+from mdclogpy import Logger
from a1.exceptions import PolicyTypeNotFound, PolicyInstanceNotFound, PolicyTypeAlreadyExists, CantDeleteNonEmptyType
-from a1 import get_module_logger
-logger = get_module_logger(__name__)
+mdc_logger = Logger(name=__name__)
INSTANCE_DELETE_NO_RESP_TTL = int(os.environ.get("INSTANCE_DELETE_NO_RESP_TTL", 5))
_clear_handlers(policy_type_id, policy_instance_id) # delete all the handlers
SDL.delete(_generate_instance_key(policy_type_id, policy_instance_id)) # delete instance
SDL.delete(_generate_instance_metadata_key(policy_type_id, policy_instance_id)) # delete instance metadata
- logger.debug("type %s instance %s deleted", policy_type_id, policy_instance_id)
- raise PolicyInstanceNotFound()
+ mdc_logger.debug("type {0} instance {1} deleted".format(policy_type_id, policy_instance_id))
# Types
# limitations under the License.
# ==================================================================================
from gevent.pywsgi import WSGIServer
-from a1 import get_module_logger, app
+from mdclogpy import Logger
+from a1 import app
from a1 import a1rmr
-logger = get_module_logger(__name__)
+mdc_logger = Logger(name=__name__)
def main():
"""Entrypoint"""
# start rmr thread
- logger.debug("Initializing rmr thread. A1s webserver will not start until rmr initialization is complete.")
+ mdc_logger.debug("Initializing rmr thread. A1s webserver will not start until rmr initialization is complete.")
a1rmr.start_rmr_thread()
# start webserver
- logger.debug("Starting gevent server")
+ mdc_logger.debug("Starting gevent server")
http_server = WSGIServer(("", 10000), app)
http_server.serve_forever()
* Changes the API for ../status to return a richer structure
* Clean up unused items in the integration tests helm chart
* Removed "RMR_RCV_RETRY_INTERVAL" leftovers since this isn't used anymore
+ * Uses the standard RIC logging library
[1.0.4]
-------
#!/bin/sh
-kubectl get pods --namespace=default | awk '{ print $1 }' | egrep '^a1-a1mediator-' | xargs kubectl logs
-kubectl get pods --namespace=default | awk '{ print $1 }' | egrep '^testreceiver-' | xargs -I X kubectl logs X testreceiver
-kubectl get pods --namespace=default | awk '{ print $1 }' | egrep '^testreceiver-' | xargs -I X kubectl logs X delayreceiver
-
+kubectl get pods --namespace=default | awk '{ print $1 }' | egrep '^a1-a1mediator-' | xargs kubectl logs > log.txt 2>&1
+kubectl get pods --namespace=default | awk '{ print $1 }' | egrep '^testreceiver-' | xargs -I X kubectl logs X testreceiver >> log.txt 2>&1
+kubectl get pods --namespace=default | awk '{ print $1 }' | egrep '^testreceiver-' | xargs -I X kubectl logs X delayreceiver >> log.txt 2>&1
url="https://gerrit.o-ran-sc.org/r/admin/repos/ric-plt/a1",
entry_points={"console_scripts": ["run.py=a1.run:main"]},
# we require jsonschema, should be in that list, but connexion already requires a specific version of it
- install_requires=["requests", "Flask", "connexion[swagger-ui]", "gevent", "msgpack", "rmr>=1.0.0"],
+ install_requires=["requests", "Flask", "connexion[swagger-ui]", "gevent", "msgpack", "rmr>=1.0.0", "mdclogpy"],
package_data={"a1": ["openapi.yaml"]},
)
new_msgs.append(fake_msg)
# insert a bad one with a malformed body to make sure we keep going
- fake_msg = {"payload": "asdf"}
- new_msgs.append(fake_msg)
+ new_msgs.append({"payload": "asdf"})
+
+ # not even a json
+ new_msgs.append("asdf")
pay = b'{"policy_type_id": 20000, "policy_instance_id": "admission_control_policy", "handler_id": "test_receiver", "status": "DELETED"}'
fake_msg = {"payload": pay}
# run apache bench
ab -n 100 -c 10 -v 4 http://localhost:10000/a1-p/healthcheck
commands_post=
-# echo "log collection"
-# integration_tests/getlogs.sh
+ echo "log collection"
+ integration_tests/getlogs.sh
echo "teardown"
helm delete testreceiver
helm del --purge testreceiver
Code Review / ric-plt / a1.git / commitdiff