feat(traces): span pipeline

Adds span creation and the end-of-span gates. PostHogTraces resolves a span's
parent (an explicit traceparent string or handle, else the active span, else a
new trace), attaches the posthogDistinctId and sessionId join keys from the
request context, bounds live spans by count and by age so a leak cannot disable
tracing, and hands each ended span to an exporter unless the client was
disabled. The `traces` option is validated key by key, falling back to the
documented default with a warning. Dropped spans are counted per reason and
reported at most once per flush interval. The export queue arrives in the next
change; this one runs against a stand-in. Not reachable from the client.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TkZAsCciW4PV8ZdcCHmAbA

Author: turnipdabeets

posthog/test/tracing/helpers.py

@@ -0,0 +1,94 @@
+"""Shared fakes for the tracing pipeline tests."""
+
+import threading
+import time
+from contextvars import ContextVar
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+
+from posthog.tracing._config import resolve_traces_config
+from posthog.tracing._drops import DropLog
+from posthog.tracing._pipeline import PostHogTraces
+
+TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736"
+SPAN_ID = "00f067aa0ba902b7"
+
+
+class FakeTimer:
+    """Records the delay it was armed with; fires only when a test says so."""
+
+    instances: list = []
+
+    def __init__(self, delay, fn):
+        self.delay = delay
+        self.fn = fn
+        self.daemon = False
+        self.started = False
+        self.cancelled = False
+        FakeTimer.instances.append(self)
+
+    def start(self):
+        self.started = True
+
+    def cancel(self):
+        self.cancelled = True
+
+    def fire(self):
+        self.fn()
+
+
+class RecordingExporter:
+    """Stands in for the export queue: keeps every record it is handed."""
+
+    def __init__(self):
+        self.records: list = []
+        self.closed = False
+        self.reinitialized = False
+
+    def enqueue(self, record):
+        self.records.append(record)
+
+    def flush(self, timeout=None):
+        pass
+
+    def close(self):
+        self.closed = True
+
+    def warn_if_queued(self):
+        pass
+
+    def reinit_after_fork(self):
+        self.reinitialized = True
+
+
+@pytest.fixture(autouse=True)
+def fake_timers():
+    FakeTimer.instances = []
+    with mock.patch.object(threading, "Timer", FakeTimer):
+        yield FakeTimer
+
+
+@pytest.fixture
+def clock():
+    state = {"now": 1000.0}
+    with mock.patch.object(time, "monotonic", lambda: state["now"]):
+        yield state
+
+
+def make(client=None, context=None, **config):
+    """A pipeline whose ended spans collect on a ``RecordingExporter``."""
+    client = client or SimpleNamespace(disabled=False, send=True)
+    exporter = RecordingExporter()
+    active: ContextVar = ContextVar("active", default=None)
+    resolved = resolve_traces_config(config)
+    drops = DropLog(resolved.flush_interval)
+    pipeline = PostHogTraces(
+        client, resolved, lambda: context or {}, active, exporter, drops
+    )
+    return pipeline, exporter, active
+
+
+def queued(pipeline):
+    return pipeline._exporter.records

posthog/test/tracing/test_config.py

