Max drop limit for TelemetryEventBuffer after which we completely drop events (#2831)

* Have a higher max limit for TelemetryEventBuffer after which we
completely drop events

* This is mainly to protect from memory blowup in high throughput
  scenarios
* Also raised the Metric buffer limits to 1k/10k as compared to 100/1k
  for logs (consistent with Python)

* Update sentry-ruby/spec/sentry/configuration_spec.rb

Co-authored-by: Peter Solnica <peter@solnica.online>

---------

Co-authored-by: Peter Solnica <peter@solnica.online>

Author: sl0thentr0py

CHANGELOG.md

@@ -60,6 +60,7 @@
 
 - Unify Logs and Metrics implementations ([#2826](https://github.com/getsentry/sentry-ruby/pull/2826))
 - Unify LogEventBuffer and MetricEventBuffer logic ([#2830](https://github.com/getsentry/sentry-ruby/pull/2830))
+- Add maximum limits on LogEventBuffer (1k) and MetricEventBuffer (10k) for protection from memory blowup ([#2831](https://github.com/getsentry/sentry-ruby/pull/2831))
 
 ## 6.2.0
 

sentry-ruby/lib/sentry/log_event_buffer.rb

@@ -10,13 +10,15 @@ module Sentry
   # @!visibility private
   class LogEventBuffer < TelemetryEventBuffer
     DEFAULT_MAX_EVENTS = 100
+    MAX_EVENTS_BEFORE_DROP = 1000
 
     def initialize(configuration, client)
       super(
         configuration,
         client,
         event_class: LogEvent,
         max_items: configuration.max_log_events || DEFAULT_MAX_EVENTS,
+        max_items_before_drop: MAX_EVENTS_BEFORE_DROP,
         envelope_type: "log",
         envelope_content_type: "application/vnd.sentry.items.log+json",
         before_send: configuration.before_send_log

sentry-ruby/lib/sentry/metric_event_buffer.rb

@@ -9,14 +9,16 @@ module Sentry
   #
   # @!visibility private
   class MetricEventBuffer < TelemetryEventBuffer
-    DEFAULT_MAX_METRICS = 100
+    DEFAULT_MAX_METRICS = 1000
+    MAX_METRICS_BEFORE_DROP = 10_000
 
     def initialize(configuration, client)
       super(
         configuration,
         client,
         event_class: MetricEvent,
         max_items: configuration.max_metric_events || DEFAULT_MAX_METRICS,
+        max_items_before_drop: MAX_METRICS_BEFORE_DROP,
         envelope_type: "trace_metric",
         envelope_content_type: "application/vnd.sentry.items.trace-metric+json",
         before_send: configuration.before_send_metric

sentry-ruby/lib/sentry/telemetry_event_buffer.rb

@@ -14,17 +14,19 @@ class TelemetryEventBuffer < ThreadedPeriodicWorker
     FLUSH_INTERVAL = 5 # seconds
 
     # @!visibility private
-    attr_reader :pending_items
+    attr_reader :pending_items, :envelope_type, :data_category
 
-    def initialize(configuration, client, event_class:, max_items:, envelope_type:, envelope_content_type:, before_send:)
+    def initialize(configuration, client, event_class:, max_items:, max_items_before_drop:, envelope_type:, envelope_content_type:, before_send:)
       super(configuration.sdk_logger, FLUSH_INTERVAL)
 
       @client = client
       @dsn = configuration.dsn
       @debug = configuration.debug
       @event_class = event_class
       @max_items = max_items
+      @max_items_before_drop = max_items_before_drop
       @envelope_type = envelope_type
+      @data_category = Sentry::Envelope::Item.data_category(@envelope_type)
       @envelope_content_type = envelope_content_type
       @before_send = before_send
 
@@ -53,10 +55,19 @@ def flush
     alias_method :run, :flush
 
     def add_item(item)
-      raise ArgumentError, "expected a #{@event_class}, got #{item.class}" unless item.is_a?(@event_class)
+      unless item.is_a?(@event_class)
+        log_debug("[#{self.class}] expected a #{@event_class}, got #{item.class}")
+        return
+      end
 
       @mutex.synchronize do
-        @pending_items << item
+        if size >= @max_items_before_drop
+          log_debug("[#{self.class}] exceeded max capacity, dropping event")
+          @client.transport.record_lost_event(:queue_overflow, @data_category)
+        else
+          @pending_items << item
+        end
+
         send_items if size >= @max_items
       end
 
@@ -103,7 +114,6 @@ def send_items
       end
 
       unless discarded_count.zero?
-        @data_category = Sentry::Envelope::Item.data_category(@envelope_type)
         @client.transport.record_lost_event(:before_send, @data_category, num: discarded_count)
       end
 

sentry-ruby/spec/sentry/configuration_spec.rb

@@ -770,8 +770,8 @@ class SentryConfigurationSample < Sentry::Configuration
   end
 
   describe "#max_metric_events" do
-    it "returns 100 by default" do
-      expect(subject.max_metric_events).to eq(100)
+    it "returns 1000 by default" do
+      expect(subject.max_metric_events).to eq(1000)
     end
 
     it "can be set to an integer value" do

sentry-ruby/spec/support/shared_examples_for_telemetry_event_buffers.rb

@@ -2,13 +2,13 @@
 
 RSpec.shared_examples "telemetry event buffer" do |event_factory:, max_items_config:, enable_config:|
   let(:string_io) { StringIO.new }
-  let(:logger) { ::Logger.new(string_io) }
-  let(:client) { double(Sentry::Client) }
+  let(:sdk_logger) { ::Logger.new(string_io) }
+  let(:client) { Sentry.get_current_client }
   let(:event) { event_factory.call }
 
   before do
     perform_basic_setup do |config|
-      config.sdk_logger = logger
+      config.sdk_logger = sdk_logger
       config.background_worker_threads = 0
       config.public_send(:"#{max_items_config}=", max_items)
       config.public_send(:"#{enable_config}=", true)
@@ -67,6 +67,49 @@
     end
   end
 
+  describe "max capacity and dropping events" do
+    let(:max_items) { 3 }
+    let(:max_items_before_drop) { 10 }
+
+    before do
+      subject.instance_variable_set(:@max_items_before_drop, max_items_before_drop)
+
+      # don't clear pending items to allow buffer to grow
+      allow(subject).to receive(:clear!)
+    end
+
+    it "adds items up to max_items_before_drop capacity" do
+      expect {
+        max_items_before_drop.times { subject.add_item(event) }
+      }.to change { subject.size }.from(0).to(max_items_before_drop)
+    end
+
+    it "drops events when buffer reaches max_items_before_drop" do
+      max_items_before_drop.times { subject.add_item(event) }
+
+      expect {
+        subject.add_item(event)
+      }.not_to change { subject.size }
+
+      expect(subject.size).to eq(max_items_before_drop)
+    end
+
+    it "records lost event when dropping due to queue overflow" do
+      max_items_before_drop.times { subject.add_item(event) }
+
+      expect(client.transport).to receive(:record_lost_event).with(:queue_overflow, subject.data_category)
+
+      subject.add_item(event)
+    end
+
+    it "logs debug message when dropping events" do
+      max_items_before_drop.times { subject.add_item(event) }
+      subject.add_item(event)
+
+      expect(string_io.string).to include("exceeded max capacity, dropping event")
+    end
+  end
+
   describe "error handling" do
     let(:max_items) { 3 }