WIP: support for error events in Telemetry Processor

Author: solnic

lib/sentry/client.ex

@@ -16,6 +16,7 @@ defmodule Sentry.Client do
     Interfaces,
     LoggerUtils,
     Options,
+    TelemetryProcessor,
     Transaction,
     Transport
   }
@@ -227,7 +228,12 @@ defmodule Sentry.Client do
         {:ok, ""}
 
       :not_collecting ->
-        :ok = Transport.Sender.send_async(client, event)
+        if Config.telemetry_processor_category?(:error) do
+          :ok = TelemetryProcessor.add(event)
+        else
+          :ok = Transport.Sender.send_async(client, event)
+        end
+
         {:ok, ""}
     end
   end

lib/sentry/config.ex

@@ -394,6 +394,24 @@ defmodule Sentry.Config do
       Use `Sentry.LogsHandler` to capture log events from Erlang's `:logger`.
       *Available since 12.0.0*.
       """
+    ],
+    telemetry_processor_categories: [
+      type: {:list, {:in, [:error, :check_in, :transaction, :log]}},
+      default: [:log],
+      doc: """
+      List of event categories that should be processed through the TelemetryProcessor.
+      Categories in this list use the TelemetryProcessor's ring buffer and weighted
+      round-robin scheduler, which provides prioritized scheduling and backpressure.
+      Categories not in this list use the original sender-based approach.
+
+      Available categories:
+        * `:error` - Error events (critical priority, batch_size=1)
+        * `:check_in` - Cron check-ins (high priority, batch_size=1)
+        * `:transaction` - Performance transactions (medium priority, batch_size=1)
+        * `:log` - Log entries (low priority, batch_size=100, 5s timeout)
+
+      *Available since 12.0.0*.
+      """
     ]
   ]
 
@@ -860,6 +878,13 @@ defmodule Sentry.Config do
   @spec transport_capacity() :: pos_integer()
   def transport_capacity, do: fetch!(:transport_capacity)
 
+  @spec telemetry_processor_categories() :: [atom()]
+  def telemetry_processor_categories, do: fetch!(:telemetry_processor_categories)
+
+  @spec telemetry_processor_category?(atom()) :: boolean()
+  def telemetry_processor_category?(category),
+    do: category in telemetry_processor_categories()
+
   @spec before_send_log() ::
           (Sentry.LogEvent.t() -> Sentry.LogEvent.t() | nil | false) | {module(), atom()} | nil
   def before_send_log, do: get(:before_send_log)

lib/sentry/telemetry/category.ex

@@ -5,25 +5,24 @@ defmodule Sentry.Telemetry.Category do
   The TelemetryProcessor uses categories to classify different types of telemetry data
   and prioritize their sending based on a weighted round-robin scheduler.
 
-  Currently, only the `:log` category is managed by the TelemetryProcessor.
-  Other categories (errors, transactions, check-ins) will be added in future versions.
-
   ## Categories
 
+    * `:error` - Error events (critical priority)
     * `:log` - Log entries (low priority)
 
   ## Priorities and Weights
 
+    * `:critical` - weight 5 (errors)
     * `:low` - weight 2 (logs)
 
   """
   @moduledoc since: "12.0.0"
 
   @typedoc "Telemetry category types."
-  @type t :: :log
+  @type t :: :error | :log
 
   @typedoc "Priority levels for categories."
-  @type priority :: :low
+  @type priority :: :critical | :low
 
   @typedoc "Buffer configuration for a category."
   @type config :: %{
@@ -32,14 +31,16 @@ defmodule Sentry.Telemetry.Category do
           timeout: pos_integer() | nil
         }
 
-  @priorities [:low]
-  @categories [:log]
+  @priorities [:critical, :low]
+  @categories [:error, :log]
 
   @weights %{
+    critical: 5,
     low: 2
   }
 
   @default_configs %{
+    error: %{capacity: 100, batch_size: 1, timeout: nil},
     log: %{capacity: 1000, batch_size: 100, timeout: 5000}
   }
 
@@ -48,11 +49,15 @@ defmodule Sentry.Telemetry.Category do
 
   ## Examples
 
+      iex> Sentry.Telemetry.Category.priority(:error)
+      :critical
+
       iex> Sentry.Telemetry.Category.priority(:log)
       :low
 
   """
   @spec priority(t()) :: priority()
+  def priority(:error), do: :critical
   def priority(:log), do: :low
 
   @doc """
@@ -82,6 +87,9 @@ defmodule Sentry.Telemetry.Category do
 
   ## Examples
 
+      iex> Sentry.Telemetry.Category.default_config(:error)
+      %{capacity: 100, batch_size: 1, timeout: nil}
+
       iex> Sentry.Telemetry.Category.default_config(:log)
       %{capacity: 1000, batch_size: 100, timeout: 5000}
 
