Add SpanProcessor for OpenTelemetry

Author: solnic

config/config.exs

@@ -16,3 +16,8 @@ if config_env() == :test do
 end
 
 config :phoenix, :json_library, if(Code.ensure_loaded?(JSON), do: JSON, else: Jason)
+
+config :opentelemetry, span_processor: {Sentry.OpenTelemetry.SpanProcessor, []}
+
+config :opentelemetry,
+  sampler: {Sentry.OpenTelemetry.Sampler, [drop: ["Elixir.Oban.Stager process"]]}

lib/sentry/application.ex

@@ -32,6 +32,7 @@ defmodule Sentry.Application do
         Sentry.Sources,
         Sentry.Dedupe,
         Sentry.ClientReport.Sender,
+        Sentry.OpenTelemetry.SpanStorage,
         {Sentry.Integrations.CheckInIDMappings,
          [
            max_expected_check_in_time:

lib/sentry/opentelemetry/sampler.ex

@@ -0,0 +1,27 @@
+defmodule Sentry.OpenTelemetry.Sampler do
+  @moduledoc false
+
+  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

lib/sentry/opentelemetry/span_processor.ex

@@ -0,0 +1,194 @@
+defmodule Sentry.OpenTelemetry.SpanProcessor do
+  @moduledoc false
+
+  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
+  @behaviour :otel_span_processor
+
+  require Logger
+
+  alias Sentry.{Transaction, OpenTelemetry.SpanStorage, OpenTelemetry.SpanRecord}
+  alias Sentry.Interfaces.Span
+
+  @impl true
+  def on_start(_ctx, otel_span, _config) do
+    span_record = SpanRecord.new(otel_span)
+
+    SpanStorage.store_span(span_record)
+
+    otel_span
+  end
+
+  @impl true
+  def on_end(otel_span, _config) do
+    span_record = SpanRecord.new(otel_span)
+
+    SpanStorage.update_span(span_record)
+
+    if span_record.parent_span_id == nil do
+      root_span_record = SpanStorage.get_root_span(span_record.span_id)
+      child_span_records = SpanStorage.get_child_spans(span_record.span_id)
+      transaction = build_transaction(root_span_record, child_span_records)
+
+      result =
+        case Sentry.send_transaction(transaction) do
+          {:ok, _id} ->
+            true
+
+          :ignored ->
+            true
+
+          {:error, error} ->
+            Logger.error("Failed to send transaction to Sentry: #{inspect(error)}")
+            {:error, :invalid_span}
+        end
+
+      :ok = SpanStorage.remove_span(span_record.span_id)
+
+      result
+    else
+      true
+    end
+  end
+
+  @impl true
+  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),
+        otel: build_otel_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 build_otel_context(span_record), do: span_record.attributes
+
+  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

lib/sentry/opentelemetry/span_record.ex

@@ -0,0 +1,70 @@
+defmodule Sentry.OpenTelemetry.SpanRecord do
+  require Record
+  require OpenTelemetry
+
+  @fields Record.extract(:span, from_lib: "opentelemetry/include/otel_span.hrl")
+  Record.defrecordp(:span, @fields)
+
+  defstruct @fields ++ [:origin]
+
+  def new(otel_span) do
+    otel_attrs = span(otel_span)
+
+    {:attributes, _, _, _, attributes} = otel_attrs[:attributes]
+
+    origin =
+      case otel_attrs[:instrumentation_scope] do
+        {:instrumentation_scope, origin, _version, _} ->
+          origin
+
+        _ ->
+          :undefined
+      end
+
+    attrs =
+      otel_attrs
+      |> Keyword.delete(:attributes)
+      |> Keyword.merge(
+        trace_id: cast_trace_id(otel_attrs[:trace_id]),
+        span_id: cast_span_id(otel_attrs[:span_id]),
+        parent_span_id: cast_span_id(otel_attrs[:parent_span_id]),
+        origin: origin,
+        start_time: cast_timestamp(otel_attrs[:start_time]),
+        end_time: cast_timestamp(otel_attrs[:end_time]),
+        attributes: normalize_attributes(attributes)
+      )
+      |> Map.new()
+
+    struct(__MODULE__, attrs)
+  end
+
+  defp normalize_attributes(attributes) do
+    Enum.map(attributes, fn {key, value} ->
+      {to_string(key), value}
+    end)
+    |> Map.new()
+  end
+
+  defp cast_span_id(nil), do: nil
+  defp cast_span_id(:undefined), do: nil
+  defp cast_span_id(span_id), do: bytes_to_hex(span_id, 16)
+
+  defp cast_trace_id(trace_id), do: bytes_to_hex(trace_id, 32)
+
+  defp cast_timestamp(:undefined), do: nil
+  defp cast_timestamp(nil), do: nil
+
+  defp cast_timestamp(timestamp) do
+    nano_timestamp = OpenTelemetry.timestamp_to_nano(timestamp)
+    {:ok, datetime} = DateTime.from_unix(div(nano_timestamp, 1_000_000), :millisecond)
+
+    DateTime.to_iso8601(datetime)
+  end
+
+  defp bytes_to_hex(bytes, length) do
+    case(:otel_utils.format_binary_string("~#{length}.16.0b", [bytes])) do
+      {:ok, result} -> result
+      {:error, _} -> raise "Failed to convert bytes to hex: #{inspect(bytes)}"
+    end
+  end
+end