From 9e06211d039c6d1b60819b6090b67224e5fb8f21 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Thu, 10 Sep 2026 21:35:28 -0400 Subject: [PATCH 1/5] feat(traces): OTLP span encoding and client-side validity Adds the OTLP JSON encoder for spans: AnyValue encoding per the traces spec (int64 as strings, out-of-range ints and non-finite floats as strings, None values and empty keys dropped, unpaired surrogates replaced with U+FFFD, a bounded walk that terminates on cycles), the span record builder with the W3C flags byte and OTel remoteness bits, the one-resource/one-scope envelope with service.name always present, and the name and timestamp sanitizers. One bad value otherwise gets the whole batch rejected. 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_otlp.py | 533 ++++++++++++++++++++++++++ posthog/test/tracing/test_sanitize.py | 141 +++++++ posthog/tracing/_otlp.py | 362 +++++++++++++++++ posthog/tracing/_sanitize.py | 150 ++++++++ 4 files changed, 1186 insertions(+) create mode 100644 posthog/test/tracing/test_otlp.py create mode 100644 posthog/test/tracing/test_sanitize.py create mode 100644 posthog/tracing/_otlp.py create mode 100644 posthog/tracing/_sanitize.py diff --git a/posthog/test/tracing/test_otlp.py b/posthog/test/tracing/test_otlp.py new file mode 100644 index 00000000..214af146 --- /dev/null +++ b/posthog/test/tracing/test_otlp.py @@ -0,0 +1,533 @@ +import json +from datetime import datetime, timezone +from unittest import mock + +import pytest + +from posthog.tracing import _otlp +from posthog.tracing._otlp import ( + CIRCULAR_VALUE, + MAX_VALUE_ITEMS, + MAX_VALUE_NODES, + TRUNCATED_VALUE, + SpanEventRecord, + SpanRecord, + SpanStatus, + build_otlp_span, + build_resource_attributes, + build_traces_payload, + host_resource_attributes, + span_kind_to_otlp, + to_any_value, + to_key_value_list, +) +from posthog.tracing._sanitize import UNSERIALIZABLE_VALUE +from posthog.version import VERSION + +TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736" +SPAN_ID = "00f067aa0ba902b7" +START_NS = 1_700_000_000_000_000_000 +END_NS = START_NS + 80_000_000 + + +def record(**overrides) -> SpanRecord: + base = dict( + trace_id=TRACE_ID, + span_id=SPAN_ID, + name="checkout", + start_ns=START_NS, + end_ns=END_NS, + ) + base.update(overrides) + return SpanRecord(**base) + + +class TestToAnyValue: + @pytest.mark.parametrize( + "value,expected", + [ + (True, {"boolValue": True}), + (False, {"boolValue": False}), + (42, {"intValue": "42"}), + (-7, {"intValue": "-7"}), + (2**63 - 1, {"intValue": str(2**63 - 1)}), + (0.25, {"doubleValue": 0.25}), + (2.0, {"doubleValue": 2.0}), + (float("nan"), {"stringValue": "NaN"}), + (float("inf"), {"stringValue": "Infinity"}), + (float("-inf"), {"stringValue": "-Infinity"}), + ("hello", {"stringValue": "hello"}), + (b"\x00\x01", {"stringValue": "b'\\x00\\x01'"}), + ( + datetime(2023, 1, 1, tzinfo=timezone.utc), + {"stringValue": "2023-01-01T00:00:00+00:00"}, + ), + ], + ) + def test_encodes_primitives(self, value, expected): + assert to_any_value(value) == expected + + def test_encodes_an_integer_beyond_int64_as_a_string(self): + assert to_any_value(2**64) == {"stringValue": str(2**64)} + assert to_any_value(-(2**63) - 1) == {"stringValue": str(-(2**63) - 1)} + + def test_encodes_arrays_and_drops_none_elements(self): + assert to_any_value([1, None, "a"]) == { + "arrayValue": {"values": [{"intValue": "1"}, {"stringValue": "a"}]} + } + + def test_encodes_mappings_as_kvlist(self): + assert to_any_value({"a": 1, "b": None}) == { + "kvlistValue": {"values": [{"key": "a", "value": {"intValue": "1"}}]} + } + + def test_replaces_a_lone_surrogate_with_the_replacement_character(self): + assert to_any_value("value \ud83d") == {"stringValue": "value \ufffd"} + assert to_any_value("\udc00x") == {"stringValue": "\ufffdx"} + + def test_keeps_a_surrogate_pair_as_the_character_it_spells(self): + assert to_any_value("a" + "\ud83d" + "\ude00") == {"stringValue": "a\U0001f600"} + + def test_encodes_an_int_subclass_as_its_number(self): + import enum + + class Code(enum.IntEnum): + OK = 200 + + class Weird(int): + def __str__(self): + return "weird" + + assert to_any_value(Code.OK) == {"intValue": "200"} + assert to_any_value(Weird(3)) == {"intValue": "3"} + + def test_falls_back_to_str_for_unknown_types(self): + class Thing: + def __str__(self): + return "thing" + + assert to_any_value(Thing()) == {"stringValue": "thing"} + + def test_encodes_a_callable_as_a_stable_marker(self): + assert to_any_value(lambda: None) == {"stringValue": "[Function]"} + + def test_survives_a_hostile_str(self): + class Hostile: + def __str__(self): + raise RuntimeError("no") + + assert to_any_value(Hostile()) == {"stringValue": UNSERIALIZABLE_VALUE} + + def test_marks_a_cycle_instead_of_recursing(self): + loop: dict = {} + loop["self"] = loop + assert to_any_value(loop) == { + "kvlistValue": { + "values": [{"key": "self", "value": {"stringValue": CIRCULAR_VALUE}}] + } + } + + def test_treats_a_repeated_sibling_as_duplication_not_a_cycle(self): + shared = {"x": 1} + encoded = to_any_value([shared, shared]) + assert ( + encoded["arrayValue"]["values"] + == [{"kvlistValue": {"values": [{"key": "x", "value": {"intValue": "1"}}]}}] + * 2 + ) + + def test_truncates_a_deep_value(self): + value: list = [] + current = value + for _ in range(30): + nested: list = [] + current.append(nested) + current = nested + encoded = to_any_value(value) + while "arrayValue" in encoded: + encoded = encoded["arrayValue"]["values"][0] + assert encoded == {"stringValue": TRUNCATED_VALUE} + + def test_truncates_a_long_array(self): + encoded = to_any_value(list(range(MAX_VALUE_ITEMS + 5))) + values = encoded["arrayValue"]["values"] + assert len(values) == MAX_VALUE_ITEMS + 1 + assert values[-1] == {"stringValue": TRUNCATED_VALUE} + + def test_bounds_total_nodes(self): + encoded = json.dumps(to_any_value([[1] * 200 for _ in range(200)])) + assert TRUNCATED_VALUE in encoded + assert 0 < encoded.count('"intValue"') < MAX_VALUE_NODES + + +class TestToKeyValueList: + def test_drops_none_values_and_empty_keys(self): + assert to_key_value_list({"": 1, "a": None, "b": 2}) == [ + {"key": "b", "value": {"intValue": "2"}} + ] + + def test_stringifies_non_string_keys(self): + assert to_key_value_list({1: "x"}) == [ + {"key": "1", "value": {"stringValue": "x"}} + ] + + def test_marks_only_the_raising_key(self): + class Explosive(dict): + def __getitem__(self, key): + if key == "bad": + raise RuntimeError("boom") + return super().__getitem__(key) + + assert to_key_value_list(Explosive(good=1, bad=2)) == [ + {"key": "good", "value": {"intValue": "1"}}, + {"key": "bad", "value": {"stringValue": UNSERIALIZABLE_VALUE}}, + ] + + def test_returns_empty_for_a_non_mapping(self): + assert to_key_value_list(["a"]) == [] + assert to_key_value_list(None) == [] + + +class TestSpanKindToOtlp: + @pytest.mark.parametrize( + "kind,code", + [ + ("internal", 1), + ("server", 2), + ("client", 3), + ("producer", 4), + ("consumer", 5), + ], + ) + def test_maps_each_kind(self, kind, code): + assert span_kind_to_otlp(kind) == code + + @pytest.mark.parametrize("kind", [None, "", "weird", 3, "__class__"]) + def test_defaults_to_internal(self, kind): + assert span_kind_to_otlp(kind) == 1 + + +class TestBuildOtlpSpan: + def test_builds_the_minimal_shape(self): + assert build_otlp_span(record()) == { + "traceId": TRACE_ID, + "spanId": SPAN_ID, + "name": "checkout", + "kind": 1, + "startTimeUnixNano": str(START_NS), + "endTimeUnixNano": str(END_NS), + "flags": 0x101, + } + + def test_omits_status_when_never_set(self): + assert "status" not in build_otlp_span(record()) + + def test_encodes_ok_and_error_status_codes(self): + assert build_otlp_span(record(status=SpanStatus("ok")))["status"] == {"code": 1} + assert build_otlp_span(record(status=SpanStatus("error", "boom")))[ + "status" + ] == { + "code": 2, + "message": "boom", + } + + def test_ignores_an_unknown_status_code(self): + assert "status" not in build_otlp_span(record(status=SpanStatus("weird"))) + + def test_includes_parent_tracestate_attributes_and_events(self): + span = build_otlp_span( + record( + parent_span_id="b7ad6b7169203331", + trace_state="vendor=abc", + attributes={"k": "v"}, + events=[ + SpanEventRecord("cache miss", START_NS + 10, {"key": "user:1"}) + ], + ) + ) + assert span["parentSpanId"] == "b7ad6b7169203331" + assert span["traceState"] == "vendor=abc" + assert span["attributes"] == [{"key": "k", "value": {"stringValue": "v"}}] + assert span["events"] == [ + { + "name": "cache miss", + "timeUnixNano": str(START_NS + 10), + "attributes": [{"key": "key", "value": {"stringValue": "user:1"}}], + } + ] + + def test_omits_event_attributes_when_empty(self): + span = build_otlp_span(record(events=[SpanEventRecord("tick", START_NS, {})])) + assert span["events"] == [{"name": "tick", "timeUnixNano": str(START_NS)}] + + def test_marks_a_root_span_as_known_not_remote(self): + assert build_otlp_span(record())["flags"] == 0x101 + + def test_marks_a_header_parent_as_remote(self): + assert build_otlp_span(record(parent_is_remote=True))["flags"] == 0x301 + + def test_propagates_an_inbound_sampled_out_flag(self): + assert build_otlp_span(record(trace_flags="00"))["flags"] == 0x100 + + def test_falls_back_to_sampled_when_the_flags_byte_is_unusable(self): + assert build_otlp_span(record(trace_flags="zz"))["flags"] == 0x101 + + def test_replaces_lone_surrogates_in_every_free_text_field(self): + lone = "value \ud83d" + span = build_otlp_span( + record( + name=lone, + trace_state=f"vendor={lone}", + status=SpanStatus("error", lone), + events=[SpanEventRecord(lone, START_NS)], + ) + ) + for text in ( + span["name"], + span["traceState"], + span["status"]["message"], + span["events"][0]["name"], + ): + assert "\ud83d" not in text + assert "\ufffd" in text + + def test_replaces_lone_surrogates_in_attribute_keys(self): + span = build_otlp_span(record(attributes={"k\ud800": "v"})) + assert span["attributes"][0]["key"] == "k\ufffd" + + def test_drops_only_an_attribute_whose_key_cannot_be_stringified(self): + class HostileKey: + def __str__(self): + raise RuntimeError("no") + + def __hash__(self): + return 1 + + assert to_key_value_list({"a": {HostileKey(): 1, "b": 2}}) == [ + { + "key": "a", + "value": { + "kvlistValue": { + "values": [{"key": "b", "value": {"intValue": "2"}}] + } + }, + } + ] + + def test_keeps_the_span_when_a_status_message_cannot_be_stringified(self): + class Hostile: + def __str__(self): + raise RuntimeError("no") + + span = build_otlp_span(record(status=SpanStatus("error", Hostile()))) + assert span["status"] == {"code": 2, "message": UNSERIALIZABLE_VALUE} + + +class TestResourceAttributes: + def test_always_emits_service_name(self): + attrs = build_resource_attributes(None, None, None, {}) + assert attrs["service.name"] == "unknown_service" + assert attrs["telemetry.sdk.name"] == "posthog-python" + assert attrs["telemetry.sdk.version"] == VERSION + + def test_uses_the_configured_service_name_and_optional_keys(self): + attrs = build_resource_attributes("api", "1.2.3", "prod", {}) + assert attrs["service.name"] == "api" + assert attrs["service.version"] == "1.2.3" + assert attrs["deployment.environment"] == "prod" + + def test_omits_environment_and_version_when_unset(self): + attrs = build_resource_attributes("api", None, None, {}) + assert "service.version" not in attrs + assert "deployment.environment" not in attrs + + def test_protects_sdk_identity_keys_from_user_attributes(self): + attrs = build_resource_attributes( + "api", None, None, {"telemetry.sdk.name": "custom", "region": "eu"} + ) + assert attrs["telemetry.sdk.name"] == "posthog-python" + assert attrs["region"] == "eu" + + def test_host_attributes_map_darwin_to_macos(self): + with ( + mock.patch.object(_otlp.platform, "system", return_value="Darwin"), + mock.patch.object(_otlp.platform, "release", return_value="23.1.0"), + ): + assert host_resource_attributes() == { + "os.name": "macOS", + "os.version": "23.1.0", + } + + def test_host_attributes_pass_unknown_names_through_and_omit_empty(self): + with ( + mock.patch.object(_otlp.platform, "system", return_value="Linux"), + mock.patch.object(_otlp.platform, "release", return_value=""), + ): + assert host_resource_attributes() == {"os.name": "Linux"} + + def test_host_attributes_report_the_windows_build_like_node(self): + with ( + mock.patch.object(_otlp.platform, "system", return_value="Windows"), + mock.patch.object(_otlp.platform, "release", return_value="11"), + mock.patch.object(_otlp.platform, "version", return_value="10.0.22631"), + ): + assert host_resource_attributes() == { + "os.name": "Windows", + "os.version": "10.0.22631", + } + + @pytest.mark.parametrize( + "system", ["CYGWIN_NT-10.0-19045", "MSYS_NT-10.0-19045", "MINGW64_NT-10.0"] + ) + def test_host_attributes_file_posix_layers_over_windows_under_windows(self, system): + with ( + mock.patch.object(_otlp.platform, "system", return_value=system), + mock.patch.object(_otlp.platform, "release", return_value="3.5.4"), + ): + assert host_resource_attributes() == { + "os.name": "Windows", + "os.version": "3.5.4", + } + + def test_an_oversized_user_attribute_does_not_cost_the_identity_keys(self): + # One user value large enough to exhaust the encoder's traversal budget. + attrs = build_resource_attributes( + "api", "1.2.3", "prod", {"huge": [list(range(1000))] * 20, "os.name": "x"} + ) + payload = build_traces_payload([], attrs) + keys = [ + kv["key"] for kv in payload["resourceSpans"][0]["resource"]["attributes"] + ] + # User keys share one budget, so "os.name" after "huge" is lost with it; + # the SDK's own keys are encoded on a budget of their own. + assert keys == [ + "huge", + "service.name", + "deployment.environment", + "service.version", + "telemetry.sdk.name", + "telemetry.sdk.version", + ] + + def test_host_attributes_survive_a_failing_platform_module(self): + with mock.patch.object(_otlp.platform, "system", side_effect=OSError("no")): + assert host_resource_attributes() == {} + + +class TestBuildTracesPayload: + def test_produces_one_resource_one_scope_n_spans(self): + spans = [build_otlp_span(record()) for _ in range(20)] + payload = build_traces_payload(spans, {"service.name": "api"}) + assert len(payload["resourceSpans"]) == 1 + assert len(payload["resourceSpans"][0]["scopeSpans"]) == 1 + assert len(payload["resourceSpans"][0]["scopeSpans"][0]["spans"]) == 20 + + def test_matches_the_shape_the_ingestion_service_accepts(self): + # Golden fixture ported from posthog-js. + payload = build_traces_payload( + [ + build_otlp_span( + record( + parent_span_id="b7ad6b7169203331", + name="GET /users/:id", + kind="server", + status=SpanStatus("error", "boom"), + attributes={ + "posthogDistinctId": "user-123", + "sessionId": "session-123", + "http.status_code": 500, + "http.duration_ratio": 0.25, + "cached": False, + }, + events=[ + SpanEventRecord( + "exception", + START_NS + 40_000_000, + { + "exception.type": "TypeError", + "exception.message": "boom", + }, + ) + ], + ) + ) + ], + {"service.name": "checkout-api", "telemetry.sdk.name": "posthog-python"}, + ) + + assert payload == { + "resourceSpans": [ + { + "resource": { + "attributes": [ + { + "key": "service.name", + "value": {"stringValue": "checkout-api"}, + }, + { + "key": "telemetry.sdk.name", + "value": {"stringValue": "posthog-python"}, + }, + ] + }, + "scopeSpans": [ + { + "scope": {"name": "posthog-python", "version": VERSION}, + "spans": [ + { + "traceId": TRACE_ID, + "spanId": SPAN_ID, + "parentSpanId": "b7ad6b7169203331", + "name": "GET /users/:id", + "kind": 2, + "startTimeUnixNano": "1700000000000000000", + "endTimeUnixNano": "1700000000080000000", + "flags": 0x101, + "attributes": [ + { + "key": "posthogDistinctId", + "value": {"stringValue": "user-123"}, + }, + { + "key": "sessionId", + "value": {"stringValue": "session-123"}, + }, + { + "key": "http.status_code", + "value": {"intValue": "500"}, + }, + { + "key": "http.duration_ratio", + "value": {"doubleValue": 0.25}, + }, + { + "key": "cached", + "value": {"boolValue": False}, + }, + ], + "events": [ + { + "name": "exception", + "timeUnixNano": "1700000000040000000", + "attributes": [ + { + "key": "exception.type", + "value": { + "stringValue": "TypeError" + }, + }, + { + "key": "exception.message", + "value": {"stringValue": "boom"}, + }, + ], + } + ], + "status": {"code": 2, "message": "boom"}, + } + ], + } + ], + } + ] + } diff --git a/posthog/test/tracing/test_sanitize.py b/posthog/test/tracing/test_sanitize.py new file mode 100644 index 00000000..f4b3b147 --- /dev/null +++ b/posthog/test/tracing/test_sanitize.py @@ -0,0 +1,141 @@ +import math +from datetime import datetime, timezone + +import pytest + +from posthog.tracing._sanitize import ( + FALLBACK_SPAN_NAME, + MAX_TIMESTAMP_NS, + UNSERIALIZABLE_VALUE, + clamp_end_ns, + copy_user_attributes, + resolve_start_ns, + resolve_supplied_ns, + sanitize_name, + to_epoch_ns, +) + +NOW_NS = 1_700_000_000_000_000_000 + + +class TestSanitizeName: + def test_keeps_a_non_empty_string(self): + assert sanitize_name("checkout", "Span name") == "checkout" + + @pytest.mark.parametrize("name", ["", " ", None, 42, ["a"]]) + def test_replaces_an_unusable_name(self, name): + assert sanitize_name(name, "Span name") == FALLBACK_SPAN_NAME + + +class TestToEpochNs: + def test_converts_an_aware_datetime(self): + dt = datetime(2023, 11, 14, 22, 13, 20, 40_000, tzinfo=timezone.utc) + assert to_epoch_ns(dt) == NOW_NS + 40_000_000 + + def test_treats_a_naive_datetime_as_local_time(self): + naive = datetime(2023, 11, 14, 22, 13, 20) + assert to_epoch_ns(naive) == int(naive.timestamp()) * 10**9 + + def test_converts_integer_seconds(self): + assert to_epoch_ns(1_700_000_000) == NOW_NS + + def test_converts_float_seconds(self): + assert to_epoch_ns(1_700_000_000.5) == NOW_NS + 500_000_000 + + @pytest.mark.parametrize( + "value", + [ + None, + True, + "1700000000", + float("nan"), + float("inf"), + -1, + MAX_TIMESTAMP_NS // 10**9 + 1, + datetime(1960, 1, 1, tzinfo=timezone.utc), + ], + ) + def test_rejects_unusable_values(self, value): + assert to_epoch_ns(value) is None + + def test_rejects_a_datetime_whose_arithmetic_raises(self): + class Hostile(datetime): + def __sub__(self, other): + raise RuntimeError("no") + + assert to_epoch_ns(Hostile(2023, 1, 1, tzinfo=timezone.utc)) is None + + +class TestResolveStartNs: + def test_uses_now_when_nothing_is_supplied(self): + assert resolve_start_ns(None, NOW_NS) == NOW_NS + + def test_uses_now_for_an_unusable_value(self): + assert resolve_start_ns("yesterday", NOW_NS) == NOW_NS + + def test_backdates_to_a_supplied_time(self): + assert resolve_start_ns(1_699_999_000, NOW_NS) == 1_699_999_000 * 10**9 + + def test_warns_when_the_server_will_clamp(self, caplog): + caplog.set_level("DEBUG", logger="posthog") + resolve_start_ns(1_700_000_000 - 48 * 3600, NOW_NS) + assert any("24 hours" in r.getMessage() for r in caplog.records) + + def test_keeps_a_future_start_and_warns(self, caplog): + caplog.set_level("DEBUG", logger="posthog") + future = 1_700_000_000 + 3600 + assert resolve_start_ns(future, NOW_NS) == future * 10**9 + assert "in the future" in caplog.text + + +class TestEndAndSuppliedTimes: + def test_clamps_an_end_before_the_start(self): + assert clamp_end_ns(NOW_NS - 1, NOW_NS) == NOW_NS + + def test_keeps_an_end_after_the_start(self): + assert clamp_end_ns(NOW_NS + 5, NOW_NS) == NOW_NS + 5 + + def test_supplied_time_wins_when_valid(self): + assert resolve_supplied_ns(1_700_000_001, NOW_NS, "end time") == NOW_NS + 10**9 + + @pytest.mark.parametrize("value", [None, "soon", math.nan, -5]) + def test_falls_back_to_the_derived_time(self, value): + assert resolve_supplied_ns(value, NOW_NS, "end time") == NOW_NS + + +class TestCopyUserAttributes: + def test_copies_a_mapping(self): + assert copy_user_attributes({"a": 1}, {"b": 2}) == {"a": 1, "b": 2} + + def test_user_keys_win_on_collision(self): + assert copy_user_attributes({"a": 1}, {"a": 2}) == {"a": 2} + + def test_ignores_none_and_non_mappings(self): + assert copy_user_attributes({"a": 1}, None) == {"a": 1} + assert copy_user_attributes({"a": 1}, ["b"]) == {"a": 1} + + def test_marks_only_the_raising_key(self): + class Explosive(dict): + def __getitem__(self, key): + if key == "bad": + raise RuntimeError("boom") + return super().__getitem__(key) + + source = Explosive(good=1, bad=2) + assert copy_user_attributes({}, source) == { + "good": 1, + "bad": UNSERIALIZABLE_VALUE, + } + + def test_stringifies_non_string_keys(self): + assert copy_user_attributes({}, {1: "x"}) == {"1": "x"} + + def test_drops_only_a_key_that_cannot_be_stringified(self): + class HostileKey: + def __str__(self): + raise RuntimeError("no") + + def __hash__(self): + return 1 + + assert copy_user_attributes({}, {HostileKey(): 1, "ok": 2}) == {"ok": 2} diff --git a/posthog/tracing/_otlp.py b/posthog/tracing/_otlp.py new file mode 100644 index 00000000..0669a481 --- /dev/null +++ b/posthog/tracing/_otlp.py @@ -0,0 +1,362 @@ +"""OTLP/JSON encoding for spans. + +Values come from application code, and one the server refuses rejects the whole +request, so the encoder produces an acceptable payload whatever it is handed. +""" + +import logging +import math +import platform +from dataclasses import dataclass, field +from datetime import date, datetime +from typing import Any, Dict, List, Mapping, Optional + +from ..version import VERSION +from ._sanitize import FUNCTION_VALUE, UNSERIALIZABLE_VALUE, attribute_key, safe_str + +log = logging.getLogger("posthog") + +SCOPE_NAME = "posthog-python" + +SPAN_KIND_TO_OTLP = { + "internal": 1, + "server": 2, + "client": 3, + "producer": 4, + "consumer": 5, +} +SPAN_STATUS_TO_OTLP = {"ok": 1, "error": 2} + +# The W3C trace flags are the low byte; OTel's parent-remoteness bits sit above. +TRACE_FLAGS_SAMPLED = 0x01 +SPAN_FLAGS_CONTEXT_HAS_IS_REMOTE = 0x100 +SPAN_FLAGS_CONTEXT_IS_REMOTE = 0x200 + +INT64_MAX = 2**63 - 1 +INT64_MIN = -(2**63) + +# Bounds on the value walk, so a self-referencing value terminates. +MAX_VALUE_DEPTH = 20 +MAX_VALUE_ITEMS = 1000 +MAX_VALUE_NODES = 10000 +CIRCULAR_VALUE = "[Circular]" +TRUNCATED_VALUE = "[Truncated]" + + +@dataclass +class SpanEventRecord: + name: str + timestamp_ns: int + attributes: Optional[Dict[str, Any]] = None + + +@dataclass +class SpanStatus: + code: str # "ok" | "error" + message: Optional[str] = None + + +@dataclass +class SpanRecord: + """A completed span in plain, pre-encoding form.""" + + trace_id: str + span_id: str + name: str + start_ns: int + end_ns: int + parent_span_id: Optional[str] = None + trace_state: Optional[str] = None + trace_flags: str = "01" + # True when the parent came from a traceparent header. + parent_is_remote: bool = False + kind: str = "internal" + status: Optional[SpanStatus] = None + attributes: Dict[str, Any] = field(default_factory=dict) + events: List[SpanEventRecord] = field(default_factory=list) + + +def sanitize_string(value: str) -> str: + """Replace unpaired surrogates, which the service rejects, with U+FFFD.""" + try: + value.encode("utf-8") + return value + except UnicodeEncodeError: + return value.encode("utf-16-le", "surrogatepass").decode("utf-16-le", "replace") + + +def wire_string(value: Any) -> str: + return sanitize_string(safe_str(value)) + + +class _EncodeState: + __slots__ = ("ancestors", "remaining_nodes") + + def __init__(self) -> None: + # Containers on the current path; a back-reference becomes a marker. + self.ancestors: set = set() + self.remaining_nodes = MAX_VALUE_NODES + + +def to_any_value(value: Any) -> dict: + """Encode one attribute value as an OTLP ``AnyValue``.""" + try: + return _encode(value, _EncodeState(), 0) + except Exception: + return {"stringValue": UNSERIALIZABLE_VALUE} + + +def to_key_value_list(attributes: Any) -> list: + """Encode an attribute mapping as an OTLP ``KeyValue`` list; ``None`` values are dropped.""" + try: + return _encode_key_value_list(attributes, _EncodeState(), 0) + except Exception: + return [] + + +def _encode(value: Any, state: _EncodeState, depth: int) -> dict: + if state.remaining_nodes <= 0: + return {"stringValue": TRUNCATED_VALUE} + state.remaining_nodes -= 1 + + # bool is an int subclass. + if isinstance(value, bool): + return {"boolValue": value} + if isinstance(value, int): + if value > INT64_MAX or value < INT64_MIN: + log.debug( + "Attribute %s is outside the int64 range; encoding it as a string", + value, + ) + return {"stringValue": str(value)} + # int(): an IntEnum stringifies as its name. + return {"intValue": str(int(value))} + if isinstance(value, float): + if not math.isfinite(value): + if math.isnan(value): + return {"stringValue": "NaN"} + return {"stringValue": "Infinity" if value > 0 else "-Infinity"} + return {"doubleValue": value} + if isinstance(value, str): + return {"stringValue": sanitize_string(value)} + if isinstance(value, (datetime, date)): + return {"stringValue": value.isoformat()} + if isinstance(value, Mapping) or isinstance(value, (list, tuple, set, frozenset)): + marker = id(value) + if marker in state.ancestors: + return {"stringValue": CIRCULAR_VALUE} + if depth >= MAX_VALUE_DEPTH: + return {"stringValue": TRUNCATED_VALUE} + state.ancestors.add(marker) + try: + if isinstance(value, Mapping): + return { + "kvlistValue": { + "values": _encode_key_value_list(value, state, depth + 1) + } + } + return {"arrayValue": {"values": _encode_array(value, state, depth + 1)}} + finally: + # Siblings sharing one object are not a cycle. + state.ancestors.discard(marker) + if callable(value): + return {"stringValue": FUNCTION_VALUE} + return {"stringValue": wire_string(value)} + + +def _encode_array(values: Any, state: _EncodeState, depth: int) -> list: + result: list = [] + count = 0 + truncated = False + for element in values: + if count >= MAX_VALUE_ITEMS or state.remaining_nodes <= 0: + truncated = True + break + count += 1 + # proto3 JSON has no null AnyValue. + if element is None: + continue + try: + result.append(_encode(element, state, depth)) + except Exception: + result.append({"stringValue": UNSERIALIZABLE_VALUE}) + if truncated: + result.append({"stringValue": TRUNCATED_VALUE}) + return result + + +def _encode_key_value_list(attributes: Any, state: _EncodeState, depth: int) -> list: + result: list = [] + if not isinstance(attributes, Mapping): + return result + for key in list(attributes.keys()): + key_str = attribute_key(key) + if key_str is None: + continue + if not key_str: + log.debug("Dropping an attribute with an empty key") + continue + if len(result) >= MAX_VALUE_ITEMS or state.remaining_nodes <= 0: + log.debug("Attributes truncated: the value exceeds the OTLP encoder budget") + break + try: + value = attributes[key] + if value is None: + continue + result.append( + {"key": sanitize_string(key_str), "value": _encode(value, state, depth)} + ) + except Exception: + result.append( + { + "key": sanitize_string(key_str), + "value": {"stringValue": UNSERIALIZABLE_VALUE}, + } + ) + return result + + +def span_kind_to_otlp(kind: Any) -> int: + if isinstance(kind, str) and kind in SPAN_KIND_TO_OTLP: + return SPAN_KIND_TO_OTLP[kind] + return SPAN_KIND_TO_OTLP["internal"] + + +def _span_flags(record: SpanRecord) -> int: + try: + w3c = int(record.trace_flags, 16) & 0xFF + except (TypeError, ValueError): + w3c = TRACE_FLAGS_SAMPLED + return ( + w3c + | SPAN_FLAGS_CONTEXT_HAS_IS_REMOTE + | (SPAN_FLAGS_CONTEXT_IS_REMOTE if record.parent_is_remote else 0) + ) + + +def _to_otlp_event(event: SpanEventRecord) -> dict: + encoded: dict = { + "name": wire_string(event.name), + "timeUnixNano": str(event.timestamp_ns), + } + if event.attributes: + attributes = to_key_value_list(event.attributes) + if attributes: + encoded["attributes"] = attributes + return encoded + + +def build_otlp_span(record: SpanRecord) -> dict: + span: dict = { + "traceId": record.trace_id, + "spanId": record.span_id, + "name": wire_string(record.name), + "kind": span_kind_to_otlp(record.kind), + "startTimeUnixNano": str(record.start_ns), + "endTimeUnixNano": str(record.end_ns), + "flags": _span_flags(record), + } + if record.parent_span_id: + span["parentSpanId"] = record.parent_span_id + if record.trace_state: + span["traceState"] = wire_string(record.trace_state) + attributes = to_key_value_list(record.attributes) + if attributes: + span["attributes"] = attributes + if record.events: + span["events"] = [_to_otlp_event(event) for event in record.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: + status["message"] = wire_string(record.status.message) + span["status"] = status + return span + + +def build_traces_payload(spans: List[dict], resource_attributes: Mapping) -> dict: + """Wrap spans in the OTLP envelope: one resource, one scope, N spans per batch.""" + return { + "resourceSpans": [ + { + "resource": { + "attributes": to_resource_key_value_list(resource_attributes) + }, + "scopeSpans": [ + { + "scope": {"name": SCOPE_NAME, "version": VERSION}, + "spans": spans, + } + ], + } + ] + } + + +def build_resource_attributes( + service_name: Optional[str], + service_version: Optional[str], + environment: Optional[str], + resource_attributes: Mapping, +) -> dict: + """OTLP resource attributes for every batch; the SDK's identity keys win. + + ``service.name`` is always emitted: the server attributes spans by it alone. + """ + attributes: dict = dict(resource_attributes) + attributes["service.name"] = service_name or "unknown_service" + if environment: + attributes["deployment.environment"] = environment + if service_version: + attributes["service.version"] = service_version + attributes["telemetry.sdk.name"] = SCOPE_NAME + attributes["telemetry.sdk.version"] = VERSION + return attributes + + +_SDK_RESOURCE_KEYS = ( + "service.name", + "deployment.environment", + "service.version", + "telemetry.sdk.name", + "telemetry.sdk.version", +) + + +def to_resource_key_value_list(attributes: Mapping) -> list: + """Encode resource attributes, the SDK's keys last on a budget of their own, + so a huge user value cannot cost the resource its ``service.name``.""" + user = dict(attributes) + sdk = {key: user.pop(key) for key in _SDK_RESOURCE_KEYS if key in user} + return to_key_value_list(user) + to_key_value_list(sdk) + + +# platform.system() spellings that differ from the os.name the other PostHog +# SDKs send; every other spelling already matches. +_OS_NAMES = {"Darwin": "macOS"} + +# POSIX layers over Windows report e.g. "CYGWIN_NT-10.0-19045"; they belong +# under the same filter as Windows, as in posthog-node. +_WINDOWS_LAYER_PREFIXES = ("CYGWIN", "MSYS", "MINGW") + + +def _os_name(system: str) -> str: + if system.upper().startswith(_WINDOWS_LAYER_PREFIXES): + return "Windows" + return _OS_NAMES.get(system, system) + + +def host_resource_attributes() -> Dict[str, str]: + """The ``os.name`` / ``os.version`` pair, each omitted when the host cannot say.""" + attributes: Dict[str, str] = {} + try: + name = platform.system() + if name: + attributes["os.name"] = _os_name(name) + # On Windows release() is just "10" or "11"; version() is the build, + # e.g. "10.0.22631", which is what posthog-node sends. + version = platform.version() if name == "Windows" else platform.release() + if version: + attributes["os.version"] = version + except Exception: + pass + return attributes diff --git a/posthog/tracing/_sanitize.py b/posthog/tracing/_sanitize.py new file mode 100644 index 00000000..b27be753 --- /dev/null +++ b/posthog/tracing/_sanitize.py @@ -0,0 +1,150 @@ +"""Client-side validity. + +The ingestion service rejects the whole request when one span is malformed (a +timestamp outside signed 64-bit nanoseconds, say), so every span is sanitized +before it is queued. +""" + +import logging +import math +from datetime import datetime, timezone +from typing import Any, Mapping, Optional, Union + +log = logging.getLogger("posthog") + +FALLBACK_SPAN_NAME = "unknown" + +# OTLP declares the timestamps fixed64, but the service parses signed 64-bit. +MAX_TIMESTAMP_NS = 2**63 - 1 +MIN_TIMESTAMP_NS = 0 + +# The server clamps timestamps more than 24 hours from receive time. +DEEP_BACKDATE_WARNING_NS = 24 * 60 * 60 * 10**9 + +UNSERIALIZABLE_VALUE = "[Unserializable]" +FUNCTION_VALUE = "[Function]" + +SpanTimeInput = Union[datetime, int, float] + +_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.""" + if isinstance(name, str) and name.strip(): + return name + log.debug('%s must be a non-empty string; using "%s"', label, FALLBACK_SPAN_NAME) + return FALLBACK_SPAN_NAME + + +def safe_str(value: Any) -> str: + """``str(value)``, or a marker when ``__str__`` raises.""" + if isinstance(value, str): + return value + try: + return str(value) + except Exception: + return UNSERIALIZABLE_VALUE + + +def attribute_key(key: Any) -> Optional[str]: + """An attribute key as a string, or ``None`` when ``str()`` raises.""" + if isinstance(key, str): + return key + try: + return str(key) + except Exception: + log.debug("Dropping an attribute whose key cannot be converted to a string") + return None + + +def to_epoch_ns(value: Any) -> Optional[int]: + """A ``datetime`` or epoch seconds as unix nanoseconds; ``None`` when unusable.""" + if value is None or isinstance(value, bool): + return None + ns: int + if isinstance(value, datetime): + try: + if value.tzinfo is None: + value = value.astimezone() + delta = value - _EPOCH + ns = ( + delta.days * 86400 + delta.seconds + ) * 10**9 + delta.microseconds * 1000 + except Exception: + return None + elif isinstance(value, int): + ns = value * 10**9 + elif isinstance(value, float): + if not math.isfinite(value): + return None + ns = int(round(value * 1e9)) + else: + return None + if ns < MIN_TIMESTAMP_NS or ns > MAX_TIMESTAMP_NS: + return None + return ns + + +def resolve_start_ns(value: Any, now_ns: int) -> int: + """A caller-supplied start time, or now; warns when the server will clamp it.""" + supplied = to_epoch_ns(value) + if supplied is None: + if value is not None: + log.debug( + "Span start_time is out of range or not a valid time; using the current time" + ) + return now_ns + if now_ns - supplied > DEEP_BACKDATE_WARNING_NS: + log.debug( + "Span start_time is more than 24 hours in the past; the server will clamp it " + "to receive time and keep the original in $originalTimestamp" + ) + elif supplied > now_ns: + log.debug( + "Span start_time is in the future; the span may export with a zero duration" + ) + return supplied + + +def clamp_end_ns(end_ns: int, start_ns: int) -> int: + return start_ns if end_ns < start_ns else end_ns + + +def resolve_supplied_ns(value: Any, derived_ns: int, label: str) -> int: + """A caller-supplied end or event time, or the span's own clock when unusable.""" + supplied = to_epoch_ns(value) + if supplied is None: + if value is not None: + log.debug( + "Span %s is out of range or not a valid time; using the derived time", + label, + ) + return derived_ns + return supplied + + +def copy_user_attributes(target: dict, source: Any) -> dict: + """Copy caller-supplied attributes onto ``target`` key by key. + + A raising accessor costs its own key rather than the whole span. + """ + if source is None: + return target + if not isinstance(source, Mapping): + log.debug("Ignoring span attributes: expected a mapping, got %s", type(source)) + return target + try: + keys = list(source.keys()) + except Exception: + return target + for key in keys: + key_str = attribute_key(key) + if key_str is None: + continue + try: + value = source[key] + except Exception: + value = UNSERIALIZABLE_VALUE + target[key_str] = value + return target From d2dc19e2ae0996a45d983e6cb233ce53176e4050 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Mon, 14 Sep 2026 18:53:31 -0400 Subject: [PATCH 2/5] fix(traces): reject a float time whose nanosecond value overflows A finite float like 1e308 becomes infinite once scaled to nanoseconds, so round() raised instead of the value falling back to the derived time. --- posthog/test/tracing/test_sanitize.py | 1 + posthog/tracing/_sanitize.py | 5 +++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/posthog/test/tracing/test_sanitize.py b/posthog/test/tracing/test_sanitize.py index f4b3b147..91357c4d 100644 --- a/posthog/test/tracing/test_sanitize.py +++ b/posthog/test/tracing/test_sanitize.py @@ -50,6 +50,7 @@ def test_converts_float_seconds(self): "1700000000", float("nan"), float("inf"), + 1e308, -1, MAX_TIMESTAMP_NS // 10**9 + 1, datetime(1960, 1, 1, tzinfo=timezone.utc), diff --git a/posthog/tracing/_sanitize.py b/posthog/tracing/_sanitize.py index b27be753..36768f70 100644 --- a/posthog/tracing/_sanitize.py +++ b/posthog/tracing/_sanitize.py @@ -76,9 +76,10 @@ def to_epoch_ns(value: Any) -> Optional[int]: elif isinstance(value, int): ns = value * 10**9 elif isinstance(value, float): - if not math.isfinite(value): + scaled = value * 1e9 + if not math.isfinite(scaled): return None - ns = int(round(value * 1e9)) + ns = int(round(scaled)) else: return None if ns < MIN_TIMESTAMP_NS or ns > MAX_TIMESTAMP_NS: From d06503008360e0b575d9b406a09830103fd27ff9 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Tue, 15 Sep 2026 10:24:43 -0400 Subject: [PATCH 3/5] test(traces): move the golden OTLP payload to a snapshot file Keeps the ingestion-shape fixture in posthog/test/snapshots alongside the other server payload snapshots instead of a 100-line inline dict. --- .../test/snapshots/otlp_traces_payload.json | 98 +++++++++++++++++++ posthog/test/tracing/test_otlp.py | 85 ++-------------- 2 files changed, 107 insertions(+), 76 deletions(-) create mode 100644 posthog/test/snapshots/otlp_traces_payload.json diff --git a/posthog/test/snapshots/otlp_traces_payload.json b/posthog/test/snapshots/otlp_traces_payload.json new file mode 100644 index 00000000..4e7c6cbe --- /dev/null +++ b/posthog/test/snapshots/otlp_traces_payload.json @@ -0,0 +1,98 @@ +{ + "resourceSpans": [ + { + "resource": { + "attributes": [ + { + "key": "service.name", + "value": { + "stringValue": "checkout-api" + } + }, + { + "key": "telemetry.sdk.name", + "value": { + "stringValue": "posthog-python" + } + } + ] + }, + "scopeSpans": [ + { + "scope": { + "name": "posthog-python", + "version": "" + }, + "spans": [ + { + "attributes": [ + { + "key": "posthogDistinctId", + "value": { + "stringValue": "user-123" + } + }, + { + "key": "sessionId", + "value": { + "stringValue": "session-123" + } + }, + { + "key": "http.status_code", + "value": { + "intValue": "500" + } + }, + { + "key": "http.duration_ratio", + "value": { + "doubleValue": 0.25 + } + }, + { + "key": "cached", + "value": { + "boolValue": false + } + } + ], + "endTimeUnixNano": "1700000000080000000", + "events": [ + { + "attributes": [ + { + "key": "exception.type", + "value": { + "stringValue": "TypeError" + } + }, + { + "key": "exception.message", + "value": { + "stringValue": "boom" + } + } + ], + "name": "exception", + "timeUnixNano": "1700000000040000000" + } + ], + "flags": 257, + "kind": 2, + "name": "GET /users/:id", + "parentSpanId": "b7ad6b7169203331", + "spanId": "00f067aa0ba902b7", + "startTimeUnixNano": "1700000000000000000", + "status": { + "code": 2, + "message": "boom" + }, + "traceId": "4bf92f3577b34da6a3ce929d0e0e4736" + } + ] + } + ] + } + ] +} diff --git a/posthog/test/tracing/test_otlp.py b/posthog/test/tracing/test_otlp.py index 214af146..815edcf9 100644 --- a/posthog/test/tracing/test_otlp.py +++ b/posthog/test/tracing/test_otlp.py @@ -1,5 +1,6 @@ import json from datetime import datetime, timezone +from pathlib import Path from unittest import mock import pytest @@ -27,6 +28,7 @@ TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736" SPAN_ID = "00f067aa0ba902b7" START_NS = 1_700_000_000_000_000_000 +SNAPSHOT_DIRECTORY = Path(__file__).parents[1] / "snapshots" END_NS = START_NS + 80_000_000 @@ -455,79 +457,10 @@ def test_matches_the_shape_the_ingestion_service_accepts(self): {"service.name": "checkout-api", "telemetry.sdk.name": "posthog-python"}, ) - assert payload == { - "resourceSpans": [ - { - "resource": { - "attributes": [ - { - "key": "service.name", - "value": {"stringValue": "checkout-api"}, - }, - { - "key": "telemetry.sdk.name", - "value": {"stringValue": "posthog-python"}, - }, - ] - }, - "scopeSpans": [ - { - "scope": {"name": "posthog-python", "version": VERSION}, - "spans": [ - { - "traceId": TRACE_ID, - "spanId": SPAN_ID, - "parentSpanId": "b7ad6b7169203331", - "name": "GET /users/:id", - "kind": 2, - "startTimeUnixNano": "1700000000000000000", - "endTimeUnixNano": "1700000000080000000", - "flags": 0x101, - "attributes": [ - { - "key": "posthogDistinctId", - "value": {"stringValue": "user-123"}, - }, - { - "key": "sessionId", - "value": {"stringValue": "session-123"}, - }, - { - "key": "http.status_code", - "value": {"intValue": "500"}, - }, - { - "key": "http.duration_ratio", - "value": {"doubleValue": 0.25}, - }, - { - "key": "cached", - "value": {"boolValue": False}, - }, - ], - "events": [ - { - "name": "exception", - "timeUnixNano": "1700000000040000000", - "attributes": [ - { - "key": "exception.type", - "value": { - "stringValue": "TypeError" - }, - }, - { - "key": "exception.message", - "value": {"stringValue": "boom"}, - }, - ], - } - ], - "status": {"code": 2, "message": "boom"}, - } - ], - } - ], - } - ] - } + scope = payload["resourceSpans"][0]["scopeSpans"][0]["scope"] + assert scope["version"] == VERSION + scope["version"] = "" + actual = ( + json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + ) + assert actual == (SNAPSHOT_DIRECTORY / "otlp_traces_payload.json").read_text() From 62b343c02f4a4577675e3c9372e64e9c1b683630 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 20:58:34 -0400 Subject: [PATCH 4/5] refactor(traces): name the OTLP sampled bit apart from the header flags TRACE_FLAGS_SAMPLED was a hex string in _traceparent and an int here. The module docstring now says which inputs the encoder trusts. --- posthog/tracing/_otlp.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/posthog/tracing/_otlp.py b/posthog/tracing/_otlp.py index 0669a481..c9af464e 100644 --- a/posthog/tracing/_otlp.py +++ b/posthog/tracing/_otlp.py @@ -1,7 +1,9 @@ """OTLP/JSON encoding for spans. -Values come from application code, and one the server refuses rejects the whole -request, so the encoder produces an acceptable payload whatever it is handed. +Attribute values come from application code, and one the server refuses +rejects the whole request, so they are encoded to an acceptable payload +whatever they are. Ids and timestamps are trusted: the span pipeline only +hands over ones it generated or validated. """ import logging @@ -28,7 +30,7 @@ SPAN_STATUS_TO_OTLP = {"ok": 1, "error": 2} # The W3C trace flags are the low byte; OTel's parent-remoteness bits sit above. -TRACE_FLAGS_SAMPLED = 0x01 +SAMPLED_FLAG_BIT = 0x01 SPAN_FLAGS_CONTEXT_HAS_IS_REMOTE = 0x100 SPAN_FLAGS_CONTEXT_IS_REMOTE = 0x200 @@ -226,7 +228,7 @@ def _span_flags(record: SpanRecord) -> int: try: w3c = int(record.trace_flags, 16) & 0xFF except (TypeError, ValueError): - w3c = TRACE_FLAGS_SAMPLED + w3c = SAMPLED_FLAG_BIT return ( w3c | SPAN_FLAGS_CONTEXT_HAS_IS_REMOTE From aeab32a0655349feb8137247cd2aff6e99f7f1d2 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 21:54:30 -0400 Subject: [PATCH 5/5] fix(traces): count attributes the encoder budget cuts, and encode the resource once When the node budget ran out, every remaining attribute vanished with a debug log and no droppedAttributesCount, so an oversized first value silently erased later keys. The cut is now counted on the span and on each event. The resource is encoded once by the caller instead of on every batch. --- posthog/test/tracing/test_otlp.py | 26 +++++++++++-- posthog/tracing/_otlp.py | 65 +++++++++++++++++++++++-------- 2 files changed, 71 insertions(+), 20 deletions(-) diff --git a/posthog/test/tracing/test_otlp.py b/posthog/test/tracing/test_otlp.py index 815edcf9..5fdf99bd 100644 --- a/posthog/test/tracing/test_otlp.py +++ b/posthog/test/tracing/test_otlp.py @@ -21,6 +21,7 @@ span_kind_to_otlp, to_any_value, to_key_value_list, + to_resource_key_value_list, ) from posthog.tracing._sanitize import UNSERIALIZABLE_VALUE from posthog.version import VERSION @@ -163,6 +164,21 @@ def test_bounds_total_nodes(self): class TestToKeyValueList: + def test_counts_the_entries_the_budget_cuts(self): + huge = [list(range(1000))] * 20 + span = build_otlp_span( + record(attributes={"huge": huge, "posthogDistinctId": "u", "n": None}) + ) + assert [kv["key"] for kv in span["attributes"]] == ["huge"] + assert span["droppedAttributesCount"] == 1 + + def test_counts_event_attributes_the_budget_cuts(self): + huge = [list(range(1000))] * 20 + span = build_otlp_span( + record(events=[SpanEventRecord("e", 1, {"huge": huge, "after": 1})]) + ) + assert span["events"][0]["droppedAttributesCount"] == 1 + def test_drops_none_values_and_empty_keys(self): assert to_key_value_list({"": 1, "a": None, "b": 2}) == [ {"key": "b", "value": {"intValue": "2"}} @@ -396,7 +412,7 @@ def test_an_oversized_user_attribute_does_not_cost_the_identity_keys(self): attrs = build_resource_attributes( "api", "1.2.3", "prod", {"huge": [list(range(1000))] * 20, "os.name": "x"} ) - payload = build_traces_payload([], attrs) + payload = build_traces_payload([], to_resource_key_value_list(attrs)) keys = [ kv["key"] for kv in payload["resourceSpans"][0]["resource"]["attributes"] ] @@ -419,7 +435,9 @@ def test_host_attributes_survive_a_failing_platform_module(self): class TestBuildTracesPayload: def test_produces_one_resource_one_scope_n_spans(self): spans = [build_otlp_span(record()) for _ in range(20)] - payload = build_traces_payload(spans, {"service.name": "api"}) + payload = build_traces_payload( + spans, to_resource_key_value_list({"service.name": "api"}) + ) assert len(payload["resourceSpans"]) == 1 assert len(payload["resourceSpans"][0]["scopeSpans"]) == 1 assert len(payload["resourceSpans"][0]["scopeSpans"][0]["spans"]) == 20 @@ -454,7 +472,9 @@ def test_matches_the_shape_the_ingestion_service_accepts(self): ) ) ], - {"service.name": "checkout-api", "telemetry.sdk.name": "posthog-python"}, + to_resource_key_value_list( + {"service.name": "checkout-api", "telemetry.sdk.name": "posthog-python"} + ), ) scope = payload["resourceSpans"][0]["scopeSpans"][0]["scope"] diff --git a/posthog/tracing/_otlp.py b/posthog/tracing/_otlp.py index c9af464e..bdfa2851 100644 --- a/posthog/tracing/_otlp.py +++ b/posthog/tracing/_otlp.py @@ -11,7 +11,7 @@ import platform from dataclasses import dataclass, field from datetime import date, datetime -from typing import Any, Dict, List, Mapping, Optional +from typing import Any, Dict, List, Mapping, Optional, Tuple from ..version import VERSION from ._sanitize import FUNCTION_VALUE, UNSERIALIZABLE_VALUE, attribute_key, safe_str @@ -108,12 +108,18 @@ def to_any_value(value: Any) -> dict: return {"stringValue": UNSERIALIZABLE_VALUE} -def to_key_value_list(attributes: Any) -> list: - """Encode an attribute mapping as an OTLP ``KeyValue`` list; ``None`` values are dropped.""" +def encode_attributes(attributes: Any) -> Tuple[list, int]: + """An attribute mapping as an OTLP ``KeyValue`` list, and how many entries + the encoder's budget cut. ``None`` values are dropped and not counted.""" try: return _encode_key_value_list(attributes, _EncodeState(), 0) except Exception: - return [] + return [], 0 + + +def to_key_value_list(attributes: Any) -> list: + """Encode an attribute mapping as an OTLP ``KeyValue`` list; ``None`` values are dropped.""" + return encode_attributes(attributes)[0] def _encode(value: Any, state: _EncodeState, depth: int) -> dict: @@ -154,7 +160,7 @@ def _encode(value: Any, state: _EncodeState, depth: int) -> dict: if isinstance(value, Mapping): return { "kvlistValue": { - "values": _encode_key_value_list(value, state, depth + 1) + "values": _encode_key_value_list(value, state, depth + 1)[0] } } return {"arrayValue": {"values": _encode_array(value, state, depth + 1)}} @@ -187,11 +193,14 @@ def _encode_array(values: Any, state: _EncodeState, depth: int) -> list: return result -def _encode_key_value_list(attributes: Any, state: _EncodeState, depth: int) -> list: +def _encode_key_value_list( + attributes: Any, state: _EncodeState, depth: int +) -> Tuple[list, int]: result: list = [] if not isinstance(attributes, Mapping): - return result - for key in list(attributes.keys()): + return result, 0 + keys = list(attributes.keys()) + for index, key in enumerate(keys): key_str = attribute_key(key) if key_str is None: continue @@ -200,7 +209,7 @@ def _encode_key_value_list(attributes: Any, state: _EncodeState, depth: int) -> continue if len(result) >= MAX_VALUE_ITEMS or state.remaining_nodes <= 0: log.debug("Attributes truncated: the value exceeds the OTLP encoder budget") - break + return result, _emittable_count(attributes, keys[index:]) try: value = attributes[key] if value is None: @@ -215,7 +224,23 @@ def _encode_key_value_list(attributes: Any, state: _EncodeState, depth: int) -> "value": {"stringValue": UNSERIALIZABLE_VALUE}, } ) - return result + return result, 0 + + +def _emittable_count(attributes: Mapping, keys: List[Any]) -> int: + """How many of ``keys`` the encoder would have emitted: a non-empty key with a value.""" + count = 0 + for key in keys: + key_str = attribute_key(key) + if not key_str: + continue + try: + if attributes[key] is None: + continue + except Exception: + pass + count += 1 + return count def span_kind_to_otlp(kind: Any) -> int: @@ -242,9 +267,11 @@ def _to_otlp_event(event: SpanEventRecord) -> dict: "timeUnixNano": str(event.timestamp_ns), } if event.attributes: - attributes = to_key_value_list(event.attributes) + attributes, cut = encode_attributes(event.attributes) if attributes: encoded["attributes"] = attributes + if cut: + encoded["droppedAttributesCount"] = cut return encoded @@ -262,9 +289,11 @@ def build_otlp_span(record: SpanRecord) -> dict: span["parentSpanId"] = record.parent_span_id if record.trace_state: span["traceState"] = wire_string(record.trace_state) - attributes = to_key_value_list(record.attributes) + 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] if record.status is not None and record.status.code in SPAN_STATUS_TO_OTLP: @@ -275,14 +304,16 @@ def build_otlp_span(record: SpanRecord) -> dict: return span -def build_traces_payload(spans: List[dict], resource_attributes: Mapping) -> dict: - """Wrap spans in the OTLP envelope: one resource, one scope, N spans per batch.""" +def build_traces_payload(spans: List[dict], resource: list) -> dict: + """Wrap spans in the OTLP envelope: one resource, one scope, N spans per batch. + + ``resource`` is the ``to_resource_key_value_list`` output, encoded once by + the caller since it is the same for every batch. + """ return { "resourceSpans": [ { - "resource": { - "attributes": to_resource_key_value_list(resource_attributes) - }, + "resource": {"attributes": resource}, "scopeSpans": [ { "scope": {"name": SCOPE_NAME, "version": VERSION},