diff --git a/agent/monitoring/events.py b/agent/monitoring/events.py index 74b0aea2a7c..0d5b964556e 100644 --- a/agent/monitoring/events.py +++ b/agent/monitoring/events.py @@ -57,6 +57,7 @@ class GatewayDiagnosticEvent: version: Optional[str] = None severity: str = "warning" ts_ns: int = field(default_factory=_now_ns) + source_logger: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return {"event": "gateway_diagnostic", **asdict(self)} diff --git a/agent/monitoring/gateway_health.py b/agent/monitoring/gateway_health.py index cb02aac95c1..752473fe31b 100644 --- a/agent/monitoring/gateway_health.py +++ b/agent/monitoring/gateway_health.py @@ -10,6 +10,7 @@ from __future__ import annotations import hashlib import logging +import re from dataclasses import dataclass from typing import Any, Dict, List, Optional @@ -38,12 +39,19 @@ _KNOWN_PLATFORM_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | { "connecting", "disconnected", "disabled", "paused", "retrying", "unknown" } _SUPERVISION_MODES = {"systemd", "s6", "container", "launchd", "manual", "unknown"} +_SOURCE_LOGGER_RE = re.compile(r"^gateway(?:\.[A-Za-z_][A-Za-z0-9_]*)*$") def _allowed_logger(name: str) -> bool: return name == "gateway" or name.startswith("gateway.") +def source_logger_for_export(name: Any) -> Optional[str]: + """Return a bounded source-controlled gateway logger name for OTLP scope.""" + value = str(name or "") + return value if len(value) <= 128 and _SOURCE_LOGGER_RE.fullmatch(value) else None + + def redact_gateway_message(message: Any) -> str: """Redact gateway diagnostic free text for operator-owned export. @@ -67,7 +75,20 @@ def classify_gateway_error(raw: Any) -> str: return "rate_limited" if "timeout" in s or "timed out" in s: return "timeout" - if any(k in s for k in ("network", "connection", "dns", "socket")): + if any( + k in s + for k in ( + "network", + "connection", + "dns", + "socket", + "connect call failed", + "failed to connect", + "cannot connect", + "unreachable", + "name resolution", + ) + ): return "network_error" if any(k in s for k in ("config", "missing", "invalid")): return "invalid_config" @@ -119,6 +140,8 @@ def _safe_instance_id(raw: Any) -> str: def subsystem_for_logger(logger_name: str) -> str: + if logger_name == "gateway.relay" or logger_name.startswith("gateway.relay."): + return "platform.relay" if logger_name.startswith("gateway.platforms."): parts = logger_name.split(".") if len(parts) >= 3 and parts[2]: @@ -417,11 +440,14 @@ class GatewayDiagnosticLogHandler(logging.Handler): return subsystem = subsystem_for_logger(record.name) message = record.getMessage() + error_class = classify_gateway_error(message) event = GatewayDiagnosticEvent( name=f"gateway.log.{record.levelname.lower()}", subsystem=subsystem, + source_logger=source_logger_for_export(record.name), platform=platform_for_subsystem(subsystem), - error_class=classify_gateway_error(message), + error_class=error_class, + error_code=error_class, profile=self.profile, version=self.version, severity=record.levelname.lower(), @@ -438,5 +464,6 @@ __all__ = [ "GatewayDiagnosticLogHandler", "build_gateway_health_snapshot", "classify_gateway_error", + "source_logger_for_export", "redact_gateway_message", ] diff --git a/agent/monitoring/gateway_health_export.py b/agent/monitoring/gateway_health_export.py index f8fefd24e2b..e10043daba6 100644 --- a/agent/monitoring/gateway_health_export.py +++ b/agent/monitoring/gateway_health_export.py @@ -17,6 +17,8 @@ from typing import Any, Dict, Optional logger = logging.getLogger(__name__) +_DEFAULT_DIAGNOSTIC_SCOPE = "hermes.gateway.diagnostics" + _RESOURCE_ATTRIBUTE_KEYS = frozenset({ "service.name", "service.namespace", @@ -381,19 +383,31 @@ class GatewayDiagnosticLogStreamer: sdk["OTLPLogExporter"](endpoint=endpoint, headers=headers or None) ) self._provider.add_log_record_processor(self._processor) - self._logger = self._provider.get_logger("hermes.gateway.diagnostics") + self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE) self._LogRecord = sdk["LogRecord"] self._sdk = sdk self.exported = 0 def __call__(self, batch: list[Dict[str, Any]]) -> None: + from agent.monitoring.gateway_health import source_logger_for_export + for ev in batch: if ev.get("event") != "gateway_diagnostic": continue attrs = _diagnostic_log_attributes(ev) - # Rendered Python log messages may contain arbitrary user IDs, names, - # paths, or configured strings. Keep the OTLP body content-free and - # carry only the structured, allowlisted attributes above. + # Preserve the source-controlled Python logger as the OTel + # instrumentation scope. This adds precise code attribution without + # turning a fluid module layout into a maintained subsystem enum. + # Rendered messages stay out because they may contain arbitrary IDs, + # names, paths, or configured strings. A future, separately gated + # ``diagnostic_detail: redacted_message`` mode may add best-effort + # free text when an observability plane defines that privacy policy. + source_logger = source_logger_for_export(ev.get("source_logger")) + otel_logger = ( + self._provider.get_logger(source_logger) + if source_logger is not None + else self._logger + ) body = "gateway diagnostic" record = self._LogRecord( timestamp=ev.get("ts_ns"), @@ -405,7 +419,7 @@ class GatewayDiagnosticLogStreamer: body=_redact_string(body), attributes=attrs, ) - self._logger.emit(record) + otel_logger.emit(record) self.exported += 1 def shutdown(self) -> None: diff --git a/agent/monitoring/redaction.py b/agent/monitoring/redaction.py index cf1564f0171..312875901e6 100644 --- a/agent/monitoring/redaction.py +++ b/agent/monitoring/redaction.py @@ -10,8 +10,9 @@ process passes through ``redact_for_export``: are rewritten to ``[email]`` / ``[phone]`` / ``[id]``. There is deliberately no setting to weaken this. The monitoring plane is -content-free by design, so the only free text it carries (log-derived -diagnostic messages) is always fully scrubbed. +content-free by design: rendered log messages are not exported, and bounded +structured strings are still scrubbed as defense-in-depth. This redactor also +remains available for a future, explicitly gated redacted-message detail mode. """ from __future__ import annotations diff --git a/tests/monitoring/test_gateway_health_export.py b/tests/monitoring/test_gateway_health_export.py index e926c5ba2b7..a48f585c3ba 100644 --- a/tests/monitoring/test_gateway_health_export.py +++ b/tests/monitoring/test_gateway_health_export.py @@ -2,6 +2,17 @@ from __future__ import annotations import logging +import pytest + + +def test_gateway_diagnostic_event_preserves_positional_error_class(): + from agent.monitoring.events import GatewayDiagnosticEvent + + event = GatewayDiagnosticEvent("gateway.log.warning", "gateway", "auth_failed") + + assert event.error_class == "auth_failed" + assert event.source_logger is None + def test_default_config_keeps_gateway_health_export_disabled(): from hermes_cli.config import DEFAULT_CONFIG @@ -180,12 +191,64 @@ def test_gateway_diagnostic_log_handler_never_carries_rendered_message(caplog): assert event["event"] == "gateway_diagnostic" assert event["name"] == "gateway.log.warning" assert event["subsystem"] == "platform.slack" + assert event["source_logger"] == "gateway.platforms.slack" assert event["error_class"] == "auth_failed" assert "redacted_message" not in event assert "acct_7f3a" not in str(event) assert "Alice Smith" not in str(event) +@pytest.mark.parametrize( + "message", + [ + "Connect call failed ('127.0.0.1', 9)", + "failed to connect to relay", + "connection refused", + "network is unreachable", + "temporary failure in name resolution", + ], +) +def test_gateway_error_classifier_recognizes_bounded_network_failures(message): + from agent.monitoring.gateway_health import classify_gateway_error + + assert classify_gateway_error(message) == "network_error" + + +def test_gateway_diagnostic_log_handler_enriches_relay_scope_without_message_content( + monkeypatch, +): + from agent.monitoring import emitter + from agent.monitoring.gateway_health import GatewayDiagnosticLogHandler + + captured = [] + + class DummyEmitter: + def emit(self, event): + captured.append(event.to_dict()) + + monkeypatch.setattr(emitter, "get_emitter", lambda: DummyEmitter()) + handler = GatewayDiagnosticLogHandler(profile="default", version="v-test") + logger = logging.getLogger("gateway.relay.adapter") + logger.addHandler(handler) + try: + logger.warning( + "Connect call failed for ws://alice@example.com/private; " + "credential=«redacted:sk-…»" + ) + finally: + logger.removeHandler(handler) + + assert len(captured) == 1 + event = captured[0] + assert event["subsystem"] == "platform.relay" + assert event["platform"] == "relay" + assert event["source_logger"] == "gateway.relay.adapter" + assert event["error_class"] == "network_error" + assert event["error_code"] == "network_error" + assert "alice@example.com" not in str(event) + assert "private" not in str(event) + + def test_runtime_status_transition_emits_lifecycle_and_platform_events(monkeypatch): from agent.monitoring import emitter from agent.monitoring.gateway_health import emit_runtime_status_transition @@ -374,6 +437,69 @@ def test_diagnostic_log_attributes_are_allowlisted_redacted_and_profile_free(): assert "top-secret-token" not in str(attrs) +def test_diagnostic_log_streamer_uses_validated_source_as_otel_scope(): + from types import SimpleNamespace + + from agent.monitoring.gateway_health_export import GatewayDiagnosticLogStreamer + + class FakeLogger: + def __init__(self): + self.records = [] + + def emit(self, record): + self.records.append(record) + + class FakeProvider: + def __init__(self): + self.loggers = {} + + def get_logger(self, name): + return self.loggers.setdefault(name, FakeLogger()) + + provider = FakeProvider() + streamer = object.__new__(GatewayDiagnosticLogStreamer) + streamer._provider = provider + streamer._logger = provider.get_logger("hermes.gateway.diagnostics") + streamer._LogRecord = lambda **kwargs: SimpleNamespace(**kwargs) + streamer._sdk = { + "INVALID_TRACE_ID": 0, + "INVALID_SPAN_ID": 0, + "TraceFlags": SimpleNamespace(DEFAULT=0), + "SeverityNumber": SimpleNamespace( + FATAL="fatal", ERROR="error", WARN="warn", INFO="info", DEBUG="debug" + ), + } + streamer.exported = 0 + + streamer([ + { + "event": "gateway_diagnostic", + "name": "gateway.log.warning", + "subsystem": "platform.relay", + "platform": "relay", + "source_logger": "gateway.relay.adapter", + "error_class": "network_error", + "severity": "warning", + }, + { + "event": "gateway_diagnostic", + "name": "gateway.log.warning", + "subsystem": "gateway", + "source_logger": "gateway.relay.adapter\nalice@example.com", + "error_class": "unknown", + "severity": "warning", + }, + ]) + + precise = provider.loggers["gateway.relay.adapter"].records + fallback = provider.loggers["hermes.gateway.diagnostics"].records + assert len(precise) == 1 + assert len(fallback) == 1 + assert precise[0].body == "gateway diagnostic" + assert "source_logger" not in precise[0].attributes + assert "alice@example.com" not in str(provider.loggers) + + def test_gateway_health_export_start_is_fail_open_when_otlp_missing(monkeypatch): from agent.monitoring import gateway_health_export from agent.monitoring.gateway_health_export import GatewayHealthExportRuntime