@@ -97,7 +105,7 @@ defmodule Sentry.Telemetry.Category do
   ## Examples
 
       iex> Sentry.Telemetry.Category.all()
-      [:log]
+      [:error, :log]
 
   """
   @spec all() :: [t()]
@@ -109,7 +117,7 @@ defmodule Sentry.Telemetry.Category do
   ## Examples
 
       iex> Sentry.Telemetry.Category.priorities()
-      [:low]
+      [:critical, :low]
 
   """
   @spec priorities() :: [priority()]

lib/sentry/telemetry/scheduler.ex

@@ -5,11 +5,9 @@ defmodule Sentry.Telemetry.Scheduler do
   The scheduler cycles through category buffers based on priority weights,
   ensuring critical telemetry gets priority over high-volume data under load.
 
-  Currently, only the `:log` category is managed. The weighted round-robin
-  structure is in place for future expansion to additional categories.
-
   ## Weights
 
+    * `:critical` - weight 5 (errors)
     * `:low` - weight 2 (logs)
 
   ## Signal-Based Wake
@@ -35,11 +33,12 @@ defmodule Sentry.Telemetry.Scheduler do
   alias __MODULE__
 
   alias Sentry.Telemetry.{Buffer, Category}
-  alias Sentry.{ClientReport, Config, Envelope, LogEvent, Transport}
+  alias Sentry.{ClientReport, Config, Envelope, Event, LogEvent, Transport}
 
   @default_capacity 1000
 
   @type buffers :: %{
+          error: GenServer.server(),
           log: GenServer.server()
         }
 
@@ -78,7 +77,7 @@ defmodule Sentry.Telemetry.Scheduler do
   ## Examples
 
       iex> Sentry.Telemetry.Scheduler.build_priority_cycle()
-      [:log, :log]
+      [:error, :error, :error, :error, :error, :log, :log]
 
   """
   @spec build_priority_cycle(map() | nil) :: [Category.t()]
@@ -151,7 +150,9 @@ defmodule Sentry.Telemetry.Scheduler do
     on_envelope = Keyword.get(opts, :on_envelope)
     capacity = Keyword.get(opts, :capacity, @default_capacity)
 
-    priority_cycle = build_priority_cycle(weights)
+    priority_cycle =
+      build_priority_cycle(weights)
+      |> Enum.filter(fn category -> Map.has_key?(buffers, category) end)
 
     state = %Scheduler{
       buffers: buffers,
@@ -232,6 +233,10 @@ defmodule Sentry.Telemetry.Scheduler do
     end
   end
 
+  defp send_items(state, :error, [%Event{} = event]) do
+    process_and_send_event(state, event, &send_envelope/2)
+  end
+
   defp send_items(state, :log, log_events) do
     process_and_send_logs(state, log_events, &send_envelope/2)
   end
@@ -242,14 +247,37 @@ defmodule Sentry.Telemetry.Scheduler do
 
       if items != [] do
         case category do
-          :log -> process_and_send_logs(state, items, &send_envelope_direct/2)
+          :error ->
+            Enum.each(items, fn event ->
+              process_and_send_event(state, event, &send_envelope_direct/2)
+            end)
+
+          :log ->
+            process_and_send_logs(state, items, &send_envelope_direct/2)
         end
       end
     end
 
     state
   end
 
+  defp process_and_send_event(%{on_envelope: on_envelope} = state, %Event{} = event, 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(event) do
+        :collected ->
+          state
+
+        :not_collecting ->
+          envelope = Envelope.from_event(event)
+          send_fn.(state, envelope)
+      end
+    else
+      envelope = Envelope.from_event(event)
+      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)
 
@@ -320,7 +348,7 @@ defmodule Sentry.Telemetry.Scheduler do
   defp send_envelope(%Scheduler{on_envelope: nil} = state, envelope) do
     if state.size >= state.capacity do
       Logger.warning(
-        "Sentry: transport queue full, dropping #{Envelope.item_count(envelope)} log item(s)"
+        "Sentry: transport queue full, dropping #{Envelope.item_count(envelope)} item(s)"
       )
 
       ClientReport.Sender.record_discarded_events(:queue_overflow, envelope.items)
@@ -352,7 +380,7 @@ defmodule Sentry.Telemetry.Scheduler do
         :ok
 
       {:error, error} ->
-        Logger.warning("Sentry: failed to send log envelope: #{Exception.message(error)}")
+        Logger.warning("Sentry: failed to send envelope: #{Exception.message(error)}")
 
         {:error, error}
     end
@@ -423,12 +451,14 @@ defmodule Sentry.Telemetry.Scheduler do
 
   defp default_weights do
     %{
+      critical: Category.weight(:critical),
       low: Category.weight(:low)
     }
   end
 
   defp category_priority_mapping do
     [
+      {:error, :critical},
       {:log, :low}
     ]
   end