feat: support for check-ins in Telemetry Processor (#989)

Author: solnic

lib/sentry/client.ex

@@ -29,20 +29,22 @@ defmodule Sentry.Client do
   @spec send_check_in(CheckIn.t(), keyword()) ::
           {:ok, check_in_id :: String.t()} | {:error, ClientError.t()}
   def send_check_in(%CheckIn{} = check_in, opts) when is_list(opts) do
-    client = Keyword.get_lazy(opts, :client, &Config.client/0)
+    if Config.telemetry_processor_category?(:check_in) do
+      :ok = TelemetryProcessor.add(check_in)
+      {:ok, check_in.check_in_id}
+    else
+      client = Keyword.get_lazy(opts, :client, &Config.client/0)
 
-    # This is a "private" option, only really used in testing.
-    request_retries =
-      Keyword.get_lazy(opts, :request_retries, fn ->
-        Application.get_env(:sentry, :request_retries, Transport.default_retries())
-      end)
+      # This is a "private" option, only really used in testing.
+      request_retries =
+        Keyword.get_lazy(opts, :request_retries, fn ->
+          Application.get_env(:sentry, :request_retries, Transport.default_retries())
+        end)
 
-    send_result =
       check_in
       |> Envelope.from_check_in()
       |> Transport.encode_and_post_envelope(client, request_retries)
-
-    send_result
+    end
   end
 
   # This is what executes the "Event Pipeline".

lib/sentry/config.ex

@@ -495,26 +495,24 @@ defmodule Sentry.Config do
       """
     ],
     telemetry_buffer_capacities: [
-      type: {:map, {:in, [:log]}, :pos_integer},
+      type: {:map, {:in, [:error, :check_in, :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.
-      Currently only the `:log` category is managed by the TelemetryProcessor.
-      Default: log=1000.
+      Default: error=100, check_in=100, log=1000.
       *Available since v12.0.0*.
       """
     ],
     telemetry_scheduler_weights: [
-      type: {:map, {:in, [:low]}, :pos_integer},
+      type: {:map, {:in, [:critical, :high, :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.
-      Currently only the `:low` priority (logs) is managed by the TelemetryProcessor.
-      Default: low=2.
+      Default: critical=5, high=4, low=2.
       *Available since v12.0.0*.
       """
     ],

lib/sentry/telemetry/category.ex

@@ -8,21 +8,23 @@ defmodule Sentry.Telemetry.Category do
   ## Categories
 
     * `:error` - Error events (critical priority)
+    * `:check_in` - Cron check-ins (high priority)
     * `:log` - Log entries (low priority)
 
   ## Priorities and Weights
 
     * `:critical` - weight 5 (errors)
+    * `:high` - weight 4 (check-ins)
     * `:low` - weight 2 (logs)
 
   """
   @moduledoc since: "12.0.0"
 
   @typedoc "Telemetry category types."
-  @type t :: :error | :log
+  @type t :: :error | :check_in | :log
 
   @typedoc "Priority levels for categories."
-  @type priority :: :critical | :low
+  @type priority :: :critical | :high | :low
 
   @typedoc "Buffer configuration for a category."
   @type config :: %{
@@ -31,16 +33,18 @@ defmodule Sentry.Telemetry.Category do
           timeout: pos_integer() | nil
         }
 
-  @priorities [:critical, :low]
-  @categories [:error, :log]
+  @priorities [:critical, :high, :low]
+  @categories [:error, :check_in, :log]
 
   @weights %{
     critical: 5,
+    high: 4,
     low: 2
   }
 
   @default_configs %{
     error: %{capacity: 100, batch_size: 1, timeout: nil},
+    check_in: %{capacity: 100, batch_size: 1, timeout: nil},
     log: %{capacity: 1000, batch_size: 100, timeout: 5000}
   }
 
@@ -52,12 +56,16 @@ defmodule Sentry.Telemetry.Category do
       iex> Sentry.Telemetry.Category.priority(:error)
       :critical
 
+      iex> Sentry.Telemetry.Category.priority(:check_in)
+      :high
+
       iex> Sentry.Telemetry.Category.priority(:log)
       :low
 
   """
   @spec priority(t()) :: priority()
   def priority(:error), do: :critical
+  def priority(:check_in), do: :high
   def priority(:log), do: :low
 
   @doc """
@@ -67,6 +75,9 @@ defmodule Sentry.Telemetry.Category do
 
   ## Examples
 
+      iex> Sentry.Telemetry.Category.weight(:high)
+      4
+
       iex> Sentry.Telemetry.Category.weight(:low)
       2
 
@@ -90,6 +101,9 @@ defmodule Sentry.Telemetry.Category do
       iex> Sentry.Telemetry.Category.default_config(:error)
       %{capacity: 100, batch_size: 1, timeout: nil}
 
+      iex> Sentry.Telemetry.Category.default_config(:check_in)
+      %{capacity: 100, batch_size: 1, timeout: nil}
+
       iex> Sentry.Telemetry.Category.default_config(:log)
       %{capacity: 1000, batch_size: 100, timeout: 5000}
 
@@ -105,7 +119,7 @@ defmodule Sentry.Telemetry.Category do
   ## Examples
 
       iex> Sentry.Telemetry.Category.all()
-      [:error, :log]
+      [:error, :check_in, :log]
 
   """
   @spec all() :: [t()]
@@ -117,7 +131,7 @@ defmodule Sentry.Telemetry.Category do
   ## Examples
 
       iex> Sentry.Telemetry.Category.priorities()
-      [:critical, :low]
+      [:critical, :high, :low]
 
   """
   @spec priorities() :: [priority()]

lib/sentry/telemetry/scheduler.ex

@@ -8,6 +8,7 @@ defmodule Sentry.Telemetry.Scheduler do
   ## Weights
 
     * `:critical` - weight 5 (errors)
+    * `:high` - weight 4 (check-ins)
     * `:low` - weight 2 (logs)
 
   ## Signal-Based Wake
@@ -33,12 +34,13 @@ defmodule Sentry.Telemetry.Scheduler do
   alias __MODULE__
 
   alias Sentry.Telemetry.{Buffer, Category}
-  alias Sentry.{ClientReport, Config, Envelope, Event, LogEvent, Transport}
+  alias Sentry.{CheckIn, ClientReport, Config, Envelope, Event, LogEvent, Transport}
 
   @default_capacity 1000
 
   @type buffers :: %{
           error: GenServer.server(),
+          check_in: GenServer.server(),
           log: GenServer.server()
         }
 
@@ -77,7 +79,7 @@ defmodule Sentry.Telemetry.Scheduler do
   ## Examples
 
       iex> Sentry.Telemetry.Scheduler.build_priority_cycle()
-      [:error, :error, :error, :error, :error, :log, :log]
+      [:error, :error, :error, :error, :error, :check_in, :check_in, :check_in, :check_in, :log, :log]
 
   """
   @spec build_priority_cycle(map() | nil) :: [Category.t()]
@@ -237,6 +239,10 @@ defmodule Sentry.Telemetry.Scheduler do
     process_and_send_event(state, event, &send_envelope/2)
   end
 
+  defp send_items(state, :check_in, [%CheckIn{} = check_in]) do
+    process_and_send_check_in(state, check_in, &send_envelope/2)
+  end
+
   defp send_items(state, :log, log_events) do
     process_and_send_logs(state, log_events, &send_envelope/2)
   end
@@ -252,6 +258,11 @@ defmodule Sentry.Telemetry.Scheduler do
               process_and_send_event(state, event, &send_envelope_direct/2)
             end)
 
+          :check_in ->
+            Enum.each(items, fn check_in ->
+              process_and_send_check_in(state, check_in, &send_envelope_direct/2)
+            end)
+
           :log ->
             process_and_send_logs(state, items, &send_envelope_direct/2)
         end
@@ -278,6 +289,11 @@ defmodule Sentry.Telemetry.Scheduler do
     end
   end
 
+  defp process_and_send_check_in(state, %CheckIn{} = check_in, send_fn) do
+    envelope = Envelope.from_check_in(check_in)
+    send_fn.(state, envelope)
+  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)
 
@@ -451,13 +467,15 @@ defmodule Sentry.Telemetry.Scheduler do
   defp default_weights do
     %{
       critical: Category.weight(:critical),
+      high: Category.weight(:high),
       low: Category.weight(:low)
     }
   end
 
   defp category_priority_mapping do
     [
       {:error, :critical},
+      {:check_in, :high},
       {:log, :low}
     ]
   end

lib/sentry/telemetry_processor.ex

@@ -11,6 +11,7 @@ defmodule Sentry.TelemetryProcessor do
   The processor starts as a supervisor with the following children:
 
     * Error Buffer - for error events (critical priority)
+    * Check-in Buffer - for cron check-ins (high priority)
     * Log Buffer - for log entries (low priority)
     * Scheduler - weighted round-robin scheduler with integrated transport queue
 
@@ -19,6 +20,9 @@ defmodule Sentry.TelemetryProcessor do
       # Add error events to the buffer
       TelemetryProcessor.add(processor, %Sentry.Event{...})
 
+      # Add check-ins to the buffer
+      TelemetryProcessor.add(processor, %Sentry.CheckIn{...})
+
       # Add log events to the buffer
       TelemetryProcessor.add(processor, %Sentry.LogEvent{...})
 
@@ -31,7 +35,7 @@ defmodule Sentry.TelemetryProcessor do
   use Supervisor
 
   alias Sentry.Telemetry.{Buffer, Category, Scheduler}
-  alias Sentry.{Event, LogEvent}
+  alias Sentry.{CheckIn, Event, LogEvent}
 
   @default_name __MODULE__
 
@@ -87,11 +91,15 @@ defmodule Sentry.TelemetryProcessor do
 
   Returns `:ok`.
   """
-  @spec add(Event.t() | LogEvent.t()) :: :ok
+  @spec add(Event.t() | CheckIn.t() | LogEvent.t()) :: :ok
   def add(%Event{} = item) do
     add(processor_name(), item)
   end
 
+  def add(%CheckIn{} = item) do
+    add(processor_name(), item)
+  end
+
   def add(%LogEvent{} = item) do
     add(processor_name(), item)
   end
@@ -103,7 +111,7 @@ defmodule Sentry.TelemetryProcessor do
 
   Returns `:ok`.
   """
-  @spec add(Supervisor.supervisor(), Event.t() | LogEvent.t()) :: :ok
+  @spec add(Supervisor.supervisor(), Event.t() | CheckIn.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))
@@ -118,6 +126,20 @@ defmodule Sentry.TelemetryProcessor do
     :ok
   end
 
+  def add(processor, %CheckIn{} = item) when is_atom(processor) do
+    Buffer.add(buffer_name(processor, :check_in), item)
+    Scheduler.signal(scheduler_name(processor))
+    :ok
+  end
+
+  def add(processor, %CheckIn{} = item) do
+    buffer = get_buffer(processor, :check_in)
+    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))
@@ -165,7 +187,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, :log] do
+  def get_buffer(processor, category) when category in [:error, :check_in, :log] do
     children = Supervisor.which_children(processor)
     buffer_id = buffer_id(category)
 
@@ -195,7 +217,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, :log] do
+  def buffer_size(category) when category in [:error, :check_in, :log] do
     buffer_size(processor_name(), category)
   end
 
@@ -205,7 +227,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, :log] do
+  def buffer_size(processor, category) when category in [:error, :check_in, :log] do
     case safe_get_buffer(processor, category) do
       {:ok, buffer} -> Buffer.size(buffer)
       :error -> 0
@@ -267,6 +289,7 @@ defmodule Sentry.TelemetryProcessor do
       [
         buffers: %{
           error: Map.fetch!(buffer_names, :error),
+          check_in: Map.fetch!(buffer_names, :check_in),
           log: Map.fetch!(buffer_names, :log)
         },
         name: scheduler_name(processor_name),
@@ -286,6 +309,7 @@ defmodule Sentry.TelemetryProcessor do
   end
 
   defp buffer_id(:error), do: :error_buffer
+  defp buffer_id(:check_in), do: :check_in_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, low=2
-      assert length(cycle) == 7
-      assert Enum.frequencies(cycle) == %{error: 5, log: 2}
+      # Default weights: critical=5, high=4, low=2
+      assert length(cycle) == 11
+      assert Enum.frequencies(cycle) == %{error: 5, check_in: 4, log: 2}
     end
 
     test "builds cycle with custom weights" do
       custom_weights = %{low: 5}
       cycle = Scheduler.build_priority_cycle(custom_weights)
 
-      assert length(cycle) == 10
-      assert Enum.frequencies(cycle) == %{error: 5, log: 5}
+      assert length(cycle) == 14
+      assert Enum.frequencies(cycle) == %{error: 5, check_in: 4, log: 5}
     end
   end