diff --git a/integration-tests/mcp-server/test-stack-reset.sh b/integration-tests/mcp-server/test-stack-reset.sh index 74cefbc..3ac6cd4 100755 --- a/integration-tests/mcp-server/test-stack-reset.sh +++ b/integration-tests/mcp-server/test-stack-reset.sh @@ -56,6 +56,10 @@ MCP_SERVER_REQUEST_TIMEOUT="${MCP_SERVER_REQUEST_TIMEOUT:-420000}" \ mcp_call odd_stack_reset > "$workdir/reset.json" assert_result_contains "$workdir/reset.json" '"running": true' +step "the reset names the services it wiped (issue #35)" +assert_result_contains "$workdir/reset.json" 'services_wiped' +assert_result_contains "$workdir/reset.json" 'reset-proof' + step "the OTLP->Loki pipeline is live again (post-reset canary)" inject_log "$CANARY" wait_for_hits "$CANARY" diff --git a/src/mcp-server/app/server.py b/src/mcp-server/app/server.py index bfd3616..4e2d99b 100644 --- a/src/mcp-server/app/server.py +++ b/src/mcp-server/app/server.py @@ -41,7 +41,7 @@ def odd_stack_up() -> dict: @mcp.tool() @telemetry.traced_tool def odd_stack_down() -> dict: - """Stop and remove the local LGTM stack; stored telemetry does not survive.""" + """Stop and remove the local LGTM stack; stored telemetry does not survive. The stack is shared by every project on this machine, so their data is destroyed too.""" return stack_ops.stack_down() @@ -55,7 +55,19 @@ def odd_stack_status() -> dict: @mcp.tool() @telemetry.traced_tool def odd_stack_reset() -> dict: - """Wipe all stored telemetry (traces, metrics, logs, profiles) and return a fresh, ready stack.""" + """Wipe ALL stored telemetry (traces, metrics, logs, profiles) and return a fresh, ready stack. + + The wipe is machine-wide and irreversible: one shared stack per machine, so + data from every project ever observed on it is destroyed, not just the + current one. The result's services_wiped field lists the service.name + values that were stored. If it may contain services outside the current + project, warn the user before calling this tool. + + services_wiped always includes oddyssey-mcp (this server observes itself + and exports to the stack it pilots) and otelcol-contrib (the embedded + collector's own metrics): those two are never another project's leftover + state - only other names are. + """ return stack_ops.stack_reset() diff --git a/src/mcp-server/app/stack.py b/src/mcp-server/app/stack.py index 486d3d1..e77fbc0 100644 --- a/src/mcp-server/app/stack.py +++ b/src/mcp-server/app/stack.py @@ -22,12 +22,31 @@ # needs to be exposed. PROMETHEUS_READY = "http://localhost:3000/api/datasources/proxy/uid/prometheus/-/ready" TEMPO_READY = "http://localhost:3000/api/datasources/proxy/uid/tempo/ready" +TEMPO_SERVICE_NAMES = ( + "http://localhost:3000/api/datasources/proxy/uid/tempo" + "/api/search/tag/service.name/values" +) +PROMETHEUS_JOB_VALUES = ( + "http://localhost:3000/api/datasources/proxy/uid/prometheus/api/v1/label/job/values" +) +LOKI_SERVICE_NAMES = ( + "http://localhost:3000/api/datasources/proxy/uid/loki" + "/loki/api/v1/label/service_name/values" +) GRAFANA_URL = "http://localhost:3000" OTLP_ENDPOINT = "http://localhost:4317" OTLP_HTTP_INGEST = "http://localhost:4318/v1/traces" STARTUP_TIMEOUT_S = 120 POLL_INTERVAL_S = 2 +# Widest lookback each backend accepts for the pre-wipe service listing: +# requests beyond the cap are rejected outright (not clamped), so the +# window sits just under Tempo's 168h search max_duration and Loki's +# 30d1h max_query_length. Signals older than these windows can be missed; +# Prometheus (queried without a range) covers its full TSDB. +TEMPO_SEARCH_WINDOW_S = 167 * 3600 +LOKI_SEARCH_WINDOW_S = 30 * 24 * 3600 + def run_args() -> list[str]: """The docker run command that creates the stack container.""" @@ -122,6 +141,59 @@ def stack_up() -> dict: ) +def stored_services(transport: httpx.BaseTransport | None = None) -> list[str]: + """Best-effort list of service.name values currently stored in the stack. + + Union across the queryable backends, since a service may have emitted + only one signal: Tempo's service.name tag values, Loki's service_name + label values, and Prometheus job values (OTLP ingestion maps + service.name onto job, prefixed by one service.namespace/ segment when + a namespace is set). Tempo and Loki need an explicit start/end pair: + unscoped, Tempo only reads its live store (flushed blocks are + invisible) and Loki defaults to a 6-hour lookback, so day-old services + would be wiped without ever being listed. The list exists to warn + before a wipe, so every failure degrades to fewer names, never to an + error. + """ + now_s = int(time.time()) + + def values(payload: object, field: str) -> list[str]: + if not isinstance(payload, dict): + return [] + items = payload.get(field) + if not isinstance(items, list): + return [] + return [v for v in items if isinstance(v, str)] + + services: set[str] = set() + with httpx.Client(timeout=3.0, transport=transport) as client: + try: + tempo = client.get( + TEMPO_SERVICE_NAMES, + params={"start": now_s - TEMPO_SEARCH_WINDOW_S, "end": now_s}, + ).json() + services.update(values(tempo, "tagValues")) + except (httpx.HTTPError, ValueError, TypeError): + pass + try: + loki = client.get( + LOKI_SERVICE_NAMES, + params={ + "start": (now_s - LOKI_SEARCH_WINDOW_S) * 1_000_000_000, + "end": now_s * 1_000_000_000, + }, + ).json() + services.update(values(loki, "data")) + except (httpx.HTTPError, ValueError, TypeError): + pass + try: + prometheus = client.get(PROMETHEUS_JOB_VALUES).json() + services.update(job.split("/", 1)[-1] for job in values(prometheus, "data")) + except (httpx.HTTPError, ValueError, TypeError): + pass + return sorted(services) + + def stack_down() -> dict: """Destroy the stack container (and its data); absent is already down.""" telemetry.force_flush() @@ -138,6 +210,20 @@ def stack_reset() -> dict: stored signal (traces, metrics, logs, profiles) by construction; a new container then starts from the image. After a reset, everything the stack contains IS the next run - no window arithmetic needed. + + The stack is shared machine-wide (issue #35), so the wipe is never + scoped to one project: the result names the services that were stored + so the destruction is at least visible to the caller. A stopped + container (normal after a host reboot) still holds telemetry but + answers nothing on :3000, so it is booted first to be enumerable - + best-effort, because wiping a container too broken to boot is also + reset's job. """ + if _container_state() == "stopped": + try: + stack_up() + except RuntimeError: + pass + services = stored_services() stack_down() - return stack_up() + return {**stack_up(), "services_wiped": services} diff --git a/tests/mcp-server/test_server.py b/tests/mcp-server/test_server.py index ec488b1..d689c4f 100644 --- a/tests/mcp-server/test_server.py +++ b/tests/mcp-server/test_server.py @@ -20,6 +20,21 @@ def test_tools_have_descriptions(): assert all(tool.description for tool in tools) +def test_reset_description_states_the_machine_wide_wipe(): + # Issue #35: the wipe is machine-wide (one shared stack per machine), + # and the tool result names the services it destroyed. Both facts must + # be visible to the calling agent through the tool description. + tools = asyncio.run(server.mcp.list_tools()) + reset = next(tool for tool in tools if tool.name == "odd_stack_reset") + assert "machine" in reset.description.lower() + assert "services_wiped" in reset.description + # Issue #35 side note: the server observes itself and the embedded + # collector self-reports, so these two names are always listed - the + # agent must not read them as another project's leftover state. + assert "oddyssey-mcp" in reset.description + assert "otelcol-contrib" in reset.description + + def test_sdk_otel_middleware_removed(): # mcp 2.0 installs its own OpenTelemetryMiddleware by default, which # duplicated every tool span (observation report finding 1). The diff --git a/tests/mcp-server/test_stack.py b/tests/mcp-server/test_stack.py index 2a7619e..f39b036 100644 --- a/tests/mcp-server/test_stack.py +++ b/tests/mcp-server/test_stack.py @@ -1,4 +1,5 @@ import httpx +from oddyssey_mcp import stack from oddyssey_mcp.stack import ( CONTAINER_NAME, IMAGE, @@ -6,6 +7,7 @@ _otlp_ingest_ready, run_args, stack_status, + stored_services, ) @@ -52,3 +54,163 @@ def refuse(request): transport = httpx.MockTransport(refuse) with httpx.Client(transport=transport) as client: assert _otlp_ingest_ready(client) is False + + +def test_stored_services_unions_tempo_prometheus_and_loki(): + # A service may have emitted only one signal, so all three queryable + # backends contribute. Prometheus OTLP ingestion maps service.name onto + # the job label, prefixed by service.namespace/ when one is set - the + # prefix must be stripped so backends report the same service.name values. + def handler(request: httpx.Request) -> httpx.Response: + url = str(request.url) + if "uid/tempo" in url: + return httpx.Response(200, json={"tagValues": ["checkout", "oddyssey-mcp"]}) + if "uid/loki" in url: + return httpx.Response( + 200, json={"status": "success", "data": ["logs-only"]} + ) + return httpx.Response( + 200, json={"status": "success", "data": ["shop/checkout", "billing"]} + ) + + services = stored_services(transport=httpx.MockTransport(handler)) + assert services == ["billing", "checkout", "logs-only", "oddyssey-mcp"] + + +def test_stored_services_queries_tempo_and_loki_with_their_widest_time_range(): + # Without explicit start/end, Tempo's tag-values endpoint only reads the + # live store (flushed blocks are invisible) and Loki defaults to a 6-hour + # lookback - a day-old project would be wiped without ever being listed. + # Both cap the queryable range (Tempo max_duration 168h, Loki + # max_query_length 30d1h) and reject wider requests outright, so the + # window must sit just under each cap, not at epoch 0. + seen: dict[str, httpx.URL] = {} + + def handler(request: httpx.Request) -> httpx.Response: + url = str(request.url) + if "uid/tempo" in url: + seen["tempo"] = request.url + return httpx.Response(200, json={"tagValues": []}) + if "uid/loki" in url: + seen["loki"] = request.url + return httpx.Response(200, json={"status": "success", "data": []}) + return httpx.Response(200, json={"status": "success", "data": []}) + + stored_services(transport=httpx.MockTransport(handler)) + + tempo_start = int(seen["tempo"].params["start"]) + tempo_end = int(seen["tempo"].params["end"]) + assert tempo_start > 0 + assert tempo_end - tempo_start == stack.TEMPO_SEARCH_WINDOW_S + + loki_start = int(seen["loki"].params["start"]) + loki_end = int(seen["loki"].params["end"]) + assert loki_start > 0 + assert loki_end - loki_start == stack.LOKI_SEARCH_WINDOW_S * 1_000_000_000 + + +def test_stored_services_strips_only_the_namespace_prefix(): + # job is "/" with a single namespace + # segment; a service.name containing "/" must survive intact. + def handler(request: httpx.Request) -> httpx.Response: + if "uid/prometheus" in str(request.url): + return httpx.Response( + 200, json={"status": "success", "data": ["eu/shop/checkout"]} + ) + return httpx.Response(200, json={"tagValues": []}) + + assert stored_services(transport=httpx.MockTransport(handler)) == ["shop/checkout"] + + +def test_stored_services_ignores_wrong_typed_json_fields(): + # A string-typed field must not be iterated char-by-char, and a null + # field must not raise: both degrade to fewer names per the contract. + def handler(request: httpx.Request) -> httpx.Response: + url = str(request.url) + if "uid/tempo" in url: + return httpx.Response(200, json={"tagValues": "checkout"}) + if "uid/loki" in url: + return httpx.Response(200, json={"status": "success", "data": None}) + return httpx.Response(200, json={"status": "success", "data": ["billing"]}) + + assert stored_services(transport=httpx.MockTransport(handler)) == ["billing"] + + +def test_stored_services_is_empty_when_stack_is_down(): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused") + + assert stored_services(transport=httpx.MockTransport(handler)) == [] + + +def test_stored_services_survives_a_malformed_backend_payload(): + # The list warns before a wipe; a broken backend answer must degrade + # to fewer names, never to an exception that blocks the reset. + def handler(request: httpx.Request) -> httpx.Response: + if "uid/tempo" in str(request.url): + return httpx.Response(200, content=b"not json") + return httpx.Response(200, json={"status": "success", "data": ["billing"]}) + + assert stored_services(transport=httpx.MockTransport(handler)) == ["billing"] + + +UP_RESULT = { + "running": True, + "grafana_url": "http://localhost:3000", + "otlp_endpoint": "http://localhost:4317", +} + + +def _trace_reset(monkeypatch, state: str, up=None) -> tuple[list[str], dict]: + """Run stack_reset with docker/backends stubbed; return (call order, result).""" + calls: list[str] = [] + monkeypatch.setattr(stack, "_container_state", lambda: state) + monkeypatch.setattr( + stack, + "stored_services", + lambda: calls.append("stored_services") or ["billing", "checkout"], + ) + monkeypatch.setattr( + stack, "stack_down", lambda: calls.append("stack_down") or {"running": False} + ) + monkeypatch.setattr( + stack, "stack_up", up or (lambda: calls.append("stack_up") or UP_RESULT) + ) + return calls, stack.stack_reset() + + +def test_stack_reset_reports_the_services_it_wiped(monkeypatch): + calls, result = _trace_reset(monkeypatch, "running") + + assert result["services_wiped"] == ["billing", "checkout"] + assert result["running"] is True + assert result["grafana_url"] == "http://localhost:3000" + # The query must happen before the wipe, or the list is always empty. + assert calls == ["stored_services", "stack_down", "stack_up"] + + +def test_stack_reset_boots_a_stopped_container_before_querying(monkeypatch): + # A stopped container (normal after a host reboot) still holds telemetry + # but answers nothing on :3000 - querying it directly would report + # services_wiped: [] while destroying real data. + calls, result = _trace_reset(monkeypatch, "stopped") + + assert calls == ["stack_up", "stored_services", "stack_down", "stack_up"] + assert result["services_wiped"] == ["billing", "checkout"] + + +def test_stack_reset_still_wipes_a_stopped_container_that_cannot_boot(monkeypatch): + # Recovering a broken stack is part of reset's job: if the pre-query + # boot fails, the wipe must proceed rather than error out. + boots: list[int] = [] + + def up(): + boots.append(1) + if len(boots) == 1: + raise RuntimeError("container will not start") + return UP_RESULT + + _, result = _trace_reset(monkeypatch, "stopped", up=up) + + assert result["running"] is True + assert result["services_wiped"] == ["billing", "checkout"]