Merge pull request #49 from minvws/log

Added single stream logging

Author: jaytaph

app.conf.example

@@ -13,15 +13,14 @@ aes_key_id = 12345678
 aes_mechanism = AES_CBC
 
 [logging]
-# Syslog addresses in host:port format (UDP)
-# Application logs go to stdout and to handler
-app_path = syslog:5514
-# Siem is for security events
-siem_path = syslog:5515
-# Logs for public inspection
-public_inspect_path = syslog:5516
-# Logs for debugging purposes
-debug_path = syslog:5517
+# All keys are optional. When syslog_path is omitted, logs only go to stdout.
+# Single syslog channel in host:port format (UDP). All streams (app, siem,
+# public_inspect, debug) are sent here; each JSON record carries a stream_id
+# field so the log server can tell them apart.
+syslog_path = syslog:5514
+# Identifies this application in the JSON records (application_id field),
+# so the log server can tell apart applications sharing the channel.
+application_id = nationale-verwijsindex-crypto-service-api
 # Include exception/stack traces in JSON logs
 include_traces = True
 # Whether to display debug logs in the console

app/config.py

@@ -33,10 +33,8 @@ class ConfigApp(BaseModel):
 
 
 class ConfigLogging(BaseModel):
-    app_path: str | None = Field(default=None)
-    siem_path: str | None = Field(default=None)
-    public_inspect_path: str | None = Field(default=None)
-    debug_path: str | None = Field(default=None)
+    syslog_path: str | None = Field(default=None)
+    application_id: str | None = Field(default=None)
     include_traces: bool = Field(default=True)
     debug_logs_in_console: bool = Field(default=False)
 

app/logging/config_builder.py

@@ -63,6 +63,29 @@ def build(self) -> dict[str, Any]:
                     "()": JsonFormatter,
                     "include_traces": True,
                 },
+                # Stream-bound formatters stamp the record with a stream_id so the
+                # log server can tell the streams apart on the shared syslog channel.
+                # APP == "stroom 2", SIEM == "stroom 3".
+                "json_app": {
+                    "()": JsonFormatter,
+                    "include_traces": False,
+                    "stream_id": "app",
+                },
+                "json_siem": {
+                    "()": JsonFormatter,
+                    "include_traces": False,
+                    "stream_id": "siem",
+                },
+                "json_public_inspect": {
+                    "()": JsonFormatter,
+                    "include_traces": False,
+                    "stream_id": "public_inspect",
+                },
+                "json_debug": {
+                    "()": JsonFormatter,
+                    "include_traces": True,
+                    "stream_id": "debug",
+                },
                 "plain": {
                     "()": PlainTextFormatter,
                 },
@@ -96,40 +119,43 @@ def build(self) -> dict[str, Any]:
             "root": {"handlers": ["console"], "level": self.loglevel},
         }
 
+        # Stamp every JSON record with the configured application id so the
+        # log server can tell apart applications sharing the syslog channel.
+        if self.logging_config.application_id:
+            for formatter in conf["formatters"].values():
+                if formatter["()"] is JsonFormatter:
+                    formatter["application_id"] = self.logging_config.application_id
+
         self._add_log_handlers(conf)
 
         return conf
 
     def _add_log_handlers(self, conf: dict[str, Any]) -> None:
+        # All streams share one syslog channel; each handler stamps its records
+        # with a stream_id via its formatter so the log server can split them.
+        path = self.logging_config.syslog_path
+        if not path:
+            return
+
         app_logger_handlers = conf["loggers"]["app"]["handlers"]
         uvicorn_handlers = conf["loggers"]["uvicorn"]["handlers"]
         uvicorn_error_handlers = conf["loggers"]["uvicorn.error"]["handlers"]
 
-        if self.logging_config.app_path:
-            conf["handlers"]["app_syslog"] = self._syslog_handler(
-                self.logging_config.app_path, filters=["app_filter"]
-            )
-            app_logger_handlers.append("app_syslog")
-            uvicorn_handlers.append("app_syslog")
-            uvicorn_error_handlers.append("app_syslog")
+        conf["handlers"]["syslog_app"] = self._syslog_handler(path, formatter="json_app", filters=["app_filter"])
+        app_logger_handlers.append("syslog_app")
+        uvicorn_handlers.append("syslog_app")
+        uvicorn_error_handlers.append("syslog_app")
 
