View Source Sentry.TelemetryProcessor (Sentry v12.0.3)

Supervisor managing telemetry buffers and scheduler for the Sentry SDK.

The TelemetryProcessor is the coordinator for telemetry data flowing through the SDK. It manages ring buffers coordinated by a weighted round-robin scheduler.

Architecture

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)
Transaction Buffer - for performance transactions (medium priority)
Log Buffer - for log entries (low priority)
Scheduler - weighted round-robin scheduler with integrated transport queue

Usage

# Add error events to the buffer
TelemetryProcessor.add(processor, %Sentry.Event{...})

# 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{...})

# Flush all pending items
TelemetryProcessor.flush(processor)

Summary

Types

option()

Functions

add(item)

Adds an event to the appropriate buffer.

add(processor, item)

Adds an event to the appropriate buffer.

buffer_size(category)

Returns the current size of a buffer for a given category.

buffer_size(processor, category)

Returns the current size of a buffer for a given category.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

default_name()

Returns the default processor name.

flush()

Flushes all buffers by draining their contents and sending all items.

flush(processor, timeout \\ 5000)

Flushes all buffers by draining their contents and sending all items.

get_buffer(processor, category)

Returns the buffer pid for a given category.

get_scheduler(processor)

Returns the scheduler pid.

start_link(opts \\ [])

Starts the TelemetryProcessor supervisor.

Types

option()

(since 12.0.0)

@type option() ::
  {:name, atom()}
  | {:buffer_capacities,
     %{required(Sentry.Telemetry.Category.t()) => pos_integer()}}
  | {:buffer_configs, %{required(Sentry.Telemetry.Category.t()) => map()}}
  | {:scheduler_weights,
     %{required(Sentry.Telemetry.Category.priority()) => pos_integer()}}
  | {:on_envelope, (Sentry.Envelope.t() -> any())}
  | {:transport_capacity, pos_integer()}

Functions

add(item)

(since 12.0.0)

@spec add(
  Sentry.Event.t()
  | Sentry.CheckIn.t()
  | Sentry.Transaction.t()
  | Sentry.LogEvent.t()
) ::
  :ok | {:ok, {:rate_limited, String.t()}}

Adds an event to the appropriate buffer.

Uses the processor from process dictionary or the default (Sentry.TelemetryProcessor). See add/2 for the version accepting a custom processor.

Returns :ok when the item was added, or {:ok, {:rate_limited, data_category}} when the item was dropped due to an active rate limit.

add(processor, item)

(since 12.0.0)

@spec add(
  Supervisor.supervisor(),
  Sentry.Event.t()
  | Sentry.CheckIn.t()
  | Sentry.Transaction.t()
  | Sentry.LogEvent.t()
) :: :ok | {:ok, {:rate_limited, String.t()}}

Adds an event to the appropriate buffer.

After adding, the scheduler is signaled to wake and process items.

Returns :ok when the item was added, or {:ok, {:rate_limited, data_category}} when the item was dropped due to an active rate limit.

buffer_size(category)

(since 12.0.0)

@spec buffer_size(Sentry.Telemetry.Category.t()) :: non_neg_integer()

Returns the current size of a buffer for a given category.

Uses the processor from process dictionary or the default. Returns 0 if the processor is not running.

buffer_size(processor, category)

(since 12.0.0)

@spec buffer_size(Supervisor.supervisor(), Sentry.Telemetry.Category.t()) ::
  non_neg_integer()

Returns the current size of a buffer for a given category.

Returns 0 if the processor is not running.

child_spec(init_arg)

(since 12.0.0)

Returns a specification to start this module under a supervisor.

See Supervisor.

default_name()

(since 12.0.0)

@spec default_name() :: atom()

Returns the default processor name.

flush()

(since 12.0.0)

@spec flush() :: :ok

Flushes all buffers by draining their contents and sending all items.

Uses the processor from process dictionary or the default (Sentry.TelemetryProcessor). This is a blocking call that returns when all items have been processed.

flush(processor, timeout \\ 5000)

(since 12.0.0)

@spec flush(Supervisor.supervisor(), timeout()) :: :ok

Flushes all buffers by draining their contents and sending all items.

This is a blocking call that returns when all items have been processed. The optional timeout specifies how long to wait (default: 5000ms).

get_buffer(processor, category)

(since 12.0.0)

@spec get_buffer(Supervisor.supervisor(), Sentry.Telemetry.Category.t()) :: pid()

Returns the buffer pid for a given category.

get_scheduler(processor)

(since 12.0.0)

@spec get_scheduler(Supervisor.supervisor()) :: pid()

Returns the scheduler pid.

start_link(opts \\ [])

(since 12.0.0)

@spec start_link([option()]) :: Supervisor.on_start()

Starts the TelemetryProcessor supervisor.

Options

:name - Name to register the supervisor under (defaults to Sentry.TelemetryProcessor)
:buffer_capacities - Map of category to capacity override (optional)
:buffer_configs - Map of category to config map with :capacity, :batch_size, :timeout (optional)
:scheduler_weights - Map of priority to weight override (optional)
:on_envelope - Callback function invoked when envelopes are ready to send (optional)
:transport_capacity - Maximum number of items the transport queue can hold (default: 1000). For log envelopes, each log event counts as one item.

Examples

TelemetryProcessor.start_link()

TelemetryProcessor.start_link(
  buffer_capacities: %{log: 2000},
  scheduler_weights: %{low: 3}
)