WIP: support for transactions in Telemetry Processor

Author: solnic

lib/sentry/client.ex

@@ -271,7 +271,12 @@ defmodule Sentry.Client do
         {:ok, ""}
 
       :not_collecting ->
-        :ok = Transport.Sender.send_async(client, transaction)
+        if Config.telemetry_processor_category?(:transaction) do
+          :ok = TelemetryProcessor.add(transaction)
+        else
+          :ok = Transport.Sender.send_async(client, transaction)
+        end
+
         {:ok, ""}
     end
   end

lib/sentry/config.ex

@@ -495,24 +495,24 @@ defmodule Sentry.Config do
       """
     ],
     telemetry_buffer_capacities: [
-      type: {:map, {:in, [:error, :check_in, :log]}, :pos_integer},
+      type: {:map, {:in, [:error, :check_in, :transaction, :log]}, :pos_integer},
       default: %{},
       type_doc: "`%{category => pos_integer()}`",
       doc: """
       Overrides for the maximum number of items each telemetry buffer can hold.
       When a buffer reaches capacity, oldest items are dropped to make room.
-      Default: error=100, check_in=100, log=1000.
+      Default: error=100, check_in=100, transaction=1000, log=1000.
       *Available since v12.0.0*.
       """
     ],
     telemetry_scheduler_weights: [
-      type: {:map, {:in, [:critical, :high, :low]}, :pos_integer},
+      type: {:map, {:in, [:critical, :high, :medium, :low]}, :pos_integer},
       default: %{},
       type_doc: "`%{priority => pos_integer()}`",
       doc: """
       Overrides for the weighted round-robin scheduler priority weights.
       Higher weights mean more sending slots for that priority level.
-      Default: critical=5, high=4, low=2.
+      Default: critical=5, high=4, medium=3, low=2.
       *Available since v12.0.0*.
       """
     ],

lib/sentry/telemetry/category.ex

@@ -9,22 +9,24 @@ defmodule Sentry.Telemetry.Category do
 
     * `:error` - Error events (critical priority)
     * `:check_in` - Cron check-ins (high priority)
+    * `:transaction` - Performance transactions (medium priority)
     * `:log` - Log entries (low priority)
 
   ## Priorities and Weights
 
     * `:critical` - weight 5 (errors)
     * `:high` - weight 4 (check-ins)
+    * `:medium` - weight 3 (transactions)
     * `:low` - weight 2 (logs)
 
   """
   @moduledoc since: "12.0.0"
 
   @typedoc "Telemetry category types."
-  @type t :: :error | :check_in | :log
+  @type t :: :error | :check_in | :transaction | :log
 
   @typedoc "Priority levels for categories."
-  @type priority :: :critical | :high | :low
+  @type priority :: :critical | :high | :medium | :low
 
   @typedoc "Buffer configuration for a category."
   @type config :: %{
@@ -33,18 +35,20 @@ defmodule Sentry.Telemetry.Category do
           timeout: pos_integer() | nil
         }
 
-  @priorities [:critical, :high, :low]
-  @categories [:error, :check_in, :log]
+  @priorities [:critical, :high, :medium, :low]
+  @categories [:error, :check_in, :transaction, :log]
 
   @weights %{
     critical: 5,
     high: 4,
+    medium: 3,
     low: 2
   }
 
   @default_configs %{
     error: %{capacity: 100, batch_size: 1, timeout: nil},
     check_in: %{capacity: 100, batch_size: 1, timeout: nil},
+    transaction: %{capacity: 1000, batch_size: 1, timeout: nil},
     log: %{capacity: 1000, batch_size: 100, timeout: 5000}
   }
 
@@ -59,13 +63,17 @@ defmodule Sentry.Telemetry.Category do
       iex> Sentry.Telemetry.Category.priority(:check_in)
       :high
 
+      iex> Sentry.Telemetry.Category.priority(:transaction)
+      :medium
+
       iex> Sentry.Telemetry.Category.priority(:log)
       :low
 
   """
   @spec priority(t()) :: priority()
   def priority(:error), do: :critical
   def priority(:check_in), do: :high
+  def priority(:transaction), do: :medium
   def priority(:log), do: :low
 
   @doc """
