7696c1c6da
Add `perf` as the 12th skinny package with full OpenTelemetry instrumentation for pipeline traces, Prometheus metrics, and Loki-correlated logs. - src/perf/: TracerProvider, MeterProvider, LoggerProvider bridge, PipelineCollector, psutil system metrics, FileExporter fallback, FastAPI middleware, auto-file Gitea issues on step/test failure - infra/otel/: OTel Collector fan-out (traces→Jaeger, metrics→Prometheus, logs→Loki), all services migrated from jaeger:4317 to collector - infra/grafana/dashboards/: 8-panel pipeline performance dashboard - Pipeline runner auto-instruments all 189 steps with zero-cost MockTracer/MockSpan no-ops when telemetry is disabled - 82 new tests, 12,055 total passing Closes #203, closes #204, closes #205, closes #206, closes #207, closes #208, closes #209, closes #210, closes #211, closes #212, closes #213
150 lines
4.2 KiB
Python
150 lines
4.2 KiB
Python
"""Integration tests — runner with perf collector, middleware wiring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from opentelemetry import trace
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import (
|
|
SimpleSpanProcessor,
|
|
SpanExporter,
|
|
SpanExportResult,
|
|
)
|
|
|
|
|
|
class _MemoryExporter(SpanExporter):
|
|
def __init__(self):
|
|
self.spans = []
|
|
|
|
def export(self, spans):
|
|
self.spans.extend(spans)
|
|
return SpanExportResult.SUCCESS
|
|
|
|
def shutdown(self):
|
|
pass
|
|
|
|
|
|
@pytest.fixture()
|
|
def trace_setup(monkeypatch):
|
|
"""Wire up real tracer for integration tests."""
|
|
monkeypatch.setenv("STACK_TELEMETRY", "false")
|
|
original = trace.get_tracer_provider()
|
|
exporter = _MemoryExporter()
|
|
provider = TracerProvider()
|
|
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
|
trace.set_tracer_provider(provider)
|
|
|
|
import perf._tracer as tmod
|
|
|
|
tmod._provider_ready = True
|
|
yield exporter
|
|
tmod._provider_ready = False
|
|
provider.shutdown()
|
|
trace._TRACER_PROVIDER = original
|
|
trace._TRACER_PROVIDER_SET_ONCE._done = False
|
|
|
|
|
|
class TestRunnerIntegration:
|
|
"""run_pipeline emits spans via PipelineCollector."""
|
|
|
|
def test_pipeline_creates_spans(self, trace_setup):
|
|
import polars as pl
|
|
|
|
from aco.pipe.runner import run_pipeline
|
|
|
|
# Simple two-step pipeline
|
|
def step_a(input_layer__eligibility):
|
|
return input_layer__eligibility.select("person_id")
|
|
|
|
def step_b(core__step_a):
|
|
return core__step_a
|
|
|
|
eligibility = pl.DataFrame({"person_id": ["P001", "P002"]})
|
|
|
|
def load(ref):
|
|
if ref == "input_layer.eligibility":
|
|
return eligibility
|
|
raise KeyError(ref)
|
|
|
|
cache = run_pipeline(
|
|
[
|
|
("core.step_a", step_a),
|
|
("core.step_b", step_b),
|
|
],
|
|
load,
|
|
)
|
|
|
|
assert "core.step_a" in cache
|
|
assert "core.step_b" in cache
|
|
assert len(cache["core.step_a"]) == 2
|
|
|
|
# Verify spans were created
|
|
names = [s.name for s in trace_setup.spans]
|
|
assert "pipeline.run" in names
|
|
assert "step.core.step_a" in names
|
|
assert "step.core.step_b" in names
|
|
|
|
def test_pipeline_records_row_counts(self, trace_setup):
|
|
import polars as pl
|
|
|
|
from aco.pipe.runner import run_pipeline
|
|
|
|
def step_one(input_layer__data):
|
|
return input_layer__data
|
|
|
|
data = pl.DataFrame({"id": list(range(50))})
|
|
|
|
run_pipeline(
|
|
[("core.step_one", step_one)],
|
|
lambda ref: data,
|
|
)
|
|
|
|
step_span = next(s for s in trace_setup.spans if s.name == "step.core.step_one")
|
|
assert step_span.attributes["rows.out"] == 50
|
|
|
|
def test_pipeline_without_perf_still_works(self, monkeypatch):
|
|
"""Verify graceful degradation if perf collector import fails."""
|
|
import polars as pl
|
|
|
|
from aco.pipe.runner import run_pipeline
|
|
|
|
def step_x(input_layer__table):
|
|
return input_layer__table
|
|
|
|
data = pl.DataFrame({"col": [1, 2, 3]})
|
|
cache = run_pipeline(
|
|
[("ns.step_x", step_x)],
|
|
lambda ref: data,
|
|
)
|
|
assert len(cache["ns.step_x"]) == 3
|
|
|
|
def test_pipeline_name_inference(self):
|
|
from aco.pipe.runner import _infer_pipeline_name
|
|
|
|
assert (
|
|
_infer_pipeline_name([("readmissions._int_enc", lambda: None)])
|
|
== "readmissions"
|
|
)
|
|
assert _infer_pipeline_name([("core.encounter", lambda: None)]) == "core"
|
|
assert _infer_pipeline_name([]) == "unknown"
|
|
|
|
|
|
class TestMiddleware:
|
|
"""perf.middleware.instrument is a safe no-op when disabled."""
|
|
|
|
def test_instrument_noop_when_disabled(self, monkeypatch):
|
|
monkeypatch.setenv("STACK_TELEMETRY", "false")
|
|
from unittest.mock import MagicMock
|
|
|
|
from perf.middleware import instrument
|
|
|
|
app = MagicMock()
|
|
instrument(app) # should not raise
|
|
|
|
def test_server_import(self):
|
|
"""Verify server.py imports cleanly with perf wiring."""
|
|
from api.server import app
|
|
|
|
assert app is not None
|
|
assert app.title == "stack"
|