-        if self.logging_config.siem_path:
-            conf["handlers"]["siem"] = self._syslog_handler(
-                self.logging_config.siem_path, filters=["siem_filter"]
-            )
-            app_logger_handlers.append("siem")
+        conf["handlers"]["syslog_siem"] = self._syslog_handler(path, formatter="json_siem", filters=["siem_filter"])
+        app_logger_handlers.append("syslog_siem")
 
-        if self.logging_config.public_inspect_path:
-            conf["handlers"]["public_inspect"] = self._syslog_handler(
-                self.logging_config.public_inspect_path, filters=["public_inspect_filter"]
-            )
-            app_logger_handlers.append("public_inspect")
+        conf["handlers"]["syslog_public_inspect"] = self._syslog_handler(
+            path, formatter="json_public_inspect", filters=["public_inspect_filter"]
+        )
+        app_logger_handlers.append("syslog_public_inspect")
 
-        if self.logging_config.debug_path:
-            conf["handlers"]["debug"] = self._syslog_handler(
-                self.logging_config.debug_path, formatter="json_traces"
-            )
-            app_logger_handlers.append("debug")
-            uvicorn_handlers.append("debug")
-            uvicorn_error_handlers.append("debug")
-            conf["root"]["handlers"].append("debug")
+        conf["handlers"]["syslog_debug"] = self._syslog_handler(path, formatter="json_debug")
+        app_logger_handlers.append("syslog_debug")
+        uvicorn_handlers.append("syslog_debug")
+        uvicorn_error_handlers.append("syslog_debug")
+        conf["root"]["handlers"].append("syslog_debug")

app/logging/formatter.py

@@ -39,8 +39,14 @@ def _collect_extras(record: logging.LogRecord) -> dict[str, Any]:
 class JsonFormatter(logging.Formatter):
     """Structured JSON formatter for the debug-json view.
 
+    All streams share a single syslog channel; ``stream_id`` tells the log
+    server which stream (app, siem, public_inspect, debug) a record belongs
+    to, and ``application_id`` which application emitted it.
+
     Example output:
         {
+            "application_id": "nationale-verwijsindex-crypto-service-api",
+            "stream_id": "app",
             "event_id": "100601",
             "timestamp": "2026-04-23T10:11:12Z",
             "level": "INFO",
@@ -54,9 +60,16 @@ class JsonFormatter(logging.Formatter):
             }
         }
     """
-    def __init__(self, include_traces: bool = True) -> None:
+    def __init__(
+        self,
+        include_traces: bool = True,
+        stream_id: str | None = None,
+        application_id: str | None = None,
+    ) -> None:
         super().__init__()
         self.include_traces = include_traces
+        self.stream_id = stream_id
+        self.application_id = application_id
 
     def format(self, record: logging.LogRecord) -> str:
         message: dict[str, Any] = {}
@@ -76,6 +89,10 @@ def format(self, record: logging.LogRecord) -> str:
             "event_description": _sanitize_message(record.getMessage()),
             "source": f"{record.module}:{record.lineno}",
         }
+        if self.application_id is not None:
+            log_record["application_id"] = self.application_id
+        if self.stream_id is not None:
+            log_record["stream_id"] = self.stream_id
         log_record["message"] = message
 
         return json.dumps(log_record, default=str)

tests/unit/logging/test_config_builder.py

@@ -46,38 +46,59 @@ def test_build_console_uses_json_when_traces_excluded() -> None:
     assert conf["handlers"]["console"]["formatter"] == "json"
 
 
+_SYSLOG_HANDLERS = ("syslog_app", "syslog_siem", "syslog_public_inspect", "syslog_debug")
+
+
 @pytest.mark.parametrize(
-    "field,handler_name,filter_name",
+    "handler_name,filter_name",
     [
-        ("app_path", "app_syslog", "app_filter"),
-        ("siem_path", "siem", "siem_filter"),
-        ("public_inspect_path", "public_inspect", "public_inspect_filter"),
+        ("syslog_app", "app_filter"),
+        ("syslog_siem", "siem_filter"),
+        ("syslog_public_inspect", "public_inspect_filter"),
     ],
 )
