Add SpanProcessor for OpenTelemetry (#875)

* Add SpanProcessor for OpenTelemetry

* Move opentelemetry config to test env

* Add tracing config option

* Minor formatting fix

* Mark SpanRecord as @moduledoc false

* Match otel span when building SpanRecord

* Clarify usage of optional integrations

* Skip loading Sampler if otel_sampler is not loaded

* Specify sampler behaviour

* Use Sentry.Case in SpanStorage tests

* No need to check if :span_storage table exists when cleaning

* Simplify span storage clean up in SpanProcessor tests

* Simplify SpanStorage.start_link/1

* Turn SpanStorage into a struct

* Move handle_info up

* Simplify how root spans are stored

* Extract SpanStorage.insert_root_span/2

* Rename remove_span to remove_root_span

* Reuse get_root_span

* Split update_span for root and child spans

* Make SamplerTest async

* fixup! Simplify how root spans are stored

* Simplify stale span cleanup

* Refactor SpanStorage to allow custom table names

* No need to guard when creating ets table

* Make SpanStorage more robust

* Minor formatting tweak

* tracing => traces_sample_rate

* Update docs for :traces_sample_rate option

* Update doc

* Add specs for pub functions in SpanStorage

* SpanProcessor test can't be async due to config change

* Make otel deps mandatory for phx test app

* Simplify span processor

* Add missing type for span record

* Fix spec for store_span

* Use System.unique_integer([:positive])

* Specify impl :otel_span_processor explicitly

* Use warning not error when sending span fails

* Replace :timer.sleep with Process.sleep in span_processor_test

* Refactor assertions in span storage tests for clarity

* Add more info to docs

* Use byte_size instead of String.length for trace_id validation

* Update span TTL calculation to use seconds for consistency

* Refactor update_span for improved readability

* Refactor store_span for improved readability

* Remove build_otel_context for now

This was redundant - we store the attributes in the
trace context

* Fix tests for 1.13

Author: solnic

config/config.exs

@@ -10,9 +10,15 @@ if config_env() == :test do
     send_result: :sync,
     send_max_attempts: 1,
     dedup_events: false,
-    test_mode: true
+    test_mode: true,
+    traces_sample_rate: 1.0
 
   config :logger, backends: []
+
+  config :opentelemetry, span_processor: {Sentry.OpenTelemetry.SpanProcessor, []}
+
+  config :opentelemetry,
+    sampler: {Sentry.OpenTelemetry.Sampler, [drop: ["Elixir.Oban.Stager process"]]}
 end
 
 config :phoenix, :json_library, if(Code.ensure_loaded?(JSON), do: JSON, else: Jason)

lib/sentry/application.ex

@@ -26,6 +26,13 @@ defmodule Sentry.Application do
 
     integrations_config = Config.integrations()
 
+    maybe_span_storage =
+      if Config.tracing?() do
+        [Sentry.OpenTelemetry.SpanStorage]
+      else
+        []
+      end
+
     children =
       [
         {Registry, keys: :unique, name: Sentry.Transport.SenderRegistry},
@@ -39,6 +46,7 @@ defmodule Sentry.Application do
          ]}
       ] ++
         maybe_http_client_spec ++
+        maybe_span_storage ++
         [Sentry.Transport.SenderPool]
 
     cache_loaded_applications()

lib/sentry/client.ex

@@ -117,7 +117,7 @@ defmodule Sentry.Client do
 
     result_type = Keyword.get_lazy(opts, :result, &Config.send_result/0)
     client = Keyword.get_lazy(opts, :client, &Config.client/0)
-    sample_rate = Keyword.get_lazy(opts, :sample_rate, &Config.sample_rate/0)
+    sample_rate = Keyword.get_lazy(opts, :sample_rate, &Config.traces_sample_rate/0)
     before_send = Keyword.get_lazy(opts, :before_send, &Config.before_send/0)
     after_send_event = Keyword.get_lazy(opts, :after_send_event, &Config.after_send_event/0)
 

lib/sentry/config.ex

