feat(logs): introduce TelemetryProcessor

Author: solnic

lib/sentry.ex

@@ -40,7 +40,7 @@ defmodule Sentry do
   > was the `:included_environments` option (a list of environments to report events for).
   > This was used together with the `:environment_name` option to determine whether to
   > send events. `:included_environments` is deprecated in v10.0.0 in favor of setting
-  > or not setting `:dsn`. It will be removed in v11.0.0.
+  > or not setting `:dsn`. It will be removed in v12.0.0.
 
   You can even rely on more specific logic to determine the environment name. It's
   not uncommon for most applications to have a "staging" environment. In order
@@ -183,7 +183,17 @@ defmodule Sentry do
   > with `:source_code_exclude_patterns`.
   """
 
-  alias Sentry.{CheckIn, Client, ClientError, ClientReport, Config, Event, LoggerUtils, Options}
+  alias Sentry.{
+    CheckIn,
+    Client,
+    ClientError,
+    ClientReport,
+    Config,
+    Event,
+    LoggerUtils,
+    Options,
+    TelemetryProcessor
+  }
 
   require Logger
 
@@ -350,7 +360,7 @@ defmodule Sentry do
   """
   @spec send_event(Event.t(), keyword()) :: send_result
   def send_event(event, options \\ []) do
-    # TODO: remove on v11.0.0, :included_environments was deprecated in 10.0.0.
+    # TODO: remove on v12.0.0, :included_environments was deprecated in 10.0.0.
     included_envs = Config.included_environments()
 
     cond do
@@ -378,7 +388,7 @@ defmodule Sentry do
   end
 
   def send_transaction(transaction, options \\ []) do
-    # TODO: remove on v11.0.0, :included_environments was deprecated in 10.0.0.
+    # TODO: remove on v12.0.0, :included_environments was deprecated in 10.0.0.
     included_envs = Config.included_environments()
 
     cond do
@@ -502,4 +512,42 @@ defmodule Sentry do
       nil -> nil
     end
   end
+
+  @doc """
+  Flushes all pending events to Sentry.
+
+  This is a blocking call that drains all the buffers and waits for the scheduler
+  to process all pending items. Useful before application shutdown to ensure
+  all telemetry events are sent.
+
+  ## Options
+
+    * `:timeout` - Maximum time to wait for flush to complete (default: 5000ms)
+
+  ## Examples
+
+      # Flush with default timeout
+      Sentry.flush()
+
+      # Flush with custom timeout
+      Sentry.flush(timeout: 10_000)
+
+  """
+  @doc since: "12.0.0"
+  @spec flush(keyword()) :: :ok
+  def flush(opts \\ []) do
+    timeout = Keyword.get(opts, :timeout, 5000)
+
+    try do
+      TelemetryProcessor.flush(TelemetryProcessor.default_name(), timeout)
+    catch
+      :exit, {:noproc, _} ->
+        Logger.warning("Sentry.flush/1 failed: TelemetryProcessor not running")
+        :ok
+
+      :exit, reason ->
+        Logger.warning("Sentry.flush/1 failed unexpectedly: #{inspect(reason)}")
+        :ok
+    end
+  end
 end

lib/sentry/application.ex

@@ -34,11 +34,15 @@ defmodule Sentry.Application do
         []
       end
 
-    maybe_log_event_buffer =
+    maybe_telemetry_processor =
       if Config.enable_logs?() do
         [
-          {Task.Supervisor, name: Sentry.LogEventBuffer.TaskSupervisor},
-          Sentry.LogEventBuffer
+          {Sentry.TelemetryProcessor,
+           [
+             buffer_capacities: Config.telemetry_buffer_capacities(),
+             scheduler_weights: Config.telemetry_scheduler_weights(),
+             transport_capacity: Config.transport_capacity()
+           ]}
         ]
       else
         []
@@ -58,7 +62,7 @@ defmodule Sentry.Application do
       ] ++
         maybe_http_client_spec ++
         maybe_span_storage ++
-        maybe_log_event_buffer ++
+        maybe_telemetry_processor ++
         maybe_rate_limiter() ++
         [Sentry.Transport.SenderPool]
 

lib/sentry/client.ex

@@ -14,7 +14,6 @@ defmodule Sentry.Client do
     Envelope,
     Event,
     Interfaces,
-    LogEvent,
     LoggerUtils,
     Options,
     Transaction,
@@ -135,69 +134,6 @@ defmodule Sentry.Client do
     end
   end
 