@@ -78,6 +86,9 @@ defmodule Sentry.Telemetry.Category do
       iex> Sentry.Telemetry.Category.weight(:high)
       4
 
+      iex> Sentry.Telemetry.Category.weight(:medium)
+      3
+
       iex> Sentry.Telemetry.Category.weight(:low)
       2
 
@@ -104,6 +115,9 @@ defmodule Sentry.Telemetry.Category do
       iex> Sentry.Telemetry.Category.default_config(:check_in)
       %{capacity: 100, batch_size: 1, timeout: nil}
 
+      iex> Sentry.Telemetry.Category.default_config(:transaction)
+      %{capacity: 1000, batch_size: 1, timeout: nil}
+
       iex> Sentry.Telemetry.Category.default_config(:log)
       %{capacity: 1000, batch_size: 100, timeout: 5000}
 
@@ -119,7 +133,7 @@ defmodule Sentry.Telemetry.Category do
   ## Examples
 
       iex> Sentry.Telemetry.Category.all()
-      [:error, :check_in, :log]
+      [:error, :check_in, :transaction, :log]
 
   """
   @spec all() :: [t()]
@@ -131,7 +145,7 @@ defmodule Sentry.Telemetry.Category do
   ## Examples
 
       iex> Sentry.Telemetry.Category.priorities()
-      [:critical, :high, :low]
+      [:critical, :high, :medium, :low]
 
   """
   @spec priorities() :: [priority()]

lib/sentry/telemetry/scheduler.ex

@@ -9,6 +9,7 @@ defmodule Sentry.Telemetry.Scheduler do
 
     * `:critical` - weight 5 (errors)
     * `:high` - weight 4 (check-ins)
+    * `:medium` - weight 3 (transactions)
     * `:low` - weight 2 (logs)
 
   ## Signal-Based Wake
@@ -34,13 +35,14 @@ defmodule Sentry.Telemetry.Scheduler do
   alias __MODULE__
 
   alias Sentry.Telemetry.{Buffer, Category}
-  alias Sentry.{CheckIn, ClientReport, Config, Envelope, Event, LogEvent, Transport}
+  alias Sentry.{CheckIn, ClientReport, Config, Envelope, Event, LogEvent, Transaction, Transport}
 
   @default_capacity 1000
 
   @type buffers :: %{
           error: GenServer.server(),
           check_in: GenServer.server(),
+          transaction: GenServer.server(),
           log: GenServer.server()
         }
 
@@ -79,7 +81,7 @@ defmodule Sentry.Telemetry.Scheduler do
   ## Examples
 
       iex> Sentry.Telemetry.Scheduler.build_priority_cycle()
