diff --git a/lib/sentry/opentelemetry/span_processor.ex b/lib/sentry/opentelemetry/span_processor.ex index 74c3e602..9f19cb1c 100644 --- a/lib/sentry/opentelemetry/span_processor.ex +++ b/lib/sentry/opentelemetry/span_processor.ex @@ -36,26 +36,39 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do end defp process_span(span_record) do - transaction_root? = - cond do - # No parent = definitely a root - span_record.parent_span_id == nil -> - true - - # Has a parent - check if it's local or remote - has_local_parent_span?(span_record.parent_span_id) -> - # Parent exists locally - this is a child span, not a transaction root - false - - true -> - # Parent is remote (distributed tracing) - treat server spans as transaction roots - server_span?(span_record) - end - - if transaction_root? do - build_and_send_transaction(span_record) - else - true + cond do + # Already reported as part of its parent's transaction (it finished + # while that transaction was being sent) - don't report it twice + SpanStorage.span_sent?(span_record.span_id) -> + true + + # No parent = definitely a root + span_record.parent_span_id == nil -> + build_and_send_transaction(span_record) + + # The parent's transaction was already sent, so this span cannot be + # attached to it anymore - report it as a follow-up transaction of + # the same trace instead + SpanStorage.span_sent?(span_record.parent_span_id) -> + build_and_send_transaction(span_record, parent_already_sent?: true) + + # Parent exists locally - this is a child span, not a transaction root + has_local_parent_span?(span_record.parent_span_id) -> + true + + # Parent is remote (distributed tracing) - treat server spans as + # transaction roots + server_span?(span_record) -> + build_and_send_transaction(span_record) + + true -> + LoggerUtils.debug(fn -> + "Discarding span #{span_record.name} (#{span_record.span_id}): its parent " <> + "#{span_record.parent_span_id} is neither in span storage nor recently sent, " <> + "so there is no transaction to attach it to" + end) + + true end end @@ -86,9 +99,26 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do Map.get(attributes, to_string(MessagingAttributes.messaging_system())) == :oban end - defp build_and_send_transaction(span_record) do - child_span_records = SpanStorage.get_child_spans(span_record.span_id) - transaction = build_transaction(span_record, child_span_records) + defp build_and_send_transaction(span_record, opts \\ []) do + # Children still running when the root ends are excluded from the + # payload: a reported span must have an end timestamp. Their records + # stay in storage until they finish. + child_span_records = + span_record.span_id + |> SpanStorage.get_child_spans() + |> Enum.filter(& &1.end_time) + + transaction = build_transaction(span_record, child_span_records, opts) + + # Every span of the transaction gets a marker - late spans may continue + # the trace from any of them, not just the root. Markers must precede + # the send: a span ending while the send is in flight must already see + # its parent as sent to be promoted. They record that the transaction + # was finalized locally - not that delivery succeeded - since once the + # records are removed below, later spans can never be attached to this + # transaction either way. + sent_span_ids = [span_record.span_id | Enum.map(child_span_records, & &1.span_id)] + :ok = SpanStorage.mark_spans_sent(sent_span_ids) result = case Sentry.send_transaction(transaction) do @@ -115,7 +145,7 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do result end - defp build_transaction(root_span_record, child_span_records) do + defp build_transaction(root_span_record, child_span_records, opts) do root_span = build_span(root_span_record) child_spans = Enum.map(child_span_records, &build_span(&1)) @@ -126,7 +156,7 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do start_timestamp: root_span_record.start_time, timestamp: root_span_record.end_time, contexts: %{ - trace: build_trace_context(root_span_record) + trace: build_trace_context(root_span_record, opts) }, spans: child_spans }) @@ -141,9 +171,18 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do defp transaction_name(span_record), do: span_record.name - defp build_trace_context(span_record) do + defp build_trace_context(span_record, opts) do {op, description} = get_op_description(span_record) + data = filter_attributes(span_record.attributes) + + data = + if Keyword.get(opts, :parent_already_sent?, false) do + Map.put(data, "sentry.parent_span_already_sent", true) + else + data + end + context = %{ trace_id: span_record.trace_id, span_id: span_record.span_id, @@ -151,7 +190,7 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do op: op, description: description, origin: span_record.origin, - data: filter_attributes(span_record.attributes) + data: data } # Add links if present (for root spans, links go in trace context) diff --git a/lib/sentry/opentelemetry/span_storage.ex b/lib/sentry/opentelemetry/span_storage.ex index 301f3058..8368c858 100644 --- a/lib/sentry/opentelemetry/span_storage.ex +++ b/lib/sentry/opentelemetry/span_storage.ex @@ -11,6 +11,14 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do @span_ttl 30 * 60 + # Sent markers let spans that outlive their parent detect that the + # parent's transaction is already closed. One is written per span of every + # sent transaction, so they are far more numerous than the in-progress + # records kept under @span_ttl and are retained for a much shorter window. + # A span finishing after its parent's marker expired can no longer be + # attached to anything and is discarded with a debug log. + @sent_span_ttl 5 * 60 + @spec start_link(keyword()) :: GenServer.on_start() def start_link(opts) when is_list(opts) do name = Keyword.get(opts, :name, __MODULE__) @@ -53,6 +61,26 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do end end + @spec mark_span_sent(String.t(), keyword()) :: :ok + def mark_span_sent(span_id, opts \\ []), do: mark_spans_sent([span_id], opts) + + @spec mark_spans_sent([String.t()], keyword()) :: :ok + def mark_spans_sent(span_ids, opts \\ []) do + table_name = Keyword.get(opts, :table_name, default_table_name()) + stored_at = System.system_time(:second) + + :ets.insert(table_name, Enum.map(span_ids, &{{:sent_span, &1}, stored_at})) + + :ok + end + + @spec span_sent?(String.t(), keyword()) :: boolean() + def span_sent?(span_id, opts \\ []) do + table_name = Keyword.get(opts, :table_name, default_table_name()) + + :ets.member(table_name, {:sent_span, span_id}) + end + @doc """ Retrieves a span by its ID, regardless of whether it's a root or child span. Returns nil if the span is not found. @@ -166,11 +194,16 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do def remove_child_spans(parent_span_id, opts) do table_name = Keyword.get(opts, :table_name, default_table_name()) + # Finished descendants at every depth were part of the sent + # transaction, so their records are removed. In-progress descendants + # are preserved so they can be reported once they finish. :ets.match_object(table_name, {{:child_span, parent_span_id, :_}, :_, :_}) |> Enum.each(fn {key, span_data, _stored_at} -> if span_data.end_time != nil do :ets.delete(table_name, key) end + + remove_child_spans(span_data.span_id, table_name: table_name) end) :ok @@ -209,6 +242,14 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do ] :ets.select_delete(table_name, child_match_spec) + + sent_cutoff_time = now - @sent_span_ttl + + sent_match_spec = [ + {{{:sent_span, :_}, :"$1"}, [{:<, :"$1", sent_cutoff_time}], [true]} + ] + + :ets.select_delete(table_name, sent_match_spec) end defp default_table_name do diff --git a/test/sentry/opentelemetry/async_traces_test.exs b/test/sentry/opentelemetry/async_traces_test.exs new file mode 100644 index 00000000..b30fc28b --- /dev/null +++ b/test/sentry/opentelemetry/async_traces_test.exs @@ -0,0 +1,482 @@ +defmodule Sentry.Opentelemetry.AsyncTracesTest do + # Asserts only on reported envelopes, never on span-processing internals, + # so this coverage holds across changes to how spans are processed. + + use Sentry.Case, async: false + + require OpenTelemetry.Tracer, as: Tracer + + import ExUnit.CaptureLog + import Sentry.Test.Assertions + import Sentry.TestHelpers + + setup do + setup_bypass() + end + + describe "spans that start after their trace's root span has ended" do + test "are reported as a follow-up transaction in the same trace", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + parent_ctx = + Tracer.with_span "sync_root" do + :otel_ctx.get_current() + end + + Task.async(fn -> + Process.sleep(25) + + token = :otel_ctx.attach(parent_ctx) + + try do + Tracer.with_span "async_parent" do + Tracer.with_span "async_child" do + Process.sleep(1) + end + end + after + :otel_ctx.detach(token) + end + end) + |> Task.await() + + transactions = collect_sentry_transactions(ref, 2, timeout: 1000) + + root_tx = find_sentry_report!(transactions, transaction: "sync_root") + async_tx = find_sentry_report!(transactions, transaction: "async_parent") + + root_trace = root_tx["contexts"]["trace"] + async_trace = async_tx["contexts"]["trace"] + + assert async_trace["trace_id"] == root_trace["trace_id"] + assert async_trace["parent_span_id"] == root_trace["span_id"] + assert async_trace["data"]["sentry.parent_span_already_sent"] == true + + assert [child_span] = async_tx["spans"] + assert child_span["op"] == "async_child" + assert child_span["parent_span_id"] == async_trace["span_id"] + assert child_span["trace_id"] == root_trace["trace_id"] + end + end + + describe "spans that continue a trace from a nested span of a sent transaction" do + test "are reported as a follow-up transaction in the same trace", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + inner_ctx = + Tracer.with_span "root" do + Tracer.with_span "level_1" do + Tracer.with_span "level_2" do + :otel_ctx.get_current() + end + end + end + + [root_tx] = collect_sentry_transactions(ref, 1) + assert root_tx["transaction"] == "root" + + level_2_span = Enum.find(root_tx["spans"], &(&1["op"] == "level_2")) + assert level_2_span + + Task.async(fn -> + token = :otel_ctx.attach(inner_ctx) + + try do + Tracer.with_span "late_async_span" do + Process.sleep(1) + end + after + :otel_ctx.detach(token) + end + end) + |> Task.await() + + assert [late_tx] = collect_sentry_transactions(ref, 1), + "no follow-up transaction was reported for the late span" + + assert late_tx["transaction"] == "late_async_span" + + late_trace = late_tx["contexts"]["trace"] + + assert late_trace["trace_id"] == root_tx["contexts"]["trace"]["trace_id"] + assert late_trace["parent_span_id"] == level_2_span["span_id"] + assert late_trace["data"]["sentry.parent_span_already_sent"] == true + end + end + + describe "a child span that is still running when the transaction root ends" do + test "is not reported without an end timestamp", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + test_pid = self() + + task = + Tracer.with_span "sync_root" do + ctx = :otel_ctx.get_current() + + task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "in_flight_child" do + send(test_pid, :child_started) + + receive do + :finish_child -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + assert_receive :child_started, 1000 + task + end + + [root_tx] = collect_sentry_transactions(ref, 1) + assert root_tx["transaction"] == "sync_root" + + assert Enum.all?(root_tx["spans"], & &1["timestamp"]), + "transaction contains spans without an end timestamp: " <> + inspect(root_tx["spans"]) + + send(task.pid, :finish_child) + Task.await(task) + end + + test "is reported as a follow-up transaction once it finishes", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + test_pid = self() + + task = + Tracer.with_span "sync_root" do + ctx = :otel_ctx.get_current() + + task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "in_flight_child" do + send(test_pid, :child_started) + + receive do + :finish_child -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + assert_receive :child_started, 1000 + task + end + + [root_tx] = collect_sentry_transactions(ref, 1) + assert root_tx["transaction"] == "sync_root" + + send(task.pid, :finish_child) + Task.await(task) + + assert [child_tx] = collect_sentry_transactions(ref, 1), + "no follow-up transaction was reported for the late-finishing child" + + assert child_tx["transaction"] == "in_flight_child" + + root_trace = root_tx["contexts"]["trace"] + child_trace = child_tx["contexts"]["trace"] + + assert child_trace["trace_id"] == root_trace["trace_id"] + assert child_trace["parent_span_id"] == root_trace["span_id"] + end + end + + describe "a grandchild span still running when both its parent and the root have ended" do + test "is reported as a follow-up transaction of its own parent", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + test_pid = self() + + Tracer.with_span "sync_root" do + root_ctx = :otel_ctx.get_current() + + parent_task = + Task.async(fn -> + root_token = :otel_ctx.attach(root_ctx) + parent_pid = self() + + try do + Tracer.with_span "async_parent" do + parent_ctx = :otel_ctx.get_current() + + {:ok, _pid} = + Task.start(fn -> + parent_token = :otel_ctx.attach(parent_ctx) + + try do + Tracer.with_span "in_flight_grandchild" do + send(parent_pid, :grandchild_started) + send(test_pid, {:grandchild, self()}) + + receive do + :finish_grandchild -> :ok + end + end + + send(test_pid, :grandchild_finished) + after + :otel_ctx.detach(parent_token) + end + end) + + receive do + :grandchild_started -> :ok + after + 1000 -> raise "grandchild span never started" + end + end + after + :otel_ctx.detach(root_token) + end + end) + + Task.await(parent_task) + end + + assert_receive {:grandchild, grandchild_pid} + + [root_tx] = collect_sentry_transactions(ref, 1) + assert root_tx["transaction"] == "sync_root" + + root_span_descriptions = Enum.map(root_tx["spans"], & &1["description"]) + assert "async_parent" in root_span_descriptions + refute "in_flight_grandchild" in root_span_descriptions + + parent_span = Enum.find(root_tx["spans"], &(&1["description"] == "async_parent")) + + send(grandchild_pid, :finish_grandchild) + assert_receive :grandchild_finished, 1000 + + assert [grandchild_tx] = collect_sentry_transactions(ref, 1), + "no follow-up transaction was reported for the in-flight grandchild" + + assert grandchild_tx["transaction"] == "in_flight_grandchild" + + grandchild_trace = grandchild_tx["contexts"]["trace"] + + assert grandchild_trace["trace_id"] == root_tx["contexts"]["trace"]["trace_id"] + assert grandchild_trace["parent_span_id"] == parent_span["span_id"] + assert grandchild_trace["data"]["sentry.parent_span_already_sent"] == true + end + end + + describe "a child span that finishes while the root transaction is being sent" do + test "is reported as a follow-up transaction", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + + test_pid = self() + + Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + send(test_pid, {:envelope, body}) + + if body =~ "sync_root" do + send(test_pid, {:root_send_started, self()}) + + receive do + :release_root -> :ok + end + end + + Plug.Conn.resp(conn, 200, ~s<{"id": "#{Sentry.UUID.uuid4_hex()}"}>) + end) + + root_task = + Task.async(fn -> + Tracer.with_span "sync_root" do + send(test_pid, {:root_ctx, :otel_ctx.get_current()}) + + receive do + :end_root -> :ok + end + end + end) + + assert_receive {:root_ctx, ctx} + + child_task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "late_child" do + send(test_pid, :child_started) + + receive do + :finish_child -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + assert_receive :child_started + send(root_task.pid, :end_root) + assert_receive {:root_send_started, handler_pid}, 1000 + + send(child_task.pid, :finish_child) + Task.await(child_task) + + send(handler_pid, :release_root) + Task.await(root_task) + + transactions = drain_transactions() + + root_tx = find_sentry_report!(transactions, transaction: "sync_root") + child_tx = find_sentry_report!(transactions, transaction: "late_child") + + child_trace = child_tx["contexts"]["trace"] + + assert child_trace["trace_id"] == root_tx["contexts"]["trace"]["trace_id"] + assert child_trace["parent_span_id"] == root_tx["contexts"]["trace"]["span_id"] + end + end + + describe "spans reported with the root transaction" do + test "are not repeated in follow-up transactions", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + test_pid = self() + + task = + Tracer.with_span "sync_root" do + ctx = :otel_ctx.get_current() + + task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "async_parent" do + Tracer.with_span "finished_grandchild" do + :ok + end + + send(test_pid, :grandchild_done) + + receive do + :finish_parent -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + assert_receive :grandchild_done + task + end + + [root_tx] = collect_sentry_transactions(ref, 1) + assert root_tx["transaction"] == "sync_root" + + root_span_descriptions = Enum.map(root_tx["spans"], & &1["description"]) + assert "finished_grandchild" in root_span_descriptions + + send(task.pid, :finish_parent) + Task.await(task) + + [followup_tx] = collect_sentry_transactions(ref, 1) + assert followup_tx["transaction"] == "async_parent" + + assert followup_tx["spans"] == [], + "spans already reported with the root transaction were repeated: " <> + inspect(followup_tx["spans"]) + end + end + + describe "when the root transaction fails to send" do + test "late spans are still reported as follow-up transactions", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + + test_pid = self() + + Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + send(test_pid, {:envelope, body}) + + if body =~ "sync_root" do + Plug.Conn.resp(conn, 500, ~s<{"error": "internal"}>) + else + Plug.Conn.resp(conn, 200, ~s<{"id": "#{Sentry.UUID.uuid4_hex()}"}>) + end + end) + + log = + capture_log(fn -> + task = + Tracer.with_span "sync_root" do + ctx = :otel_ctx.get_current() + + task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "late_child" do + send(test_pid, :child_started) + + receive do + :finish_child -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + assert_receive :child_started + task + end + + send(task.pid, :finish_child) + Task.await(task) + end) + + assert log =~ "Failed to send transaction to Sentry" + + transactions = drain_transactions() + + root_tx = find_sentry_report!(transactions, transaction: "sync_root") + child_tx = find_sentry_report!(transactions, transaction: "late_child") + + child_trace = child_tx["contexts"]["trace"] + + assert child_trace["trace_id"] == root_tx["contexts"]["trace"]["trace_id"] + assert child_trace["parent_span_id"] == root_tx["contexts"]["trace"]["span_id"] + end + end + + defp drain_transactions(acc \\ []) do + receive do + {:envelope, body} -> drain_transactions([body | acc]) + after + 500 -> + acc + |> Enum.reverse() + |> Enum.map(&decode_envelope!/1) + |> extract_transactions() + end + end +end diff --git a/test/sentry/opentelemetry/span_processor_test.exs b/test/sentry/opentelemetry/span_processor_test.exs index c479ab47..808a061b 100644 --- a/test/sentry/opentelemetry/span_processor_test.exs +++ b/test/sentry/opentelemetry/span_processor_test.exs @@ -807,8 +807,6 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do assert retrieved_child != nil assert retrieved_child.end_time == "2024-01-01T00:00:03.000Z" - # When child's on_end runs, it won't find its parent in storage, - # so it should become a transaction root itself (tested via has_local_parent_span?) refute SpanStorage.span_exists?("parent_http_span", table_name: table_name) end @@ -1048,4 +1046,58 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do assert link_with_attrs["attributes"] == %{"order" => "second"} end end + + describe "a span whose parent is gone and no longer marked as sent" do + @tag span_storage: true + test "is discarded with a diagnostic log", %{bypass: bypass} do + put_test_config(environment_name: "test", traces_sample_rate: 1.0) + ref = setup_bypass_envelope_collector(bypass) + + test_pid = self() + + task = + Tracer.with_span "sync_root" do + ctx = :otel_ctx.get_current() + + task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "abandoned_child" do + send(test_pid, :child_started) + + receive do + :finish_child -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + assert_receive :child_started, 1000 + task + end + + [root_tx] = collect_sentry_transactions(ref, 1) + root_span_id = root_tx["contexts"]["trace"]["span_id"] + + # Stand in for the sent marker expiring before the child finishes. + :ets.delete( + Sentry.OpenTelemetry.SpanStorage.ETSTable, + {:sent_span, root_span_id} + ) + + log = + capture_log([level: :debug], fn -> + send(task.pid, :finish_child) + Task.await(task) + end) + + assert log =~ "Discarding span abandoned_child" + assert log =~ root_span_id + assert [] == collect_sentry_transactions(ref, 1, timeout: 300) + end + end end diff --git a/test/sentry/opentelemetry/span_storage_test.exs b/test/sentry/opentelemetry/span_storage_test.exs index bb117074..ef3a5c62 100644 --- a/test/sentry/opentelemetry/span_storage_test.exs +++ b/test/sentry/opentelemetry/span_storage_test.exs @@ -347,6 +347,30 @@ defmodule Sentry.OpenTelemetry.SpanStorageTest do assert SpanStorage.get_child_spans("root123", table_name: table_name) == [] end + describe "sent span markers" do + @tag span_storage: true + test "span_sent? reflects mark_span_sent", %{table_name: table_name} do + refute SpanStorage.span_sent?("span1", table_name: table_name) + + SpanStorage.mark_span_sent("span1", table_name: table_name) + + assert SpanStorage.span_sent?("span1", table_name: table_name) + end + + @tag span_storage: [cleanup_interval: 100] + test "markers expire after their TTL", %{table_name: table_name} do + old_time = System.system_time(:second) - 6 * 60 + :ets.insert(table_name, {{:sent_span, "old_marker"}, old_time}) + + SpanStorage.mark_span_sent("fresh_marker", table_name: table_name) + + Process.sleep(200) + + refute SpanStorage.span_sent?("old_marker", table_name: table_name) + assert SpanStorage.span_sent?("fresh_marker", table_name: table_name) + end + end + describe "stale span cleanup" do @tag span_storage: [cleanup_interval: 100] test "cleans up stale spans", %{table_name: table_name} do @@ -826,6 +850,118 @@ defmodule Sentry.OpenTelemetry.SpanStorageTest do end end + describe "recursive cleanup of sent descendants" do + @tag span_storage: true + test "remove_transaction_root_span removes finished descendants at every depth", %{ + table_name: table_name + } do + spans = [ + %SpanRecord{ + span_id: "root", + parent_span_id: nil, + trace_id: "trace123", + name: "root_span", + start_time: "2024-01-01T00:00:00.000Z", + end_time: "2024-01-01T00:00:04.000Z" + }, + %SpanRecord{ + span_id: "child", + parent_span_id: "root", + trace_id: "trace123", + name: "child_span", + start_time: "2024-01-01T00:00:01.000Z", + end_time: "2024-01-01T00:00:03.000Z" + }, + %SpanRecord{ + span_id: "grandchild", + parent_span_id: "child", + trace_id: "trace123", + name: "grandchild_span", + start_time: "2024-01-01T00:00:01.500Z", + end_time: "2024-01-01T00:00:02.500Z" + }, + %SpanRecord{ + span_id: "great_grandchild", + parent_span_id: "grandchild", + trace_id: "trace123", + name: "great_grandchild_span", + start_time: "2024-01-01T00:00:01.700Z", + end_time: "2024-01-01T00:00:02.000Z" + } + ] + + Enum.each(spans, &SpanStorage.store_span(&1, table_name: table_name)) + + SpanStorage.remove_transaction_root_span("root", nil, table_name: table_name) + + refute SpanStorage.span_exists?("root", table_name: table_name) + refute SpanStorage.span_exists?("child", table_name: table_name) + refute SpanStorage.span_exists?("grandchild", table_name: table_name) + refute SpanStorage.span_exists?("great_grandchild", table_name: table_name) + + assert :ets.tab2list(table_name) == [] + end + + @tag span_storage: true + test "remove_transaction_root_span preserves in-progress descendants at every depth", %{ + table_name: table_name + } do + spans = [ + %SpanRecord{ + span_id: "root", + parent_span_id: nil, + trace_id: "trace123", + name: "root_span", + start_time: "2024-01-01T00:00:00.000Z", + end_time: "2024-01-01T00:00:04.000Z" + }, + %SpanRecord{ + span_id: "finished_child", + parent_span_id: "root", + trace_id: "trace123", + name: "finished_child_span", + start_time: "2024-01-01T00:00:01.000Z", + end_time: "2024-01-01T00:00:03.000Z" + }, + %SpanRecord{ + span_id: "in_progress_grandchild", + parent_span_id: "finished_child", + trace_id: "trace123", + name: "in_progress_grandchild_span", + start_time: "2024-01-01T00:00:01.500Z", + end_time: nil + }, + %SpanRecord{ + span_id: "in_progress_child", + parent_span_id: "root", + trace_id: "trace123", + name: "in_progress_child_span", + start_time: "2024-01-01T00:00:02.000Z", + end_time: nil + }, + %SpanRecord{ + span_id: "finished_grandchild", + parent_span_id: "in_progress_child", + trace_id: "trace123", + name: "finished_grandchild_span", + start_time: "2024-01-01T00:00:02.200Z", + end_time: "2024-01-01T00:00:02.800Z" + } + ] + + Enum.each(spans, &SpanStorage.store_span(&1, table_name: table_name)) + + SpanStorage.remove_transaction_root_span("root", nil, table_name: table_name) + + refute SpanStorage.span_exists?("root", table_name: table_name) + refute SpanStorage.span_exists?("finished_child", table_name: table_name) + refute SpanStorage.span_exists?("finished_grandchild", table_name: table_name) + + assert SpanStorage.span_exists?("in_progress_child", table_name: table_name) + assert SpanStorage.span_exists?("in_progress_grandchild", table_name: table_name) + end + end + describe "race condition: in-progress child spans" do @tag span_storage: true test "remove_child_spans preserves in-progress child spans", %{ diff --git a/test_integrations/phoenix_app/lib/phoenix_app_web/live/async_report_live.ex b/test_integrations/phoenix_app/lib/phoenix_app_web/live/async_report_live.ex new file mode 100644 index 00000000..cb24ed71 --- /dev/null +++ b/test_integrations/phoenix_app/lib/phoenix_app_web/live/async_report_live.ex @@ -0,0 +1,48 @@ +defmodule PhoenixAppWeb.AsyncReportLive do + use PhoenixAppWeb, :live_view + + require OpenTelemetry.Tracer, as: Tracer + + @impl true + def mount(params, _session, socket) do + start_report_task(params["test_process"]) + {:ok, socket} + end + + @impl true + def render(assigns) do + ~H""" +
Report scheduled
+ """ + end + + defp start_report_task(nil), do: :ok + + defp start_report_task(test_process) do + notify = String.to_existing_atom(test_process) + ctx = :otel_ctx.get_current() + + {:ok, _pid} = + Task.start(fn -> + token = :otel_ctx.attach(ctx) + + try do + send(notify, {:report_task, self()}) + + receive do + :generate -> :ok + end + + Tracer.with_span "generate_report" do + :ok + end + + send(notify, :report_generated) + after + :otel_ctx.detach(token) + end + end) + + :ok + end +end diff --git a/test_integrations/phoenix_app/lib/phoenix_app_web/router.ex b/test_integrations/phoenix_app/lib/phoenix_app_web/router.ex index 8887473b..0badd3b3 100644 --- a/test_integrations/phoenix_app/lib/phoenix_app_web/router.ex +++ b/test_integrations/phoenix_app/lib/phoenix_app_web/router.ex @@ -38,6 +38,7 @@ defmodule PhoenixAppWeb.Router do get "/logs-with-structs", PageController, :logs_with_structs live "/test-worker", TestWorkerLive + live "/async-report", AsyncReportLive live "/tracing-test", TracingTestLive live "/users", UserLive.Index, :index diff --git a/test_integrations/phoenix_app/test/phoenix_app/async_traces_test.exs b/test_integrations/phoenix_app/test/phoenix_app/async_traces_test.exs new file mode 100644 index 00000000..1e714488 --- /dev/null +++ b/test_integrations/phoenix_app/test/phoenix_app/async_traces_test.exs @@ -0,0 +1,189 @@ +defmodule PhoenixApp.AsyncTracesTest do + use PhoenixAppWeb.ConnCase, async: false + use Oban.Testing, repo: PhoenixApp.Repo + + import Sentry.Test.Assertions + import Sentry.TestHelpers + + setup do + Sentry.Test.setup_sentry(collect_envelopes: true, traces_sample_rate: 1.0) + end + + defmodule ReportWorker do + use Oban.Worker + + require OpenTelemetry.Tracer, as: Tracer + + @impl Oban.Worker + def perform(%Oban.Job{}) do + runner = self() + ctx = :otel_ctx.get_current() + + task = + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + Tracer.with_span "deliver_report" do + send(runner, :report_started) + + receive do + :deliver -> :ok + end + end + after + :otel_ctx.detach(token) + end + end) + + receive do + :report_started -> :ok + end + + {:ok, task} + end + end + + defmodule BatchWorker do + use Oban.Worker + + require OpenTelemetry.Tracer, as: Tracer + + @impl Oban.Worker + def perform(%Oban.Job{}) do + task = + Tracer.with_span "process_batch" do + PhoenixApp.Repo.query!("SELECT 1") + ctx = :otel_ctx.get_current() + + Task.async(fn -> + token = :otel_ctx.attach(ctx) + + try do + receive do + :finalize -> :ok + end + + Tracer.with_span "finalize_batch" do + :ok + end + after + :otel_ctx.detach(token) + end + end) + end + + {:ok, task} + end + end + + test "an Oban job transaction excludes spans that are still in flight", %{ref: ref} do + {:ok, task} = perform_job(ReportWorker, %{}) + + job_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "#{inspect(ReportWorker)}" + ) + + assert job_tx["contexts"]["trace"]["origin"] == "opentelemetry_oban" + + assert Enum.all?(job_tx["spans"], & &1["timestamp"]), + "job transaction contains spans without an end timestamp" + + send(task.pid, :deliver) + Task.await(task) + end + + test "work outliving an Oban job is reported as a follow-up transaction", %{ref: ref} do + {:ok, task} = perform_job(ReportWorker, %{}) + + job_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "#{inspect(ReportWorker)}" + ) + + send(task.pid, :deliver) + Task.await(task) + + report_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "deliver_report" + ) + + job_trace = job_tx["contexts"]["trace"] + report_trace = report_tx["contexts"]["trace"] + + assert report_trace["trace_id"] == job_trace["trace_id"] + assert report_trace["parent_span_id"] == job_trace["span_id"] + assert report_trace["data"]["sentry.parent_span_already_sent"] == true + end + + test "late work continuing from a nested span of an Oban job is reported", %{ref: ref} do + {:ok, task} = perform_job(BatchWorker, %{}) + + job_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "#{inspect(BatchWorker)}" + ) + + batch_span = Enum.find(job_tx["spans"], &(&1["description"] == "process_batch")) + db_span = Enum.find(job_tx["spans"], &(&1["op"] == "db")) + + assert batch_span + assert db_span["parent_span_id"] == batch_span["span_id"] + + send(task.pid, :finalize) + Task.await(task) + + finalize_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "finalize_batch" + ) + + finalize_trace = finalize_tx["contexts"]["trace"] + + assert finalize_trace["trace_id"] == job_tx["contexts"]["trace"]["trace_id"] + assert finalize_trace["parent_span_id"] == batch_span["span_id"] + assert finalize_trace["data"]["sentry.parent_span_already_sent"] == true + end + + test "async work started in a LiveView mount is reported after the mount transaction", %{ + conn: conn, + ref: ref + } do + name = :"async_report_test_#{System.unique_integer([:positive])}" + Process.register(self(), name) + + get(conn, ~p"/async-report?test_process=#{name}") + + mount_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "PhoenixAppWeb.AsyncReportLive.mount" + ) + + assert mount_tx["contexts"]["trace"]["origin"] == "opentelemetry_phoenix" + + assert_receive {:report_task, task_pid} + send(task_pid, :generate) + assert_receive :report_generated, 1000 + + report_tx = + find_sentry_report!( + collect_sentry_transactions(ref, 100, timeout: 500), + transaction: "generate_report" + ) + + mount_trace = mount_tx["contexts"]["trace"] + report_trace = report_tx["contexts"]["trace"] + + assert report_trace["trace_id"] == mount_trace["trace_id"] + assert report_trace["parent_span_id"] == mount_trace["span_id"] + assert report_trace["data"]["sentry.parent_span_already_sent"] == true + end +end