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 new file mode 100644 index 00000000..815edcf9 --- /dev/null +++ b/posthog/test/tracing/test_otlp.py @@ -0,0 +1,466 @@ +import json +from datetime import datetime, timezone +from pathlib import Path +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 +SNAPSHOT_DIRECTORY = Path(__file__).parents[1] / "snapshots" +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"}, + ) + + 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() diff --git a/posthog/test/tracing/test_sanitize.py b/posthog/test/tracing/test_sanitize.py new file mode 100644 index 00000000..91357c4d --- /dev/null +++ b/posthog/test/tracing/test_sanitize.py @@ -0,0 +1,142 @@ +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"), + 1e308, + -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..36768f70 --- /dev/null +++ b/posthog/tracing/_sanitize.py @@ -0,0 +1,151 @@ +"""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): + scaled = value * 1e9 + if not math.isfinite(scaled): + return None + ns = int(round(scaled)) + 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