-  @doc """
-  Sends a batch of log events to Sentry.
-
-  Log events are sent asynchronously and do not support sampling.
-  They are buffered and sent in batches according to the Sentry Logs Protocol.
-
-  If a `:before_send_log` callback is configured, it will be called for each log event.
-  If the callback returns `nil` or `false`, the log event is not sent. If it returns an
-  updated `Sentry.LogEvent`, that will be used instead.
-
-  Returns `{:ok, envelope_id}` on success or `{:error, reason}` on failure.
-  """
-  @doc since: "12.0.0"
-  @spec send_log_batch([LogEvent.t()]) ::
-          {:ok, envelope_id :: String.t()} | {:error, ClientError.t()}
-  def send_log_batch([]), do: {:ok, ""}
-
-  def send_log_batch(log_events) when is_list(log_events) do
-    case Sentry.Test.maybe_collect_logs(log_events) do
-      :collected ->
-        {:ok, ""}
-
-      :not_collecting ->
-        log_events
-        |> run_before_send_log_callbacks()
-        |> send_log_events()
-    end
-  end
-
-  defp run_before_send_log_callbacks(log_events) do
-    callback = Config.before_send_log()
-
-    if callback do
-      for log_event <- log_events,
-          %LogEvent{} = modified_event <- [call_before_send_log(log_event, callback)] do
-        modified_event
-      end
-    else
-      log_events
-    end
-  end
-
-  defp call_before_send_log(log_event, function) when is_function(function, 1) do
-    function.(log_event)
-  end
-
-  defp call_before_send_log(log_event, {mod, fun}) do
-    apply(mod, fun, [log_event])
-  end
-
-  defp send_log_events([]), do: {:ok, ""}
-
-  defp send_log_events(log_events) do
-    client = Config.client()
-
-    request_retries =
-      Application.get_env(:sentry, :request_retries, Transport.default_retries())
-
-    log_events
-    |> Envelope.from_log_events()
-    |> Transport.encode_and_post_envelope(client, request_retries)
-  end
-
   defp sample_event(sample_rate) do
     cond do
       sample_rate == 1 -> :ok

lib/sentry/client_report/sender.ex

@@ -31,6 +31,7 @@ defmodule Sentry.ClientReport.Sender do
                | Sentry.CheckIn.t()
                | ClientReport.t()
                | Sentry.Event.t()
+               | Sentry.Transaction.t()
   def record_discarded_events(reason, event_items, genserver)
       when is_list(event_items) do
     # We silently ignore events whose reasons aren't valid because we have to add it to the allowlist in Snuba

lib/sentry/config.ex

@@ -394,17 +394,6 @@ defmodule Sentry.Config do
       Use `Sentry.LogsHandler` to capture log events from Erlang's `:logger`.
       *Available since 12.0.0*.
       """
-    ],
-    max_log_events: [
-      type: :non_neg_integer,
-      default: 100,
-      doc: """
-      The maximum number of log events to buffer before flushing to Sentry.
-      Log events are buffered and sent in batches to reduce network overhead.
-      When the buffer reaches this size, it will be flushed immediately.
-      Otherwise, logs are flushed every 5 seconds. Only applies when `:enable_logs`
-      is `true`. *Available since 12.0.0*.
-      """
     ]
   ]
 
@@ -486,6 +475,41 @@ defmodule Sentry.Config do
       connections to keep in the pool. Only applied if `:client` is set to
       `Sentry.HackneyClient`.
       """
+    ],
+    telemetry_buffer_capacities: [
+      type: {:map, {: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.
+      *Available since v12.0.0*.
+      """
+    ],
+    telemetry_scheduler_weights: [
+      type: {:map, {:in, [: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.
+      *Available since v12.0.0*.
+      """
+    ],
+    transport_capacity: [
+      type: :pos_integer,
+      default: 1000,
+      doc: """
+      Maximum number of items the transport queue can hold. For log envelopes,
+      each log event counts as one item toward capacity. When the queue is full,
+      the scheduler stops dequeuing from buffers until space becomes available.
+      The transport queue processes one envelope at a time.
+      *Available since v12.0.0*.
+      """
     ]
   ]
 
@@ -827,8 +851,14 @@ defmodule Sentry.Config do
   @spec enable_logs?() :: boolean()
   def enable_logs?, do: fetch!(:enable_logs)
 
-  @spec max_log_events() :: non_neg_integer()
-  def max_log_events, do: fetch!(:max_log_events)
+  @spec telemetry_buffer_capacities() :: %{Sentry.Telemetry.Category.t() => pos_integer()}
+  def telemetry_buffer_capacities, do: fetch!(:telemetry_buffer_capacities)
+
+  @spec telemetry_scheduler_weights() :: %{Sentry.Telemetry.Category.priority() => pos_integer()}
+  def telemetry_scheduler_weights, do: fetch!(:telemetry_scheduler_weights)
+
+  @spec transport_capacity() :: pos_integer()
+  def transport_capacity, do: fetch!(:transport_capacity)
 
   @spec before_send_log() ::
           (Sentry.LogEvent.t() -> Sentry.LogEvent.t() | nil | false) | {module(), atom()} | nil

lib/sentry/envelope.ex

@@ -114,6 +114,20 @@ defmodule Sentry.Envelope do
   def get_data_category(%Event{}), do: "error"
   def get_data_category(%LogBatch{}), do: "log_item"
 
+  @doc """
+  Returns the total number of payload items in the envelope.
+
+  For log envelopes, this counts individual log events within the LogBatch.
+  For other envelope types, each item counts as 1.
+  """
+  @spec item_count(t()) :: non_neg_integer()
+  def item_count(%__MODULE__{items: items}) do
+    Enum.reduce(items, 0, fn
+      %LogBatch{log_events: log_events}, acc -> acc + length(log_events)
+      _other, acc -> acc + 1
+    end)
+  end
+
   @doc """
   Encodes the envelope into its binary representation.