@@ -125,6 +125,20 @@ defmodule Sentry.Config do
       be used as the value for this option.
       """
     ],
+    traces_sample_rate: [
+      type: {:custom, __MODULE__, :__validate_traces_sample_rate__, []},
+      default: 0.0,
+      doc: """
+      The sample rate for transaction events. A value between `0.0` and `1.0` (inclusive).
+      A value of `0.0` means no transactions will be sampled, while `1.0` means all transactions
+      will be sampled. This value is also used to determine if tracing is enabled: if it's
+      greater than `0`, tracing is enabled.
+
+      Tracing requires OpenTelemetry packages to work. See [the
+      OpenTelemetry setup documentation](https://opentelemetry.io/docs/languages/erlang/getting-started/)
+      for guides on how to set it up.
+      """
+    ],
     included_environments: [
       type: {:or, [{:in, [:all]}, {:list, {:or, [:atom, :string]}}]},
       deprecated: "Use :dsn to control whether to send events to Sentry.",
@@ -590,6 +604,9 @@ defmodule Sentry.Config do
   @spec sample_rate() :: float()
   def sample_rate, do: fetch!(:sample_rate)
 
+  @spec traces_sample_rate() :: float()
+  def traces_sample_rate, do: fetch!(:traces_sample_rate)
+
   @spec hackney_opts() :: keyword()
   def hackney_opts, do: fetch!(:hackney_opts)
 
@@ -627,6 +644,9 @@ defmodule Sentry.Config do
   @spec integrations() :: keyword()
   def integrations, do: fetch!(:integrations)
 
+  @spec tracing?() :: boolean()
+  def tracing?, do: fetch!(:traces_sample_rate) > 0.0
+
   @spec put_config(atom(), term()) :: :ok
   def put_config(key, value) when is_atom(key) do
     unless key in @valid_keys do
@@ -726,6 +746,15 @@ defmodule Sentry.Config do
     end
   end
 
+  def __validate_traces_sample_rate__(float) do
+    if is_float(float) and float >= 0.0 and float <= 1.0 do
+      {:ok, float}
+    else
+      {:error,
+       "expected :traces_sample_rate to be a float between 0.0 and 1.0 (included), got: #{inspect(float)}"}
+    end
+  end
+
   def __validate_json_library__(nil) do
     {:error, "nil is not a valid value for the :json_library option"}
   end

lib/sentry/opentelemetry/sampler.ex

@@ -0,0 +1,31 @@
+if Code.ensure_loaded?(:otel_sampler) do
+  defmodule Sentry.OpenTelemetry.Sampler do
+    @moduledoc false
+
+    @behaviour :otel_sampler
+
+    def setup(config) do
+      config
+    end
+
+    def description(_) do
+      "SentrySampler"
+    end
+
+    def should_sample(
+          _ctx,
+          _trace_id,
+          _links,
+          span_name,
+          _span_kind,
+          _attributes,
+          config
+        ) do
+      if span_name in config[:drop] do
+        {:drop, [], []}
+      else
+        {:record_and_sample, [], []}
+      end
+    end
+  end
+end

lib/sentry/opentelemetry/span_processor.ex

@@ -0,0 +1,193 @@
+if Code.ensure_loaded?(OpenTelemetry) do
+  defmodule Sentry.OpenTelemetry.SpanProcessor do
+    @moduledoc false
+
+    @behaviour :otel_span_processor
+
+    require OpenTelemetry.SemConv.ClientAttributes, as: ClientAttributes
+    require OpenTelemetry.SemConv.Incubating.DBAttributes, as: DBAttributes
+    require OpenTelemetry.SemConv.Incubating.HTTPAttributes, as: HTTPAttributes
+    require OpenTelemetry.SemConv.Incubating.URLAttributes, as: URLAttributes
+    require OpenTelemetry.SemConv.Incubating.MessagingAttributes, as: MessagingAttributes
+
+    require Logger
+
+    alias Sentry.{Transaction, OpenTelemetry.SpanStorage, OpenTelemetry.SpanRecord}
+    alias Sentry.Interfaces.Span
+
+    # This can be a no-op since we can postpone inserting the span into storage until on_end
+    @impl :otel_span_processor
+    def on_start(_ctx, otel_span, _config) do
+      otel_span
+    end
+
+    @impl :otel_span_processor
+    def on_end(otel_span, _config) do
+      span_record = SpanRecord.new(otel_span)
+
+      SpanStorage.store_span(span_record)
+
+      if span_record.parent_span_id == nil do
+        child_span_records = SpanStorage.get_child_spans(span_record.span_id)
+        transaction = build_transaction(span_record, child_span_records)
+
+        result =
+          case Sentry.send_transaction(transaction) do
+            {:ok, _id} ->
+              true
+
+            :ignored ->
+              true
+
+            {:error, error} ->
+              Logger.warning("Failed to send transaction to Sentry: #{inspect(error)}")
+              {:error, :invalid_span}
+          end
+
+        :ok = SpanStorage.remove_root_span(span_record.span_id)
+
+        result
+      else
+        true
+      end
+    end
+
+    @impl :otel_span_processor
+    def force_flush(_config) do
+      :ok
+    end
+
+    defp build_transaction(root_span_record, child_span_records) do
+      root_span = build_span(root_span_record)
+      child_spans = Enum.map(child_span_records, &build_span(&1))
+
+      Transaction.new(%{
+        span_id: root_span.span_id,
+        transaction: transaction_name(root_span_record),
+        transaction_info: %{source: :custom},
+        start_timestamp: root_span_record.start_time,
+        timestamp: root_span_record.end_time,
+        contexts: %{
+          trace: build_trace_context(root_span_record)
+        },
+        spans: child_spans
+      })
+    end
+
+    defp transaction_name(
+           %{attributes: %{unquote(to_string(MessagingAttributes.messaging_system())) => :oban}} =
+             span_record
+         ) do
+      span_record.attributes["oban.job.worker"]
+    end
+
+    defp transaction_name(span_record), do: span_record.name
+
+    defp build_trace_context(span_record) do
+      {op, description} = get_op_description(span_record)
+
+      %{
+        trace_id: span_record.trace_id,
+        span_id: span_record.span_id,
+        parent_span_id: span_record.parent_span_id,
+        op: op,
+        description: description,
+        origin: span_record.origin,
+        data: span_record.attributes
+      }
+    end
+
+    defp get_op_description(
+           %{
+             attributes: %{
+               unquote(to_string(HTTPAttributes.http_request_method())) => http_request_method
+             }
+           } = span_record
+         ) do
+      op = "http.#{span_record.kind}"
+
+      client_address =
+        Map.get(span_record.attributes, to_string(ClientAttributes.client_address()))
+
+      url_path = Map.get(span_record.attributes, to_string(URLAttributes.url_path()))
+
+      description =
+        to_string(http_request_method) <>
+          ((client_address && " from #{client_address}") || "") <>
+          ((url_path && " #{url_path}") || "")
+
+      {op, description}
+    end
+
+    defp get_op_description(
+           %{attributes: %{unquote(to_string(DBAttributes.db_system())) => _db_system}} =
+             span_record
+         ) do
+      db_query_text = Map.get(span_record.attributes, "db.statement")
+
+      {"db", db_query_text}
+    end
+
+    defp get_op_description(%{
+           attributes:
+             %{unquote(to_string(MessagingAttributes.messaging_system())) => :oban} = attributes
+         }) do
+      {"queue.process", attributes["oban.job.worker"]}
+    end
+
+    defp get_op_description(span_record) do
+      {span_record.name, span_record.name}
+    end
+
+    defp build_span(span_record) do
+      {op, description} = get_op_description(span_record)
+
+      %Span{
+        op: op,
+        description: description,
+        start_timestamp: span_record.start_time,
+        timestamp: span_record.end_time,
+        trace_id: span_record.trace_id,
+        span_id: span_record.span_id,
+        parent_span_id: span_record.parent_span_id,
+        origin: span_record.origin,
+        data: Map.put(span_record.attributes, "otel.kind", span_record.kind),
+        status: span_status(span_record)
+      }
+    end
+
+    defp span_status(%{
+           attributes: %{
+             unquote(to_string(HTTPAttributes.http_response_status_code())) =>
+               http_response_status_code
+           }
+         }) do
+      to_status(http_response_status_code)
+    end
+
+    defp span_status(_span_record), do: nil
+
+    # WebSocket upgrade spans doesn't have a HTTP status
+    defp to_status(nil), do: nil
+
+    defp to_status(status) when status in 200..299, do: "ok"
+
+    for {status, string} <- %{
+          400 => "invalid_argument",
+          401 => "unauthenticated",
+          403 => "permission_denied",
+          404 => "not_found",
+          409 => "already_exists",
+          429 => "resource_exhausted",
+          499 => "cancelled",
+          500 => "internal_error",
+          501 => "unimplemented",
+          503 => "unavailable",
+          504 => "deadline_exceeded"
+        } do
+      defp to_status(unquote(status)), do: unquote(string)
+    end
+
+    defp to_status(_any), do: "unknown_error"
+  end
+end