fix: Propagate uvicorn logs (#1250)

Logs produced by Uvicorn's internals and blueapi are currently processed
by different handlers. Due to this nothing produced by Uvicorn is being
propagated to Graylog.

This PR relegates handling of Uvicorn logs to BlueAPI's handlers and
standardises terminal formatting.

Fixes #1248

---------

Co-authored-by: Joseph Ware <[email protected]>

Author: dan-fernandes

src/blueapi/log.py

@@ -1,7 +1,10 @@
+import enum
 import logging
 import os
 from contextlib import contextmanager
+from copy import copy
 
+import click
 from graypy import GELFTCPHandler
 
 from blueapi.config import LoggingConfig
@@ -91,6 +94,9 @@ def set_up_stream_handler(
     stream_handler = logging.StreamHandler()
     stream_handler.setLevel(logging_config.level)
 
+    formatter = ColorFormatter("%(asctime)s %(levelname)s %(message)s")
+    stream_handler.setFormatter(formatter)
+
     for filter in filters:
         stream_handler.addFilter(filter)
 
@@ -120,3 +126,47 @@ def set_up_graylog_handler(
 
     logger.addHandler(graylog_handler)
     return graylog_handler
+
+
+class IBMColorBlindSafeColors(enum.Enum):
+    ultramarine = (100, 143, 255)
+    indigo = (120, 94, 240)
+    magenta = (220, 38, 127)
+    orange = (254, 97, 0)
+    gold = (255, 176, 0)
+
+
+class ColorFormatter(logging.Formatter):
+    """Colors level_name of log using IBM color blind safe palette."""
+
+    def color_level_name(self, level_name: str, level_no: int) -> str:
+        match level_no:
+            case logging.DEBUG:
+                return click.style(
+                    str(level_name), fg=IBMColorBlindSafeColors.ultramarine.value
+                )
+            case logging.INFO:
+                return click.style(
+                    str(level_name), fg=IBMColorBlindSafeColors.indigo.value
+                )
+            case logging.WARNING:
+                return click.style(
+                    str(level_name), fg=IBMColorBlindSafeColors.gold.value
+                )
+            case logging.ERROR:
+                return click.style(
+                    str(level_name), fg=IBMColorBlindSafeColors.magenta.value
+                )
+            case logging.CRITICAL:
+                return click.style(
+                    str(level_name), fg=IBMColorBlindSafeColors.orange.value
+                )
+        return level_name
+
+    def formatMessage(self, record: logging.LogRecord) -> str:  # noqa: N802
+        # Copy record to avoid modifying for other handlers etc.
+        recordcopy = copy(record)
+        recordcopy.levelname = self.color_level_name(
+            recordcopy.levelname, recordcopy.levelno
+        )
+        return super().formatMessage(recordcopy)

src/blueapi/service/main.py

@@ -555,13 +555,13 @@ def start(config: ApplicationConfig):
     import uvicorn
     from uvicorn.config import LOGGING_CONFIG
 
-    LOGGING_CONFIG["formatters"]["default"]["fmt"] = (
-        "%(asctime)s %(levelprefix)s %(message)s"
-    )
+    # Enable propagation and disable default handlers, so that blueapi handles all logs
+    LOGGING_CONFIG["loggers"]["uvicorn"]["handlers"] = []
     LOGGING_CONFIG["formatters"]["access"]["fmt"] = (
         "%(asctime)s %(levelprefix)s %(client_addr)s"
         + " - '%(request_line)s' %(status_code)s"
     )
+    LOGGING_CONFIG["loggers"]["uvicorn"]["propagate"] = True
     app = get_app(config)
 
     FastAPIInstrumentor().instrument_app(

tests/unit_tests/test_log.py

@@ -7,7 +7,12 @@
 from dodal.log import LOGGER as DODAL_LOGGER
 
 from blueapi.config import GraylogConfig, LoggingConfig
-from blueapi.log import PlanTagFilter, plan_tag_filter_context, set_up_logging
+from blueapi.log import (
+    ColorFormatter,
+    PlanTagFilter,
+    plan_tag_filter_context,
+    set_up_logging,
+)
 
 
 def clear_all_loggers_and_handlers(logger):
@@ -124,3 +129,56 @@ def test_filter_tags_plan_name(logger, mock_handler_emit):
     logger.addFilter(PlanTagFilter("foo"))
     logger.info("FOO")
     assert mock_handler_emit.call_args[0][0].plan_name == "foo"
+
+
+@pytest.fixture
+def color_formatter():
+    return ColorFormatter("%(levelname)s %(message)s")
+
+
+@pytest.fixture
+def foo_record():
+    return logging.LogRecord("foo", logging.INFO, "foo", 0, "foo", None, None)
+
+
+class TestColorFormatter:
+    def test_debug(self, color_formatter, foo_record):
+        foo_record.levelno = logging.DEBUG
+
+        assert (
+            color_formatter.format(foo_record)
+            == "\x1b[38;2;100;143;255mINFO\x1b[0m foo"
+        )
+
+    def test_info(self, color_formatter, foo_record):
+        foo_record.levelno = logging.INFO
+
+        assert (
+            color_formatter.format(foo_record) == "\x1b[38;2;120;94;240mINFO\x1b[0m foo"
+        )
+
+    def test_warning(self, color_formatter, foo_record):
+        foo_record.levelno = logging.WARNING
+
+        assert (
+            color_formatter.format(foo_record) == "\x1b[38;2;255;176;0mINFO\x1b[0m foo"
+        )
+
+    def test_error(self, color_formatter, foo_record):
+        foo_record.levelno = logging.ERROR
+
+        assert (
+            color_formatter.format(foo_record) == "\x1b[38;2;220;38;127mINFO\x1b[0m foo"
+        )
+
+    def test_critical(self, color_formatter, foo_record):
+        foo_record.levelno = logging.CRITICAL
+
+        assert (
+            color_formatter.format(foo_record) == "\x1b[38;2;254;97;0mINFO\x1b[0m foo"
+        )
+
+    def test_other(self, color_formatter, foo_record):
+        foo_record.levelno = -1
+
+        assert color_formatter.format(foo_record) == "INFO foo"