From d0bd9d7fb4f1fbdfb5179aeb6332cf2a96af0cae Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Thu, 10 Sep 2026 21:41:34 -0400 Subject: [PATCH 1/4] feat(traces): per-span limits and exception stacktraces Bounds what one span can hold, per the traces spec. A span keeps at most max_attributes_per_span user attributes and max_events_per_span events (128 each, earliest-set wins), and each event at most 128 attributes; what the caps refuse is reported as droppedAttributesCount / droppedEventsCount, clamped to uint32. The posthogDistinctId and sessionId join keys are exempt. max_attribute_value_length (8192) bounds every string an attribute holds, nested ones included, along with span and event names, status messages and resource attributes, so one large value cannot get a span dropped as too large. Recorded exceptions now carry exception.stacktrace, keeping the tail of the traceback where Python puts the raising frame. Not reachable from the client. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01TkZAsCciW4PV8ZdcCHmAbA --- posthog/test/tracing/test_config.py | 39 +++++ posthog/test/tracing/test_export.py | 17 +++ posthog/test/tracing/test_limits.py | 151 +++++++++++++++++++ posthog/test/tracing/test_otlp.py | 33 +++++ posthog/test/tracing/test_pipeline.py | 40 +++++ posthog/test/tracing/test_span.py | 205 +++++++++++++++++++++++++- posthog/tracing/_config.py | 19 +++ posthog/tracing/_export.py | 14 +- posthog/tracing/_limits.py | 150 +++++++++++++++++++ posthog/tracing/_otlp.py | 35 ++++- posthog/tracing/_pipeline.py | 25 +++- posthog/tracing/_sanitize.py | 6 +- posthog/tracing/_span.py | 118 +++++++++++++-- posthog/tracing/span.py | 6 +- 14 files changed, 826 insertions(+), 32 deletions(-) create mode 100644 posthog/test/tracing/test_limits.py create mode 100644 posthog/tracing/_limits.py diff --git a/posthog/test/tracing/test_config.py b/posthog/test/tracing/test_config.py index f9e28536..29fee059 100644 --- a/posthog/test/tracing/test_config.py +++ b/posthog/test/tracing/test_config.py @@ -1,6 +1,9 @@ import pytest from posthog.tracing._config import ( + DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH, + DEFAULT_MAX_ATTRIBUTES_PER_SPAN, + DEFAULT_MAX_EVENTS_PER_SPAN, DEFAULT_FLUSH_INTERVAL_SECONDS, DEFAULT_MAX_EXPORT_BATCH_SIZE, DEFAULT_MAX_LIVE_SPANS, @@ -190,3 +193,39 @@ def __str__(self): assert resolved.service_name == "api" assert resolved.resource_attributes["team"] == "x" assert all(isinstance(key, str) for key in resolved.resource_attributes) + + +class TestSpanLimitKnobs: + def test_defaults_to_opentelemetrys_counts_and_a_finite_value_length(self): + resolved = resolve_traces_config({}) + assert ( + resolved.max_attributes_per_span == DEFAULT_MAX_ATTRIBUTES_PER_SPAN == 128 + ) + assert resolved.max_events_per_span == DEFAULT_MAX_EVENTS_PER_SPAN == 128 + assert resolved.max_attribute_value_length == DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH + assert DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH == 8192 + + def test_honours_explicit_values(self): + resolved = resolve_traces_config( + { + "max_attributes_per_span": 10, + "max_events_per_span": 5, + "max_attribute_value_length": 100, + } + ) + assert resolved.max_attributes_per_span == 10 + assert resolved.max_events_per_span == 5 + assert resolved.max_attribute_value_length == 100 + + @pytest.mark.parametrize("value", [0, -1, 1.5, "128", None, True]) + def test_an_unusable_value_falls_back_rather_than_dropping_every_span(self, value): + resolved = resolve_traces_config( + { + "max_attributes_per_span": value, + "max_events_per_span": value, + "max_attribute_value_length": value, + } + ) + assert resolved.max_attributes_per_span == DEFAULT_MAX_ATTRIBUTES_PER_SPAN + assert resolved.max_events_per_span == DEFAULT_MAX_EVENTS_PER_SPAN + assert resolved.max_attribute_value_length == DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH diff --git a/posthog/test/tracing/test_export.py b/posthog/test/tracing/test_export.py index aa1cc15c..b4aebf2d 100644 --- a/posthog/test/tracing/test_export.py +++ b/posthog/test/tracing/test_export.py @@ -950,6 +950,23 @@ def test_a_forked_child_drops_the_inherited_queue_and_timer(self): assert [r.name for r in queued(pipeline)] == ["child-span"] +class TestResourceAttributes: + def test_bounds_resource_attributes_on_every_batch(self): + sender = FakeSender(SendOutcome("ok")) + pipeline, _, _ = make_traces( + sender=sender, + max_attribute_value_length=5, + resource_attributes={"team": "platform-infrastructure"}, + ) + pipeline.start_span("a").end() + pipeline.flush() + resource = { + kv["key"]: kv["value"] + for kv in sender.payloads[0]["resourceSpans"][0]["resource"]["attributes"] + } + assert resource["team"] == {"stringValue": "platf"} + + def waits_advance(clock, pipeline): """Make the exporter's backoff wait move the fake clock instead of sleeping.""" waited = [] diff --git a/posthog/test/tracing/test_limits.py b/posthog/test/tracing/test_limits.py new file mode 100644 index 00000000..cc1d95d5 --- /dev/null +++ b/posthog/test/tracing/test_limits.py @@ -0,0 +1,151 @@ +import pytest + +from posthog.tracing._limits import ( + bound_attributes, + truncate_attribute_value, + truncate_attributes, +) +from posthog.tracing._otlp import ( + CIRCULAR_VALUE, + MAX_VALUE_ITEMS, + MAX_VALUE_NODES, + TRUNCATED_VALUE, + to_any_value, +) +from posthog.tracing._sanitize import UNSERIALIZABLE_VALUE + + +class TestTruncateAttributeValue: + def test_truncates_a_long_string(self): + assert truncate_attribute_value("x" * 40000, 8192) == "x" * 8192 + + def test_returns_a_short_string_unchanged(self): + assert truncate_attribute_value("short", 8192) == "short" + + @pytest.mark.parametrize("value", [42, 1.5, True, None, 2**70]) + def test_leaves_numbers_booleans_and_none_alone(self, value): + assert truncate_attribute_value(value, 3) == value + + def test_reaches_strings_nested_in_mappings_and_lists(self): + value = {"body": "x" * 40000, "items": ["y" * 20, {"deep": "z" * 20}]} + assert truncate_attribute_value(value, 8) == { + "body": "x" * 8, + "items": ["y" * 8, {"deep": "z" * 8}], + } + + def test_does_not_mutate_the_callers_value(self): + value = {"body": "x" * 20} + truncate_attribute_value(value, 4) + assert value == {"body": "x" * 20} + + def test_a_self_referencing_value_terminates_with_the_encoders_marker(self): + value: dict = {"name": "n" * 20} + value["self"] = value + assert truncate_attribute_value(value, 4) == { + "name": "nnnn", + "self": CIRCULAR_VALUE, + } + + def test_siblings_sharing_one_object_are_not_a_cycle(self): + shared = {"k": "v" * 10} + assert truncate_attribute_value([shared, shared], 2) == [ + {"k": "vv"}, + {"k": "vv"}, + ] + + def test_marks_items_past_the_encoders_item_cap(self): + bounded = truncate_attribute_value(["a"] * (MAX_VALUE_ITEMS + 5), 8) + assert len(bounded) == MAX_VALUE_ITEMS + 1 + assert bounded[-1] == TRUNCATED_VALUE + # The encoder emits the same shape it would have for the original. + assert to_any_value(bounded) == to_any_value(["a"] * (MAX_VALUE_ITEMS + 5)) + + def test_stringifies_and_bounds_a_type_the_encoder_would_stringify(self): + class Big: + def __str__(self): + return "b" * 100 + + assert truncate_attribute_value(Big(), 10) == "b" * 10 + assert truncate_attribute_value(b"\x00" * 100, 10) == "b'\\x00\\x00" + + def test_a_key_the_encoder_skips_does_not_spend_the_walks_budget(self): + # The encoder drops "" without charging for its value, so the walk must + # too, or "x" would ship unbounded once the walk's budget ran out. + value = {"": list(range(999)), "a": [list(range(999))] * 9, "x": "A" * 1000} + bounded = truncate_attribute_value(value, 100) + assert "" not in bounded + assert bounded["x"] == "A" * 100 + encoded = to_any_value(bounded)["kvlistValue"]["values"] + x = next(kv for kv in encoded if kv["key"] == "x") + assert len(x["value"]["stringValue"]) == 100 + + def test_a_raising_str_costs_only_that_value(self): + class Hostile: + def __str__(self): + raise RuntimeError("no") + + assert truncate_attribute_value({"a": Hostile(), "b": "ok"}, 8) == { + "a": UNSERIALIZABLE_VALUE, + "b": "ok", + } + + def test_a_raising_accessor_costs_only_that_key(self): + class Explosive(dict): + def __getitem__(self, key): + if key == "bad": + raise RuntimeError("no") + return super().__getitem__(key) + + assert truncate_attribute_value(Explosive(good="g" * 9, bad=1), 3) == { + "good": "ggg", + "bad": UNSERIALIZABLE_VALUE, + } + + +class TestBoundAttributes: + def test_keeps_the_earliest_entries_and_counts_the_rest(self): + source = {f"k{i}": i for i in range(130)} + attributes, dropped = bound_attributes(source, 128, 8192) + assert list(attributes) == [f"k{i}" for i in range(128)] + assert dropped == 2 + + def test_a_none_value_spends_no_slot(self): + attributes, dropped = bound_attributes({"a": None, "b": 1, "c": 2}, 2, 8) + assert attributes == {"b": 1, "c": 2} + assert dropped == 0 + + def test_bounds_each_value(self): + attributes, _ = bound_attributes({"a": "x" * 20}, 2, 5) + assert attributes == {"a": "xxxxx"} + + def test_a_non_mapping_yields_nothing(self): + assert bound_attributes(["a"], 2, 5) == ({}, 0) + + +class TestTruncateAttributes: + def test_bounds_every_value_as_a_copy(self): + source = {"service.name": "api", "blob": "x" * 20} + assert truncate_attributes(source, 4) == {"service.name": "api", "blob": "xxxx"} + assert source["blob"] == "x" * 20 + + +class TestWalkBounds: + def test_walks_no_more_strings_than_the_encoder_would_emit(self): + # A thousand paths to one shared list of a thousand strings. Leaves are + # charged against the node budget, as in the encoder, so the walk does + # not copy every string on every path. + inner = ["x" * 50] * 1000 + bounded = truncate_attribute_value([inner] * 1000, 8) + walked = sum( + 1 + for items in bounded + if items is not inner + for item in items + if item == "x" * 8 + ) + assert 0 < walked <= MAX_VALUE_NODES + + def test_stops_walking_a_mapping_at_the_encoders_item_cap(self): + value = {f"k{i}": "v" * 50 for i in range(MAX_VALUE_ITEMS + 50)} + bounded = truncate_attribute_value(value, 4) + assert len(bounded) == MAX_VALUE_ITEMS diff --git a/posthog/test/tracing/test_otlp.py b/posthog/test/tracing/test_otlp.py index 5fdf99bd..4f5999a4 100644 --- a/posthog/test/tracing/test_otlp.py +++ b/posthog/test/tracing/test_otlp.py @@ -284,6 +284,39 @@ def test_marks_a_root_span_as_known_not_remote(self): def test_marks_a_header_parent_as_remote(self): assert build_otlp_span(record(parent_is_remote=True))["flags"] == 0x301 + def test_omits_dropped_counts_when_nothing_was_dropped(self): + span = build_otlp_span(record(events=[SpanEventRecord("e", START_NS)])) + assert "droppedAttributesCount" not in span + assert "droppedEventsCount" not in span + assert "droppedAttributesCount" not in span["events"][0] + + def test_emits_dropped_counts_on_the_span_and_its_events(self): + span = build_otlp_span( + record( + dropped_attributes_count=2, + dropped_events_count=3, + events=[SpanEventRecord("e", START_NS, {"k": 1}, 4)], + ) + ) + assert span["droppedAttributesCount"] == 2 + assert span["droppedEventsCount"] == 3 + assert span["events"][0]["droppedAttributesCount"] == 4 + + @pytest.mark.parametrize( + "value,expected", + [ + (2**40, 0xFFFFFFFF), + (-1, 0), + (1.9, 1), + ("3", 0), + (True, 0), + (float("inf"), 0), + ], + ) + def test_clamps_a_dropped_count_to_uint32(self, value, expected): + span = build_otlp_span(record(dropped_attributes_count=value)) + assert span.get("droppedAttributesCount", 0) == expected + def test_propagates_an_inbound_sampled_out_flag(self): assert build_otlp_span(record(trace_flags="00"))["flags"] == 0x100 diff --git a/posthog/test/tracing/test_pipeline.py b/posthog/test/tracing/test_pipeline.py index 3b245c57..7920cb6b 100644 --- a/posthog/test/tracing/test_pipeline.py +++ b/posthog/test/tracing/test_pipeline.py @@ -11,14 +11,17 @@ from posthog.test.tracing.helpers import ( SPAN_ID, TRACE_ID, + FakeSender, clock, fake_timers, make, + make_traces, queued, ) from posthog.tracing import _pipeline as pipeline_module from posthog.tracing import _span as span_module from posthog.tracing._drops import DropLog +from posthog.tracing._transport import SendOutcome from posthog.tracing._span import NOOP_SPAN, PassThroughSpan, RecordingSpan __all__ = ["clock", "fake_timers"] @@ -414,6 +417,17 @@ def test_lets_user_attributes_win_on_collision(self): pipeline.start_span("a", attributes={"posthogDistinctId": "override"}).end() assert queued(pipeline)[0].attributes["posthogDistinctId"] == "override" + def test_the_join_keys_survive_a_span_at_its_attribute_cap(self): + pipeline, _, _ = make( + context={"distinct_id": "user-1", "session_id": "sess-1"}, + max_attributes_per_span=2, + ) + pipeline.start_span("a", attributes={"x": 1, "y": 2, "z": 3}).end() + record = queued(pipeline)[0] + assert record.attributes["posthogDistinctId"] == "user-1" + assert record.attributes["sessionId"] == "sess-1" + assert record.dropped_attributes_count == 1 + def test_still_records_the_span_when_reading_context_raises(self): pipeline, _, _ = make() pipeline._get_context = mock.Mock(side_effect=RuntimeError("no context")) @@ -586,3 +600,29 @@ def test_reinit_after_fork_replaces_locks_without_acquiring_them(self): pipeline.reinit_after_fork() assert not pipeline._lock.locked() pipeline.start_span("a").end() + + +class TestLimitsReachTheExport: + def test_bounds_names_and_attributes_with_the_configured_length(self): + sender = FakeSender(SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_attribute_value_length=5) + pipeline.start_span("a long name", attributes={"k": "a long value"}).end() + pipeline.flush() + (span,) = sender.batches()[0] + assert span["name"] == "a lon" + assert span["attributes"] == [{"key": "k", "value": {"stringValue": "a lon"}}] + + def test_reports_a_spans_limit_drops_once_at_debug(self, caplog): + caplog.set_level("DEBUG", logger="posthog") + pipeline, _, _ = make(max_attributes_per_span=1, max_events_per_span=1) + span = pipeline.start_span("capped", attributes={"a": 1, "b": 2}) + span.add_event("e1", {"k": 1}).add_event("e2") + span.end() + messages = [ + r.getMessage() for r in caplog.records if "Span limits" in r.getMessage() + ] + assert len(messages) == 1 + assert messages[0].endswith( + 'Span limits discarded data from "capped": 1 attributes, 1 events, ' + "0 event attributes" + ) diff --git a/posthog/test/tracing/test_span.py b/posthog/test/tracing/test_span.py index ac845178..b855c30f 100644 --- a/posthog/test/tracing/test_span.py +++ b/posthog/test/tracing/test_span.py @@ -11,6 +11,7 @@ import pytest from posthog.tracing import _span as span_module +from posthog.tracing._config import MAX_ATTRIBUTES_PER_EVENT from posthog.tracing._otlp import SpanRecord from posthog.tracing._sanitize import FALLBACK_SPAN_NAME, UNSERIALIZABLE_VALUE from posthog.tracing._span import ( @@ -356,10 +357,13 @@ def test_records_a_raised_exception_and_reraises_it_unchanged(self): assert records[0].status is not None assert records[0].status.code == "error" assert records[0].status.message == "boom" - assert records[0].events[0].attributes == { - "exception.type": "TypeError", - "exception.message": "boom", - } + attributes = records[0].events[0].attributes + assert attributes["exception.type"] == "TypeError" + assert attributes["exception.message"] == "boom" + assert attributes["exception.stacktrace"].startswith( + "Traceback (most recent call last):" + ) + assert "TypeError: boom" in attributes["exception.stacktrace"] @pytest.mark.parametrize( "error", [GeneratorExit(), asyncio.CancelledError(), KeyboardInterrupt()] @@ -650,3 +654,196 @@ def test_a_dropped_handle_is_collectable(self): del span gc.collect() assert ref() is None + + +class TestSpanLimits: + def test_keeps_the_first_attributes_and_counts_the_overflow(self): + records: list = [] + span = make_span(records, max_attributes=128) + for i in range(130): + span.set_attribute(f"k{i}", i) + span.end() + assert len(records[0].attributes) == 128 + assert "k127" in records[0].attributes and "k128" not in records[0].attributes + assert records[0].dropped_attributes_count == 2 + + def test_overwriting_a_key_does_not_spend_a_slot(self): + records: list = [] + span = make_span(records, max_attributes=1) + span.set_attribute("a", 1).set_attribute("a", 2).set_attribute("b", 3) + span.end() + assert records[0].attributes == {"a": 2} + assert records[0].dropped_attributes_count == 1 + + def test_none_removes_a_key_and_frees_its_slot(self): + records: list = [] + span = make_span(records, max_attributes=1) + span.set_attribute("a", 1).set_attribute("a", None).set_attribute("b", 2) + span.end() + assert records[0].attributes == {"b": 2} + assert records[0].dropped_attributes_count == 0 + + def test_auto_context_is_exempt_from_the_cap_and_never_evicted(self): + records: list = [] + span = make_span( + records, + attributes={"posthogDistinctId": "user-1", "sessionId": "s-1"}, + auto_attribute_keys=["posthogDistinctId", "sessionId"], + max_attributes=2, + ) + span.set_attributes({"a": 1, "b": 2, "c": 3}) + span.end() + assert records[0].attributes == { + "posthogDistinctId": "user-1", + "sessionId": "s-1", + "a": 1, + "b": 2, + } + assert records[0].dropped_attributes_count == 1 + + def test_keeps_the_first_events_and_counts_the_overflow(self): + records: list = [] + span = make_span(records, max_events=2) + span.add_event("a").add_event("b").add_event("c") + span.end() + assert [e.name for e in records[0].events] == ["a", "b"] + assert records[0].dropped_events_count == 1 + + def test_a_recorded_exception_spends_an_event_slot(self): + records: list = [] + span = make_span(records, max_events=1) + span.add_event("a") + span.record_exception(ValueError("boom")) + span.end() + assert [e.name for e in records[0].events] == ["a"] + assert records[0].dropped_events_count == 1 + assert records[0].status.code == "error" + + def test_caps_each_events_attributes(self): + records: list = [] + span = make_span(records) + span.add_event("batch", {f"k{i}": i for i in range(130)}) + span.end() + event = records[0].events[0] + assert len(event.attributes) == 128 + assert event.dropped_attributes_count == 2 + + def test_truncates_a_long_attribute_value_without_counting_a_drop(self): + records: list = [] + span = make_span(records, max_attribute_value_length=8192) + span.set_attribute("payload", "x" * 40000) + span.set_attribute("nested", {"body": "y" * 40000}) + span.end() + assert records[0].attributes["payload"] == "x" * 8192 + assert records[0].attributes["nested"] == {"body": "y" * 8192} + assert records[0].dropped_attributes_count == 0 + + def test_truncates_event_attributes_names_and_status_messages(self): + records: list = [] + span = make_span(records, max_attribute_value_length=4) + span.add_event("long event name", {"k": "long value"}) + span.update_name("long span name") + span.set_status("error", "long message") + span.end() + record = records[0] + assert record.name == "long" + assert record.events[0].name == "long" + assert record.events[0].attributes == {"k": "long"} + assert record.status.message == "long" + + def test_a_self_referencing_attribute_ends_the_span(self): + records: list = [] + value: dict = {} + value["self"] = value + span = make_span(records) + span.set_attribute("loop", value) + span.end() + assert len(records) == 1 + + +class TestExceptionStacktrace: + def test_record_exception_attaches_the_stack_of_a_raised_exception(self): + records: list = [] + span = make_span(records) + try: + raise KeyError("missing") + except KeyError as e: + span.record_exception(e) + span.end() + stack = records[0].events[0].attributes["exception.stacktrace"] + assert stack.startswith("Traceback (most recent call last):") + assert "KeyError: 'missing'" in stack + + def test_omits_the_stack_of_an_exception_that_was_never_raised(self): + records: list = [] + span = make_span(records) + span.record_exception(ValueError("constructed, not raised")) + span.end() + assert "exception.stacktrace" not in records[0].events[0].attributes + + def test_bounds_the_stack_keeping_the_crash_site(self): + # Python lists the most recent call last, so the tail is what matters: + # the raising frame and the exception line. + def recurse(depth): + if depth == 0: + raise ValueError("the actual crash site") + recurse(depth - 1) + + records: list = [] + span = make_span(records, max_attribute_value_length=300) + try: + try: + recurse(60) + except ValueError as cause: + raise RuntimeError("wrapped") from cause + except RuntimeError as e: + span.record_exception(e) + span.end() + stack = records[0].events[0].attributes["exception.stacktrace"] + assert len(stack) == 300 + assert stack.rstrip().endswith("RuntimeError: wrapped") + + def test_a_stack_that_cannot_be_formatted_costs_only_the_attribute(self): + records: list = [] + span = make_span(records) + try: + raise ValueError("boom") + except ValueError as e: + with mock.patch.object( + span_module.traceback, + "format_exception", + side_effect=RuntimeError("no"), + ): + span.record_exception(e) + span.end() + attributes = records[0].events[0].attributes + assert attributes["exception.message"] == "boom" + assert "exception.stacktrace" not in attributes + + +class TestSpanLimitInteractions: + def test_the_attribute_and_event_caps_are_independent(self): + records: list = [] + span = make_span(records, max_attributes=1, max_events=1) + span.set_attributes({"a": 1, "b": 2}) + span.add_event("e1", {"x": 1, "y": 2}) + span.add_event("e2") + span.end() + record = records[0] + assert record.attributes == {"a": 1} + assert record.events[0].attributes == {"x": 1, "y": 2} + assert (record.dropped_attributes_count, record.dropped_events_count) == (1, 1) + + def test_a_raising_key_in_add_event_costs_only_that_key(self): + class HostileKey: + def __str__(self): + raise RuntimeError("no") + + records: list = [] + span = make_span(records) + span.add_event("e", {HostileKey(): 1, "ok": 2}) + span.end() + assert records[0].events[0].attributes == {"ok": 2} + + def test_the_per_event_attribute_cap_is_opentelemetrys_default(self): + assert MAX_ATTRIBUTES_PER_EVENT == 128 diff --git a/posthog/tracing/_config.py b/posthog/tracing/_config.py index c953e844..71574d63 100644 --- a/posthog/tracing/_config.py +++ b/posthog/tracing/_config.py @@ -18,6 +18,13 @@ DEFAULT_MAX_EXPORT_BATCH_SIZE = 512 DEFAULT_MAX_QUEUE_SIZE = 2048 +DEFAULT_MAX_ATTRIBUTES_PER_SPAN = 128 +DEFAULT_MAX_EVENTS_PER_SPAN = 128 +MAX_ATTRIBUTES_PER_EVENT = 128 +# OpenTelemetry leaves this unlimited, but one huge value gets the whole span +# dropped as too large. 8192 fits a deep stack trace. +DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH = 8192 + # Well above realistic concurrency; production traces routinely exceed ten minutes. DEFAULT_MAX_LIVE_SPANS = 10_000 DEFAULT_MAX_SPAN_AGE_SECONDS = 3600.0 @@ -37,6 +44,9 @@ class ResolvedTracesConfig: max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE max_live_spans: int = DEFAULT_MAX_LIVE_SPANS max_span_age: float = DEFAULT_MAX_SPAN_AGE_SECONDS + max_attributes_per_span: int = DEFAULT_MAX_ATTRIBUTES_PER_SPAN + max_events_per_span: int = DEFAULT_MAX_EVENTS_PER_SPAN + max_attribute_value_length: int = DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH _KNOWN_KEYS = frozenset(field.name for field in fields(ResolvedTracesConfig)) @@ -163,4 +173,13 @@ def resolve_traces_config( max_span_age=_positive_number( config, "max_span_age", DEFAULT_MAX_SPAN_AGE_SECONDS ), + max_attributes_per_span=_positive_int( + config, "max_attributes_per_span", DEFAULT_MAX_ATTRIBUTES_PER_SPAN + ), + max_events_per_span=_positive_int( + config, "max_events_per_span", DEFAULT_MAX_EVENTS_PER_SPAN + ), + max_attribute_value_length=_positive_int( + config, "max_attribute_value_length", DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH + ), ) diff --git a/posthog/tracing/_export.py b/posthog/tracing/_export.py index 18e86dbc..f7dbdacf 100644 --- a/posthog/tracing/_export.py +++ b/posthog/tracing/_export.py @@ -15,6 +15,7 @@ from ..capture_v1 import _MAX_BACKOFF_SECONDS from ._config import ResolvedTracesConfig from ._drops import DropLog +from ._limits import truncate_attributes from ._otlp import ( SpanRecord, build_otlp_span, @@ -104,11 +105,14 @@ def __init__( self._send = send # Encoded once: the resource is the same for every batch. self._resource = to_resource_key_value_list( - build_resource_attributes( - config.service_name, - config.service_version, - config.environment, - config.resource_attributes, + truncate_attributes( + build_resource_attributes( + config.service_name, + config.service_version, + config.environment, + config.resource_attributes, + ), + config.max_attribute_value_length, ) ) diff --git a/posthog/tracing/_limits.py b/posthog/tracing/_limits.py new file mode 100644 index 00000000..67ebb267 --- /dev/null +++ b/posthog/tracing/_limits.py @@ -0,0 +1,150 @@ +"""Per-span limits: the attribute and event caps, and the value-length bound. + +The length bound reaches every string a value contains, nested ones included: +one huge value otherwise gets the whole span dropped as too large. The walk +mirrors ``_otlp._encode`` (depth, item and node budgets, cycle marker, skipped +keys) so it bounds exactly what the encoder will emit. +""" + +from datetime import date +from typing import Any, Dict, Mapping, Tuple + +from ._otlp import ( + CIRCULAR_VALUE, + MAX_VALUE_DEPTH, + MAX_VALUE_ITEMS, + MAX_VALUE_NODES, + TRUNCATED_VALUE, +) +from ._sanitize import UNSERIALIZABLE_VALUE, attribute_key + + +class _WalkState: + __slots__ = ("ancestors", "remaining_nodes") + + def __init__(self) -> None: + self.ancestors: set = set() + self.remaining_nodes = MAX_VALUE_NODES + + +def truncate_string(value: str, max_length: int) -> str: + return value[:max_length] if len(value) > max_length else value + + +def truncate_attribute_value(value: Any, max_length: int) -> Any: + """Bound every string reachable from ``value`` to ``max_length`` characters. + + A value that cannot be walked is returned as it is. + """ + try: + return _truncate(value, max_length, _WalkState(), 0) + except Exception: + return value + + +def _truncate(value: Any, max_length: int, state: _WalkState, depth: int) -> Any: + if value is None: + return None + is_container = isinstance(value, (Mapping, list, tuple, set, frozenset)) + if not is_container: + # Leaves are charged too: a shared subtree is walked once per path. + if state.remaining_nodes <= 0: + return value + state.remaining_nodes -= 1 + if isinstance(value, str): + return truncate_string(value, max_length) + if isinstance(value, (bool, int, float, date)): + return value + # The encoder stringifies anything else, so bound that text. + try: + return truncate_string(str(value), max_length) + except Exception: + return UNSERIALIZABLE_VALUE + + marker = id(value) + if marker in state.ancestors: + # The marker, not the ancestor: inside a copied parent the encoder would + # no longer see the cycle. + return CIRCULAR_VALUE + if state.remaining_nodes <= 0 or depth >= MAX_VALUE_DEPTH: + return value + state.remaining_nodes -= 1 + state.ancestors.add(marker) + try: + if isinstance(value, Mapping): + return _truncate_mapping(value, max_length, state, depth) + return _truncate_items(value, max_length, state, depth) + finally: + state.ancestors.discard(marker) + + +def _truncate_mapping( + value: Mapping, max_length: int, state: _WalkState, depth: int +) -> dict: + bounded: dict = {} + emittable = 0 + for key in list(value.keys()): + if emittable >= MAX_VALUE_ITEMS: + break + key_str = attribute_key(key) + if not key_str: + continue + try: + item = _truncate(value[key], max_length, state, depth + 1) + except Exception: + item = UNSERIALIZABLE_VALUE + if item is not None: + emittable += 1 + bounded[key] = item + return bounded + + +def _truncate_items(value: Any, max_length: int, state: _WalkState, depth: int) -> list: + bounded: list = [] + for index, element in enumerate(value): + if index >= MAX_VALUE_ITEMS: + bounded.append(TRUNCATED_VALUE) + break + try: + bounded.append(_truncate(element, max_length, state, depth + 1)) + except Exception: + bounded.append(UNSERIALIZABLE_VALUE) + return bounded + + +def bound_attributes( + source: Any, max_count: int, max_length: int +) -> Tuple[Dict[str, Any], int]: + """A copy of ``source`` with at most ``max_count`` entries, each bounded, and + how many entries the cap refused. A ``None`` value spends no slot.""" + if not isinstance(source, Mapping): + return {}, 0 + try: + keys = list(source.keys()) + except Exception: + return {}, 0 + attributes: Dict[str, Any] = {} + dropped = 0 + for key in keys: + key_str = attribute_key(key) + if key_str is None: + continue + if len(attributes) >= max_count and key_str not in attributes: + dropped += 1 + continue + try: + value = truncate_attribute_value(source[key], max_length) + except Exception: + value = UNSERIALIZABLE_VALUE + if value is None: + continue + attributes[key_str] = value + return attributes, dropped + + +def truncate_attributes(attributes: Mapping, max_length: int) -> Dict[str, Any]: + """``truncate_attribute_value`` across an attribute mapping, as a copy.""" + return { + key: truncate_attribute_value(value, max_length) + for key, value in attributes.items() + } diff --git a/posthog/tracing/_otlp.py b/posthog/tracing/_otlp.py index bdfa2851..7801beb9 100644 --- a/posthog/tracing/_otlp.py +++ b/posthog/tracing/_otlp.py @@ -50,6 +50,7 @@ class SpanEventRecord: name: str timestamp_ns: int attributes: Optional[Dict[str, Any]] = None + dropped_attributes_count: int = 0 @dataclass @@ -76,6 +77,24 @@ class SpanRecord: status: Optional[SpanStatus] = None attributes: Dict[str, Any] = field(default_factory=dict) events: List[SpanEventRecord] = field(default_factory=list) + # User attributes and events the per-span caps refused. + dropped_attributes_count: int = 0 + dropped_events_count: int = 0 + + +MAX_UINT32 = 0xFFFFFFFF + + +def non_negative_count(value: Any) -> int: + """A dropped count as the ``uint32`` the wire declares, or 0 for anything else. + + One that overflows the field is refused for the whole request. + """ + if isinstance(value, bool) or not isinstance(value, (int, float)): + return 0 + if not math.isfinite(value) or value <= 0: + return 0 + return min(int(value), MAX_UINT32) def sanitize_string(value: str) -> str: @@ -266,12 +285,14 @@ def _to_otlp_event(event: SpanEventRecord) -> dict: "name": wire_string(event.name), "timeUnixNano": str(event.timestamp_ns), } + cut = 0 if event.attributes: attributes, cut = encode_attributes(event.attributes) if attributes: encoded["attributes"] = attributes - if cut: - encoded["droppedAttributesCount"] = cut + dropped = non_negative_count(non_negative_count(event.dropped_attributes_count) + cut) + if dropped: + encoded["droppedAttributesCount"] = dropped return encoded @@ -292,10 +313,16 @@ def build_otlp_span(record: SpanRecord) -> dict: attributes, cut = encode_attributes(record.attributes) if attributes: span["attributes"] = attributes - if cut: - span["droppedAttributesCount"] = cut if record.events: span["events"] = [_to_otlp_event(event) for event in record.events] + dropped_attributes = non_negative_count( + non_negative_count(record.dropped_attributes_count) + cut + ) + if dropped_attributes: + span["droppedAttributesCount"] = dropped_attributes + dropped_events = non_negative_count(record.dropped_events_count) + if dropped_events: + span["droppedEventsCount"] = dropped_events if record.status is not None and record.status.code in SPAN_STATUS_TO_OTLP: status: dict = {"code": SPAN_STATUS_TO_OTLP[record.status.code]} if record.status.message: diff --git a/posthog/tracing/_pipeline.py b/posthog/tracing/_pipeline.py index d09c05a7..65bebd83 100644 --- a/posthog/tracing/_pipeline.py +++ b/posthog/tracing/_pipeline.py @@ -199,11 +199,12 @@ def _build_span( start_ns = resolve_start_ns(start_time, now_ns) auto_attributes = self._auto_context_attributes() span_attributes = copy_user_attributes(dict(auto_attributes), attributes) + config = self._config return RecordingSpan( trace_id=parent_context.trace_id if parent_context else new_trace_id(), span_id=span_id, - name=sanitize_name(name, "Span name"), + name=sanitize_name(name, "Span name", config.max_attribute_value_length), start_ns=start_ns, backdated=start_ns != now_ns, on_end=self._on_span_end, @@ -218,6 +219,10 @@ def _build_span( clock_anchor=parent_context.clock_anchor if parent_context and to_epoch_ns(start_time) is None else None, + auto_attribute_keys=auto_attributes.keys(), + max_attributes=config.max_attributes_per_span, + max_events=config.max_events_per_span, + max_attribute_value_length=config.max_attribute_value_length, ) def _is_closed(self) -> bool: @@ -296,5 +301,23 @@ def _on_span_end(self, record: SpanRecord) -> None: if disabled: self._drops.record(1, "the client is disabled") else: + _report_limit_drops(record) self._exporter.enqueue(record) self._drops.warn_if_due() + + +def _report_limit_drops(record: SpanRecord) -> None: + event_attributes = sum(event.dropped_attributes_count for event in record.events) + if ( + record.dropped_attributes_count + or record.dropped_events_count + or event_attributes + ): + log.debug( + 'Span limits discarded data from "%s": %s attributes, %s events, ' + "%s event attributes", + record.name, + record.dropped_attributes_count, + record.dropped_events_count, + event_attributes, + ) diff --git a/posthog/tracing/_sanitize.py b/posthog/tracing/_sanitize.py index 36768f70..d87299da 100644 --- a/posthog/tracing/_sanitize.py +++ b/posthog/tracing/_sanitize.py @@ -29,10 +29,10 @@ _EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) -def sanitize_name(name: Any, label: str) -> str: - """A non-empty name; an unusable one becomes ``unknown`` rather than dropping the span.""" +def sanitize_name(name: Any, label: str, max_length: Optional[int] = None) -> str: + """A non-empty name, truncated to ``max_length``; an unusable one becomes ``unknown``.""" if isinstance(name, str) and name.strip(): - return name + return name if max_length is None else name[:max_length] log.debug('%s must be a non-empty string; using "%s"', label, FALLBACK_SPAN_NAME) return FALLBACK_SPAN_NAME diff --git a/posthog/tracing/_span.py b/posthog/tracing/_span.py index 3418e46e..81708eff 100644 --- a/posthog/tracing/_span.py +++ b/posthog/tracing/_span.py @@ -3,10 +3,18 @@ import logging import threading import time +import traceback from contextvars import ContextVar, Token from dataclasses import dataclass -from typing import Any, Callable, Dict, List, Mapping, Optional +from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional +from ._config import ( + DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH, + DEFAULT_MAX_ATTRIBUTES_PER_SPAN, + DEFAULT_MAX_EVENTS_PER_SPAN, + MAX_ATTRIBUTES_PER_EVENT, +) +from ._limits import bound_attributes, truncate_attribute_value, truncate_string from ._otlp import SpanEventRecord, SpanRecord, SpanStatus from ._sanitize import ( SpanTimeInput, @@ -165,6 +173,33 @@ def describe_error(error: Any) -> "tuple[str, str]": return type(error).__name__, "" +def describe_stacktrace(error: Any) -> Optional[str]: + """The OTel ``exception.stacktrace``, or ``None`` for an exception never raised.""" + try: + if isinstance(error, BaseException) and error.__traceback__ is not None: + return "".join( + traceback.format_exception(type(error), error, error.__traceback__) + ) + except Exception: + pass + return None + + +def _exception_event_attributes( + error: Any, max_length: int +) -> "tuple[Dict[str, Any], str]": + exc_type, message = describe_error(error) + attributes: Dict[str, Any] = { + "exception.type": exc_type, + "exception.message": message, + } + # The tail is kept: Python lists the raising frame last. + stacktrace = describe_stacktrace(error) + if stacktrace: + attributes["exception.stacktrace"] = stacktrace[-max_length:] + return attributes, message + + class RecordingSpan(_Activatable, Span): """A span that records and, on ``end()``, hands one record to the pipeline.""" @@ -185,6 +220,10 @@ def __init__( parent_is_remote: bool = False, active_var: Optional[ContextVar] = None, clock_anchor: Optional[ClockAnchor] = None, + auto_attribute_keys: Iterable[str] = (), + max_attributes: int = DEFAULT_MAX_ATTRIBUTES_PER_SPAN, + max_events: int = DEFAULT_MAX_EVENTS_PER_SPAN, + max_attribute_value_length: int = DEFAULT_MAX_ATTRIBUTE_VALUE_LENGTH, ) -> None: self._trace_id = trace_id self._span_id = span_id @@ -208,7 +247,18 @@ def __init__( self._name = name self._kind = kind - self._attributes: Dict[str, Any] = attributes if attributes is not None else {} + # The SDK's own join keys, exempt from the attribute cap. + self._auto_keys = frozenset(auto_attribute_keys) + self._max_attributes = max_attributes + self._max_events = max_events + self._max_attribute_value_length = max_attribute_value_length + self._user_attribute_count = 0 + self._user_event_count = 0 + self._dropped_attributes = 0 + self._dropped_events = 0 + self._attributes: Dict[str, Any] = {} + for key, value in (attributes or {}).items(): + self._write_attribute(key, value) self._events: List[SpanEventRecord] = [] self._status: Optional[SpanStatus] = None self._ended = False @@ -225,16 +275,35 @@ def _mutable(self, operation: str) -> bool: return False return True + def _write_attribute(self, key: str, value: Any) -> None: + """Write an attribute unless the span is at its cap of distinct user keys.""" + if value is None: + # None removes the key, freeing its slot. + if key in self._attributes and key not in self._auto_keys: + self._user_attribute_count -= 1 + self._attributes.pop(key, None) + return + # Checked before the value is walked, which is the costly part. + if key not in self._auto_keys and key not in self._attributes: + if self._user_attribute_count >= self._max_attributes: + self._dropped_attributes += 1 + return + self._user_attribute_count += 1 + self._attributes[key] = truncate_attribute_value( + value, self._max_attribute_value_length + ) + def set_attribute(self, key: str, value: Any) -> "Span": if self._mutable("set_attribute"): key_str = attribute_key(key) if key_str is not None: - self._attributes[key_str] = value + self._write_attribute(key_str, value) return self def set_attributes(self, attributes: Mapping[str, Any]) -> "Span": if self._mutable("set_attributes"): - copy_user_attributes(self._attributes, attributes) + for key, value in copy_user_attributes({}, attributes).items(): + self._write_attribute(key, value) return self def add_event( @@ -244,15 +313,29 @@ def add_event( timestamp: Optional[SpanTimeInput] = None, ) -> "Span": if self._mutable("add_event"): + # A recorded exception spends a slot like any other event. + if self._user_event_count >= self._max_events: + self._dropped_events += 1 + return self + self._user_event_count += 1 + bounded: Optional[Dict[str, Any]] = None + dropped = 0 + if attributes is not None: + bounded, dropped = bound_attributes( + attributes, + MAX_ATTRIBUTES_PER_EVENT, + self._max_attribute_value_length, + ) self._events.append( SpanEventRecord( - name=sanitize_name(name, "Span event name"), + name=sanitize_name( + name, "Span event name", self._max_attribute_value_length + ), timestamp_ns=resolve_supplied_ns( timestamp, self._now_ns(), "event timestamp" ), - attributes=copy_user_attributes({}, attributes) - if attributes is not None - else None, + attributes=bounded, + dropped_attributes_count=dropped, ) ) return self @@ -263,7 +346,12 @@ def set_status(self, code: str, message: Optional[str] = None) -> "Span": log.debug('Ignoring an unknown span status; expected "ok" or "error"') return self text = None if message is None else safe_str(message) - self._status = SpanStatus(code, text or None) + self._status = SpanStatus( + code, + truncate_string(text, self._max_attribute_value_length) + if text + else None, + ) return self @property @@ -276,17 +364,19 @@ def record_exception(self, exception: BaseException) -> "Span": return self def _record_exception(self, exception: BaseException, keep_ok: bool) -> None: - exc_type, message = describe_error(exception) - self.add_event( - "exception", {"exception.type": exc_type, "exception.message": message} + attributes, message = _exception_event_attributes( + exception, self._max_attribute_value_length ) + self.add_event("exception", attributes) # Only the scoped form treats an explicit `ok` as final. if not (keep_ok and self._status_is_explicitly_ok): self.set_status("error", message) def update_name(self, name: str) -> "Span": if self._mutable("update_name"): - self._name = sanitize_name(name, "Span name") + self._name = sanitize_name( + name, "Span name", self._max_attribute_value_length + ) return self def traceparent(self) -> Optional[str]: @@ -327,6 +417,8 @@ def end(self, end_time: Optional[SpanTimeInput] = None) -> None: events=self._events, start_ns=self._start_ns, end_ns=clamp_end_ns(resolved, self._start_ns), + dropped_attributes_count=self._dropped_attributes, + dropped_events_count=self._dropped_events, ) try: self._on_end(record) diff --git a/posthog/tracing/span.py b/posthog/tracing/span.py index 3cddf2fd..255d99a9 100644 --- a/posthog/tracing/span.py +++ b/posthog/tracing/span.py @@ -81,8 +81,10 @@ def set_status(self, code: str, message: Optional[str] = None) -> "Span": def record_exception(self, exception: BaseException) -> "Span": """Record an exception as an ``exception`` event and mark the span ``error``. - Ignored after ``end()``. The event carries ``exception.type`` and - ``exception.message``. Returns the span, so calls chain. Inside + Ignored after ``end()``. The event carries ``exception.type``, + ``exception.message`` and, for a raised exception, + ``exception.stacktrace`` (its last ``max_attribute_value_length`` + characters). Returns the span, so calls chain. Inside ``with span:`` a raised ``Exception`` is recorded automatically, so this is for exceptions that are caught and handled. From 1f8d472775bfc15f7f5031fbbf034b0437bfb1a7 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 14:06:28 -0400 Subject: [PATCH 2/4] fix(traces): keep callables for the encoder's marker and stop charging None event attributes past the cap A callable was stringified by the value-length walk, so the OTLP encoder emitted its repr instead of the stable [Function] marker. And an event attribute bag with None past the per-event cap counted that None as a drop even though the encoder never emits it. --- posthog/test/tracing/test_limits.py | 26 ++++++++++++++-- posthog/test/tracing/test_span.py | 46 +++++++++++++++++++++++++++-- posthog/tracing/_limits.py | 12 ++++---- 3 files changed, 73 insertions(+), 11 deletions(-) diff --git a/posthog/test/tracing/test_limits.py b/posthog/test/tracing/test_limits.py index cc1d95d5..8caaefa6 100644 --- a/posthog/test/tracing/test_limits.py +++ b/posthog/test/tracing/test_limits.py @@ -12,7 +12,7 @@ TRUNCATED_VALUE, to_any_value, ) -from posthog.tracing._sanitize import UNSERIALIZABLE_VALUE +from posthog.tracing._sanitize import FUNCTION_VALUE, UNSERIALIZABLE_VALUE class TestTruncateAttributeValue: @@ -79,6 +79,16 @@ def test_a_key_the_encoder_skips_does_not_spend_the_walks_budget(self): x = next(kv for kv in encoded if kv["key"] == "x") assert len(x["value"]["stringValue"]) == 100 + def test_leaves_a_callable_for_the_encoders_marker(self): + def handler(): + pass + + assert truncate_attribute_value(handler, 3) is handler + assert truncate_attribute_value({"fn": handler}, 3) == {"fn": handler} + assert to_any_value(truncate_attribute_value(handler, 3)) == { + "stringValue": FUNCTION_VALUE + } + def test_a_raising_str_costs_only_that_value(self): class Hostile: def __str__(self): @@ -109,11 +119,21 @@ def test_keeps_the_earliest_entries_and_counts_the_rest(self): assert list(attributes) == [f"k{i}" for i in range(128)] assert dropped == 2 - def test_a_none_value_spends_no_slot(self): - attributes, dropped = bound_attributes({"a": None, "b": 1, "c": 2}, 2, 8) + @pytest.mark.parametrize( + "source", + [{"a": None, "b": 1, "c": 2}, {"b": 1, "c": 2, "a": None}], + ids=["before-the-cap", "past-the-cap"], + ) + def test_a_none_value_spends_no_slot_and_counts_no_drop(self, source): + attributes, dropped = bound_attributes(source, 2, 8) assert attributes == {"b": 1, "c": 2} assert dropped == 0 + def test_a_real_value_past_the_cap_counts_a_drop(self): + attributes, dropped = bound_attributes({"b": 1, "c": 2, "a": 3}, 2, 8) + assert attributes == {"b": 1, "c": 2} + assert dropped == 1 + def test_bounds_each_value(self): attributes, _ = bound_attributes({"a": "x" * 20}, 2, 5) assert attributes == {"a": "xxxxx"} diff --git a/posthog/test/tracing/test_span.py b/posthog/test/tracing/test_span.py index b855c30f..796d1c88 100644 --- a/posthog/test/tracing/test_span.py +++ b/posthog/test/tracing/test_span.py @@ -12,8 +12,12 @@ from posthog.tracing import _span as span_module from posthog.tracing._config import MAX_ATTRIBUTES_PER_EVENT -from posthog.tracing._otlp import SpanRecord -from posthog.tracing._sanitize import FALLBACK_SPAN_NAME, UNSERIALIZABLE_VALUE +from posthog.tracing._otlp import SpanRecord, build_otlp_span +from posthog.tracing._sanitize import ( + FALLBACK_SPAN_NAME, + FUNCTION_VALUE, + UNSERIALIZABLE_VALUE, +) from posthog.tracing._span import ( NOOP_SPAN, ClockAnchor, @@ -738,6 +742,44 @@ def test_truncates_a_long_attribute_value_without_counting_a_drop(self): assert records[0].attributes["nested"] == {"body": "y" * 8192} assert records[0].dropped_attributes_count == 0 + def test_a_callable_reaches_the_encoder_as_its_marker(self): + def handler(): + pass + + records: list = [] + span = make_span(records) + span.set_attribute("fn", handler) + span.add_event("e", {"fn": handler, "nested": {"fn": handler}}) + span.end() + encoded = build_otlp_span(records[0]) + assert encoded["attributes"] == [ + {"key": "fn", "value": {"stringValue": FUNCTION_VALUE}} + ] + assert encoded["events"][0]["attributes"] == [ + {"key": "fn", "value": {"stringValue": FUNCTION_VALUE}}, + { + "key": "nested", + "value": { + "kvlistValue": { + "values": [ + {"key": "fn", "value": {"stringValue": FUNCTION_VALUE}} + ] + } + }, + }, + ] + + def test_a_none_event_attribute_past_the_cap_counts_no_drop(self): + records: list = [] + span = make_span(records) + attributes = {f"k{i}": i for i in range(MAX_ATTRIBUTES_PER_EVENT)} + attributes["late"] = None + span.add_event("batch", attributes) + span.end() + event = records[0].events[0] + assert len(event.attributes) == MAX_ATTRIBUTES_PER_EVENT + assert event.dropped_attributes_count == 0 + def test_truncates_event_attributes_names_and_status_messages(self): records: list = [] span = make_span(records, max_attribute_value_length=4) diff --git a/posthog/tracing/_limits.py b/posthog/tracing/_limits.py index 67ebb267..35ad7d00 100644 --- a/posthog/tracing/_limits.py +++ b/posthog/tracing/_limits.py @@ -53,7 +53,7 @@ def _truncate(value: Any, max_length: int, state: _WalkState, depth: int) -> Any state.remaining_nodes -= 1 if isinstance(value, str): return truncate_string(value, max_length) - if isinstance(value, (bool, int, float, date)): + if isinstance(value, (bool, int, float, date)) or callable(value): return value # The encoder stringifies anything else, so bound that text. try: @@ -129,16 +129,16 @@ def bound_attributes( key_str = attribute_key(key) if key_str is None: continue - if len(attributes) >= max_count and key_str not in attributes: - dropped += 1 - continue try: - value = truncate_attribute_value(source[key], max_length) + value = source[key] except Exception: value = UNSERIALIZABLE_VALUE if value is None: continue - attributes[key_str] = value + if len(attributes) >= max_count and key_str not in attributes: + dropped += 1 + continue + attributes[key_str] = truncate_attribute_value(value, max_length) return attributes, dropped From e719d84ad1c34c653f6c03290570000aad0fcb53 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 21:05:53 -0400 Subject: [PATCH 3/4] fix(traces): an empty attribute key spends no slot The encoder drops an empty key, so charging it against the cap lost a real attribute. Scalars skip the truncation walk, event attributes go through the shared copier so a non-mapping logs, and the traceback test is named for what it asserts: the outermost exception survives the cut. --- posthog/test/tracing/test_limits.py | 16 ++++++++++++-- posthog/test/tracing/test_span.py | 32 ++++++++++++++++++++++------ posthog/tracing/_limits.py | 33 ++++++++++++----------------- posthog/tracing/_span.py | 9 ++++++-- posthog/tracing/span.py | 3 ++- 5 files changed, 62 insertions(+), 31 deletions(-) diff --git a/posthog/test/tracing/test_limits.py b/posthog/test/tracing/test_limits.py index 8caaefa6..123c57fc 100644 --- a/posthog/test/tracing/test_limits.py +++ b/posthog/test/tracing/test_limits.py @@ -1,5 +1,8 @@ +from unittest import mock + import pytest +from posthog.tracing import _limits as limits_module from posthog.tracing._limits import ( bound_attributes, truncate_attribute_value, @@ -19,6 +22,13 @@ class TestTruncateAttributeValue: def test_truncates_a_long_string(self): assert truncate_attribute_value("x" * 40000, 8192) == "x" * 8192 + def test_never_walks_a_scalar(self): + with mock.patch.object(limits_module, "_truncate") as walk: + assert truncate_attribute_value("short", 8) == "short" + assert truncate_attribute_value(7, 8) == 7 + assert truncate_attribute_value(None, 8) is None + assert not walk.called + def test_returns_a_short_string_unchanged(self): assert truncate_attribute_value("short", 8192) == "short" @@ -138,8 +148,10 @@ def test_bounds_each_value(self): attributes, _ = bound_attributes({"a": "x" * 20}, 2, 5) assert attributes == {"a": "xxxxx"} - def test_a_non_mapping_yields_nothing(self): - assert bound_attributes(["a"], 2, 5) == ({}, 0) + def test_an_empty_key_spends_no_slot_and_counts_no_drop(self): + attributes, dropped = bound_attributes({"": 1, "b": 2, "c": 3}, 2, 8) + assert attributes == {"b": 2, "c": 3} + assert dropped == 0 class TestTruncateAttributes: diff --git a/posthog/test/tracing/test_span.py b/posthog/test/tracing/test_span.py index 796d1c88..712cba93 100644 --- a/posthog/test/tracing/test_span.py +++ b/posthog/test/tracing/test_span.py @@ -12,7 +12,7 @@ from posthog.tracing import _span as span_module from posthog.tracing._config import MAX_ATTRIBUTES_PER_EVENT -from posthog.tracing._otlp import SpanRecord, build_otlp_span +from posthog.tracing._otlp import CIRCULAR_VALUE, SpanRecord, build_otlp_span from posthog.tracing._sanitize import ( FALLBACK_SPAN_NAME, FUNCTION_VALUE, @@ -793,14 +793,34 @@ def test_truncates_event_attributes_names_and_status_messages(self): assert record.events[0].attributes == {"k": "long"} assert record.status.message == "long" - def test_a_self_referencing_attribute_ends_the_span(self): + def test_a_self_referencing_attribute_is_stored_as_the_encoders_marker(self): records: list = [] value: dict = {} value["self"] = value span = make_span(records) span.set_attribute("loop", value) span.end() - assert len(records) == 1 + assert records[0].attributes["loop"] == {"self": CIRCULAR_VALUE} + + def test_an_empty_key_spends_no_slot(self): + records: list = [] + span = make_span(records, max_attributes=1) + span.set_attribute("", 1).set_attribute("real", 2) + span.set_attributes({"": 3}) + span.add_event("e", {"": 1, "k": 2}) + span.end() + assert records[0].attributes == {"real": 2} + assert records[0].dropped_attributes_count == 0 + assert records[0].events[0].attributes == {"k": 2} + + def test_add_event_with_a_non_mapping_logs_and_keeps_the_event(self, caplog): + records: list = [] + span = make_span(records) + with caplog.at_level("DEBUG", logger="posthog"): + span.add_event("e", ["not", "a", "mapping"]) + span.end() + assert records[0].events[0].attributes == {} + assert "expected a mapping" in caplog.text class TestExceptionStacktrace: @@ -823,9 +843,9 @@ def test_omits_the_stack_of_an_exception_that_was_never_raised(self): span.end() assert "exception.stacktrace" not in records[0].events[0].attributes - def test_bounds_the_stack_keeping_the_crash_site(self): - # Python lists the most recent call last, so the tail is what matters: - # the raising frame and the exception line. + def test_bounds_the_stack_keeping_the_outermost_exception(self): + # Python prints a cause first and the outermost exception last, so + # the tail keeps the outermost raise and may cut the cause off. def recurse(depth): if depth == 0: raise ValueError("the actual crash site") diff --git a/posthog/tracing/_limits.py b/posthog/tracing/_limits.py index 35ad7d00..cdec65ad 100644 --- a/posthog/tracing/_limits.py +++ b/posthog/tracing/_limits.py @@ -36,6 +36,11 @@ def truncate_attribute_value(value: Any, max_length: int) -> Any: A value that cannot be walked is returned as it is. """ + # The common case never allocates a walk. + if isinstance(value, str): + return truncate_string(value, max_length) + if value is None or isinstance(value, (bool, int, float)): + return value try: return _truncate(value, max_length, _WalkState(), 0) except Exception: @@ -113,32 +118,20 @@ def _truncate_items(value: Any, max_length: int, state: _WalkState, depth: int) def bound_attributes( - source: Any, max_count: int, max_length: int + source: Mapping[str, Any], max_count: int, max_length: int ) -> Tuple[Dict[str, Any], int]: - """A copy of ``source`` with at most ``max_count`` entries, each bounded, and - how many entries the cap refused. A ``None`` value spends no slot.""" - if not isinstance(source, Mapping): - return {}, 0 - try: - keys = list(source.keys()) - except Exception: - return {}, 0 + """A copy of a ``copy_user_attributes`` result with at most ``max_count`` + entries, each bounded, and how many entries the cap refused. An empty key + or a ``None`` value spends no slot.""" attributes: Dict[str, Any] = {} dropped = 0 - for key in keys: - key_str = attribute_key(key) - if key_str is None: - continue - try: - value = source[key] - except Exception: - value = UNSERIALIZABLE_VALUE - if value is None: + for key, value in source.items(): + if not key or value is None: continue - if len(attributes) >= max_count and key_str not in attributes: + if len(attributes) >= max_count and key not in attributes: dropped += 1 continue - attributes[key_str] = truncate_attribute_value(value, max_length) + attributes[key] = truncate_attribute_value(value, max_length) return attributes, dropped diff --git a/posthog/tracing/_span.py b/posthog/tracing/_span.py index 81708eff..c287eaa3 100644 --- a/posthog/tracing/_span.py +++ b/posthog/tracing/_span.py @@ -193,7 +193,8 @@ def _exception_event_attributes( "exception.type": exc_type, "exception.message": message, } - # The tail is kept: Python lists the raising frame last. + # The tail is kept: Python lists the raising frame last. For a chained + # exception that is the outermost one; a cause printed above it goes first. stacktrace = describe_stacktrace(error) if stacktrace: attributes["exception.stacktrace"] = stacktrace[-max_length:] @@ -277,6 +278,10 @@ def _mutable(self, operation: str) -> bool: def _write_attribute(self, key: str, value: Any) -> None: """Write an attribute unless the span is at its cap of distinct user keys.""" + if not key: + # The encoder drops it, so it must not spend a slot. + log.debug("Dropping an attribute with an empty key") + return if value is None: # None removes the key, freeing its slot. if key in self._attributes and key not in self._auto_keys: @@ -322,7 +327,7 @@ def add_event( dropped = 0 if attributes is not None: bounded, dropped = bound_attributes( - attributes, + copy_user_attributes({}, attributes), MAX_ATTRIBUTES_PER_EVENT, self._max_attribute_value_length, ) diff --git a/posthog/tracing/span.py b/posthog/tracing/span.py index 255d99a9..8bd401ec 100644 --- a/posthog/tracing/span.py +++ b/posthog/tracing/span.py @@ -84,7 +84,8 @@ def record_exception(self, exception: BaseException) -> "Span": Ignored after ``end()``. The event carries ``exception.type``, ``exception.message`` and, for a raised exception, ``exception.stacktrace`` (its last ``max_attribute_value_length`` - characters). Returns the span, so calls chain. Inside + characters, so a chained traceback keeps the outermost exception and + may lose its cause). Returns the span, so calls chain. Inside ``with span:`` a raised ``Exception`` is recorded automatically, so this is for exceptions that are caught and handled. From 3820d7d41e4cd8c28d357855cae6cbbfb45e2ba2 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 22:03:02 -0400 Subject: [PATCH 4/4] fix(traces): guard span writes and end() with the handle's lock A same-key write from two threads could reserve two slots, and a write that lost the race to end() could land after the record was taken. The cap check, the event count and the end() snapshot now run under the handle's lock; the value walk still runs outside it. --- posthog/test/tracing/test_span.py | 63 ++++++++++++++++ posthog/tracing/_otlp.py | 4 +- posthog/tracing/_span.py | 120 ++++++++++++++++++------------ 3 files changed, 140 insertions(+), 47 deletions(-) diff --git a/posthog/test/tracing/test_span.py b/posthog/test/tracing/test_span.py index 712cba93..8dae529a 100644 --- a/posthog/test/tracing/test_span.py +++ b/posthog/test/tracing/test_span.py @@ -3,6 +3,7 @@ import gc import sys import threading +import time import weakref from contextvars import ContextVar from datetime import datetime, timezone @@ -823,6 +824,68 @@ def test_add_event_with_a_non_mapping_logs_and_keeps_the_event(self, caplog): assert "expected a mapping" in caplog.text +class TestConcurrentWrites: + def _run(self, worker, count=8): + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(30) + + def test_writes_to_one_key_from_many_threads_spend_one_slot(self): + records: list = [] + span = make_span(records, max_attributes=1) + + def worker(i): + for n in range(300): + span.set_attribute("k", n) + + self._run(worker) + span.end() + assert list(records[0].attributes) == ["k"] + assert records[0].dropped_attributes_count == 0 + + def test_events_from_many_threads_never_exceed_the_cap(self): + records: list = [] + span = make_span(records, max_events=100) + + def worker(i): + for n in range(50): + span.add_event(f"{i}-{n}") + + self._run(worker) + span.end() + assert len(records[0].events) == 100 + assert records[0].dropped_events_count == 300 + + def test_a_write_racing_end_never_lands_after_the_record(self): + records: list = [] + span = make_span(records) + stop = threading.Event() + + def writer(i): + n = 0 + while not stop.is_set(): + span.set_attribute("k", n) + span.add_event("e") + n += 1 + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(4)] + for thread in threads: + thread.start() + time.sleep(0.02) + span.end() + stop.set() + for thread in threads: + thread.join(30) + record = records[0] + assert len(records) == 1 + assert span._events == record.events + assert {k: v for k, v in span._attributes.items() if v is not None} == ( + record.attributes + ) + + class TestExceptionStacktrace: def test_record_exception_attaches_the_stack_of_a_raised_exception(self): records: list = [] diff --git a/posthog/tracing/_otlp.py b/posthog/tracing/_otlp.py index 7801beb9..735ced0b 100644 --- a/posthog/tracing/_otlp.py +++ b/posthog/tracing/_otlp.py @@ -290,7 +290,9 @@ def _to_otlp_event(event: SpanEventRecord) -> dict: attributes, cut = encode_attributes(event.attributes) if attributes: encoded["attributes"] = attributes - dropped = non_negative_count(non_negative_count(event.dropped_attributes_count) + cut) + dropped = non_negative_count( + non_negative_count(event.dropped_attributes_count) + cut + ) if dropped: encoded["droppedAttributesCount"] = dropped return encoded diff --git a/posthog/tracing/_span.py b/posthog/tracing/_span.py index c287eaa3..ed22c4db 100644 --- a/posthog/tracing/_span.py +++ b/posthog/tracing/_span.py @@ -74,11 +74,12 @@ class _Activatable: _active_var: Optional[ContextVar] _tokens: List[Token] - _tokens_lock: threading.Lock + # Guards the tokens and, on a recording span, every write and end(). + _lock: threading.Lock def _activate(self) -> None: if self._active_var is not None: - with self._tokens_lock: + with self._lock: self._tokens.append(self._active_var.set(self)) def _deactivate(self) -> None: @@ -86,7 +87,7 @@ def _deactivate(self) -> None: return # The same handle can be entered in several threads or tasks at once, # and a token only resets in the context that created it. - with self._tokens_lock: + with self._lock: for index in range(len(self._tokens) - 1, -1, -1): try: self._active_var.reset(self._tokens[index]) @@ -117,7 +118,7 @@ def __init__( self._tracestate = tracestate self._active_var = active_var self._tokens = [] - self._tokens_lock = threading.Lock() + self._lock = threading.Lock() def traceparent(self) -> Optional[str]: return self._traceparent @@ -244,7 +245,7 @@ def __init__( self._on_end = on_end self._active_var = active_var self._tokens = [] - self._tokens_lock = threading.Lock() + self._lock = threading.Lock() self._name = name self._kind = kind @@ -258,11 +259,11 @@ def __init__( self._dropped_attributes = 0 self._dropped_events = 0 self._attributes: Dict[str, Any] = {} - for key, value in (attributes or {}).items(): - self._write_attribute(key, value) self._events: List[SpanEventRecord] = [] self._status: Optional[SpanStatus] = None self._ended = False + for key, value in (attributes or {}).items(): + self._write_attribute(key, value) def _now_ns(self) -> int: """Now, on this span's clock basis: start plus monotonic elapsed, else wall clock.""" @@ -282,21 +283,27 @@ def _write_attribute(self, key: str, value: Any) -> None: # The encoder drops it, so it must not spend a slot. log.debug("Dropping an attribute with an empty key") return - if value is None: - # None removes the key, freeing its slot. - if key in self._attributes and key not in self._auto_keys: - self._user_attribute_count -= 1 - self._attributes.pop(key, None) - return - # Checked before the value is walked, which is the costly part. - if key not in self._auto_keys and key not in self._attributes: - if self._user_attribute_count >= self._max_attributes: - self._dropped_attributes += 1 + with self._lock: + if self._ended: return - self._user_attribute_count += 1 - self._attributes[key] = truncate_attribute_value( - value, self._max_attribute_value_length - ) + if value is None: + # None removes the key, freeing its slot. + if key in self._attributes and key not in self._auto_keys: + self._user_attribute_count -= 1 + self._attributes.pop(key, None) + return + # Checked before the value is walked, which is the costly part. + if key not in self._auto_keys and key not in self._attributes: + if self._user_attribute_count >= self._max_attributes: + self._dropped_attributes += 1 + return + self._user_attribute_count += 1 + # Reserved, so a concurrent write to the same key sees it taken. + self._attributes[key] = None + bounded = truncate_attribute_value(value, self._max_attribute_value_length) + with self._lock: + if not self._ended: + self._attributes[key] = bounded def set_attribute(self, key: str, value: Any) -> "Span": if self._mutable("set_attribute"): @@ -318,11 +325,14 @@ def add_event( timestamp: Optional[SpanTimeInput] = None, ) -> "Span": if self._mutable("add_event"): - # A recorded exception spends a slot like any other event. - if self._user_event_count >= self._max_events: - self._dropped_events += 1 - return self - self._user_event_count += 1 + with self._lock: + if self._ended: + return self + # A recorded exception spends a slot like any other event. + if self._user_event_count >= self._max_events: + self._dropped_events += 1 + return self + self._user_event_count += 1 bounded: Optional[Dict[str, Any]] = None dropped = 0 if attributes is not None: @@ -331,18 +341,19 @@ def add_event( MAX_ATTRIBUTES_PER_EVENT, self._max_attribute_value_length, ) - self._events.append( - SpanEventRecord( - name=sanitize_name( - name, "Span event name", self._max_attribute_value_length - ), - timestamp_ns=resolve_supplied_ns( - timestamp, self._now_ns(), "event timestamp" - ), - attributes=bounded, - dropped_attributes_count=dropped, - ) + event = SpanEventRecord( + name=sanitize_name( + name, "Span event name", self._max_attribute_value_length + ), + timestamp_ns=resolve_supplied_ns( + timestamp, self._now_ns(), "event timestamp" + ), + attributes=bounded, + dropped_attributes_count=dropped, ) + with self._lock: + if not self._ended: + self._events.append(event) return self def set_status(self, code: str, message: Optional[str] = None) -> "Span": @@ -351,12 +362,15 @@ def set_status(self, code: str, message: Optional[str] = None) -> "Span": log.debug('Ignoring an unknown span status; expected "ok" or "error"') return self text = None if message is None else safe_str(message) - self._status = SpanStatus( + status = SpanStatus( code, truncate_string(text, self._max_attribute_value_length) if text else None, ) + with self._lock: + if not self._ended: + self._status = status return self @property @@ -379,9 +393,12 @@ def _record_exception(self, exception: BaseException, keep_ok: bool) -> None: def update_name(self, name: str) -> "Span": if self._mutable("update_name"): - self._name = sanitize_name( + sanitized = sanitize_name( name, "Span name", self._max_attribute_value_length ) + with self._lock: + if not self._ended: + self._name = sanitized return self def traceparent(self) -> Optional[str]: @@ -400,11 +417,22 @@ def _child_context(self) -> ParentContext: ) def end(self, end_time: Optional[SpanTimeInput] = None) -> None: - with self._tokens_lock: + with self._lock: if self._ended: log.debug("Ignoring end() on a span that has already ended") return self._ended = True + # Snapshotted under the lock: a write that lost the race to end() + # must not land in the record or after it. + attributes = { + key: value + for key, value in self._attributes.items() + if value is not None + } + events = list(self._events) + name, status = self._name, self._status + dropped_attributes = self._dropped_attributes + dropped_events = self._dropped_events derived = self._now_ns() resolved = resolve_supplied_ns(end_time, derived, "end time") @@ -415,15 +443,15 @@ def end(self, end_time: Optional[SpanTimeInput] = None) -> None: trace_state=self._trace_state, trace_flags=self._trace_flags, parent_is_remote=self._parent_is_remote, - name=self._name, + name=name, kind=self._kind, - status=self._status, - attributes=dict(self._attributes), - events=self._events, + status=status, + attributes=attributes, + events=events, start_ns=self._start_ns, end_ns=clamp_end_ns(resolved, self._start_ns), - dropped_attributes_count=self._dropped_attributes, - dropped_events_count=self._dropped_events, + dropped_attributes_count=dropped_attributes, + dropped_events_count=dropped_events, ) try: self._on_end(record)