-      [:error, :error, :error, :error, :error, :check_in, :check_in, :check_in, :check_in, :log, :log]
+      [:error, :error, :error, :error, :error, :check_in, :check_in, :check_in, :check_in, :transaction, :transaction, :transaction, :log, :log]
 
   """
   @spec build_priority_cycle(map() | nil) :: [Category.t()]
@@ -243,6 +245,10 @@ defmodule Sentry.Telemetry.Scheduler do
     process_and_send_check_in(state, check_in, &send_envelope/2)
   end
 
+  defp send_items(state, :transaction, [%Transaction{} = transaction]) do
+    process_and_send_transaction(state, transaction, &send_envelope/2)
+  end
+
   defp send_items(state, :log, log_events) do
     process_and_send_logs(state, log_events, &send_envelope/2)
   end
@@ -263,6 +269,11 @@ defmodule Sentry.Telemetry.Scheduler do
               process_and_send_check_in(state, check_in, &send_envelope_direct/2)
             end)
 
+          :transaction ->
+            Enum.each(items, fn transaction ->
+              process_and_send_transaction(state, transaction, &send_envelope_direct/2)
+            end)
+
           :log ->
             process_and_send_logs(state, items, &send_envelope_direct/2)
         end
@@ -294,6 +305,27 @@ defmodule Sentry.Telemetry.Scheduler do
     send_fn.(state, envelope)
   end
 
+  defp process_and_send_transaction(
+         %{on_envelope: on_envelope} = state,
+         %Transaction{} = transaction,
+         send_fn
+       ) do
+    # Skip test collection when on_envelope is set (used by unit tests)
+    if is_nil(on_envelope) do
+      case Sentry.Test.maybe_collect(transaction) do
+        :collected ->
+          state
+
+        :not_collecting ->
+          envelope = Envelope.from_transaction(transaction)
+          send_fn.(state, envelope)
+      end
+    else
+      envelope = Envelope.from_transaction(transaction)
+      send_fn.(state, envelope)
+    end
+  end
+
   defp process_and_send_logs(%{on_envelope: on_envelope} = state, log_events, send_fn) do
     processed_logs = apply_before_send_log_callbacks(log_events)
 
@@ -469,6 +501,7 @@ defmodule Sentry.Telemetry.Scheduler do
     %{
       critical: Category.weight(:critical),
       high: Category.weight(:high),
+      medium: Category.weight(:medium),
       low: Category.weight(:low)
     }
   end
@@ -477,6 +510,7 @@ defmodule Sentry.Telemetry.Scheduler do
     [
       {:error, :critical},
       {:check_in, :high},
+      {:transaction, :medium},
       {:log, :low}
     ]
   end

lib/sentry/telemetry_processor.ex

@@ -12,6 +12,7 @@ defmodule Sentry.TelemetryProcessor do
 
     * Error Buffer - for error events (critical priority)
     * Check-in Buffer - for cron check-ins (high priority)
+    * Transaction Buffer - for performance transactions (medium priority)
     * Log Buffer - for log entries (low priority)
     * Scheduler - weighted round-robin scheduler with integrated transport queue
 
@@ -23,6 +24,9 @@ defmodule Sentry.TelemetryProcessor do
       # Add check-ins to the buffer
       TelemetryProcessor.add(processor, %Sentry.CheckIn{...})
 
+      # Add transactions to the buffer
+      TelemetryProcessor.add(processor, %Sentry.Transaction{...})
+
       # Add log events to the buffer
       TelemetryProcessor.add(processor, %Sentry.LogEvent{...})
 
@@ -35,7 +39,7 @@ defmodule Sentry.TelemetryProcessor do
   use Supervisor
 
   alias Sentry.Telemetry.{Buffer, Category, Scheduler}
-  alias Sentry.{CheckIn, Event, LogEvent}
+  alias Sentry.{CheckIn, Event, LogEvent, Transaction}
 
   @default_name __MODULE__
 
@@ -91,7 +95,7 @@ defmodule Sentry.TelemetryProcessor do
 
   Returns `:ok`.
   """
-  @spec add(Event.t() | CheckIn.t() | LogEvent.t()) :: :ok
+  @spec add(Event.t() | CheckIn.t() | Transaction.t() | LogEvent.t()) :: :ok
   def add(%Event{} = item) do
     add(processor_name(), item)
   end
@@ -100,6 +104,10 @@ defmodule Sentry.TelemetryProcessor do
     add(processor_name(), item)
   end
 
+  def add(%Transaction{} = item) do
+    add(processor_name(), item)
+  end
+
   def add(%LogEvent{} = item) do
     add(processor_name(), item)
   end
