wip - initial work on oban support

Author: solnic

lib/sentry/opentelemetry/span_processor.ex

@@ -172,6 +172,22 @@ defmodule Sentry.OpenTelemetry.SpanProcessor do
     })
   end
 
+  defp build_transaction(%SpanRecord{origin: "opentelemetry_oban"} = root_span, child_spans) do
+    Transaction.new(%{
+      transaction: root_span.name |> String.split(" ") |> List.first(),
+      start_timestamp: root_span.start_time,
+      timestamp: root_span.end_time,
+      transaction_info: %{
+        source: "task"
+      },
+      contexts: %{
+        trace: build_trace_context(root_span)
+      },
+      measurements: %{},
+      spans: Enum.map(child_spans, &build_span(&1))
+    })
+  end
+
   defp build_trace_context(
          %SpanRecord{origin: "opentelemetry_phoenix", attributes: attributes} = root_span
        )
@@ -217,6 +233,47 @@ defmodule Sentry.OpenTelemetry.SpanProcessor do
     }
   end
 
+  defp build_trace_context(
+         %SpanRecord{
+           origin: "opentelemetry_oban",
+           attributes: %{"oban.plugin" => Oban.Stager} = _attributes
+         } = root_span
+       ) do
+    %{
+      trace_id: root_span.trace_id,
+      span_id: root_span.span_id,
+      parent_span_id: root_span.parent_span_id,
+      op: "queue.process",
+      origin: root_span.origin,
+      data: %{
+        "oban.plugin" => "stager"
+      }
+    }
+  end
+
+  defp build_trace_context(
+         %SpanRecord{origin: "opentelemetry_oban", attributes: attributes} = root_span
+       ) do
+    now = DateTime.utc_now()
+    {:ok, scheduled_at, _} = DateTime.from_iso8601(attributes["oban.job.scheduled_at"])
+
+    latency = DateTime.diff(now, scheduled_at, :millisecond)
+
+    %{
+      trace_id: root_span.trace_id,
+      span_id: root_span.span_id,
+      parent_span_id: root_span.parent_span_id,
+      op: "queue.process",
+      origin: root_span.origin,
+      data: %{
+        id: attributes["oban.job.job_id"],
+        queue: attributes["messaging.destination"],
+        retry_count: attributes["oban.job.attempt"],
+        latency: latency
+      }
+    }
+  end
+
   defp build_trace_context(%SpanRecord{attributes: attributes} = root_span) do
     %{
       trace_id: root_span.trace_id,

mix.exs

@@ -114,8 +114,8 @@ defmodule Sentry.Mixfile do
       {:quantum, "~> 3.0", only: [:test]},
 
       # Required by Tracing
-      {:opentelemetry, "~> 1.4", only: [:test]},
-      {:opentelemetry_api, "~> 1.3", only: [:test]}
+      {:opentelemetry, "~> 1.5"},
+      {:opentelemetry_api, "~> 1.3"}
     ]
   end
 

mix.lock

