diff --git a/mcp-server/src/streamops_mcp/agent/main.py b/mcp-server/src/streamops_mcp/agent/main.py index 71dbc34..60fbb3b 100644 --- a/mcp-server/src/streamops_mcp/agent/main.py +++ b/mcp-server/src/streamops_mcp/agent/main.py @@ -17,11 +17,9 @@ from streamops_mcp.agent.monitor import MonitorAgent from streamops_mcp.config import config +from streamops_mcp.logging_setup import configure_logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(name)s] %(levelname)s %(message)s", -) +configure_logging() logger = logging.getLogger("streamops-mcp.agent") diff --git a/mcp-server/src/streamops_mcp/agent/monitor.py b/mcp-server/src/streamops_mcp/agent/monitor.py index 5046880..f8c9394 100644 --- a/mcp-server/src/streamops_mcp/agent/monitor.py +++ b/mcp-server/src/streamops_mcp/agent/monitor.py @@ -34,6 +34,11 @@ ) from streamops_mcp.agent.tools import ALL_TOOLS, DIAGNOSTIC_TOOLS from streamops_mcp.config import config +from streamops_mcp.logging_setup import ( + new_correlation_id, + reset_correlation_id, + set_correlation_id, +) from streamops_mcp.prompts import load_prompt, load_runbook logger = logging.getLogger("streamops-mcp.monitor") @@ -102,69 +107,81 @@ async def run_cycle(self) -> IncidentReport | None: """Run one monitoring cycle: poll, detect, diagnose, report. Returns an IncidentReport if an anomaly was found, None if healthy. - """ - logger.info("Starting monitoring cycle (multi_agent=%s)", self.multi_agent) - - detection = await self._detect_anomalies() - if detection is None: - logger.info("No anomalies detected, infrastructure healthy") - return None - - if self.multi_agent: - try: - diagnosis = await self._retry_subagent( - "Diagnostic Agent", - lambda: self._spawn_diagnostic_agent(detection), - ) - except Exception as exc: - logger.error( - "Diagnostic Agent failed after retries, cycle aborted: %s", exc, - ) - return None - self._log_confidence_distribution(diagnosis) + A correlation id is bound for the whole cycle so every log line the + cycle emits (this coordinator, both sub-agents, the tool executor, and + escalation) shares one greppable id. See issue #84. + """ + cycle_id = new_correlation_id() + token = set_correlation_id(cycle_id) + try: + logger.info( + "Starting monitoring cycle (cycle_id=%s, multi_agent=%s)", + cycle_id, self.multi_agent, + ) - if self._all_claims_low_confidence(diagnosis): - logger.warning( - "All %d claims are LOW or UNSOURCED confidence; " - "downgrading to warning-level log, skipping report agent", - len(diagnosis.claims), - ) + detection = await self._detect_anomalies() + if detection is None: + logger.info("No anomalies detected, infrastructure healthy") return None - low_claims = self._extract_low_confidence_claims(diagnosis) + if self.multi_agent: + try: + diagnosis = await self._retry_subagent( + "Diagnostic Agent", + lambda: self._spawn_diagnostic_agent(detection), + ) + except Exception as exc: + logger.error( + "Diagnostic Agent failed after retries, cycle aborted: %s", exc, + ) + return None + + self._log_confidence_distribution(diagnosis) + + if self._all_claims_low_confidence(diagnosis): + logger.warning( + "All %d claims are LOW or UNSOURCED confidence; " + "downgrading to warning-level log, skipping report agent", + len(diagnosis.claims), + ) + return None + + low_claims = self._extract_low_confidence_claims(diagnosis) + + try: + report = await self._retry_subagent( + "Report Agent", + lambda: self._spawn_report_agent(diagnosis), + ) + except Exception as exc: + logger.warning( + "Report Agent failed after retries, producing fallback report: %s", exc, + ) + report = self._fallback_report(diagnosis) + + if low_claims and not report.low_confidence_claims: + report.low_confidence_claims = low_claims + else: + diagnosis = self._extract_diagnosis_from_detection(detection) + report = await self._spawn_report_agent(diagnosis) - try: - report = await self._retry_subagent( - "Report Agent", - lambda: self._spawn_report_agent(diagnosis), - ) - except Exception as exc: + if diagnosis.conflicts: logger.warning( - "Report Agent failed after retries, producing fallback report: %s", exc, + "Coordinator received %d unresolved conflict(s) from Diagnostic Agent", + len(diagnosis.conflicts), ) - report = self._fallback_report(diagnosis) - - if low_claims and not report.low_confidence_claims: - report.low_confidence_claims = low_claims - else: - diagnosis = self._extract_diagnosis_from_detection(detection) - report = await self._spawn_report_agent(diagnosis) - - if diagnosis.conflicts: - logger.warning( - "Coordinator received %d unresolved conflict(s) from Diagnostic Agent", - len(diagnosis.conflicts), - ) - for conflict in diagnosis.conflicts: - logger.warning( - "Conflict %s [%s]: claims %s vs %s", - conflict.conflict_id, conflict.topic, - conflict.claim_a_id, conflict.claim_b_id, - ) - - await escalate(report, diagnosis=diagnosis) - return report + for conflict in diagnosis.conflicts: + logger.warning( + "Conflict %s [%s]: claims %s vs %s", + conflict.conflict_id, conflict.topic, + conflict.claim_a_id, conflict.claim_b_id, + ) + + await escalate(report, diagnosis=diagnosis) + return report + finally: + reset_correlation_id(token) @staticmethod def _fallback_report(diagnosis: DiagnosisReport) -> IncidentReport: diff --git a/mcp-server/src/streamops_mcp/logging_setup.py b/mcp-server/src/streamops_mcp/logging_setup.py new file mode 100644 index 0000000..54c6a5c --- /dev/null +++ b/mcp-server/src/streamops_mcp/logging_setup.py @@ -0,0 +1,84 @@ +"""Correlated logging setup. + +Binds a per-cycle correlation id to a ContextVar so every log line emitted +anywhere in a monitoring cycle's async call tree (monitor, diagnostic and +report sub-agents, tool executor, escalation) carries the same id, without +threading it through every function signature. ContextVars propagate across +``await`` within the same asyncio task, so setting the id once at the top of +``run_cycle`` covers the whole cycle. + +A log line with no correlation id is orphaned the moment two cycles overlap +or two incidents run concurrently: the lines interleave with no way to +reconstruct one incident's story. See issue #84. +""" + +import contextvars +import logging +import uuid + +# Default sentinel for lines emitted outside any cycle (startup, config, the +# MCP server's own request handling). Never left unset, so the formatter's +# %(correlation_id)s always resolves. +_UNSET = "-" + +_correlation_id: contextvars.ContextVar[str] = contextvars.ContextVar( + "correlation_id", default=_UNSET +) + +# Includes the correlation id so a single cycle's lines are greppable by cid=. +LOG_FORMAT = "%(asctime)s [%(name)s] [cid=%(correlation_id)s] %(levelname)s %(message)s" + + +def new_correlation_id(prefix: str = "cyc") -> str: + """Generate a correlation id that is unique across process restarts. + + A monotonic per-process counter would collide across restarts (two cold + starts both produce cycle 1); the uuid suffix makes each cycle's id + globally distinct. + """ + return f"{prefix}-{uuid.uuid4().hex[:12]}" + + +def set_correlation_id(cid: str) -> contextvars.Token: + """Bind a correlation id to the current context; returns a reset token.""" + return _correlation_id.set(cid) + + +def reset_correlation_id(token: contextvars.Token) -> None: + """Restore the correlation id to its previous value using the token.""" + _correlation_id.reset(token) + + +def get_correlation_id() -> str: + """Return the correlation id bound to the current context (or the sentinel).""" + return _correlation_id.get() + + +class CorrelationIdFilter(logging.Filter): + """Injects the current correlation id onto every LogRecord. + + Attached to the handler so the attribute is always present before the + formatter runs, including for third-party loggers that never heard of the + correlation id. + """ + + def filter(self, record: logging.LogRecord) -> bool: + record.correlation_id = _correlation_id.get() + return True + + +def configure_logging(level: int = logging.INFO) -> None: + """Configure root logging with the correlation-id format and filter. + + Idempotent: replaces the root handlers so calling it from multiple entry + points (agent main, MCP server) does not double-log or drop the filter. + """ + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter(LOG_FORMAT)) + handler.addFilter(CorrelationIdFilter()) + + root = logging.getLogger() + root.setLevel(level) + for existing in list(root.handlers): + root.removeHandler(existing) + root.addHandler(handler) diff --git a/mcp-server/src/streamops_mcp/server.py b/mcp-server/src/streamops_mcp/server.py index cd87d4f..67f0a81 100644 --- a/mcp-server/src/streamops_mcp/server.py +++ b/mcp-server/src/streamops_mcp/server.py @@ -13,15 +13,13 @@ from mcp.server.fastmcp import FastMCP +from streamops_mcp.logging_setup import configure_logging from streamops_mcp.tools.events import register_event_tools from streamops_mcp.tools.flink import register_flink_tools from streamops_mcp.tools.kafka import register_kafka_tools from streamops_mcp.tools.prometheus import register_prometheus_tools -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(name)s] %(levelname)s %(message)s", -) +configure_logging() logger = logging.getLogger("streamops-mcp") mcp = FastMCP( diff --git a/mcp-server/tests/test_logging_setup.py b/mcp-server/tests/test_logging_setup.py new file mode 100644 index 0000000..71a5322 --- /dev/null +++ b/mcp-server/tests/test_logging_setup.py @@ -0,0 +1,150 @@ +"""Tests for correlated logging (issue #84). + +Verifies that a per-cycle correlation id is bound to the context, propagates +to every log record emitted while it is set (including from other modules via +the shared filter), is unique across cycles, and falls back to a sentinel +outside any cycle. +""" + +import logging + +import pytest + +from streamops_mcp.logging_setup import ( + CorrelationIdFilter, + configure_logging, + get_correlation_id, + new_correlation_id, + reset_correlation_id, + set_correlation_id, +) + + +@pytest.fixture +def captured_records(): + """Attach a capturing handler (with the correlation filter) to the root logger. + + Mirrors the production handler: the filter injects correlation_id onto every + record so the format string never hits a missing attribute. + """ + records: list[logging.LogRecord] = [] + + class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + handler = _Capture() + handler.addFilter(CorrelationIdFilter()) + root = logging.getLogger() + root.addHandler(handler) + prev_level = root.level + root.setLevel(logging.INFO) + try: + yield records + finally: + root.removeHandler(handler) + root.setLevel(prev_level) + + +class TestCorrelationId: + def test_new_correlation_id_is_unique_across_calls(self): + # Arrange / Act + a = new_correlation_id() + b = new_correlation_id() + + # Assert: two cold cycles never collide (the restart-collision bug the + # old int counter had). + assert a != b + assert a.startswith("cyc-") + + def test_records_carry_the_bound_correlation_id(self, captured_records): + # Arrange + cid = new_correlation_id() + token = set_correlation_id(cid) + + # Act: log from two different module loggers within the same cycle. + try: + logging.getLogger("streamops-mcp.monitor").info("cycle start") + logging.getLogger("streamops-mcp.executor").info("tool call") + finally: + reset_correlation_id(token) + + # Assert: every line from the cycle shares the one id, regardless of + # which module emitted it. + assert len(captured_records) == 2 + assert {r.correlation_id for r in captured_records} == {cid} + + def test_two_cycles_get_distinct_ids(self, captured_records): + # Arrange / Act + cid1 = new_correlation_id() + token = set_correlation_id(cid1) + logging.getLogger("streamops-mcp.monitor").info("cycle 1") + reset_correlation_id(token) + + cid2 = new_correlation_id() + token = set_correlation_id(cid2) + logging.getLogger("streamops-mcp.monitor").info("cycle 2") + reset_correlation_id(token) + + # Assert + assert captured_records[0].correlation_id == cid1 + assert captured_records[1].correlation_id == cid2 + assert cid1 != cid2 + + def test_record_outside_any_cycle_uses_sentinel(self, captured_records): + # Arrange / Act: no correlation id has been set in this context. + logging.getLogger("streamops-mcp").info("startup line") + + # Assert: sentinel, not a crash on a missing attribute. + assert captured_records[0].correlation_id == "-" + + def test_reset_restores_previous_value(self): + # Arrange + assert get_correlation_id() == "-" + + # Act + token = set_correlation_id("cyc-abc") + assert get_correlation_id() == "cyc-abc" + reset_correlation_id(token) + + # Assert + assert get_correlation_id() == "-" + + +class TestConfigureLogging: + def test_configure_logging_is_idempotent(self): + # Arrange / Act: calling twice must not stack duplicate handlers. + configure_logging() + count_after_first = len(logging.getLogger().handlers) + configure_logging() + count_after_second = len(logging.getLogger().handlers) + + # Assert + assert count_after_first == 1 + assert count_after_second == 1 + + def test_format_renders_correlation_id_without_error(self): + # Arrange + configure_logging() + handler = logging.getLogger().handlers[0] + cid = new_correlation_id() + token = set_correlation_id(cid) + record = logging.LogRecord( + name="streamops-mcp.monitor", + level=logging.INFO, + pathname=__file__, + lineno=1, + msg="hello", + args=(), + exc_info=None, + ) + + # Act: the filter must populate the attribute before the formatter reads it. + try: + handler.filters[0].filter(record) + rendered = handler.formatter.format(record) + finally: + reset_correlation_id(token) + + # Assert + assert f"cid={cid}" in rendered diff --git a/mcp-server/tests/test_monitor.py b/mcp-server/tests/test_monitor.py index d75bcc0..2de65c8 100644 --- a/mcp-server/tests/test_monitor.py +++ b/mcp-server/tests/test_monitor.py @@ -669,3 +669,56 @@ async def test_diagnostic_failure_aborts_cycle(self, agent, monkeypatch): # Assert: cycle aborts cleanly, nothing escalated assert result is None mock_escalate.assert_not_awaited() + + @pytest.mark.asyncio + async def test_cycle_binds_correlation_id_and_resets_after(self, agent): + # Arrange: capture the correlation id as seen deep in the cycle (at escalation) + from streamops_mcp.logging_setup import get_correlation_id + + agent.client = _mock_client( + _api_response("Anomaly detected: consumer lag spike on partition 2"), + _api_response(_DIAGNOSIS_JSON), + _api_response(_REPORT_JSON), + ) + seen: dict[str, str] = {} + + async def _capture(*args, **kwargs): + seen["cid"] = get_correlation_id() + + # Act + with patch( + "streamops_mcp.agent.monitor.escalate", new=_capture, + ): + await agent.run_cycle() + + # Assert: a real cycle id was bound during the cycle, and it is cleared after + assert seen["cid"].startswith("cyc-") + assert get_correlation_id() == "-" + + @pytest.mark.asyncio + async def test_two_cycles_get_distinct_correlation_ids(self, agent): + # Arrange + from streamops_mcp.logging_setup import get_correlation_id + + seen: list[str] = [] + + async def _capture(*args, **kwargs): + seen.append(get_correlation_id()) + + def _fresh_client(): + return _mock_client( + _api_response("Anomaly detected: consumer lag spike on partition 2"), + _api_response(_DIAGNOSIS_JSON), + _api_response(_REPORT_JSON), + ) + + # Act: run two independent cycles + with patch("streamops_mcp.agent.monitor.escalate", new=_capture): + agent.client = _fresh_client() + await agent.run_cycle() + agent.client = _fresh_client() + await agent.run_cycle() + + # Assert: each cycle got its own id (no cross-cycle bleed) + assert len(seen) == 2 + assert seen[0] != seen[1]