diff --git a/.sampo/changesets/application-span-context.md b/.sampo/changesets/application-span-context.md new file mode 100644 index 0000000..aa9d9f8 --- /dev/null +++ b/.sampo/changesets/application-span-context.md @@ -0,0 +1,5 @@ +--- +hex/posthog: minor +--- + +Add an optional OpenTelemetry processor that adds session and person IDs to new application spans from the active process context or PostHog header attributes already captured by application instrumentation. diff --git a/README.md b/README.md index 2a6908f..1f0df97 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ SDK usage examples and code snippets live in the official documentation so they - [Elixir library docs](https://posthog.com/docs/libraries/elixir) - [Phoenix framework docs](https://posthog.com/docs/libraries/phoenix) +- [OpenTelemetry span context (HexDocs)](https://hexdocs.pm/posthog/PostHog.OpenTelemetry.ContextSpanProcessor.html) - [API reference (HexDocs)](https://hexdocs.pm/posthog) ## Contributing diff --git a/lib/posthog/integrations/plug.ex b/lib/posthog/integrations/plug.ex index 06301ed..763dc7b 100644 --- a/lib/posthog/integrations/plug.ex +++ b/lib/posthog/integrations/plug.ex @@ -20,6 +20,8 @@ defmodule PostHog.Integrations.Plug do end """ + alias PostHog.TracingHeaders + @doc false def init(opts), do: opts @@ -35,8 +37,6 @@ defmodule PostHog.Integrations.Plug do {"x-posthog-distinct-id", :distinct_id}, {"x-posthog-session-id", :"$session_id"} ] - @max_header_value_length 1000 - @control_chars_regex ~r/[\x{00}-\x{1F}\x{7F}-\x{9F}]/u @doc false def conn_to_context(conn) when is_struct(conn, Plug.Conn) do @@ -68,7 +68,7 @@ defmodule PostHog.Integrations.Plug do defp header_value(conn, header_name) when is_struct(conn, Plug.Conn) do case req_header_values(conn, header_name) do - [value | _] when is_binary(value) -> sanitize_value(value) + [value | _] when is_binary(value) -> TracingHeaders.sanitize_value(value) _ -> nil end end @@ -98,16 +98,6 @@ defmodule PostHog.Integrations.Plug do end end - defp sanitize_value(value) when is_binary(value) do - value = - value - |> String.replace(@control_chars_regex, "") - |> String.trim() - |> String.slice(0, @max_header_value_length) - - if value == "", do: nil, else: value - end - defp put_if_present(map, _key, nil), do: map defp put_if_present(map, key, value), do: Map.put(map, key, value) diff --git a/lib/posthog/open_telemetry/context_span_processor.ex b/lib/posthog/open_telemetry/context_span_processor.ex new file mode 100644 index 0000000..ddd11a5 --- /dev/null +++ b/lib/posthog/open_telemetry/context_span_processor.ex @@ -0,0 +1,113 @@ +if Code.ensure_loaded?(:otel_span_processor) do + defmodule PostHog.OpenTelemetry.ContextSpanProcessor do + @moduledoc """ + Adds PostHog IDs to application spans when they start. + + Requires the optional `{:opentelemetry, "~> 1.7"}` dependency. Add + `{PostHog.OpenTelemetry.ContextSpanProcessor, %{supervisor_name: PostHog}}` + before the existing export processors in your OpenTelemetry `:processors` + configuration. Keep the application's provider, exporter, and sampler. + + The processor reads the current process's PostHog context. `:supervisor_name` + selects a named PostHog instance and defaults to `PostHog`. Spans stay + unchanged when that instance is disabled or not started. It adds the + `sessionId` and `posthogDistinctId` attributes when their values are present. + Existing attributes with these names take precedence, with atom or string + keys. Empty and non-string IDs are omitted. + + Initial span attributes named `http.request.header.x-posthog-session-id` + and `http.request.header.x-posthog-distinct-id` provide fallback IDs when + process context has none. The processor sanitizes those values with the + same rules as the PostHog plug. Existing canonical attributes take precedence, + followed by process context, then captured header attributes. + + With Cowboy, explicitly allow only `x-posthog-session-id` and + `x-posthog-distinct-id` in the application's OpenTelemetry `request_headers` + option to cover its early request span. This processor does not enable + header capture. Keep `PostHog.Integrations.Plug` before the router for child + spans. A span that starts without context or captured headers remains + unchanged, even if PostHog extracts headers later. + + Context has the same lifetime as existing PostHog process metadata. A + standard request process loses its context when it exits. Reusing a process + for unrelated work requires explicit context cleanup. Tasks and spawned + processes do not automatically inherit PostHog context. Set context in the + destination process when that propagation is required. + """ + + @behaviour :otel_span_processor + + alias PostHog.TracingHeaders + + require Record + + Record.defrecordp( + :span, + Record.extract(:span, from_lib: "opentelemetry/include/otel_span.hrl") + ) + + @context_attributes [ + {:sessionId, :"$session_id", :"http.request.header.x-posthog-session-id"}, + {:posthogDistinctId, :distinct_id, :"http.request.header.x-posthog-distinct-id"} + ] + + @impl true + @doc false + def on_start(_parent_context, current_span, config) do + name = Map.get(config, :supervisor_name, PostHog) + + case PostHog.config(name) do + %{enabled: true} -> add_context(current_span, PostHog.Context.get(name)) + _ -> current_span + end + rescue + _ -> current_span + catch + _, _ -> current_span + end + + defp add_context(current_span, context) do + attributes = span(current_span, :attributes) + existing = :otel_attributes.map(attributes) + + attributes = + Enum.reduce(@context_attributes, attributes, fn {key, context_key, header_key}, + attributes -> + name = Atom.to_string(key) + value = context_id(context, context_key, existing, header_key) + + if is_binary(value) and value != "" and + not (Map.has_key?(existing, name) or Map.has_key?(existing, key)) do + :otel_attributes.set(name, value, attributes) + else + attributes + end + end) + + span(current_span, attributes: attributes) + end + + defp context_id(context, context_key, attributes, header_key) do + case Map.get(context, context_key) do + value when is_binary(value) and value != "" -> value + _ -> header_id(attributes, header_key) + end + end + + defp header_id(attributes, key) do + case Map.get(attributes, Atom.to_string(key), Map.get(attributes, key)) do + [value | _] when is_binary(value) -> TracingHeaders.sanitize_value(value) + value when is_binary(value) -> TracingHeaders.sanitize_value(value) + _ -> nil + end + end + + @impl true + @doc false + def on_end(_span, _config), do: true + + @impl true + @doc false + def force_flush(_config), do: :ok + end +end diff --git a/lib/posthog/tracing_headers.ex b/lib/posthog/tracing_headers.ex new file mode 100644 index 0000000..3a817ef --- /dev/null +++ b/lib/posthog/tracing_headers.ex @@ -0,0 +1,16 @@ +defmodule PostHog.TracingHeaders do + @moduledoc false + + @max_header_value_length 1000 + @control_chars_regex ~r/[\x{00}-\x{1F}\x{7F}-\x{9F}]/u + + def sanitize_value(value) when is_binary(value) do + value = + value + |> String.replace(@control_chars_regex, "") + |> String.trim() + |> String.slice(0, @max_header_value_length) + + if value == "", do: nil, else: value + end +end diff --git a/mix.exs b/mix.exs index 3a369fb..6f78b96 100644 --- a/mix.exs +++ b/mix.exs @@ -70,12 +70,13 @@ defmodule PostHog.MixProject do {:nimble_ownership, "~> 1.0"}, {:uuid_v7, "~> 0.6"}, {:open_feature, "~> 0.1", optional: true}, + {:opentelemetry, "~> 1.7", optional: true, runtime: false}, + {:opentelemetry_api, "~> 1.5", optional: true, runtime: false}, # Development tools {:ex_doc, "~> 0.37", only: :dev, runtime: false}, {:logger_handler_kit, "~> 0.4", only: :test}, {:credo, "~> 1.7", only: [:dev, :test], runtime: false}, - {:mox, "~> 1.1", only: :test}, - {:opentelemetry_api, "~> 1.5", only: :test, runtime: false} + {:mox, "~> 1.1", only: :test} ] end diff --git a/mix.lock b/mix.lock index c51801b..88f6d2d 100644 --- a/mix.lock +++ b/mix.lock @@ -24,6 +24,7 @@ "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, "open_feature": {:hex, :open_feature, "0.1.3", "b03ea38818010dd04c3e690059a8baf14a797c9322af06eb40ac3f81c733474e", [:mix], [], "hexpm", "2dafd79e94384328469c655ef2ddc66647c2531a53b57634e012f952b6afcaf9"}, + "opentelemetry": {:hex, :opentelemetry, "1.7.0", "20d0f12d3d1c398d3670fd44fd1a7c495dd748ab3e5b692a7906662e2fb1a38a", [:rebar3], [{:opentelemetry_api, "~> 1.5.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "a9173b058c4549bf824cbc2f1d2fa2adc5cdedc22aa3f0f826951187bbd53131"}, "opentelemetry_api": {:hex, :opentelemetry_api, "1.5.0", "1a676f3e3340cab81c763e939a42e11a70c22863f645aa06aafefc689b5550cf", [:mix, :rebar3], [], "hexpm", "f53ec8a1337ae4a487d43ac89da4bd3a3c99ddf576655d071deed8b56a2d5dda"}, "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"}, "plug_cowboy": {:hex, :plug_cowboy, "2.9.0", "87e21e0d9054ced99c36d128f49e3ea2cd8b745fffb97de50bff99706087af4f", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "2002bafba4f3a45b55a58e68d70211b153a7ed18d37edb1ceb6e96e7a92c422e"}, diff --git a/public_api.snapshot b/public_api.snapshot index aff3f9e..2c3302c 100644 --- a/public_api.snapshot +++ b/public_api.snapshot @@ -217,6 +217,13 @@ PostHog.OpenFeature.Provider macros: (none) +PostHog.OpenTelemetry.ContextSpanProcessor + module_doc: documented + functions: + (none) + macros: + (none) + PostHog.Supervisor module_doc: documented functions: diff --git a/test/posthog/context_span_processor_test.exs b/test/posthog/context_span_processor_test.exs new file mode 100644 index 0000000..b5dd563 --- /dev/null +++ b/test/posthog/context_span_processor_test.exs @@ -0,0 +1,317 @@ +defmodule PostHog.OpenTelemetry.ContextSpanProcessorTest do + use ExUnit.Case, async: false + + require Record + Record.defrecordp(:span, Record.extract(:span, from_lib: "opentelemetry/include/otel_span.hrl")) + + alias PostHog.Integrations.Plug, as: PostHogPlug + alias PostHog.OpenTelemetry.ContextSpanProcessor + + defmodule CaptureProcessor do + @behaviour :otel_span_processor + def on_start(_context, span, _config), do: span + + def on_end(span, %{pid: pid}) do + send(pid, {:finished_span, span}) + true + end + + def force_flush(_config), do: :ok + end + + setup context do + Application.stop(:opentelemetry) + previous = Application.get_all_env(:opentelemetry) + + Mox.stub_with(PostHog.API.Mock, PostHog.API.Stub) + start_posthog(:span_context_test, "my_api_key") + instances = context[:processor_instances] || [:span_context_test] + if :span_context_disabled in instances, do: start_posthog(:span_context_disabled, "") + processors = for name <- instances, do: {ContextSpanProcessor, %{supervisor_name: name}} + + Application.put_env( + :opentelemetry, + :processors, + processors ++ [{CaptureProcessor, %{pid: self()}}] + ) + + Application.put_env(:opentelemetry, :sampler, context[:sampler] || :always_on) + Application.put_env(:opentelemetry, :attribute_count_limit, context[:attribute_limit] || 128) + {:ok, _} = Application.ensure_all_started(:opentelemetry) + + on_exit(fn -> + Application.stop(:opentelemetry) + + for {key, _} <- Application.get_all_env(:opentelemetry), + do: Application.delete_env(:opentelemetry, key) + + Application.put_all_env(opentelemetry: previous) + end) + + %{tracer: :opentelemetry.get_tracer(context.test)} + end + + test "snapshots process context before later changes and cleanup", %{tracer: tracer} do + PostHog.set_context(:span_context_test, %{ + distinct_id: "person-one", + "$session_id": "session-one" + }) + + active = :otel_tracer.start_span(tracer, "child", %{}) + + PostHog.set_context(:span_context_test, %{ + distinct_id: "person-two", + "$session_id": "session-two" + }) + + :logger.unset_process_metadata() + :otel_span.end_span(active) + assert attributes() == %{"sessionId" => "session-one", "posthogDistinctId" => "person-one"} + finish_span(tracer, "outside") + assert attributes() == %{} + end + + for explicit <- [ + %{"sessionId" => "explicit-session"}, + %{posthogDistinctId: "explicit-person"}, + %{"sessionId" => "", posthogDistinctId: false} + ] do + @explicit explicit + test "preserves explicit attributes #{inspect(explicit)}", %{tracer: tracer} do + PostHog.set_context(:span_context_test, %{distinct_id: "person", "$session_id": "session"}) + finish_span(tracer, "explicit", %{attributes: @explicit}) + actual = attributes() + for {key, value} <- @explicit, do: assert(actual[key] == value) + assert map_size(actual) == 2 + end + end + + @tag attribute_limit: 1 + test "keeps existing attributes when span is full", %{tracer: tracer} do + PostHog.set_context(:span_context_test, %{distinct_id: "person", "$session_id": "session"}) + finish_span(tracer, "full", %{attributes: %{"application" => "keep"}}) + assert attributes() == %{"application" => "keep"} + end + + for value <- [nil, "", 10, false] do + @value value + test "omits invalid and absent IDs #{inspect(value)}", %{tracer: tracer} do + PostHog.set_context(:span_context_test, %{distinct_id: @value, "$session_id": @value}) + finish_span(tracer, "missing") + assert attributes() == %{} + end + end + + for outcome <- [:success, :raise] do + @outcome outcome + test "Plug headers reach new request spans on #{outcome}", %{tracer: tracer} do + conn = + Plug.Test.conn(:get, "/") + |> Plug.Conn.put_req_header("x-posthog-distinct-id", "header-person") + |> Plug.Conn.put_req_header("x-posthog-session-id", "header-session") + + request = fn -> + Plug.run(conn, [ + {PostHogPlug, []}, + fn conn -> + :otel_tracer.with_span(tracer, "request", %{kind: :server}, fn _ -> + finish_span(tracer, "child") + if @outcome == :raise, do: raise("request failure") + conn + end) + end + ]) + end + + if @outcome == :raise do + assert_raise RuntimeError, "request failure", request + else + assert request.() == conn + end + + for _ <- 1..2 do + assert attributes() == %{ + "sessionId" => "header-session", + "posthogDistinctId" => "header-person" + } + end + end + end + + test "application context takes precedence over header identity", %{tracer: tracer} do + PostHog.set_context(:span_context_test, %{distinct_id: "application-person"}) + + Plug.Test.conn(:get, "/") + |> Plug.Conn.put_req_header("x-posthog-distinct-id", "header-person") + |> Plug.Conn.put_req_header("x-posthog-session-id", "header-session") + |> PostHogPlug.call([]) + + finish_span(tracer, "child") + + assert attributes() == %{ + "sessionId" => "header-session", + "posthogDistinctId" => "application-person" + } + end + + for {label, raw, expected} <- [ + {"trim and controls", [" \0session\n-\u0085123\r\t "], "session-123"}, + {"first value", ["first", "second"], "first"}, + {"single string", "single", "single"}, + {"empty", [" \n\t "], nil}, + {"empty array", [], nil}, + {"wrong type", true, nil}, + {"length limit", [String.duplicate("a", 1001)], String.duplicate("a", 1000)} + ], + key_type <- [:atom, :string] do + @raw raw + @expected expected + @key_type key_type + test "uses initial #{key_type} header attributes: #{label}", %{tracer: tracer} do + header = :"http.request.header.x-posthog-session-id" + key = if @key_type == :string, do: Atom.to_string(header), else: header + finish_span(tracer, "request", %{attributes: %{key => @raw}}) + assert attributes()["sessionId"] == @expected + end + end + + for {label, context, explicit, expected} <- [ + {"header fallback", %{}, %{}, + %{"sessionId" => "header-session", "posthogDistinctId" => "header-person"}}, + {"application context", + %{distinct_id: "application-person", "$session_id": "application-session"}, %{}, + %{"sessionId" => "application-session", "posthogDistinctId" => "application-person"}}, + {"explicit canonical attributes", %{distinct_id: "application-person"}, + %{"sessionId" => "explicit-session", "posthogDistinctId" => "explicit-person"}, + %{"sessionId" => "explicit-session", "posthogDistinctId" => "explicit-person"}} + ] do + @context context + @explicit explicit + @expected expected + test "initial header fallback respects #{label}", %{tracer: tracer} do + PostHog.set_context(:span_context_test, @context) + + initial = + Map.merge( + %{ + :"http.request.header.x-posthog-session-id" => ["header-session"], + :"http.request.header.x-posthog-distinct-id" => ["header-person"] + }, + @explicit + ) + + finish_span(tracer, "request", %{attributes: initial}) + actual = attributes() + assert Map.take(actual, ["sessionId", "posthogDistinctId"]) == @expected + for {key, value} <- initial, do: assert(actual[key] == value) + end + end + + @tag attribute_limit: 1 + test "header fallback does not evict captured header attributes", %{tracer: tracer} do + initial = %{:"http.request.header.x-posthog-session-id" => ["header-session"]} + finish_span(tracer, "request", %{attributes: initial}) + assert attributes() == initial + end + + test "keeps an earlier incoming span unchanged", %{tracer: tracer} do + :otel_tracer.with_span(tracer, "existing-request", %{kind: :server}, fn _ -> + Plug.Test.conn(:get, "/") + |> Plug.Conn.put_req_header("x-posthog-session-id", "header-session") + |> PostHogPlug.call([]) + + finish_span(tracer, "child") + end) + + assert attributes() == %{"sessionId" => "header-session"} + assert attributes() == %{} + end + + test "isolates concurrent request processes and their tasks", %{tracer: tracer} do + tasks = + for id <- ["one", "two"] do + Task.async(fn -> + conn = Plug.Test.conn(:get, "/") |> Plug.Conn.put_req_header("x-posthog-session-id", id) + PostHogPlug.call(conn, []) + finish_span(tracer, id) + Task.async(fn -> finish_span(tracer, "detached") end) |> Task.await() + end) + end + + Enum.each(tasks, &Task.await/1) + results = for _ <- 1..4, do: attributes() + + assert Enum.sort(results) == + Enum.sort([%{"sessionId" => "one"}, %{"sessionId" => "two"}, %{}, %{}]) + end + + @tag sampler: :always_off + test "does not change application sampling", %{tracer: tracer} do + PostHog.set_context(:span_context_test, %{distinct_id: "person", "$session_id": "session"}) + active = :otel_tracer.start_span(tracer, "sampled-out", %{}) + refute :otel_span.is_recording(active) + :otel_span.end_span(active) + refute_receive {:finished_span, _}, 0 + end + + @headers %{ + :"http.request.header.x-posthog-session-id" => ["header-session"], + :"http.request.header.x-posthog-distinct-id" => ["header-person"] + } + + for {label, instances, context_instances, initial, expected} <- [ + {"missing instance", [:span_context_missing], [:span_context_missing], %{}, %{}}, + {"disabled instance with process context", [:span_context_disabled], + [:span_context_disabled], %{}, %{}}, + {"disabled instance with header fallback", [:span_context_disabled], [], @headers, + @headers}, + {"enabled test-mode instance", [:span_context_test], [:span_context_test], %{}, + %{ + "sessionId" => "span_context_test-session", + "posthogDistinctId" => "span_context_test-person" + }}, + {"enabled instance beside a disabled one", [:span_context_disabled, :span_context_test], + [:span_context_disabled, :span_context_test], %{}, + %{ + "sessionId" => "span_context_test-session", + "posthogDistinctId" => "span_context_test-person" + }} + ] do + @context_instances context_instances + @initial initial + @expected expected + @tag processor_instances: instances + test "selected instance state controls enrichment: #{label}", %{tracer: tracer} do + for name <- @context_instances do + PostHog.set_context(name, %{ + distinct_id: "#{name}-person", + "$session_id": "#{name}-session" + }) + end + + finish_span(tracer, "request", %{attributes: @initial}) + assert attributes() == @expected + end + end + + defp start_posthog(name, api_key) do + config = + PostHog.Config.validate!( + api_key: api_key, + api_client_module: PostHog.API.Mock, + supervisor_name: name, + test_mode: true + ) + + start_supervised!({PostHog.Supervisor, config}, id: name) + end + + defp finish_span(tracer, name, options \\ %{}) do + tracer |> :otel_tracer.start_span(name, options) |> :otel_span.end_span() + end + + defp attributes do + assert_receive {:finished_span, finished} + finished |> span(:attributes) |> :otel_attributes.map() + end +end