feat(logging): structured logs with request correlation via Powertools

.github/copilot-instructions.md

@@ -18,7 +18,9 @@ Python style
 - Use type aliases for complex types
 - Use built-in generics for type hints
 - Use `logging.getLogger(__name__)`, not print
-- Lazy % formatting in logging: `logger.info("msg %s", var)`
+- Structured logging: constant message + data in `extra`: `logger.info("Message accepted.", extra={"writer": name})`
+- Do not re-log `topic`, `user`, `resource`, `http_method` or the Lambda context; they are bound per request in `src/utils/observability.py`
+- `logger.exception()` only inside an `except` block; outside it pass `exc_info=exc` to `logger.error()`
 - F-strings in exceptions: `raise ValueError(f"Error {var}")`
 - All imports at top of file (never inside functions)
 - Apache 2.0 license header in every .py file (including `__init__.py`)
@@ -27,6 +29,8 @@ Python style
 - Use single backticks in docstrings (`value`), never double backticks (`` ``value`` ``)
 - Do not use `# -----------` separator comments to divide sections
 - End all log messages with a period: `logger.info("Message.")`
+- Every non-2xx response must produce exactly one log line explaining the cause
+- Levels: TRACE payloads, DEBUG steps, INFO request outcomes, WARNING rejected requests, ERROR actionable failures
 
 Patterns
 - `__init__` methods must not raise exceptions; defer validation and connection to first use (lazy init)

DEVELOPER.md