@@ -32,8 +32,8 @@
   "nimble_ownership": {:hex, :nimble_ownership, "1.0.0", "3f87744d42c21b2042a0aa1d48c83c77e6dd9dd357e425a038dd4b49ba8b79a1", [:mix], [], "hexpm", "7c16cc74f4e952464220a73055b557a273e8b1b7ace8489ec9d86e9ad56cb2cc"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
   "oban": {:hex, :oban, "2.18.3", "1608c04f8856c108555c379f2f56bc0759149d35fa9d3b825cb8a6769f8ae926", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "36ca6ca84ef6518f9c2c759ea88efd438a3c81d667ba23b02b062a0aa785475e"},
-  "opentelemetry": {:hex, :opentelemetry, "1.4.0", "f928923ed80adb5eb7894bac22e9a198478e6a8f04020ae1d6f289fdcad0b498", [:rebar3], [{:opentelemetry_api, "~> 1.3.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}], "hexpm", "50b32ce127413e5d87b092b4d210a3449ea80cd8224090fe68d73d576a3faa15"},
-  "opentelemetry_api": {:hex, :opentelemetry_api, "1.3.1", "83b4713593f80562d9643c4ab0b6f80f3c5fa4c6d0632c43e11b2ccb6b04dfa7", [:mix, :rebar3], [{:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}], "hexpm", "9e8a5cc38671e3ac61be48abe5f6b3afdbbb50a1dc08b7950c56f169611505c1"},
+  "opentelemetry": {:hex, :opentelemetry, "1.5.0", "7dda6551edfc3050ea4b0b40c0d2570423d6372b97e9c60793263ef62c53c3c2", [:rebar3], [{:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "cdf4f51d17b592fc592b9a75f86a6f808c23044ba7cf7b9534debbcc5c23b0ee"},
+  "opentelemetry_api": {:hex, :opentelemetry_api, "1.4.0", "63ca1742f92f00059298f478048dfb826f4b20d49534493d6919a0db39b6db04", [:mix, :rebar3], [], "hexpm", "3dfbbfaa2c2ed3121c5c483162836c4f9027def469c41578af5ef32589fcfc58"},
   "opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "0.2.0", "b67fe459c2938fcab341cb0951c44860c62347c005ace1b50f8402576f241435", [:mix, :rebar3], [], "hexpm", "d61fa1f5639ee8668d74b527e6806e0503efc55a42db7b5f39939d84c07d6895"},
   "parse_trans": {:hex, :parse_trans, "3.4.1", "6e6aa8167cb44cc8f39441d05193be6e6f4e7c2946cb2759f015f8c56b76e5ff", [:rebar3], [], "hexpm", "620a406ce75dada827b82e453c19cf06776be266f5a67cff34e1ef2cbb60e49a"},
   "phoenix": {:hex, :phoenix, "1.7.17", "2fcdceecc6fb90bec26fab008f96abbd0fd93bc9956ec7985e5892cf545152ca", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.7", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "50e8ad537f3f7b0efb1509b2f75b5c918f697be6a45d48e49a30d3b7c0e464c9"},

test_integrations/phoenix_app/config/dev.exs

@@ -89,3 +89,8 @@ config :sentry,
   environment_name: :dev,
   enable_source_code_context: true,
   send_result: :sync
+
+config :phoenix_app, Oban,
+  repo: PhoenixApp.Repo,
+  engine: Oban.Engines.Lite,
+  queues: [default: 10, background: 5]

test_integrations/phoenix_app/config/test.exs

@@ -35,3 +35,8 @@ config :sentry,
   enable_source_code_context: true,
   send_result: :sync,
   test_mode: true
+
+config :phoenix_app, Oban,
+  repo: PhoenixApp.Repo,
+  engine: Oban.Engines.Lite,
+  queues: [default: 10, background: 5]

test_integrations/phoenix_app/lib/phoenix_app/application.ex

@@ -15,6 +15,7 @@ defmodule PhoenixApp.Application do
 
     # OpentelemetryBandit.setup()
     OpentelemetryPhoenix.setup()
+    OpentelemetryOban.setup()
     OpentelemetryEcto.setup([:phoenix_app, :repo], db_statement: :enabled)
 
     children = [
@@ -26,8 +27,8 @@ defmodule PhoenixApp.Application do
       {Phoenix.PubSub, name: PhoenixApp.PubSub},
       # Start the Finch HTTP client for sending emails
       {Finch, name: PhoenixApp.Finch},
-      # Start a worker by calling: PhoenixApp.Worker.start_link(arg)
-      # {PhoenixApp.Worker, arg},
+      # Start Oban
+      {Oban, Application.fetch_env!(:phoenix_app, Oban)},
       # Start to serve requests, typically the last entry
       PhoenixAppWeb.Endpoint
     ]

test_integrations/phoenix_app/mix.exs

@@ -64,13 +64,15 @@ defmodule PhoenixApp.MixProject do
       {:dns_cluster, "~> 0.1.1"},
       {:bandit, "~> 1.5"},
       {:bypass, "~> 2.1", only: :test},
-      {:opentelemetry, "~> 1.4"},
+      {:opentelemetry, "~> 1.5"},
       {:opentelemetry_api, "~> 1.3"},
       {:opentelemetry_phoenix, "~> 1.2"},
+      {:opentelemetry_oban, "~> 1.1"},
       # {:opentelemetry_bandit, "~> 0.1.4", github: "solnic/opentelemetry-bandit", depth: 1},
       {:opentelemetry_ecto, "~> 1.2"},
       {:sentry, path: "../.."},
-      {:hackney, "~> 1.18"}
+      {:hackney, "~> 1.18"},
+      {:oban, "~> 2.10"}
     ]
   end
 

test_integrations/phoenix_app/mix.lock

@@ -33,10 +33,12 @@
   "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
   "nimble_ownership": {:hex, :nimble_ownership, "1.0.0", "3f87744d42c21b2042a0aa1d48c83c77e6dd9dd357e425a038dd4b49ba8b79a1", [:mix], [], "hexpm", "7c16cc74f4e952464220a73055b557a273e8b1b7ace8489ec9d86e9ad56cb2cc"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},
+  "oban": {:hex, :oban, "2.18.3", "1608c04f8856c108555c379f2f56bc0759149d35fa9d3b825cb8a6769f8ae926", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "36ca6ca84ef6518f9c2c759ea88efd438a3c81d667ba23b02b062a0aa785475e"},
   "opentelemetry": {:hex, :opentelemetry, "1.5.0", "7dda6551edfc3050ea4b0b40c0d2570423d6372b97e9c60793263ef62c53c3c2", [:rebar3], [{:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "cdf4f51d17b592fc592b9a75f86a6f808c23044ba7cf7b9534debbcc5c23b0ee"},
   "opentelemetry_api": {:hex, :opentelemetry_api, "1.4.0", "63ca1742f92f00059298f478048dfb826f4b20d49534493d6919a0db39b6db04", [:mix, :rebar3], [], "hexpm", "3dfbbfaa2c2ed3121c5c483162836c4f9027def469c41578af5ef32589fcfc58"},
   "opentelemetry_bandit": {:git, "https://github.com/solnic/opentelemetry-bandit.git", "1e00505fb3bb02001a3400f8a807cd1c7f7f957d", []},
   "opentelemetry_ecto": {:hex, :opentelemetry_ecto, "1.2.0", "2382cb47ddc231f953d3b8263ed029d87fbf217915a1da82f49159d122b64865", [:mix], [{:opentelemetry_api, "~> 1.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_process_propagator, "~> 0.2", [hex: :opentelemetry_process_propagator, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "70dfa2e79932e86f209df00e36c980b17a32f82d175f0068bf7ef9a96cf080cf"},
+  "opentelemetry_oban": {:hex, :opentelemetry_oban, "1.1.1", "519e9ba60d3dc3483ad2df3fade131d47056e0dae74f0724c8a40b9718f089d1", [:mix], [{:oban, "~> 2.0", [hex: :oban, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.2", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:opentelemetry_telemetry, "~> 1.1", [hex: :opentelemetry_telemetry, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ae6aed431626a94a4bb6bf5b268247ced687ec8f99eced6887e3754f9d3a2089"},
   "opentelemetry_phoenix": {:hex, :opentelemetry_phoenix, "1.2.0", "b8a53ee595b24970571a7d2fcaef3e4e1a021c68e97cac163ca5d9875fad5e9f", [:mix], [{:nimble_options, "~> 0.5 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_process_propagator, "~> 0.2", [hex: :opentelemetry_process_propagator, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:opentelemetry_telemetry, "~> 1.0", [hex: :opentelemetry_telemetry, repo: "hexpm", optional: false]}, {:plug, ">= 1.11.0", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "acab991d14ed3efc3f780c5a20cabba27149cf731005b1cc6454c160859debe5"},
   "opentelemetry_process_propagator": {:hex, :opentelemetry_process_propagator, "0.3.0", "ef5b2059403a1e2b2d2c65914e6962e56371570b8c3ab5323d7a8d3444fb7f84", [:mix, :rebar3], [{:opentelemetry_api, "~> 1.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "7243cb6de1523c473cba5b1aefa3f85e1ff8cc75d08f367104c1e11919c8c029"},
   "opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "0.2.0", "b67fe459c2938fcab341cb0951c44860c62347c005ace1b50f8402576f241435", [:mix, :rebar3], [], "hexpm", "d61fa1f5639ee8668d74b527e6806e0503efc55a42db7b5f39939d84c07d6895"},

test_integrations/phoenix_app/test/phoenix_app/oban_test.exs

@@ -0,0 +1,44 @@
+defmodule Sentry.Integrations.Phoenix.ObanTest do
+  use PhoenixAppWeb.ConnCase, async: false
+  use Oban.Testing, repo: PhoenixApp.Repo
+
+  import Sentry.TestHelpers
+
+  setup do
+    put_test_config(dsn: "http://public:secret@localhost:8080/1")
+    Sentry.Test.start_collecting_sentry_reports()
+
+    :ok
+  end
+
+  defmodule TestWorker do
+    use Oban.Worker
+
+    @impl Oban.Worker
+    def perform(_args) do
+      :timer.sleep(100)
+    end
+  end
+
+  test "captures Oban worker execution as transaction" do
+    :ok = perform_job(TestWorker, %{test: "args"})
+
+    transactions = Sentry.Test.pop_sentry_transactions()
+    assert length(transactions) == 1
+
+    [transaction] = transactions
+
+    assert transaction.transaction == "Sentry.Integrations.Phoenix.ObanTest.TestWorker"
+    assert transaction.transaction_info == %{source: "task"}
+
+    trace = transaction.contexts.trace
+    assert trace.origin == "opentelemetry_oban"
+    assert trace.op == "queue.process"
+    assert trace.data.id
+    assert trace.data.queue == "default"
+    assert trace.data.retry_count == 1
+    assert trace.data.latency > 0
+
+    assert [] = transaction.spans
+  end
+end