@@ -111,7 +119,8 @@ defmodule Sentry.TelemetryProcessor do
 
   Returns `:ok`.
   """
-  @spec add(Supervisor.supervisor(), Event.t() | CheckIn.t() | LogEvent.t()) :: :ok
+  @spec add(Supervisor.supervisor(), Event.t() | CheckIn.t() | Transaction.t() | LogEvent.t()) ::
+          :ok
   def add(processor, %Event{} = item) when is_atom(processor) do
     Buffer.add(buffer_name(processor, :error), item)
     Scheduler.signal(scheduler_name(processor))
@@ -140,6 +149,20 @@ defmodule Sentry.TelemetryProcessor do
     :ok
   end
 
+  def add(processor, %Transaction{} = item) when is_atom(processor) do
+    Buffer.add(buffer_name(processor, :transaction), item)
+    Scheduler.signal(scheduler_name(processor))
+    :ok
+  end
+
+  def add(processor, %Transaction{} = item) do
+    buffer = get_buffer(processor, :transaction)
+    Buffer.add(buffer, item)
+    scheduler = get_scheduler(processor)
+    Scheduler.signal(scheduler)
+    :ok
+  end
+
   def add(processor, %LogEvent{} = item) when is_atom(processor) do
     Buffer.add(buffer_name(processor, :log), item)
     Scheduler.signal(scheduler_name(processor))
@@ -187,7 +210,7 @@ defmodule Sentry.TelemetryProcessor do
   Returns the buffer pid for a given category.
   """
   @spec get_buffer(Supervisor.supervisor(), Category.t()) :: pid()
-  def get_buffer(processor, category) when category in [:error, :check_in, :log] do
+  def get_buffer(processor, category) when category in [:error, :check_in, :transaction, :log] do
     children = Supervisor.which_children(processor)
     buffer_id = buffer_id(category)
 
@@ -217,7 +240,7 @@ defmodule Sentry.TelemetryProcessor do
   Returns 0 if the processor is not running.
   """
   @spec buffer_size(Category.t()) :: non_neg_integer()
-  def buffer_size(category) when category in [:error, :check_in, :log] do
+  def buffer_size(category) when category in [:error, :check_in, :transaction, :log] do
     buffer_size(processor_name(), category)
   end
 
@@ -227,7 +250,7 @@ defmodule Sentry.TelemetryProcessor do
   Returns 0 if the processor is not running.
   """
   @spec buffer_size(Supervisor.supervisor(), Category.t()) :: non_neg_integer()
-  def buffer_size(processor, category) when category in [:error, :check_in, :log] do
+  def buffer_size(processor, category) when category in [:error, :check_in, :transaction, :log] do
     case safe_get_buffer(processor, category) do
       {:ok, buffer} -> Buffer.size(buffer)
       :error -> 0
@@ -290,6 +313,7 @@ defmodule Sentry.TelemetryProcessor do
         buffers: %{
           error: Map.fetch!(buffer_names, :error),
           check_in: Map.fetch!(buffer_names, :check_in),
+          transaction: Map.fetch!(buffer_names, :transaction),
           log: Map.fetch!(buffer_names, :log)
         },
         name: scheduler_name(processor_name),
@@ -310,6 +334,7 @@ defmodule Sentry.TelemetryProcessor do
 
   defp buffer_id(:error), do: :error_buffer
   defp buffer_id(:check_in), do: :check_in_buffer
+  defp buffer_id(:transaction), do: :transaction_buffer
   defp buffer_id(:log), do: :log_buffer
 
   @doc false

test/sentry/telemetry/scheduler_test.exs

@@ -19,17 +19,17 @@ defmodule Sentry.Telemetry.SchedulerTest do
     test "builds cycle with correct weights for all categories" do
       cycle = Scheduler.build_priority_cycle()
 
-      # Default weights: critical=5, high=4, low=2
-      assert length(cycle) == 11
-      assert Enum.frequencies(cycle) == %{error: 5, check_in: 4, log: 2}
+      # Default weights: critical=5, high=4, medium=3, low=2
+      assert length(cycle) == 14
+      assert Enum.frequencies(cycle) == %{error: 5, check_in: 4, transaction: 3, log: 2}
     end
 
     test "builds cycle with custom weights" do
       custom_weights = %{low: 5}
       cycle = Scheduler.build_priority_cycle(custom_weights)
 
-      assert length(cycle) == 14
-      assert Enum.frequencies(cycle) == %{error: 5, check_in: 4, log: 5}
+      assert length(cycle) == 17
+      assert Enum.frequencies(cycle) == %{error: 5, check_in: 4, transaction: 3, log: 5}
     end
   end