@@ -208,6 +208,45 @@ View container logs in pytest output by increasing log level:
 pytest tests/integration/ -v --log-cli-level=DEBUG
 ```
 
+## Logging Conventions
+
+Logging is structured. `src/utils/observability.py` attaches the AWS Lambda Powertools JSON handler to the root logger, so modules keep using the standard library logger and inherit the format, the Lambda execution context and the request correlation id.
+
+```python
+import logging
+
+logger = logging.getLogger(__name__)
+```
+
+Rules:
+
+- The message is a constant sentence ending with a period. Variable data goes into `extra`, never into the sentence.
+  ```python
+  logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(known_topics)})
+  ```
+- Do not add `topic`, `user`, `resource`, `http_method`, `correlation_id` or the Lambda context to `extra`; they are bound once per request by `bind_request_context()` and `append_request_keys()`.
+- Every non-2xx response must produce exactly one log line explaining the cause.
+- Levels: `TRACE` payloads, `DEBUG` steps, `INFO` request outcomes, `WARNING` rejected requests and soft failures, `ERROR` failures that need action. See the level table in [README](./README.md#logging--correlation).
+- Never log tokens, passwords or full message payloads outside `TRACE`. `TRACE` payload logging goes through `log_payload_at_trace()`, which redacts and size caps the payload.
+- `logger.exception()` is only valid inside an `except` block. Outside one, pass the captured exception: `logger.error("...", exc_info=exc)`.
+- Durations are logged as milliseconds with an explicit key (`duration_ms`, `writer_duration_ms`, `query_duration_ms`).
+
+Assert on structured fields in tests, not on formatted strings:
+
+```python
+def test_rejects_unknown_topic(caplog):
+    caplog.set_level(logging.WARNING)
+    ...
+    assert "Request rejected: unknown topic." == caplog.records[-1].message
+```
+
+Keys bound with `append_request_keys()` live on the Powertools formatter rather than on the `LogRecord`. To assert on them, render the record:
+
+```python
+payload = json.loads(logger.registered_formatter.format(caplog.records[-1]))
+assert "run-42" == payload["correlation_id"]
+```
+
 ## Run All Quality Gates
 
 Run Black, Pylint, mypy, unit tests (with coverage), and integration tests in a single command:

README.md

@@ -128,11 +128,52 @@ Supporting configs:
 - `topic_schemas/*.json` – each file contains a JSON Schema for a topic. In the current code these are explicitly loaded inside `event_gate_lambda.py`. (Future enhancement: auto-discover or index file.)
 
 Environment variables:
-- `LOG_LEVEL` (optional) – defaults to `INFO`.
+- `LOG_LEVEL` (optional) – defaults to `INFO`. Accepts `TRACE`, `DEBUG`, `INFO`, `WARNING`, `ERROR`. An unknown value falls back to `INFO` and is reported with a warning.
+- `POWERTOOLS_LOG_LEVEL` (optional) – takes precedence over `LOG_LEVEL`.
+- `POWERTOOLS_SERVICE_NAME` (optional) – value of the `service` key in every log line. Defaults to `eventgate`.
+- `TRACE_REDACT_KEYS` (optional) – comma separated message keys redacted from `TRACE` payload logs. Defaults to `password,secret,token,key,apikey,api_key`.
+- `TRACE_MAX_BYTES` (optional) – maximum size of a logged `TRACE` payload. Defaults to `10000`.
 - `CONF_DIR` (optional) – directory containing `config.json` and `access.json`. Defaults to `conf`.
 - `POSTGRES_SECRET_NAME` (optional) – AWS Secrets Manager secret name holding PostgreSQL connection credentials (host, port, database, user, password). Required for Postgres writer and stats reader.
 - `POSTGRES_SECRET_REGION` (optional) – AWS region of the Secrets Manager secret. Must be set together with `POSTGRES_SECRET_NAME`.
 
+## Logging & Correlation
+
+Both lambdas emit structured JSON logs through [AWS Lambda Powertools](https://docs.aws.amazon.com/powertools/python/latest/core/logger/). Every line carries the service name, log level, the Lambda execution context (`function_name`, `function_request_id`, `cold_start`) and a `correlation_id`.
+
+The correlation id is resolved per request in this order:
+1. The `X-Correlation-ID` request header, when it matches `^[A-Za-z0-9._:-]{1,128}$`.
+2. The `X-Request-ID` request header, with the same constraint.
+3. The API Gateway request id (`requestContext.requestId`).
+
+Callers that already have a run or job id should send it as `X-Correlation-ID` so their logs and EventGate logs can be joined. The resolved id is returned in the `X-Correlation-ID` response header of every response, including errors.
+
+Log levels used by the service:
+
+| Level     | Content                                                                                       |
+|-----------|-----------------------------------------------------------------------------------------------|
+| `TRACE`   | Full message payloads, redacted and size capped. Never enable by default.                      |
+| `DEBUG`   | Configuration loading, lazy initialization, per-writer send attempts, connection reuse.        |
+| `INFO`    | One line per request outcome, cold start initialization, accepted messages, stats query results.|
+| `WARNING` | Rejected requests (auth, authorization, validation), degraded health, Kafka flush retries.     |
+| `ERROR`   | Writer failures, partial fan-out failures, failed queries, unhandled request errors.           |
+
+Every non-2xx response has exactly one log line explaining the cause.
+
+Example CloudWatch Logs Insights queries:
+
+```text
+fields @timestamp, level, message, status_code, duration_ms
+| filter correlation_id = "<id>"
+| sort @timestamp asc
+```
+
+```text
+fields @timestamp, topic, user, message
+| filter level = "WARNING" and status_code = 403
+| stats count() by user, topic
+```
+
 ## Local Development & Testing
 
 | Purpose                       | Relative link                                                         |

adr/001-observability/001-observability.md

@@ -0,0 +1,84 @@
+# ADR 001. Observability -- structured logging and request correlation
+
+## Decision
+
+Adopt [AWS Lambda Powertools for Python](https://docs.aws.amazon.com/powertools/python/latest/core/logger/) as the logging backend for both EventGate lambdas. Logs are emitted as JSON, enriched with the Lambda execution context, and carry a request correlation id that is also returned to the caller. AWS X-Ray tracing is deliberately **not** adopted at this point.
+
+### Background
+
+EventGate ran with 79 log statements, of which 52 were `DEBUG` and none were `INFO`. Production runs with `LOG_LEVEL=INFO`, so the service emitted effectively nothing per request: no request line, no outcome, no reason for a rejection. Raising the level to `DEBUG` was the only alternative and produced unusable output, because the level was applied to the root logger and `botocore`, `urllib3` and `s3transfer` flooded the log group.
+
+The concrete operational failures this caused:
+
+- A client receiving `401`, `403` or `400` produced **no log line at all**. Support could not answer "why was I rejected?".
+- A partial fan-out failure (Kafka accepted, Postgres failed) returned `500` without recording which sink had already accepted the message, so reconciliation had to be done by hand.
+- Nothing tied log lines from one invocation together, and nothing tied EventGate's lines to the caller's job or run.
+- No durations were recorded anywhere, so "which writer is slow" was unanswerable.
+
+See issue [#193](https://github.com/AbsaOSS/EventGate/issues/193).
+
+### Considered options
+
+**1. Plain standard library logging with a hand-written JSON formatter.**
+No new dependency, but the Lambda execution context, cold start detection, correlation id plumbing and log sampling would all have to be written and maintained here.
+
+**2. AWS Lambda Powertools (chosen).**
+Purpose-built for this runtime. Provides the JSON formatter, `cold_start`, `function_request_id`, correlation id handling and sampling out of the box. The lambdas are deployed as container images, so it is a single `requirements.txt` entry with no layer ARNs to pin per region. Cost is one production dependency and roughly 2-4 MB of image size, negligible next to the librdkafka build already in the image.
+
+**3. Powertools plus X-Ray tracing.**
+Deferred. X-Ray usage in the target account is not decided, it carries its own cost, and `confluent-kafka` is not patchable by the X-Ray SDK, so Kafka spans would need manual instrumentation. The correlation id covers the immediate need, which is joining log lines across an invocation and across services.
+
+### Integration approach
+
+Powertools' `Logger` is created once in `src/utils/observability.py`, and its handler is attached to the **root** logger by `configure_root_logging()`. Modules keep using `logging.getLogger(__name__)`.
+
+This matters for two reasons:
+
+- Existing call sites did not have to change to gain JSON output, so the change stayed reviewable.
+- Persistent keys (including `correlation_id`) live on the shared `LambdaPowertoolsFormatter` instance, so a key appended once per request appears on every module's log lines without threading a logger through the call stack.
+
+The alternative, `copy_config_to_registered_loggers()`, sets `propagate = False` on each module logger, which breaks `caplog` in the unit tests. Attaching at the root avoids that.
+
+`inject_lambda_context` is **not** used as a decorator. It raises `AttributeError` when the Lambda context is `None`, which is how both unit and integration tests invoke `lambda_handler`. `bind_request_context()` performs the same binding and tolerates a missing context.
+
+### Log level policy
+
+| Level     | Content                                                                                        |
+|-----------|------------------------------------------------------------------------------------------------|
+| `TRACE`   | Full message payloads, redacted and size capped. Never enabled by default.                       |
+| `DEBUG`   | Configuration loading, lazy initialization, per-writer send attempts, connection reuse.          |
+| `INFO`    | One line per request outcome, cold start initialization, accepted messages, stats query results. |
+| `WARNING` | Rejected requests (auth, authorization, validation), degraded health, Kafka flush retries.       |
+| `ERROR`   | Writer failures, partial fan-out failures, failed queries, unhandled request errors.             |
+
+The binding rule is: **every non-2xx response produces exactly one log line explaining the cause.**
+
+Third-party loggers (`boto3`, `botocore`, `urllib3`, `s3transfer`, `aiosql`, `confluent_kafka`) are capped at `WARNING` so that running the service at `DEBUG` or `TRACE` stays readable.
+
+The custom `TRACE` level (5) introduced earlier is retained. It is registered with the standard `logging` module before the Powertools `Logger` is constructed, and the level is passed numerically, so Powertools' level validation resolves it. A unit test pins this behaviour, because an unrecognised level would silently fall back to `INFO` and disable payload logging without any error.
+
+### Correlation id contract
+
+Resolved per request, in order:
+
+1. `X-Correlation-ID` request header.
+2. `X-Request-ID` request header.
+3. API Gateway request id (`requestContext.requestId`).
+
+Header values are accepted only when they match `^[A-Za-z0-9._:-]{1,128}$`. Inbound header values are written into the log stream, so newlines and control characters are rejected to prevent log injection, and the length is capped so the id cannot bloat every line.
+
+The resolved id is returned in the `X-Correlation-ID` response header of every response, success or error, so a client can quote a single id when reporting a problem. It is stamped centrally in `dispatch_request()`, which is the single point every response passes through.
+
+### Consequences
+
+- Log output changes from plain text to JSON. Nothing downstream parses these logs today; they are read by humans and queried in CloudWatch Logs Insights, where JSON is strictly easier to query.
+- `dispatch_request()` now catches `Exception` at the boundary instead of a fixed list of six exception types. Previously a `psycopg2.Error` or `KafkaException` escaping a handler reached the runtime, producing an unstructured traceback and an opaque `502` from API Gateway. `SystemExit` still propagates, so the `/terminate` route is unaffected.
+- A malformed request body on `POST /topics/{topic_name}` now returns `400 validation` instead of `500 internal`. It is a client error and was only reaching the internal error path because `json.loads` was left to the boundary handler.
+- Callers gain a response header they did not have before. This is additive and does not change the response body contract.
+- CloudWatch cost rises slightly. `INFO` stays at roughly one to three lines per invocation; `POWERTOOLS_LOGGER_SAMPLE_RATE` is available if `DEBUG` detail is wanted in production without enabling it for every invocation.
+
+### Not in scope
+
+- **X-Ray tracing.** If it is enabled later, `Tracer` from the same library plugs into `observability.py`; the function needs `Tracing: Active` and the `xray:PutTraceSegments` permission, and Kafka spans need manual subsegments.
+- **EMF metrics** (writer failures, auth failures, cold starts) as alarm sources. Worth a separate issue.
+- **Downstream correlation propagation**: Kafka message headers and the EventBridge `TraceHeader`. Useful, but only once a consumer is ready to read them.

requirements.txt

@@ -1,3 +1,4 @@
+aws-lambda-powertools==3.31.1
 cryptography==49.0.0
 jsonschema==4.26.0
 PyJWT==2.13.0

src/event_gate_lambda.py

@@ -16,8 +16,9 @@
 
 """AWS Lambda entry point for the EventGate service."""
 
-import os
+import logging
 import sys
+import time
 from typing import Any
 
 import boto3
@@ -27,22 +28,22 @@
 from src.handlers.handler_health import HandlerHealth
 from src.handlers.handler_token import HandlerToken
 from src.handlers.handler_topic import HandlerTopic
-from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
+from src.utils.conf_path import CONF_DIR, log_configuration_source
 from src.utils.config_loader import load_config
 from src.utils.constants import SSL_CA_BUNDLE_KEY
-from src.utils.logging_levels import init_root_logger
+from src.utils.observability import configured_log_level, init_lambda_logging
 from src.utils.utils import dispatch_request
 from src.writers.writer_eventbridge import WriterEventBridge
 from src.writers.writer_kafka import WriterKafka
 from src.writers.writer_postgres import WriterPostgres
 
-logger = init_root_logger(__name__)
-logger.debug("Initialized logger with level %s.", os.environ.get("LOG_LEVEL", "INFO"))
+logger = init_lambda_logging(__name__)
+
+_INIT_STARTED_AT = time.perf_counter()
+logger.info("Initializing EventGate lambda.", extra={"log_level": configured_log_level()})
 
 # Load main configuration
-logger.debug("Using CONF_DIR=%s.", CONF_DIR)
-if INVALID_CONF_ENV:
-    logger.warning("CONF_DIR env var set to non-existent path: %s; fell back to %s.", INVALID_CONF_ENV, CONF_DIR)
+log_configuration_source(logger)
 config = load_config(CONF_DIR)
 logger.debug("Loaded main configuration.")
 
@@ -73,6 +74,15 @@
 handler_health = HandlerHealth(writers)
 handler_api = HandlerApi().with_api_definition_loaded()
 
+logger.info(
+    "EventGate lambda initialized.",
+    extra={
+        "init_duration_ms": round((time.perf_counter() - _INIT_STARTED_AT) * 1000, 2),
+        "writers": sorted(writers),
+        "topics": sorted(handler_topic.topics),
+    },
+)
+
 # Route to handler function mapping
 ROUTE_MAP: dict[str, Any] = {
     "/api": lambda _: handler_api.get_api(),
@@ -85,12 +95,12 @@
 }
 
 
-def lambda_handler(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
+def lambda_handler(event: dict[str, Any], context: Any = None) -> dict[str, Any]:
     """AWS Lambda entry point for EventGate. Dispatches based on API Gateway proxy resource field.
     Args:
         event: The event data from API Gateway.
-        _context: The mandatory context argument for AWS Lambda invocation (unused).
+        context: The AWS Lambda context object, used to enrich logs with the execution context.
     Returns:
         A dictionary compatible with API Gateway Lambda Proxy integration.
     """
-    return dispatch_request(event, ROUTE_MAP, logger)
+    return dispatch_request(event, ROUTE_MAP, logger, context)

src/event_stats_lambda.py

@@ -16,24 +16,25 @@
 
 """AWS Lambda entry point for the EventStats service."""
 
-import os
+import logging
+import time
 from typing import Any
 
 from src.handlers.handler_health import HandlerHealth
 from src.handlers.handler_stats import HandlerStats
 from src.readers.reader_postgres import ReaderPostgres
-from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
+from src.utils.conf_path import CONF_DIR, log_configuration_source
 from src.utils.config_loader import load_topic_names
-from src.utils.logging_levels import init_root_logger
+from src.utils.observability import configured_log_level, init_lambda_logging
 from src.utils.utils import dispatch_request
 
-logger = init_root_logger(__name__)
-logger.debug("Initialized EventStats logger with level %s.", os.environ.get("LOG_LEVEL", "INFO"))
+logger = init_lambda_logging(__name__)
+
+_INIT_STARTED_AT = time.perf_counter()
+logger.info("Initializing EventStats lambda.", extra={"log_level": configured_log_level()})
 
 # Load main configuration
-logger.debug("Using CONF_DIR=%s.", CONF_DIR)
-if INVALID_CONF_ENV:
-    logger.warning("CONF_DIR env var set to non-existent path: %s; fell back to %s.", INVALID_CONF_ENV, CONF_DIR)
+log_configuration_source(logger)
 
 # Load topic names.
 topic_names = load_topic_names(CONF_DIR)
@@ -46,19 +47,27 @@
 handler_stats = HandlerStats(topics, reader_postgres)
 handler_health = HandlerHealth({"postgres_reader": reader_postgres})
 
+logger.info(
+    "EventStats lambda initialized.",
+    extra={
+        "init_duration_ms": round((time.perf_counter() - _INIT_STARTED_AT) * 1000, 2),
+        "topics": sorted(topics),
+    },
+)
+
 # Route to handler function mapping
 ROUTE_MAP: dict[str, Any] = {
     "/stats/{topic_name}": handler_stats.handle_request,
     "/health": lambda _: handler_health.get_health(),
 }
 
 
-def lambda_handler(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
+def lambda_handler(event: dict[str, Any], context: Any = None) -> dict[str, Any]:
     """AWS Lambda entry point for EventStats. Dispatches based on API Gateway proxy resource field.
     Args:
         event: The event data from API Gateway.
-        _context: The mandatory context argument for AWS Lambda invocation (unused).
+        context: The AWS Lambda context object, used to enrich logs with the execution context.
     Returns:
         A dictionary compatible with API Gateway Lambda Proxy integration.
     """
-    return dispatch_request(event, ROUTE_MAP, logger)
+    return dispatch_request(event, ROUTE_MAP, logger, context)