feat: add structured logging helpers and instrument get_issue (#14)
docker / test (pull_request) Successful in 29s
test / test (push) Successful in 38s
docker / lint (pull_request) Successful in 39s
lint / lint (push) Successful in 39s
docker / docker-test (pull_request) Successful in 12s
docker / docker-publish (pull_request) Has been skipped
lint / lint (pull_request) Successful in 28s
test / test (pull_request) Successful in 22s

Adds reusable, secret-safe logging helpers to `logging_utils`:
- `log_event(logger, level, event, **context)` emits a named event with a
  sanitized `context` mapping (sensitive keys masked as `***`).
- `log_nullable_field(...)` records whether a parsed field is None plus its
  runtime type, without dumping its contents.
- `sanitize_context(...)` is the shared masking primitive.

The JSON formatter now serializes a record's `context` into the payload.

`get_issue_tool` is instrumented at DEBUG (`get_issue.start`,
`get_issue.payload_shape`, `get_issue.field_check` for labels/assignees/user)
so the nullable-field parsing that caused #13 is diagnosable going forward.

Adds tests for the helpers, the formatter, and the get_issue instrumentation,
and documents the pattern in docs/observability.md.
This commit is contained in:
2026-06-22 15:09:59 +02:00
parent cd309ee290
commit f53e1a3a5a
4 changed files with 225 additions and 1 deletions
+63 -1
View File
@@ -5,16 +5,22 @@ from __future__ import annotations
import json
import logging
from datetime import datetime, timezone
from typing import Any
from aegis_gitea_mcp.request_context import get_request_id
# Context keys whose values must never be written to logs verbatim.
SENSITIVE_CONTEXT_KEYS = frozenset(
{"token", "authorization", "cookie", "password", "secret", "api_key"}
)
class JsonLogFormatter(logging.Formatter):
"""Format log records as JSON documents."""
def format(self, record: logging.LogRecord) -> str:
"""Serialize a log record to JSON."""
payload = {
payload: dict[str, Any] = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
@@ -22,6 +28,10 @@ class JsonLogFormatter(logging.Formatter):
"request_id": get_request_id(),
}
context = getattr(record, "context", None)
if isinstance(context, dict) and context:
payload["context"] = context
if record.exc_info:
# Security decision: include only exception type to avoid stack leakage.
exception_type = record.exc_info[0]
@@ -46,3 +56,55 @@ def configure_logging(level: str) -> None:
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(JsonLogFormatter())
logger.addHandler(stream_handler)
def sanitize_context(context: dict[str, Any]) -> dict[str, Any]:
"""Redact sensitive values from a logging context mapping.
Keys whose lower-cased name is in ``SENSITIVE_CONTEXT_KEYS`` have their
value replaced with ``"***"`` so credentials never reach the log sink.
Args:
context: Arbitrary key/value pairs intended for structured logging.
Returns:
A new mapping with sensitive values masked.
"""
return {
key: ("***" if key.lower() in SENSITIVE_CONTEXT_KEYS else value)
for key, value in context.items()
}
def log_event(logger: logging.Logger, level: int, event: str, **context: Any) -> None:
"""Emit a structured log event with a sanitized context payload.
Args:
logger: Logger to emit on.
level: Standard ``logging`` level (e.g. ``logging.DEBUG``).
event: Stable, machine-friendly event name (e.g. ``get_issue.start``).
**context: Extra structured fields; sensitive keys are masked.
"""
logger.log(level, event, extra={"context": sanitize_context(context)})
def log_nullable_field(logger: logging.Logger, event: str, field_name: str, value: Any) -> None:
"""Log whether a parsed response field is ``None`` and its runtime type.
Makes null/nullable field failures (such as a ``null`` ``labels`` array)
obvious in logs without dumping the field's full contents.
Args:
logger: Logger to emit on.
event: Stable event name for the field check.
field_name: Name of the field being inspected.
value: The parsed value to characterize.
"""
log_event(
logger,
logging.DEBUG,
event,
field=field_name,
is_none=value is None,
value_type=(type(value).__name__ if value is not None else None),
)
+22
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from aegis_gitea_mcp.gitea_client import (
@@ -10,6 +11,7 @@ from aegis_gitea_mcp.gitea_client import (
GiteaClient,
GiteaError,
)
from aegis_gitea_mcp.logging_utils import log_event, log_nullable_field
from aegis_gitea_mcp.response_limits import limit_items, limit_text
from aegis_gitea_mcp.tools.arguments import (
CommitDiffArgs,
@@ -38,6 +40,8 @@ from aegis_gitea_mcp.tools.arguments import (
SearchCodeArgs,
)
logger = logging.getLogger(__name__)
async def search_code_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
"""Search repository code and return bounded result snippets."""
@@ -265,8 +269,26 @@ async def list_issues_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dic
async def get_issue_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
"""Get issue details."""
parsed = IssueArgs.model_validate(arguments)
log_event(
logger,
logging.DEBUG,
"get_issue.start",
owner=parsed.owner,
repo=parsed.repo,
issue_number=parsed.issue_number,
)
try:
issue = await gitea.get_issue(parsed.owner, parsed.repo, parsed.issue_number)
log_event(
logger,
logging.DEBUG,
"get_issue.payload_shape",
top_level_keys=sorted(issue.keys()) if issue else None,
)
# Surface nullable collections that previously broke parsing (see #13).
log_nullable_field(logger, "get_issue.field_check", "labels", issue.get("labels"))
log_nullable_field(logger, "get_issue.field_check", "assignees", issue.get("assignees"))
log_nullable_field(logger, "get_issue.field_check", "user", issue.get("user"))
return {
"number": issue.get("number", 0),
"title": limit_text(str(issue.get("title", ""))),