fix: Propagate uvicorn logs

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:
+        # 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"