basic json log formatter

atst/utils/logging.py

@@ -0,0 +1,38 @@
+import datetime
+import json
+import logging
+
+
+class ContextFilter(logging.Filter):
+    # this should impart the request_id and user_id if available
+    pass
+
+
+def epoch_to_iso8601(ts):
+    dt = datetime.datetime.utcfromtimestamp(ts)
+    return dt.replace(tzinfo=datetime.timezone.utc).isoformat()
+
+
+class JsonFormatter(logging.Formatter):
+    _DEFAULT_RECORD_FIELDS = [
+        ("timestamp", lambda r: epoch_to_iso8601(r.created)),
+        ("version", lambda r: 1),
+        ("request_id", lambda r: r.__dict__.get("request_id")),
+        ("user_id", lambda r: r.__dict__.get("user_id")),
+        ("severity", lambda r: r.levelname),
+        ("tags", lambda r: r.__dict__.get("tags")),
+        ("message", lambda r: r.msg),
+    ]
+
+    def format(self, record):
+        message_dict = {}
+        for field, func in self._DEFAULT_RECORD_FIELDS:
+            message_dict[field] = func(record)
+
+        if record.__dict__.get("exc_info") is not None:
+            message_dict["details"] = {
+                "backtrace": self.formatException(record.exc_info),
+                "exception": str(record.exc_info[1]),
+            }
+
+        return json.dumps(message_dict)

tests/utils/test_logging.py

@@ -0,0 +1,57 @@
+from io import StringIO
+import json
+import logging
+
+import pytest
+
+from atst.utils.logging import JsonFormatter
+
+
+@pytest.fixture
+def log_stream():
+    return StringIO()
+
+
+@pytest.fixture
+def log_stream_content(log_stream):
+    def _log_stream_content():
+        log_stream.seek(0)
+        return log_stream.read()
+
+    return _log_stream_content
+
+
+@pytest.fixture
+def logger(log_stream):
+    logger = logging.getLogger()
+    for handler in logger.handlers:
+        logger.removeHandler(handler)
+
+    logHandler = logging.StreamHandler(log_stream)
+    formatter = JsonFormatter()
+    logHandler.setFormatter(formatter)
+    logger.addHandler(logHandler)
+
+    return logger
+
+
+def test_json_formatter(logger, log_stream_content):
+    logger.warning("do or do not", extra={"tags": ["wisdom", "jedi"]})
+
+    log = json.loads(log_stream_content())
+
+    assert log["tags"] == ["wisdom", "jedi"]
+    assert log["message"] == "do or do not"
+    assert log["severity"] == "WARNING"
+    assert log.get("details") is None
+
+
+def test_json_formatter_for_exceptions(logger, log_stream_content):
+    try:
+        raise Exception()
+    except Exception:
+        logger.exception("you found the ventilation shaft!")
+
+    log = json.loads(log_stream_content())
+    assert log["severity"] == "ERROR"
+    assert log.get("details")