-def test_build_path_adds_handler_with_filter(field: str, handler_name: str, filter_name: str) -> None:
-    conf = _build(**{field: "host:514"})
+def test_build_syslog_path_adds_handler_with_filter(handler_name: str, filter_name: str) -> None:
+    conf = _build(syslog_path="host:514")
     assert handler_name in conf["handlers"]
     handler = conf["handlers"][handler_name]
     assert handler["address"] == ("host", 514)
     assert filter_name in handler["filters"]
 
 
-def test_build_debug_path_adds_handler_to_root_and_app() -> None:
-    conf = _build(debug_path="host:516")
-    assert "debug" in conf["handlers"]
-    assert "debug" in conf["loggers"]["app"]["handlers"]
-    assert "debug" in conf["root"]["handlers"]
-    assert conf["handlers"]["debug"]["formatter"] == "json_traces"
+def test_build_syslog_debug_handler_added_to_root_and_app() -> None:
+    conf = _build(syslog_path="host:516")
+    assert "syslog_debug" in conf["handlers"]
+    assert "syslog_debug" in conf["loggers"]["app"]["handlers"]
+    assert "syslog_debug" in conf["root"]["handlers"]
+    assert conf["handlers"]["syslog_debug"]["formatter"] == "json_debug"
+
+
+def test_all_streams_share_the_single_syslog_channel() -> None:
+    conf = _build(syslog_path="logserver:514")
+
+    assert {"console", *_SYSLOG_HANDLERS} == set(conf["handlers"].keys())
+    for name in _SYSLOG_HANDLERS:
+        assert conf["handlers"][name]["address"] == ("logserver", 514)
+
+    formatters = {conf["handlers"][name]["formatter"] for name in _SYSLOG_HANDLERS}
+    stream_ids = {conf["formatters"][formatter].get("stream_id") for formatter in formatters}
+    assert stream_ids == {"app", "siem", "public_inspect", "debug"}
+
+
+def test_application_id_is_stamped_on_all_json_formatters() -> None:
+    conf = _build(syslog_path="logserver:514", application_id="nationale-verwijsindex-crypto-service-api")
+
+    for name, formatter in conf["formatters"].items():
+        if name == "plain":
+            assert "application_id" not in formatter
+        else:
+            assert formatter["application_id"] == "nationale-verwijsindex-crypto-service-api"
 
 
-def test_build_all_paths_configured() -> None:
-    conf = _build(
-        app_path="h:514",
-        siem_path="h:515",
-        public_inspect_path="h:516",
-        debug_path="h:517",
-    )
-    assert {"console", "app_syslog", "siem", "public_inspect", "debug"} == set(conf["handlers"].keys())
+def test_no_application_id_without_config() -> None:
+    conf = _build(syslog_path="logserver:514")
+    for formatter in conf["formatters"].values():
+        assert "application_id" not in formatter
 
 
 def test_syslog_handler_parses_host_port_and_filters() -> None:

tests/unit/logging/test_formatter.py

@@ -93,3 +93,22 @@ def test_plaintext_formatter_includes_basic_fields(context_vars: None) -> None:
 def test_plaintext_formatter_appends_exception() -> None:
     out = PlainTextFormatter().format(_exc_record())
     assert "ValueError" in out
+
+
+def test_json_formatter_stamps_stream_id_and_application_id() -> None:
+    # On the shared syslog channel the log server tells streams and
+    # applications apart by the stream_id/application_id stamped per record.
+    out = json.loads(
+        JsonFormatter(
+            stream_id="app",
+            application_id="nationale-verwijsindex-crypto-service-api",
+        ).format(_record())
+    )
+    assert out["stream_id"] == "app"
+    assert out["application_id"] == "nationale-verwijsindex-crypto-service-api"
+
+
+def test_json_formatter_omits_stream_id_and_application_id_when_unset() -> None:
+    out = json.loads(JsonFormatter().format(_record()))
+    assert "stream_id" not in out
+    assert "application_id" not in out