@@ -0,0 +1,184 @@
+import pytest
+
+from posthog.tracing._config import (
+    DEFAULT_FLUSH_INTERVAL_SECONDS,
+    DEFAULT_MAX_EXPORT_BATCH_SIZE,
+    DEFAULT_MAX_LIVE_SPANS,
+    DEFAULT_MAX_QUEUE_SIZE,
+    DEFAULT_MAX_SPAN_AGE_SECONDS,
+    ResolvedTracesConfig,
+    resolve_traces_config,
+)
+
+
+class TestDefaults:
+    def test_applies_the_documented_defaults(self):
+        assert resolve_traces_config({}) == ResolvedTracesConfig(
+            flush_interval=DEFAULT_FLUSH_INTERVAL_SECONDS,
+            max_export_batch_size=DEFAULT_MAX_EXPORT_BATCH_SIZE,
+            max_queue_size=DEFAULT_MAX_QUEUE_SIZE,
+            max_live_spans=DEFAULT_MAX_LIVE_SPANS,
+            max_span_age=DEFAULT_MAX_SPAN_AGE_SECONDS,
+        )
+
+    def test_leaves_service_name_unset_so_the_encoder_supplies_unknown_service(self):
+        assert resolve_traces_config({}).service_name is None
+
+    @pytest.mark.parametrize("config", [None, "nope", 42, ["a"]])
+    def test_a_non_dict_config_falls_back_to_defaults(self, config):
+        assert resolve_traces_config(config) == resolve_traces_config({})
+
+
+class TestExplicitValues:
+    def test_honours_explicit_values(self):
+        resolved = resolve_traces_config(
+            {
+                "service_name": "api",
+                "service_version": "1.2.3",
+                "environment": "prod",
+                "flush_interval": 2,
+                "max_export_batch_size": 100,
+                "max_queue_size": 400,
+                "max_live_spans": 50,
+                "max_span_age": 60,
+            }
+        )
+        assert resolved == ResolvedTracesConfig(
+            service_name="api",
+            service_version="1.2.3",
+            environment="prod",
+            flush_interval=2.0,
+            max_export_batch_size=100,
+            max_queue_size=400,
+            max_live_spans=50,
+            max_span_age=60.0,
+        )
+
+    @pytest.mark.parametrize(
+        "value",
+        [0, -1, 0.5, 512.7, float("nan"), float("inf"), "512", True, None],
+    )
+    def test_falls_back_for_an_unusable_batch_size(self, value):
+        assert (
+            resolve_traces_config(
+                {"max_export_batch_size": value}
+            ).max_export_batch_size
+            == DEFAULT_MAX_EXPORT_BATCH_SIZE
+        )
+
+    def test_accepts_a_whole_number_float_batch_size(self):
+        assert (
+            resolve_traces_config(
+                {"max_export_batch_size": 100.0}
+            ).max_export_batch_size
+            == 100
+        )
+
+    def test_an_unusable_knob_keeps_the_rest_of_the_config(self):
+        resolved = resolve_traces_config(
+            {"service_name": "api", "max_live_spans": float("inf")}
+        )
+        assert resolved.service_name == "api"
+        assert resolved.max_live_spans == DEFAULT_MAX_LIVE_SPANS
+
+    @pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), "5", False])
+    def test_falls_back_for_an_unusable_flush_interval(self, value):
+        assert (
+            resolve_traces_config({"flush_interval": value}).flush_interval
+            == DEFAULT_FLUSH_INTERVAL_SECONDS
+        )
+
+    @pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf")])
+    def test_falls_back_for_unusable_live_span_bounds(self, value):
+        resolved = resolve_traces_config(
+            {"max_live_spans": value, "max_span_age": value}
+        )
+        assert resolved.max_live_spans == DEFAULT_MAX_LIVE_SPANS
+        assert resolved.max_span_age == DEFAULT_MAX_SPAN_AGE_SECONDS
+
+    def test_keeps_the_queue_at_least_as_large_as_the_export_batch(self):
+        resolved = resolve_traces_config({"max_export_batch_size": 4096})
+        assert resolved.max_queue_size == 4096
+
+    def test_floors_an_explicit_queue_size_at_the_batch_size(self):
+        resolved = resolve_traces_config(
+            {"max_export_batch_size": 10, "max_queue_size": 3}
+        )
+        assert resolved.max_queue_size == 10
+
+    def test_ignores_a_non_string_named_field(self):
+        assert resolve_traces_config({"service_name": 42}).service_name is None
+
+
+class TestResourceAttributes:
+    def test_lets_otlp_resource_attributes_override_the_named_fields(self):
+        resolved = resolve_traces_config(
+            {
+                "service_name": "named",
+                "resource_attributes": {"service.name": "from-attrs", "region": "eu"},
+            }
+        )
+        assert resolved.service_name == "from-attrs"
+        assert resolved.resource_attributes == {
+            "service.name": "from-attrs",
+            "region": "eu",
+        }
+
+    def test_attaches_host_attributes_and_lets_user_attributes_override_them(self):
+        resolved = resolve_traces_config(
+            {"resource_attributes": {"os.name": "Custom"}},
+            {"os.name": "Linux", "os.version": "6.1"},
+        )
+        assert resolved.resource_attributes == {
+            "os.name": "Custom",
+            "os.version": "6.1",
+        }
+
+    def test_ignores_a_non_dict_value(self):
+        assert (
+            resolve_traces_config({"resource_attributes": ["a"]}).resource_attributes
+            == {}
+        )
+
+    def test_drops_an_identity_key_that_is_not_a_string(self):
+        resolved = resolve_traces_config(
+            {
+                "service_name": "named",
+                "resource_attributes": {
+                    "service.name": 42,
+                    "deployment.environment": 1,
+                },
+            }
+        )
+        assert resolved.service_name == "named"
+        assert resolved.environment is None
+        assert "service.name" not in resolved.resource_attributes
+
+    def test_keeps_the_readable_attributes_when_one_accessor_raises(self):
+        class Explosive(dict):
+            def __getitem__(self, key):
+                if key == "bad":
+                    raise RuntimeError("boom")
+                return super().__getitem__(key)
+
+        resolved = resolve_traces_config(
+            {"resource_attributes": Explosive(good=1, bad=2)}
+        )
+        assert resolved.resource_attributes == {"good": 1}
+
+
+class TestHostileResourceAttributeKeys:
+    def test_drops_only_a_key_that_cannot_be_stringified(self):
+        class HostileKey:
+            def __str__(self):
+                raise RuntimeError("no")
+
+        resolved = resolve_traces_config(
+            {
+                "service_name": "api",
+                "resource_attributes": {HostileKey(): 1, "team": "x"},
+            }
+        )
+        assert resolved.service_name == "api"
+        assert resolved.resource_attributes["team"] == "x"
+        assert all(isinstance(key, str) for key in resolved.resource_attributes)