"""obs.py: the three things that are wrong by default. Not tested here: that structlog logs, that prometheus counts, that OTEL traces. Those are the libraries' tests. What is tested is every place where the default is a bug — the histogram buckets, the context that does not cross a queue, and the contextvars that leak between tasks. """ from __future__ import annotations import io import json import logging import re from collections.abc import Iterator from pathlib import Path from typing import Any from uuid import uuid4 import pytest import structlog import yaml from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from prometheus_client import REGISTRY from svcforge_core import obs from svcforge_core.settings import Settings # W3C traceparent: version-traceid-spanid-flags. # # The flags byte is matched loosely and the sampled bit checked separately, on purpose. # Module 7's acceptance line says `00-<32 hex>-<16 hex>-01`, and current SDKs emit `-03`: # bit 0x01 is `sampled`, and bit 0x02 is `random-trace-id` from trace-context level 2. An # assertion pinned to `-01` fails on a spec revision that changed nothing we care about. # What we care about is the trace being sampled, which is bit 0x01 and nothing else. TRACEPARENT_RE = re.compile( r"^00-(?P[0-9a-f]{32})-(?P[0-9a-f]{16})-(?P[0-9a-f]{2})$" ) SAMPLED_BIT = 0x01 # A provision takes minutes. prometheus_client's default buckets end at 10 seconds. DEFAULT_TOP_BUCKET = 10.0 def _settings(**over: object) -> Settings: return Settings(pg_dsn="postgresql://u:p@localhost:5432/db", **over) # type: ignore[arg-type] @pytest.fixture def logs() -> Iterator[io.StringIO]: """setup() against a captured stream, restoring the process-wide config afterwards. setup() is deliberately global and deliberately once-only, which makes it deliberately awkward to test. Reaching into the module flag is the honest price of that; the alternative is a seam that exists only for tests. """ stream = io.StringIO() saved_handlers = logging.getLogger().handlers[:] saved_level = logging.getLogger().level saved_config = structlog.get_config() obs._configured = False obs.setup("test-service", _settings()) # Re-point the handler setup() installed at our buffer; everything else it configured — # processors, formatter, the JSON renderer last — is exactly what production gets. handler = logging.getLogger().handlers[0] assert isinstance(handler, logging.StreamHandler) handler.setStream(stream) try: yield stream finally: structlog.contextvars.clear_contextvars() structlog.configure(**saved_config) logging.getLogger().handlers = saved_handlers logging.getLogger().setLevel(saved_level) obs._configured = False def _lines(stream: io.StringIO) -> list[dict[str, Any]]: return [json.loads(line) for line in stream.getvalue().splitlines() if line.strip()] # --- Buckets ------------------------------------------------------------------------------ def test_provision_histogram_has_a_bucket_for_a_thirty_minute_provision() -> None: """The acceptance check, as a unit test: `le="1800"` exists. With the library defaults the top finite bucket is 10s, every provision lands in +Inf, and histogram_quantile interpolates across a bucket spanning 10s to infinity. The p95 it returns is meaningless, and it is meaningless silently, which is why this is asserted rather than eyeballed on a dashboard. """ bounds = obs.PROVISION_TIME._upper_bounds assert 1800.0 in bounds assert bounds[-1] == float("inf") assert max(b for b in bounds if b != float("inf")) > DEFAULT_TOP_BUCKET def test_provision_histogram_exports_the_1800_bucket() -> None: """Same claim, checked through the exposition format the scrape actually reads.""" obs.PROVISION_TIME.observe(42.0) bucket = REGISTRY.get_sample_value("svcforge_provision_duration_seconds_bucket", {"le": "1800.0"}) assert bucket is not None def test_every_metric_the_alerts_query_actually_exists() -> None: """Parse the alerts out of values.yaml and check each metric is really registered. Reading the chart is the whole point. An earlier version of this test asserted against a hardcoded list, which meant it could only catch a rename on the *code* side — rename a metric in values.yaml and it stayed green while the alert silently never fired again. A test that duplicates the thing it is meant to cross-check is not a cross-check. """ values = yaml.safe_load( (Path(__file__).resolve().parents[2] / "deploy" / "chart" / "values.yaml").read_text() ) rules = (values.get("prometheusRule") or {}).get("rules") or [] assert rules, "no alert rules found under prometheusRule.rules — this test would silently pass" # Prometheus appends _bucket/_sum/_count to a histogram's series; the collector is # registered under the base name. suffixes = ("_bucket", "_sum", "_count") registered = set(REGISTRY._names_to_collectors) checked = 0 for rule in rules: for token in re.findall(r"\bsvcforge_[a-z0-9_]+\b", str(rule.get("expr", ""))): base = next((token[: -len(s)] for s in suffixes if token.endswith(s)), token) assert base in registered, ( f"alert {rule.get('alert')!r} queries {token!r}, but no such metric is registered" ) checked += 1 assert checked >= len(rules), "expected at least one metric reference per alert" # --- Trace context across the queue ------------------------------------------------------- def test_traceparent_round_trips_through_a_string() -> None: """inject -> a W3C string -> extract -> the same trace. This is the queue crossing.""" provider = TracerProvider() tracer = provider.get_tracer("test") with tracer.start_as_current_span("api.post") as span: traceparent = obs.inject_traceparent() api_trace_id = span.get_span_context().trace_id assert traceparent is not None match = TRACEPARENT_RE.match(traceparent) assert match is not None, traceparent assert match["trace_id"] == format(api_trace_id, "032x") assert int(match["flags"], 16) & SAMPLED_BIT, "unsampled: the worker's span would be dropped" # The worker's side: a fresh context, minutes later, in another process. ctx = obs.context_from_traceparent(traceparent) with tracer.start_as_current_span("worker.claim", context=ctx) as worker_span: assert worker_span.get_span_context().trace_id == api_trace_id def test_inject_returns_none_without_a_span() -> None: """A task the reconciler enqueued has no inbound request. Null column, not an error.""" assert obs.inject_traceparent() is None def test_context_from_traceparent_survives_none_and_garbage() -> None: """A malformed traceparent must start a new trace, never fail a provision. Extract does not raise on bad input — it returns a context with no span. Asserted here because the alternative would be a tenant's provision failing over a telemetry header. """ for bad in (None, "", "not-a-traceparent", "00-tooshort-01"): ctx = obs.context_from_traceparent(bad) assert not trace.get_current_span(ctx).get_span_context().is_valid # --- Contextvars -------------------------------------------------------------------------- def test_every_line_carries_instance_id_task_id_and_team(logs: io.StringIO) -> None: """The acceptance check: bind once at claim, and the keys ride on every line after.""" instance_id = uuid4() obs.bind_task_context(instance_id, 42, "platform") obs.get_logger("t").info("provision.started") obs.get_logger("t").warning("helm.slow") for line in _lines(logs): assert line["instance_id"] == str(instance_id) assert line["task_id"] == 42 assert line["team"] == "platform" def test_foreign_stdlib_logs_are_json_and_carry_the_context(logs: io.StringIO) -> None: """psycopg and uvicorn log through stdlib `logging`, and their lines must parse too. Without the ProcessorFormatter bridge these arrive as bare text on the same stdout, and every one of them is a parse failure in the collector. """ obs.bind_task_context(uuid4(), 7, "payments") logging.getLogger("some.library").warning("connection reset") line = _lines(logs)[-1] assert line["event"] == "connection reset" assert line["task_id"] == 7 assert line["team"] == "payments" def test_bind_task_context_clears_the_previous_task(logs: io.StringIO) -> None: """The bug this prevents: task 2's log line naming task 1's tenant. A worker coroutine reuses its context across claim-loop iterations. Bind without clearing and the stale instance_id survives into the next task, which means the log for the incident you are debugging points at the wrong customer. """ obs.bind_task_context(uuid4(), 1, "team-a") second = uuid4() obs.bind_task_context(second, 2, "team-b") obs.get_logger("t").info("claimed") line = _lines(logs)[-1] assert line["instance_id"] == str(second) assert line["task_id"] == 2 assert line["team"] == "team-b" def test_json_renderer_is_last_and_output_is_one_object_per_line(logs: io.StringIO) -> None: """JSONRenderer last in the chain, and nothing after it. A processor appended after the renderer receives a `str` where it expects a dict and raises. The symptom is not a crash — structlog's default is to fail the log call, so the line simply never appears. """ obs.get_logger("t").info("hello", extra_key="value") lines = _lines(logs) assert len(lines) == 1 assert lines[0]["event"] == "hello" assert lines[0]["extra_key"] == "value" assert lines[0]["level"] == "info" assert lines[0]["service"] == "test-service" assert "timestamp" in lines[0] def test_setup_is_idempotent(logs: io.StringIO) -> None: """Called twice, one handler, one line. Two handlers would be two of every line. Each service calls setup() from its entrypoint, and an entrypoint that imports another entrypoint (the CLI shelling into the reconciler) calls it twice. """ obs.setup("test-service", _settings()) obs.setup("test-service", _settings()) # Count ours, not pytest's — its capture handler is on the root logger too. ours = [ h for h in logging.getLogger().handlers if isinstance(h.formatter, structlog.stdlib.ProcessorFormatter) ] assert len(ours) == 1 obs.get_logger("t").info("once") assert len(_lines(logs)) == 1 def test_metrics_are_single_process_in_memory() -> None: """One process per pod, scale with replicas. Multiprocess mode is not in use. `ValueClass` is prometheus_client's fork in the road, chosen at import from PROMETHEUS_MULTIPROC_DIR: MutexValue keeps counters in memory, MultiProcessValue mmaps them into a shared directory. This repo takes the other fix for `uvicorn --workers 4` corrupting counters — one process per pod — so MutexValue is the correct answer, and a stray PROMETHEUS_MULTIPROC_DIR in a Deployment's env would silently change it to the other one along with the meaning of every gauge. """ from prometheus_client import values assert values.ValueClass is values.MutexValue # And the registry the services expose is the default in-memory one, not a # MultiProcessCollector reading files off disk. obs.QUEUE_DEPTH.set(3) assert REGISTRY.get_sample_value("svcforge_queue_depth") == 3.0