diff --git a/lib/sentry/client_report/sender.ex b/lib/sentry/client_report/sender.ex index c688ada26..94b129964 100644 --- a/lib/sentry/client_report/sender.ex +++ b/lib/sentry/client_report/sender.ex @@ -6,7 +6,19 @@ defmodule Sentry.ClientReport.Sender do use GenServer - alias Sentry.{Client, ClientReport, Config, Envelope, Transaction} + alias Sentry.{ + Client, + ClientReport, + Config, + Envelope, + LogBatch, + LogEvent, + Metric, + MetricBatch, + Transaction + } + + alias Sentry.Telemetry.Category @send_interval 30_000 @@ -39,17 +51,21 @@ defmodule Sentry.ClientReport.Sender do | Sentry.CheckIn.t() | ClientReport.t() | Sentry.Event.t() + | LogBatch.t() + | LogEvent.t() + | Metric.t() + | MetricBatch.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 # https://develop.sentry.dev/sdk/client-reports/ if Enum.member?(@client_report_reasons, reason) do - Enum.each(event_items, fn item -> - for {category, quantity} <- data_categories(item) do - GenServer.cast(genserver, {:record_discarded_events, reason, category, quantity}) - end - end) + # The items are cast as-is: expanding them into outcomes can mean a full + # JSON encode (for the byte categories), and callers are hot paths such as + # every `Logger` call under an active rate limit. That work belongs to this + # GenServer, not to the producer. + GenServer.cast(genserver, {:record_discarded_items, reason, event_items}) end :ok @@ -65,6 +81,38 @@ defmodule Sentry.ClientReport.Sender do [{Envelope.get_data_category(transaction), 1}, {"span", span_count}] end + defp data_categories(%LogEvent{} = log_event) do + [ + {Category.data_category(:log), 1}, + {Category.byte_data_category(:log), Envelope.item_byte_size(log_event)} + ] + end + + defp data_categories(%Metric{} = metric) do + [ + {Category.data_category(:metric), 1}, + {Category.byte_data_category(:metric), Envelope.item_byte_size(metric)} + ] + end + + defp data_categories(%LogBatch{log_events: log_events}) do + bytes = Enum.reduce(log_events, 0, &(Envelope.item_byte_size(&1) + &2)) + + [ + {Category.data_category(:log), length(log_events)}, + {Category.byte_data_category(:log), bytes} + ] + end + + defp data_categories(%MetricBatch{metrics: metrics}) do + bytes = Enum.reduce(metrics, 0, &(Envelope.item_byte_size(&1) + &2)) + + [ + {Category.data_category(:metric), length(metrics)}, + {Category.byte_data_category(:metric), bytes} + ] + end + defp data_categories(item) do [{Envelope.get_data_category(item), 1}] end @@ -84,7 +132,26 @@ defmodule Sentry.ClientReport.Sender do @impl true def handle_cast({:record_discarded_events, reason, category, quantity}, discarded_events) do - {:noreply, Map.update(discarded_events, {reason, category}, quantity, &(&1 + quantity))} + {:noreply, record(discarded_events, reason, category, quantity)} + end + + def handle_cast({:record_discarded_items, reason, items}, discarded_events) do + discarded_events = + Enum.reduce(items, discarded_events, fn item, acc -> + Enum.reduce(data_categories(item), acc, fn + # Client reports type `quantity` as a positive integer, so an outcome we + # can't quantify (an empty batch, or an item that failed to encode) is + # left out rather than reported as a zero. + {_category, 0}, acc -> acc + {category, quantity}, acc -> record(acc, reason, category, quantity) + end) + end) + + {:noreply, discarded_events} + end + + defp record(discarded_events, reason, category, quantity) do + Map.update(discarded_events, {reason, category}, quantity, &(&1 + quantity)) end @impl true diff --git a/lib/sentry/envelope.ex b/lib/sentry/envelope.ex index 0e83597c2..e54eaaeb7 100644 --- a/lib/sentry/envelope.ex +++ b/lib/sentry/envelope.ex @@ -10,6 +10,7 @@ defmodule Sentry.Envelope do Event, LogBatch, LogEvent, + LoggerUtils, Metric, MetricBatch, Transaction, @@ -138,6 +139,42 @@ defmodule Sentry.Envelope do def get_data_category(%LogBatch{}), do: "log_item" def get_data_category(%MetricBatch{}), do: "trace_metric" + @doc """ + Approximates the serialized byte size of a single log event or metric. + + The size is computed by encoding the item with the same serialization used when + it is placed in an envelope item (`Sentry.LogEvent.to_map/1` or + `Sentry.Metric.to_map/1`), then measuring the byte size of the resulting JSON. + Per the client report spec, a serialized-size approximation like this is + acceptable for `log_byte` and `trace_metric_byte` outcomes. + + Returns `0` (and logs) if the item cannot be encoded. + """ + @spec item_byte_size(LogEvent.t() | Metric.t()) :: non_neg_integer() + def item_byte_size(%LogEvent{} = log_event) do + log_event |> LogEvent.to_map() |> encoded_byte_size(log_event) + end + + def item_byte_size(%Metric{} = metric) do + metric |> Metric.to_map() |> encoded_byte_size(metric) + end + + defp encoded_byte_size(map, item) do + case Sentry.JSON.encode(map, Config.json_library()) do + {:ok, encoded} -> + byte_size(encoded) + + {:error, reason} -> + # A byte outcome is quota data reported to Sentry, so an unencodable item + # means we under-report rather than report nothing — worth surfacing. + LoggerUtils.log(fn -> + "Failed to compute the byte size of #{inspect(item.__struct__)}: #{inspect(reason)}" + end) + + 0 + end + end + @doc """ Returns the total number of payload items in the envelope. diff --git a/lib/sentry/logger_handler/logs_backend.ex b/lib/sentry/logger_handler/logs_backend.ex index ad50ee137..84f33deed 100644 --- a/lib/sentry/logger_handler/logs_backend.ex +++ b/lib/sentry/logger_handler/logs_backend.ex @@ -37,8 +37,8 @@ defmodule Sentry.LoggerHandler.LogsBackend do log_event_struct = LogEvent.from_logger_event(log_event, attributes, parameters) case TelemetryProcessor.add(log_event_struct) do - {:ok, {:rate_limited, data_category}} -> - Sentry.ClientReport.Sender.record_discarded_events(:ratelimit_backoff, data_category) + {:ok, {:rate_limited, _data_category}} -> + Sentry.ClientReport.Sender.record_discarded_events(:ratelimit_backoff, [log_event_struct]) :ok -> :ok diff --git a/lib/sentry/metrics.ex b/lib/sentry/metrics.ex index 5c8d946d9..abc31fc67 100644 --- a/lib/sentry/metrics.ex +++ b/lib/sentry/metrics.ex @@ -135,8 +135,8 @@ defmodule Sentry.Metrics do metric = Metric.attach_default_attributes(metric) case TelemetryProcessor.add(metric) do - {:ok, {:rate_limited, data_category}} -> - ClientReport.Sender.record_discarded_events(:ratelimit_backoff, data_category) + {:ok, {:rate_limited, _data_category}} -> + ClientReport.Sender.record_discarded_events(:ratelimit_backoff, [metric]) :ok -> :ok diff --git a/lib/sentry/telemetry/buffer.ex b/lib/sentry/telemetry/buffer.ex index b6244e147..b281de720 100644 --- a/lib/sentry/telemetry/buffer.ex +++ b/lib/sentry/telemetry/buffer.ex @@ -8,7 +8,7 @@ defmodule Sentry.Telemetry.Buffer do ## Options - * `:category` - The telemetry category (required), currently only `:log` + * `:category` - The telemetry category (required), one of `Sentry.Telemetry.Category.t/0` * `:name` - The name to register the GenServer under (optional) * `:capacity` - Buffer capacity (defaults to category default) * `:batch_size` - Items per batch (defaults to category default) @@ -24,6 +24,7 @@ defmodule Sentry.Telemetry.Buffer do alias Sentry.ClientReport alias Sentry.Telemetry.Category + alias Sentry.{LogEvent, Metric} @enforce_keys [:category, :capacity, :batch_size] defstruct [ @@ -178,12 +179,9 @@ defmodule Sentry.Telemetry.Buffer do defp offer(%Buffer{size: size, capacity: capacity} = state, item) when size >= capacity do - {{:value, _dropped}, items} = :queue.out(state.items) + {{:value, dropped}, items} = :queue.out(state.items) - ClientReport.Sender.record_discarded_events( - :cache_overflow, - Category.data_category(state.category) - ) + record_overflow_discard(state.category, dropped) %{state | items: :queue.in(item, items)} end @@ -192,6 +190,23 @@ defmodule Sentry.Telemetry.Buffer do %{state | items: :queue.in(item, state.items), size: state.size + 1} end + # Logs and metrics are handed to the recorder as structs so it can pair the + # count outcome with the byte one; no other category has a byte companion. + defp record_overflow_discard(:log, %LogEvent{} = dropped) do + ClientReport.Sender.record_discarded_events(:cache_overflow, [dropped]) + end + + defp record_overflow_discard(:metric, %Metric{} = dropped) do + ClientReport.Sender.record_discarded_events(:cache_overflow, [dropped]) + end + + defp record_overflow_discard(category, _dropped) do + ClientReport.Sender.record_discarded_events( + :cache_overflow, + Category.data_category(category) + ) + end + defp poll_batch(state, count), do: poll_batch(state, count, []) defp poll_batch(state, 0, acc), do: {Enum.reverse(acc), state} defp poll_batch(%{size: 0} = state, _count, acc), do: {Enum.reverse(acc), state} diff --git a/lib/sentry/telemetry/category.ex b/lib/sentry/telemetry/category.ex index 51acee5f5..218353003 100644 --- a/lib/sentry/telemetry/category.ex +++ b/lib/sentry/telemetry/category.ex @@ -186,4 +186,28 @@ defmodule Sentry.Telemetry.Category do def data_category(:transaction), do: "transaction" def data_category(:log), do: "log_item" def data_category(:metric), do: "trace_metric" + + @doc """ + Returns the byte-based Sentry data category string for a given telemetry category. + + Some data categories have a companion "byte" category used to report the total + serialized size of dropped items in client reports and to honor byte-based rate + limits. Only `:log` and `:metric` currently have such companion categories. + + These strings are used in client reports and rate limiting alongside the + count-based category returned by `data_category/1`. + + ## Examples + + iex> Sentry.Telemetry.Category.byte_data_category(:log) + "log_byte" + + iex> Sentry.Telemetry.Category.byte_data_category(:metric) + "trace_metric_byte" + + """ + @doc since: "13.4.0" + @spec byte_data_category(:log | :metric) :: String.t() + def byte_data_category(:log), do: "log_byte" + def byte_data_category(:metric), do: "trace_metric_byte" end diff --git a/lib/sentry/telemetry/scheduler.ex b/lib/sentry/telemetry/scheduler.ex index d7b890e85..72cd025ed 100644 --- a/lib/sentry/telemetry/scheduler.ex +++ b/lib/sentry/telemetry/scheduler.ex @@ -534,9 +534,9 @@ defmodule Sentry.Telemetry.Scheduler do items == [] -> :ok - # Transactions carry spans, so pass the actual structs through the - # list-based recorder to also record the discarded "span" outcomes. - category == :transaction -> + # These categories expand to paired outcomes (span / log_byte / + # trace_metric_byte) that only the struct-based recorder emits. + category in [:transaction, :log, :metric] -> ClientReport.Sender.record_discarded_events(:ratelimit_backoff, items) true -> @@ -554,8 +554,9 @@ defmodule Sentry.Telemetry.Scheduler do defp category_rate_limited?(%{on_envelope: cb}, _category) when is_function(cb, 1), do: false defp category_rate_limited?(_state, category) do - data_category = Category.data_category(category) - RateLimiter.rate_limited?(data_category) + category + |> Category.data_category() + |> RateLimiter.rate_limited_for_category?() end defp default_weights do diff --git a/lib/sentry/telemetry_processor.ex b/lib/sentry/telemetry_processor.ex index de78419a5..df1a7926a 100644 --- a/lib/sentry/telemetry_processor.ex +++ b/lib/sentry/telemetry_processor.ex @@ -253,7 +253,7 @@ defmodule Sentry.TelemetryProcessor do defp add_to_buffer(processor, category, item) when is_atom(processor) do data_category = Category.data_category(category) - if RateLimiter.rate_limited?(data_category) do + if RateLimiter.rate_limited_for_category?(data_category) do {:ok, {:rate_limited, data_category}} else Buffer.add(buffer_name(processor, category), item) @@ -265,7 +265,7 @@ defmodule Sentry.TelemetryProcessor do defp add_to_buffer(processor, category, item) do data_category = Category.data_category(category) - if RateLimiter.rate_limited?(data_category) do + if RateLimiter.rate_limited_for_category?(data_category) do {:ok, {:rate_limited, data_category}} else buffer = get_buffer(processor, category) diff --git a/lib/sentry/transport.ex b/lib/sentry/transport.ex index 36ae4fb0f..fe5cb473d 100644 --- a/lib/sentry/transport.ex +++ b/lib/sentry/transport.ex @@ -86,8 +86,9 @@ defmodule Sentry.Transport do defp check_rate_limited(envelope_items) do rate_limited? = Enum.any?(envelope_items, fn item -> - category = Envelope.get_data_category(item) - RateLimiter.rate_limited?(category) + item + |> Envelope.get_data_category() + |> RateLimiter.rate_limited_for_category?() end) if rate_limited?, do: {:error, :rate_limited}, else: :ok diff --git a/lib/sentry/transport/rate_limiter.ex b/lib/sentry/transport/rate_limiter.ex index d79904127..16e9cff2d 100644 --- a/lib/sentry/transport/rate_limiter.ex +++ b/lib/sentry/transport/rate_limiter.ex @@ -90,6 +90,27 @@ defmodule Sentry.Transport.RateLimiter do rate_limited?(category, now) or rate_limited?(:global, now) end + @doc """ + Checks whether sending items of the given data category is currently limited. + + Logs and metrics have a companion byte category (`log_byte` / + `trace_metric_byte`) that Sentry can limit independently of the count + category, so a limit on either one must suppress sending. Every other category + gates on itself alone. + + So an active `log_byte` limit makes this return `true` for `"log_item"`, even + though `rate_limited?("log_item")` on its own is `false`. + """ + @spec rate_limited_for_category?(String.t()) :: boolean() + def rate_limited_for_category?("log_item"), + do: rate_limited?("log_item") or rate_limited?("log_byte") + + def rate_limited_for_category?("trace_metric"), + do: rate_limited?("trace_metric") or rate_limited?("trace_metric_byte") + + def rate_limited_for_category?(category) when is_binary(category), + do: rate_limited?(category) + @doc """ Updates global rate limit from a `Retry-After` header value. diff --git a/test/envelope_test.exs b/test/envelope_test.exs index feea8995f..5d356abdb 100644 --- a/test/envelope_test.exs +++ b/test/envelope_test.exs @@ -1,6 +1,7 @@ defmodule Sentry.EnvelopeTest do use Sentry.Case, async: false + import ExUnit.CaptureLog import Sentry.TestHelpers alias Sentry.{Attachment, CheckIn, ClientReport, Envelope, Event, LogEvent, Metric} @@ -336,4 +337,68 @@ defmodule Sentry.EnvelopeTest do assert Envelope.get_data_category(metric_batch) == "trace_metric" end end + + describe "item_byte_size/1 for log events" do + test "approximates the serialized size of a single log event" do + log_event = %LogEvent{ + timestamp: 1_588_601_261.535_386, + level: :info, + body: "something happened" + } + + assert Envelope.item_byte_size(log_event) > 0 + end + + test "matches the size of the serialized envelope payload for the log event" do + log_event = %LogEvent{ + timestamp: 1_588_601_261.535_386, + level: :info, + body: "something happened" + } + + # Computed independently of the function under test, so that a change in + # what gets serialized fails here instead of silently agreeing with itself. + expected = byte_size(encode!(LogEvent.to_map(log_event))) + + assert Envelope.item_byte_size(log_event) == expected + end + + test "grows with the size of the log body" do + base = %LogEvent{timestamp: 1_588_601_261.535_386, level: :info, body: "hi"} + large = %LogEvent{base | body: String.duplicate("x", 1_000)} + + assert Envelope.item_byte_size(large) > + Envelope.item_byte_size(base) + 900 + end + + @tag :capture_log + test "returns 0 and logs when the log event cannot be encoded" do + log_event = %LogEvent{timestamp: 1_588_601_261.535_386, level: :info, body: <<0xFF>>} + + assert capture_log(fn -> + assert Envelope.item_byte_size(log_event) == 0 + end) =~ "Failed to compute the byte size of Sentry.LogEvent" + end + end + + describe "item_byte_size/1 for metrics" do + test "approximates the serialized size of a single metric" do + metric = %Metric{ + type: :counter, + name: "test.counter", + value: 1, + timestamp: 1_588_601_261.535_386 + } + + assert Envelope.item_byte_size(metric) > 0 + end + + test "grows with the size of the metric name" do + base = %Metric{type: :counter, name: "m", value: 1, timestamp: 1_588_601_261.535_386} + large = %Metric{base | name: String.duplicate("n", 1_000)} + + assert Envelope.item_byte_size(large) > + Envelope.item_byte_size(base) + 900 + end + end end diff --git a/test/sentry/client_report/sender_test.exs b/test/sentry/client_report/sender_test.exs index 940a72cf1..00c3dc80f 100644 --- a/test/sentry/client_report/sender_test.exs +++ b/test/sentry/client_report/sender_test.exs @@ -4,7 +4,7 @@ defmodule Sentry.ClientReportTest do import Sentry.TestHelpers alias Sentry.ClientReport.Sender - alias Sentry.Event + alias Sentry.{Envelope, Event, LogBatch, LogEvent, Metric, MetricBatch} setup do setup_bypass() @@ -139,5 +139,127 @@ defmodule Sentry.ClientReportTest do {:before_send, "span"} => 1 } end + + test "records both log_item and log_byte outcomes when a log event is discarded" do + start_supervised!({Sender, name: :test_log_report}) + + log_event = make_log_event("hello world") + expected_bytes = Envelope.item_byte_size(log_event) + assert expected_bytes > 0 + + assert :ok = + Sender.record_discarded_events(:ratelimit_backoff, [log_event], :test_log_report) + + assert :sys.get_state(:test_log_report) == %{ + {:ratelimit_backoff, "log_item"} => 1, + {:ratelimit_backoff, "log_byte"} => expected_bytes + } + end + + test "records both trace_metric and trace_metric_byte outcomes when a metric is discarded" do + start_supervised!({Sender, name: :test_metric_report}) + + metric = make_metric("requests", 1) + expected_bytes = Envelope.item_byte_size(metric) + assert expected_bytes > 0 + + assert :ok = + Sender.record_discarded_events(:ratelimit_backoff, [metric], :test_metric_report) + + assert :sys.get_state(:test_metric_report) == %{ + {:ratelimit_backoff, "trace_metric"} => 1, + {:ratelimit_backoff, "trace_metric_byte"} => expected_bytes + } + end + + test "records aggregate log_item and log_byte outcomes when a log batch is discarded" do + start_supervised!({Sender, name: :test_log_batch_report}) + + log_events = [make_log_event("first"), make_log_event("second")] + expected_bytes = Enum.reduce(log_events, 0, &(Envelope.item_byte_size(&1) + &2)) + assert expected_bytes > 0 + log_batch = %LogBatch{log_events: log_events} + + assert :ok = + Sender.record_discarded_events(:send_error, [log_batch], :test_log_batch_report) + + assert :sys.get_state(:test_log_batch_report) == %{ + {:send_error, "log_item"} => 2, + {:send_error, "log_byte"} => expected_bytes + } + end + + test "records aggregate trace_metric and trace_metric_byte outcomes when a metric batch is discarded" do + start_supervised!({Sender, name: :test_metric_batch_report}) + + metrics = [make_metric("a", 1), make_metric("b", 2)] + expected_bytes = Enum.reduce(metrics, 0, &(Envelope.item_byte_size(&1) + &2)) + assert expected_bytes > 0 + metric_batch = %MetricBatch{metrics: metrics} + + assert :ok = + Sender.record_discarded_events( + :send_error, + [metric_batch], + :test_metric_batch_report + ) + + assert :sys.get_state(:test_metric_batch_report) == %{ + {:send_error, "trace_metric"} => 2, + {:send_error, "trace_metric_byte"} => expected_bytes + } + end + + test "does not record zero-quantity outcomes for an empty batch" do + start_supervised!({Sender, name: :test_empty_batch_report}) + + assert :ok = + Sender.record_discarded_events( + :send_error, + [%LogBatch{log_events: []}, %MetricBatch{metrics: []}], + :test_empty_batch_report + ) + + assert :sys.get_state(:test_empty_batch_report) == %{} + end + + test "expands items into outcomes inside the Sender, not on the calling process" do + start_supervised!({Sender, name: :test_encoding_process_report}) + put_test_config(json_library: ReportingJSONLibrary) + + ReportingJSONLibrary.report_to(self()) + sender = Process.whereis(:test_encoding_process_report) + + assert :ok = + Sender.record_discarded_events( + :ratelimit_backoff, + [make_log_event("hello world")], + :test_encoding_process_report + ) + + # Flush the cast so the encode has definitely happened by now. + _ = :sys.get_state(:test_encoding_process_report) + + assert_received {:encoded, encoding_pid} + assert encoding_pid == sender + end + end + + defp make_log_event(body) do + %LogEvent{ + timestamp: System.system_time(:nanosecond) / 1_000_000_000, + level: :info, + body: body + } + end + + defp make_metric(name, value) do + %Metric{ + type: :counter, + name: name, + value: value, + timestamp: System.system_time(:nanosecond) / 1_000_000_000, + attributes: %{} + } end end diff --git a/test/sentry/telemetry/category_test.exs b/test/sentry/telemetry/category_test.exs new file mode 100644 index 000000000..20c9c50d9 --- /dev/null +++ b/test/sentry/telemetry/category_test.exs @@ -0,0 +1,5 @@ +defmodule Sentry.Telemetry.CategoryTest do + use Sentry.Case, async: true + + doctest Sentry.Telemetry.Category +end diff --git a/test/sentry/telemetry_processor_integration_test.exs b/test/sentry/telemetry_processor_integration_test.exs index 61be3b53d..46b811822 100644 --- a/test/sentry/telemetry_processor_integration_test.exs +++ b/test/sentry/telemetry_processor_integration_test.exs @@ -3,6 +3,8 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do import Sentry.TestHelpers + require Logger + alias Sentry.TelemetryProcessor alias Sentry.Telemetry.Buffer alias Sentry.{LogEvent, Metric, Transaction} @@ -245,7 +247,12 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do describe "buffer overflow client reports" do setup ctx do - Sentry.Test.setup_telemetry_processor(buffer_configs: %{log: %{capacity: 2, batch_size: 1}}) + Sentry.Test.setup_telemetry_processor( + buffer_configs: %{ + log: %{capacity: 2, batch_size: 1}, + metric: %{capacity: 2, batch_size: 1} + } + ) Sentry.ClientReport.Sender.flush() flush_ref_messages(ctx.ref) @@ -257,12 +264,14 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do scheduler = TelemetryProcessor.get_scheduler(ctx.processor) :sys.suspend(scheduler) - TelemetryProcessor.add(ctx.processor, make_log_event("log-1")) + # The evicted (oldest) event is deliberately much larger than the ones that + # stay, so reporting the wrong item fails loudly instead of by a byte or two. + dropped_log = make_log_event(String.duplicate("x", 500)) + TelemetryProcessor.add(ctx.processor, dropped_log) TelemetryProcessor.add(ctx.processor, make_log_event("log-2")) TelemetryProcessor.add(ctx.processor, make_log_event("log-3")) - log_buffer = TelemetryProcessor.get_buffer(ctx.processor, :log) - _ = Buffer.size(log_buffer) + _ = Buffer.size(TelemetryProcessor.get_buffer(ctx.processor, :log)) Sentry.ClientReport.Sender.flush() @@ -272,11 +281,53 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do items = decode_envelope!(body) assert [{%{"type" => "client_report"}, client_report}] = items - cache_overflow = - Enum.find(client_report["discarded_events"], &(&1["reason"] == "cache_overflow")) + discarded = client_report["discarded_events"] + + log_item = + Enum.find(discarded, &(&1["reason"] == "cache_overflow" and &1["category"] == "log_item")) + + log_byte = + Enum.find(discarded, &(&1["reason"] == "cache_overflow" and &1["category"] == "log_byte")) - assert cache_overflow["category"] == "log_item" - assert cache_overflow["quantity"] == 1 + assert log_item["quantity"] == 1 + assert log_byte["quantity"] == Sentry.Envelope.item_byte_size(dropped_log) + + :sys.resume(scheduler) + end + + test "sends trace_metric and trace_metric_byte reports when metric buffer overflows", ctx do + scheduler = TelemetryProcessor.get_scheduler(ctx.processor) + :sys.suspend(scheduler) + + dropped_metric = make_metric(String.duplicate("m", 500), 1) + TelemetryProcessor.add(ctx.processor, dropped_metric) + TelemetryProcessor.add(ctx.processor, make_metric("metric-2", 2)) + TelemetryProcessor.add(ctx.processor, make_metric("metric-3", 3)) + + _ = Buffer.size(TelemetryProcessor.get_buffer(ctx.processor, :metric)) + + Sentry.ClientReport.Sender.flush() + + ref = ctx.ref + assert_receive {:bypass_envelope, ^ref, body}, 2000 + + assert [{%{"type" => "client_report"}, client_report}] = decode_envelope!(body) + discarded = client_report["discarded_events"] + + trace_metric = + Enum.find( + discarded, + &(&1["reason"] == "cache_overflow" and &1["category"] == "trace_metric") + ) + + trace_metric_byte = + Enum.find( + discarded, + &(&1["reason"] == "cache_overflow" and &1["category"] == "trace_metric_byte") + ) + + assert trace_metric["quantity"] == 1 + assert trace_metric_byte["quantity"] == Sentry.Envelope.item_byte_size(dropped_metric) :sys.resume(scheduler) end @@ -307,56 +358,88 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do test "rate-limited HTTP response causes subsequent events to be dropped with client report", ctx do - test_pid = self() - ref = make_ref() - request_count = :counters.new(1, []) - put_test_config(client: Sentry.FinchClient) - Bypass.expect(ctx.bypass, "POST", "/api/1/envelope/", fn conn -> - count = :counters.get(request_count, 1) - :counters.add(request_count, 1, 1) - {:ok, body, conn} = Plug.Conn.read_body(conn) - send(test_pid, {ref, body}) - - if count == 0 do - conn - |> Plug.Conn.put_resp_header("X-Sentry-Rate-Limits", "60:error:organization") - |> Plug.Conn.resp(200, ~s<{"id": "340"}>) - else - Plug.Conn.resp(conn, 200, ~s<{"id": "340"}>) - end - end) + ref = install_rate_limit_response(ctx.bypass, "error") Sentry.capture_message("first-error", result: :none) - assert_receive {^ref, body}, 2000 + assert_receive {:bypass_envelope, ^ref, body}, 2000 assert [{%{"type" => "event"}, event}] = decode_envelope!(body) assert event["message"]["formatted"] == "first-error" - scheduler = TelemetryProcessor.get_scheduler(ctx.processor) - - poll_until(fn -> - %{active_ref: active_ref} = :sys.get_state(scheduler) - is_nil(active_ref) - end) + wait_for_scheduler_idle(ctx.processor) Sentry.capture_message("rate-limited-error", result: :none) - refute_receive {^ref, _body}, 200 + outcomes = collect_discarded_outcomes(ref, "ratelimit_backoff") + assert outcomes["error"] == 1 + end + end + + # The queue-worker process applies the response header to the *global* rate + # limiter table, while producers read this test's isolated one — so items still + # enter the buffer here and are dropped later, when the scheduler drains it. + describe "byte-based rate limits returned by Sentry" do + setup ctx do + Sentry.Test.setup_telemetry_processor( + buffer_configs: %{log: %{batch_size: 1}, metric: %{batch_size: 1}} + ) + + put_test_config(client: Sentry.FinchClient) Sentry.ClientReport.Sender.flush() + flush_ref_messages(ctx.ref) - assert_receive {^ref, body}, 2000 - items = decode_envelope!(body) - assert [{%{"type" => "client_report"}, client_report}] = items + on_exit(fn -> + for category <- ~w(log_item log_byte trace_metric trace_metric_byte) do + try do + :ets.delete(Sentry.Transport.RateLimiter, category) + catch + :error, :badarg -> :ok + end + end + end) - ratelimit_event = - Enum.find(client_report["discarded_events"], &(&1["reason"] == "ratelimit_backoff")) + :ok + end - assert ratelimit_event != nil - assert ratelimit_event["category"] == "error" - assert ratelimit_event["quantity"] == 1 + test "a trace_metric_byte limit stops further metrics and reports paired outcomes", ctx do + put_test_config(enable_metrics: true) + + ref = install_rate_limit_response(ctx.bypass, "trace_metric_byte") + + Sentry.Metrics.count("first.metric", 1) + assert_receive {:bypass_envelope, ^ref, _body}, 5000 + + wait_for_scheduler_idle(ctx.processor) + + Sentry.Metrics.count("rate.limited.metric", 1) + + outcomes = collect_discarded_outcomes(ref, "ratelimit_backoff") + + assert outcomes["trace_metric"] == 1 + assert is_integer(outcomes["trace_metric_byte"]) and outcomes["trace_metric_byte"] > 0 + end + + @tag :capture_log + test "a log_byte limit stops further logs and reports paired outcomes", ctx do + put_test_config(enable_logs: true, logs: [level: :info]) + attach_sentry_logs_handler() + + ref = install_rate_limit_response(ctx.bypass, "log_byte") + + Logger.info("first log") + assert_receive {:bypass_envelope, ^ref, _body}, 5000 + + wait_for_scheduler_idle(ctx.processor) + + Logger.info("rate limited log") + + outcomes = collect_discarded_outcomes(ref, "ratelimit_backoff") + + assert outcomes["log_item"] == 1 + assert is_integer(outcomes["log_byte"]) and outcomes["log_byte"] > 0 end end @@ -374,6 +457,8 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do :ets.delete(rate_limiter_table, "monitor") :ets.delete(rate_limiter_table, "transaction") :ets.delete(rate_limiter_table, "trace_metric") + :ets.delete(rate_limiter_table, "log_byte") + :ets.delete(rate_limiter_table, "trace_metric_byte") catch :error, :badarg -> :ok end @@ -484,6 +569,46 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do assert Buffer.size(metric_buffer) == 0 end + + # These drive the public APIs rather than `TelemetryProcessor.add/2` because + # the paired byte outcome is recorded by `add/2`'s caller, which runs in the + # emitting process and so reads this test's isolated rate limiter table. + @tag :capture_log + test "a log_byte rate limit drops logs emitted via Logger with paired outcomes", ctx do + put_test_config(enable_logs: true, logs: [level: :info]) + attach_sentry_logs_handler() + + log_buffer = TelemetryProcessor.get_buffer(ctx.processor, :log) + + :ets.insert(ctx.rate_limiter_table, {"log_byte", System.system_time(:second) + 60}) + + Logger.info("dropped by a log_byte limit") + + assert Buffer.size(log_buffer) == 0 + + outcomes = collect_discarded_outcomes(ctx.ref, "ratelimit_backoff") + + assert outcomes["log_item"] == 1 + assert is_integer(outcomes["log_byte"]) and outcomes["log_byte"] > 0 + end + + test "a trace_metric_byte rate limit drops metrics emitted via Sentry.Metrics with paired outcomes", + ctx do + put_test_config(enable_metrics: true) + + metric_buffer = TelemetryProcessor.get_buffer(ctx.processor, :metric) + + :ets.insert(ctx.rate_limiter_table, {"trace_metric_byte", System.system_time(:second) + 60}) + + Sentry.Metrics.count("dropped.by.byte.limit", 1) + + assert Buffer.size(metric_buffer) == 0 + + outcomes = collect_discarded_outcomes(ctx.ref, "ratelimit_backoff") + + assert outcomes["trace_metric"] == 1 + assert is_integer(outcomes["trace_metric_byte"]) and outcomes["trace_metric_byte"] > 0 + end end describe "scheduler draining a rate-limited buffer" do @@ -498,6 +623,10 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do on_exit(fn -> try do :ets.delete(Sentry.Transport.RateLimiter, "transaction") + :ets.delete(Sentry.Transport.RateLimiter, "log_item") + :ets.delete(Sentry.Transport.RateLimiter, "log_byte") + :ets.delete(Sentry.Transport.RateLimiter, "trace_metric") + :ets.delete(Sentry.Transport.RateLimiter, "trace_metric_byte") catch :error, :badarg -> :ok end @@ -545,9 +674,75 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do assert outcomes["transaction"] == 1 assert outcomes["span"] == 3 end + + test "records log_item and log_byte outcomes when a buffered log is dropped by rate limiting", + ctx do + scheduler = TelemetryProcessor.get_scheduler(ctx.processor) + log_buffer = TelemetryProcessor.get_buffer(ctx.processor, :log) + + :sys.suspend(scheduler) + + dropped_log = make_log_event("rate-limited-log") + TelemetryProcessor.add(ctx.processor, dropped_log) + assert Buffer.size(log_buffer) == 1 + + :ets.insert(Sentry.Transport.RateLimiter, {"log_item", System.system_time(:second) + 60}) + + :sys.resume(scheduler) + GenServer.cast(scheduler, :signal) + + poll_until(fn -> Buffer.size(log_buffer) == 0 end) + + Sentry.ClientReport.Sender.flush() + + ref = ctx.ref + assert_receive {:bypass_envelope, ^ref, body}, 2000 + assert [{%{"type" => "client_report"}, client_report}] = decode_envelope!(body) + + outcomes = + for event <- client_report["discarded_events"], + event["reason"] == "ratelimit_backoff", + into: %{}, + do: {event["category"], event["quantity"]} + + assert outcomes["log_item"] == 1 + assert outcomes["log_byte"] == Sentry.Envelope.item_byte_size(dropped_log) + end end - defp make_transaction do + defp install_rate_limit_response(bypass, category) do + test_pid = self() + ref = make_ref() + request_count = :counters.new(1, []) + + Bypass.expect(bypass, "POST", "/api/1/envelope/", fn conn -> + count = :counters.get(request_count, 1) + :counters.add(request_count, 1, 1) + {:ok, body, conn} = Plug.Conn.read_body(conn) + send(test_pid, {:bypass_envelope, ref, body}) + + if count == 0 do + conn + |> Plug.Conn.put_resp_header("X-Sentry-Rate-Limits", "60:#{category}:organization") + |> Plug.Conn.resp(200, ~s<{"id": "340"}>) + else + Plug.Conn.resp(conn, 200, ~s<{"id": "340"}>) + end + end) + + ref + end + + defp wait_for_scheduler_idle(processor) do + scheduler = TelemetryProcessor.get_scheduler(processor) + + poll_until(fn -> + %{active_ref: active_ref} = :sys.get_state(scheduler) + is_nil(active_ref) + end) + end + + defp make_transaction(opts \\ []) do now = System.system_time(:microsecond) %Transaction{ @@ -555,18 +750,34 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do span_id: Sentry.UUID.uuid4_hex() |> binary_part(0, 16), start_timestamp: (now - 1_000_000) / 1_000_000, timestamp: now / 1_000_000, - spans: [] + spans: Keyword.get(opts, :spans, []) } end - defp make_metric(name, value) do - %Metric{ - type: :counter, - name: name, - value: value, - timestamp: System.system_time(:nanosecond) / 1_000_000_000, - attributes: %{} - } + # Outcomes are recorded with a cast from whichever process dropped the item, so + # wait for the Sender to have something to report before flushing — an empty + # buffer is not enough, since items are drained before the outcome is recorded. + defp collect_discarded_outcomes(ref, reason) do + poll_until(fn -> :sys.get_state(Sentry.ClientReport.Sender) != %{} end) + + Sentry.ClientReport.Sender.flush() + + for event <- await_client_report(ref)["discarded_events"], + event["reason"] == reason, + into: %{}, + do: {event["category"], event["quantity"]} + end + + defp await_client_report(ref) do + receive do + {:bypass_envelope, ^ref, body} -> + case decode_envelope!(body) do + [{%{"type" => "client_report"}, client_report}] -> client_report + _other -> await_client_report(ref) + end + after + 2000 -> flunk("no client report envelope received") + end end defp flush_ref_messages(ref) do @@ -577,6 +788,11 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do end end + defp decoded_envelope_category([{%{"type" => "event"}, _} | _]), do: :error + defp decoded_envelope_category([{%{"type" => "check_in"}, _} | _]), do: :check_in + defp decoded_envelope_category([{%{"type" => "transaction"}, _} | _]), do: :transaction + defp decoded_envelope_category([{%{"type" => "log"}, _} | _]), do: :log + defp make_log_event(body) do %LogEvent{ timestamp: System.system_time(:nanosecond) / 1_000_000_000, @@ -585,10 +801,36 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do } end - defp decoded_envelope_category([{%{"type" => "event"}, _} | _]), do: :error - defp decoded_envelope_category([{%{"type" => "check_in"}, _} | _]), do: :check_in - defp decoded_envelope_category([{%{"type" => "transaction"}, _} | _]), do: :transaction - defp decoded_envelope_category([{%{"type" => "log"}, _} | _]), do: :log + defp make_metric(name, value) do + %Metric{ + type: :counter, + name: name, + value: value, + timestamp: System.system_time(:nanosecond) / 1_000_000_000, + attributes: %{} + } + end + + defp attach_sentry_logs_handler do + logs = Sentry.Config.logs() + + config = [ + enable_logs: true, + capture_log_messages: Keyword.fetch!(logs, :capture_log_messages), + capture_level: Keyword.fetch!(logs, :capture_level), + capture_metadata: Keyword.fetch!(logs, :capture_metadata), + capture_excluded_domains: Keyword.fetch!(logs, :capture_excluded_domains), + logs_level: Keyword.fetch!(logs, :level), + logs_metadata: Keyword.fetch!(logs, :metadata), + logs_excluded_domains: Keyword.fetch!(logs, :excluded_domains) + ] + + handler_name = :"sentry_logs_handler_#{System.unique_integer([:positive])}" + :ok = :logger.add_handler(handler_name, Sentry.LoggerHandler, %{config: config}) + on_exit(fn -> _ = :logger.remove_handler(handler_name) end) + + handler_name + end defp poll_until(fun, timeout \\ 2000) do deadline = System.monotonic_time(:millisecond) + timeout diff --git a/test/sentry/transport/rate_limiter_test.exs b/test/sentry/transport/rate_limiter_test.exs index 9ac055deb..ca7e92322 100644 --- a/test/sentry/transport/rate_limiter_test.exs +++ b/test/sentry/transport/rate_limiter_test.exs @@ -60,6 +60,40 @@ defmodule Sentry.Transport.RateLimiterTest do assert RateLimiter.rate_limited?("error") == true assert RateLimiter.rate_limited?("transaction") == true end + + test "a byte category parsed from the header gates its count category" do + # X-Sentry-Rate-Limits: 60:log_byte:organization + RateLimiter.update_rate_limits("60:log_byte:organization") + + assert RateLimiter.rate_limited?("log_byte") == true + assert RateLimiter.rate_limited_for_category?("log_item") == true + end + end + + describe "rate_limited_for_category?/1" do + test "gates log_item on the log_byte limit as well" do + now = System.system_time(:second) + :ets.insert(table_name(), {"log_byte", now + 60}) + + assert RateLimiter.rate_limited_for_category?("log_item") == true + assert RateLimiter.rate_limited?("log_item") == false + end + + test "gates trace_metric on the trace_metric_byte limit as well" do + now = System.system_time(:second) + :ets.insert(table_name(), {"trace_metric_byte", now + 60}) + + assert RateLimiter.rate_limited_for_category?("trace_metric") == true + assert RateLimiter.rate_limited?("trace_metric") == false + end + + test "gates a category on itself when it has no companion byte category" do + now = System.system_time(:second) + :ets.insert(table_name(), {"error", now + 60}) + + assert RateLimiter.rate_limited_for_category?("error") == true + assert RateLimiter.rate_limited_for_category?("transaction") == false + end end describe "update_rate_limits/1" do diff --git a/test/sentry/transport_test.exs b/test/sentry/transport_test.exs index 9b222449b..3e5d29345 100644 --- a/test/sentry/transport_test.exs +++ b/test/sentry/transport_test.exs @@ -393,6 +393,58 @@ defmodule Sentry.TransportTest do # Other categories should not be rate-limited refute Transport.RateLimiter.rate_limited?("session") end + + test "drops log envelopes when a log_byte rate limit is active", %{bypass: bypass} do + Bypass.expect_once(bypass, "POST", "/api/1/envelope/", fn conn -> + conn + |> Plug.Conn.put_resp_header("X-Sentry-Rate-Limits", "60:log_byte:key") + |> Plug.Conn.resp(200, ~s<{"id":"first"}>) + end) + + first = Envelope.from_event(Event.create_event(message: "First")) + assert {:ok, "first"} = Transport.encode_and_post_envelope(first, HackneyClient) + assert Transport.RateLimiter.rate_limited?("log_byte") + + log_envelope = Envelope.from_log_events([make_log_event("dropped")]) + + assert {:error, %ClientError{reason: :rate_limited}} = + Transport.encode_and_post_envelope(log_envelope, HackneyClient, _retries = []) + end + + test "drops metric envelopes when a trace_metric_byte rate limit is active", %{bypass: bypass} do + Bypass.expect_once(bypass, "POST", "/api/1/envelope/", fn conn -> + conn + |> Plug.Conn.put_resp_header("X-Sentry-Rate-Limits", "60:trace_metric_byte:key") + |> Plug.Conn.resp(200, ~s<{"id":"first"}>) + end) + + first = Envelope.from_event(Event.create_event(message: "First")) + assert {:ok, "first"} = Transport.encode_and_post_envelope(first, HackneyClient) + assert Transport.RateLimiter.rate_limited?("trace_metric_byte") + + metric_envelope = Envelope.from_metric_events([make_metric("dropped", 1)]) + + assert {:error, %ClientError{reason: :rate_limited}} = + Transport.encode_and_post_envelope(metric_envelope, HackneyClient, _retries = []) + end + end + + defp make_log_event(body) do + %Sentry.LogEvent{ + timestamp: System.system_time(:nanosecond) / 1_000_000_000, + level: :info, + body: body + } + end + + defp make_metric(name, value) do + %Sentry.Metric{ + type: :counter, + name: name, + value: value, + timestamp: System.system_time(:nanosecond) / 1_000_000_000, + attributes: %{} + } end defp error(fun) do diff --git a/test/support/reporting_json_library.ex b/test/support/reporting_json_library.ex new file mode 100644 index 000000000..d0ca7b97c --- /dev/null +++ b/test/support/reporting_json_library.ex @@ -0,0 +1,41 @@ +defmodule ReportingJSONLibrary do + @moduledoc false + + # A `:json_library` drop-in that reports which process performed an encode. + # + # Used to assert that expensive work (like computing byte-based client report + # outcomes) happens off the caller's process. Delegates to the real library so + # behaviour is otherwise unchanged. + + @key {__MODULE__, :report_to} + + @doc """ + Sends `{:encoded, encoding_pid}` to `pid` on every non-trivial encode. + + Empty maps are ignored so that `Sentry.Config`'s own validation encode (which + runs on whichever process calls `put_test_config/1`) isn't reported. + """ + @spec report_to(pid()) :: :ok + def report_to(pid) when is_pid(pid) do + :persistent_term.put(@key, pid) + end + + def encode(data) do + if data != %{} do + case :persistent_term.get(@key, nil) do + nil -> :ok + pid -> send(pid, {:encoded, self()}) + end + end + + Sentry.JSON.encode(data, default_library()) + end + + def decode(binary) do + Sentry.JSON.decode(binary, default_library()) + end + + defp default_library do + if Code.ensure_loaded?(JSON), do: JSON, else: Jason + end +end