feat(observability): add OpenTelemetry request tracing
Build and Push Docker Container / build-and-push (push) Successful in 1m34s
Build and Push Docker Container / build-and-push (push) Successful in 1m34s
- Add opt-in OTLP/HTTP export with bounded background batching. - Propagate W3C context and correlate wide events with response trace IDs. - Update my_helpers for Convex child spans and native request IDs. - Respect parent sampling and shut down the exporter off the event loop. - Keep raw URLs, payloads and exception text out of spans. - Cover worker isolation, retries, sampling, privacy and exporter lifecycle. - Document collector configuration and bump NanoShare to 1.27.5. - Refresh locked dependencies, including pytest and watchdog. - Validate the changes with 293 passing tests.
This commit is contained in:
@@ -0,0 +1,182 @@
|
||||
import asyncio
|
||||
|
||||
import httpx
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.trace import SpanKind, StatusCode
|
||||
from quart import Quart, g
|
||||
|
||||
from my_helpers.db.convex.ConvexJob import ConvexJob
|
||||
from my_helpers.db.convex.ConvexWorkerPool import ConvexWorkerPool
|
||||
from my_helpers.db.convex import telemetry
|
||||
from my_modules.observability import register_request_tracing
|
||||
|
||||
def tracing_provider(monkeypatch):
|
||||
exporter = InMemorySpanExporter()
|
||||
provider = TracerProvider(resource=Resource.create({"service.name": "test-api"}))
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
monkeypatch.setattr(telemetry, "_TRACER", provider.get_tracer("convex-test"))
|
||||
return provider, exporter
|
||||
|
||||
def response(value=None, *, error=None):
|
||||
body = {"status": "success", "value": value}
|
||||
if error:
|
||||
body = {"status": "error", "errorMessage": error}
|
||||
return httpx.Response(200, json=body, request=httpx.Request("POST", "https://convex.example.com/api/query"))
|
||||
|
||||
def test_tracing_is_disabled_without_endpoint(monkeypatch):
|
||||
monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", raising=False)
|
||||
app = Quart(__name__)
|
||||
original = app.asgi_app
|
||||
assert register_request_tracing(app) is None
|
||||
assert app.asgi_app == original
|
||||
|
||||
def test_request_to_worker_to_convex_keeps_trace_and_payload_private(monkeypatch):
|
||||
provider, exporter = tracing_provider(monkeypatch)
|
||||
captured = []
|
||||
events = []
|
||||
|
||||
def post(url, *, json, headers, timeout):
|
||||
captured.append({"json": json, "headers": headers})
|
||||
return response({"ok": True})
|
||||
|
||||
monkeypatch.setattr(httpx, "post", post)
|
||||
|
||||
async def run():
|
||||
app = Quart(__name__)
|
||||
pool = ConvexWorkerPool("https://convex.example.com", workers=1)
|
||||
|
||||
@app.before_request
|
||||
async def wide_event():
|
||||
g.wide_event = {"request_id": "req-test"}
|
||||
|
||||
register_request_tracing(app, provider=provider)
|
||||
|
||||
@app.get("/items/<item_id>")
|
||||
async def items(item_id):
|
||||
await pool.submit(ConvexJob("job", "query", "items", "get", {"secret": "never-export"}), timeout=2)
|
||||
events.append(dict(g.wide_event))
|
||||
return {"ok": True}
|
||||
|
||||
await pool.start(health_log_interval=0)
|
||||
try:
|
||||
result = await app.test_client().get(
|
||||
"/items/private-value?token=never-export",
|
||||
headers={"traceparent": "00-11111111111111111111111111111111-2222222222222222-01"},
|
||||
)
|
||||
assert result.status_code == 200
|
||||
assert result.headers["x-trace-id"] == "1" * 32
|
||||
finally:
|
||||
await pool.stop()
|
||||
|
||||
asyncio.run(run())
|
||||
spans = exporter.get_finished_spans()
|
||||
server = next(s for s in spans if s.kind == SpanKind.SERVER)
|
||||
client = next(s for s in spans if s.kind == SpanKind.CLIENT)
|
||||
assert server.name == "GET /items/<item_id>"
|
||||
assert server.parent.span_id == int("2" * 16, 16)
|
||||
assert client.parent.span_id == server.context.span_id
|
||||
assert client.context.trace_id == server.context.trace_id == int("1" * 32, 16)
|
||||
native_id = f"otel-{'1' * 32}-{client.context.span_id:016x}"
|
||||
assert captured[0]["headers"]["convex-request-id"] == native_id
|
||||
assert captured[0]["headers"]["traceparent"] == f"00-{'1' * 32}-{client.context.span_id:016x}-01"
|
||||
assert client.attributes["convex.request_id"] == native_id
|
||||
assert captured[0]["json"]["args"] == {"secret": "never-export"}
|
||||
assert events[0]["trace_id"] == "1" * 32
|
||||
exported = " ".join(str((s.name, dict(s.attributes), s.events, s.status.description)) for s in spans)
|
||||
assert "never-export" not in exported
|
||||
assert "private-value" not in exported
|
||||
assert "token=" not in exported
|
||||
provider.shutdown()
|
||||
|
||||
def test_concurrent_requests_do_not_inherit_worker_or_other_request_context(monkeypatch):
|
||||
provider, exporter = tracing_provider(monkeypatch)
|
||||
captured = []
|
||||
|
||||
def post(url, *, json, headers, timeout):
|
||||
captured.append((json["args"]["request"], headers["convex-request-id"]))
|
||||
return response(None)
|
||||
|
||||
monkeypatch.setattr(httpx, "post", post)
|
||||
|
||||
async def run():
|
||||
app = Quart(__name__)
|
||||
register_request_tracing(app, provider=provider)
|
||||
pool = ConvexWorkerPool("https://convex.example.com", workers=1)
|
||||
|
||||
@app.get("/query/<key>")
|
||||
async def query(key):
|
||||
await pool.submit(ConvexJob(key, "query", "items", "get", {"request": key}), timeout=2)
|
||||
return "ok"
|
||||
|
||||
await pool.start(health_log_interval=0)
|
||||
try:
|
||||
await asyncio.gather(*[
|
||||
app.test_client().get(f"/query/{key}", headers={"traceparent": f"00-{key * 32}-{'3' * 16}-01"})
|
||||
for key in ["a", "b"]
|
||||
])
|
||||
finally:
|
||||
await pool.stop()
|
||||
|
||||
asyncio.run(run())
|
||||
assert len(captured) == 2
|
||||
for key, native_id in captured:
|
||||
assert native_id.startswith(f"otel-{key * 32}-")
|
||||
servers = {s.context.trace_id: s for s in exporter.get_finished_spans() if s.kind == SpanKind.SERVER}
|
||||
for span in exporter.get_finished_spans():
|
||||
if span.kind == SpanKind.CLIENT:
|
||||
assert span.parent.span_id == servers[span.context.trace_id].context.span_id
|
||||
provider.shutdown()
|
||||
|
||||
def test_retry_attempts_get_unique_ids_and_do_not_export_exception_text(monkeypatch):
|
||||
provider, exporter = tracing_provider(monkeypatch)
|
||||
calls = []
|
||||
|
||||
def post(url, *, json, headers, timeout):
|
||||
calls.append(headers["convex-request-id"])
|
||||
if len(calls) == 1:
|
||||
return response(error="credential=never-export")
|
||||
return response("ok")
|
||||
|
||||
monkeypatch.setattr(httpx, "post", post)
|
||||
|
||||
async def run():
|
||||
pool = ConvexWorkerPool("https://convex.example.com", max_retries=1, retry_backoff_secs=0)
|
||||
await pool.start(health_log_interval=0)
|
||||
try:
|
||||
with provider.get_tracer("test").start_as_current_span("request"):
|
||||
assert await pool.submit(ConvexJob("id", "query", "items", "get", {}), timeout=2) == "ok"
|
||||
finally:
|
||||
await pool.stop()
|
||||
|
||||
asyncio.run(run())
|
||||
clients = [s for s in exporter.get_finished_spans() if s.kind == SpanKind.CLIENT]
|
||||
assert len(clients) == 2
|
||||
assert calls[0] != calls[1]
|
||||
assert clients[0].context.trace_id == clients[1].context.trace_id
|
||||
assert clients[0].status.status_code == StatusCode.ERROR
|
||||
assert clients[0].attributes["error.type"] == "ConvexExecutionError"
|
||||
assert clients[0].events == ()
|
||||
assert clients[0].status.description is None
|
||||
provider.shutdown()
|
||||
|
||||
def test_invalid_incoming_traceparent_is_not_used(monkeypatch):
|
||||
provider, exporter = tracing_provider(monkeypatch)
|
||||
|
||||
async def run():
|
||||
app = Quart(__name__)
|
||||
register_request_tracing(app, provider=provider)
|
||||
|
||||
@app.get("/")
|
||||
async def home():
|
||||
return "ok"
|
||||
|
||||
result = await app.test_client().get("/", headers={"traceparent": f"00-{'0' * 32}-{'0' * 16}-01"})
|
||||
assert int(result.headers["x-trace-id"], 16) != 0
|
||||
|
||||
asyncio.run(run())
|
||||
assert exporter.get_finished_spans()[0].parent is None
|
||||
provider.shutdown()
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Exporter configuration, sampling and ASGI lifecycle regression tests."""
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
|
||||
from opentelemetry.trace import StatusCode
|
||||
from quart import Quart, g
|
||||
|
||||
from my_modules.observability import RequestTracingMiddleware, register_request_tracing
|
||||
from quart_common.web.wide_event import register_wide_event_logging
|
||||
|
||||
@pytest.mark.parametrize("rate", ["-0.1", "1.1", "nan", "invalid"])
|
||||
def test_invalid_sample_ratio_rejected(monkeypatch, rate):
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.example:4318")
|
||||
monkeypatch.setenv("OTEL_TRACES_SAMPLER_ARG", rate)
|
||||
with pytest.raises(ValueError):
|
||||
register_request_tracing(Quart(__name__))
|
||||
|
||||
@pytest.mark.parametrize("service,legacy,expected", [
|
||||
(None, None, "nanoshare"),
|
||||
(None, "legacy-name", "legacy-name"),
|
||||
("otel-name", "legacy-name", "otel-name"),
|
||||
])
|
||||
def test_exporter_configuration_and_shutdown(monkeypatch, service, legacy, expected):
|
||||
from opentelemetry.exporter.otlp.proto.http import trace_exporter
|
||||
from my_modules import observability
|
||||
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://collector.example:4318/v1/traces")
|
||||
monkeypatch.setenv("OTEL_TRACES_SAMPLER_ARG", "0.25")
|
||||
for key, value in [("OTEL_SERVICE_NAME", service), ("SERVICE_NAME", legacy)]:
|
||||
if value is None:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
else:
|
||||
monkeypatch.setenv(key, value)
|
||||
exporter = InMemorySpanExporter()
|
||||
timeouts = []
|
||||
|
||||
def make_exporter(**kwargs):
|
||||
timeouts.append(kwargs["timeout"])
|
||||
return exporter
|
||||
|
||||
providers = []
|
||||
monkeypatch.setattr(trace_exporter, "OTLPSpanExporter", make_exporter)
|
||||
monkeypatch.setattr(observability.trace, "set_tracer_provider", providers.append)
|
||||
app = Quart(__name__)
|
||||
provider = register_request_tracing(app)
|
||||
try:
|
||||
assert providers == [provider]
|
||||
assert provider.resource.attributes["service.name"] == expected
|
||||
assert provider.sampler._root.rate == 0.25
|
||||
assert timeouts == [2]
|
||||
shutdowns = []
|
||||
monkeypatch.setattr(provider, "shutdown", lambda: shutdowns.append(True))
|
||||
asyncio.run(app.after_serving_funcs[0]())
|
||||
assert shutdowns == [True]
|
||||
finally:
|
||||
# Restore the real shutdown before releasing the batch processor thread.
|
||||
monkeypatch.undo()
|
||||
provider.shutdown()
|
||||
|
||||
def test_wide_event_and_response_use_the_active_trace(monkeypatch):
|
||||
monkeypatch.setenv("WIDE_EVENT_LOGGING", "true")
|
||||
monkeypatch.setenv("WIDE_EVENT_SAMPLE_RATE", "1")
|
||||
provider = TracerProvider()
|
||||
exporter = InMemorySpanExporter()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
events = []
|
||||
|
||||
class Logger:
|
||||
async def info(self, message):
|
||||
events.append(message)
|
||||
|
||||
error = info
|
||||
warning = info
|
||||
|
||||
async def run():
|
||||
app = Quart(__name__)
|
||||
register_wide_event_logging(app, Logger())
|
||||
register_request_tracing(app, provider=provider)
|
||||
|
||||
@app.get("/files/<file_id>")
|
||||
async def download(file_id):
|
||||
return {"trace_id": g.wide_event["trace_id"]}
|
||||
|
||||
result = await app.test_client().get("/files/private-id", headers={"X-Trace-ID": "untrusted-id"})
|
||||
trace_id = (await result.get_json())["trace_id"]
|
||||
assert result.headers["X-Trace-ID"] == trace_id
|
||||
assert len(trace_id) == 32
|
||||
assert result.headers["X-Request-ID"]
|
||||
return trace_id
|
||||
|
||||
try:
|
||||
trace_id = asyncio.run(run())
|
||||
assert events
|
||||
assert trace_id in events[0]
|
||||
span = exporter.get_finished_spans()[0]
|
||||
assert span.attributes["request.id"]
|
||||
assert span.name == "GET /files/<file_id>"
|
||||
finally:
|
||||
provider.shutdown()
|
||||
|
||||
@pytest.mark.parametrize("parent_sampled,expected_count", [(False, 0), (True, 1)])
|
||||
def test_upstream_sampling_is_respected(parent_sampled, expected_count):
|
||||
provider = TracerProvider(sampler=ParentBased(TraceIdRatioBased(1 if not parent_sampled else 0)))
|
||||
exporter = InMemorySpanExporter()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
|
||||
async def run():
|
||||
app = Quart(__name__)
|
||||
register_request_tracing(app, provider=provider)
|
||||
|
||||
@app.get("/")
|
||||
async def home():
|
||||
return "ok"
|
||||
|
||||
flag = "01" if parent_sampled else "00"
|
||||
result = await app.test_client().get("/", headers={"traceparent": f"00-{'1' * 32}-{'2' * 16}-{flag}"})
|
||||
assert result.headers["X-Trace-ID"] == "1" * 32
|
||||
|
||||
try:
|
||||
asyncio.run(run())
|
||||
assert len(exporter.get_finished_spans()) == expected_count
|
||||
finally:
|
||||
provider.shutdown()
|
||||
|
||||
def test_asgi_error_does_not_export_exception_text():
|
||||
provider = TracerProvider()
|
||||
exporter = InMemorySpanExporter()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
|
||||
async def broken(scope, receive, send):
|
||||
raise RuntimeError("credential=never-export")
|
||||
|
||||
async def run():
|
||||
middleware = RequestTracingMiddleware(broken, provider.get_tracer("test"))
|
||||
with pytest.raises(RuntimeError):
|
||||
await middleware({"type": "http", "method": "GET"}, None, None)
|
||||
|
||||
try:
|
||||
asyncio.run(run())
|
||||
span = exporter.get_finished_spans()[0]
|
||||
assert span.status.status_code == StatusCode.ERROR
|
||||
assert span.attributes["error.type"] == "RuntimeError"
|
||||
assert span.events == ()
|
||||
assert span.status.description is None
|
||||
finally:
|
||||
provider.shutdown()
|
||||
|
||||
def test_non_http_scopes_pass_through():
|
||||
provider = TracerProvider()
|
||||
exporter = InMemorySpanExporter()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
seen = []
|
||||
|
||||
async def app(scope, receive, send):
|
||||
seen.append(scope)
|
||||
|
||||
async def run():
|
||||
middleware = RequestTracingMiddleware(app, provider.get_tracer("test"))
|
||||
for kind in ("websocket", "lifespan"):
|
||||
await middleware({"type": kind}, None, None)
|
||||
|
||||
try:
|
||||
asyncio.run(run())
|
||||
assert [scope["type"] for scope in seen] == ["websocket", "lifespan"]
|
||||
assert not exporter.get_finished_spans()
|
||||
finally:
|
||||
provider.shutdown()
|
||||
Reference in New Issue
Block a user