fix(metrics, logs): Ensure TelemetryBuffer uses own thread for both size based and periodic flushing

sentry-ruby/lib/sentry/telemetry_event_buffer.rb

@@ -41,30 +41,26 @@ def initialize(configuration, client, event_class:, max_items:, max_items_before
     end
 
     def flush
-      pending_items = @mutex.synchronize do
-        next if @pending_items.empty?
-
-        items = @pending_items
-        @pending_items = []
-        items
-      end
-
-      return unless pending_items
+      wait_until_idle
+      result = flush_pending_items
+      wait_until_idle
+      result
+    end
 
-      log_debug("[#{self.class}] flushing #{pending_items.size} #{@event_class}")
-      send_items(pending_items)
-      self
+    def run
+      flush_pending_items
     end
-    alias_method :run, :flush
 
     def add_item(item)
+      # the buffer thread can never add telemetry itself to prevent recursion
+      return self if Thread.current == thread
       # Prevent ThreadError from re-entrant locking (e.g. transport instrumentation calling Sentry.metrics.*)
       return self if @mutex.owned?
 
+      return unless ensure_thread
+
       dropped = false
       size_exceeded = @mutex.synchronize do
-        return unless ensure_thread
-
         if size >= @max_items_before_drop
           dropped = true
         else
@@ -83,7 +79,7 @@ def add_item(item)
         )
       end
 
-      flush if size_exceeded
+      wake if size_exceeded
       self
     end
 
@@ -101,6 +97,22 @@ def clear!
 
     private
 
+    def flush_pending_items
+      pending_items = @mutex.synchronize do
+        next if @pending_items.empty?
+
+        items = @pending_items
+        @pending_items = []
+        items
+      end
+
+      return unless pending_items
+
+      log_debug("[#{self.class}] flushing #{pending_items.size} #{@event_class}")
+      send_items(pending_items)
+      self
+    end
+
     def send_items(pending_items)
       envelope = Envelope.new(sent_at: Sentry.utc_now.iso8601)
 

sentry-ruby/lib/sentry/test_helper.rb

@@ -127,13 +127,17 @@ def sentry_envelopes
     end
 
     def sentry_logs
+      Sentry.get_current_client&.log_event_buffer&.flush
+
       sentry_envelopes
         .flat_map(&:items)
         .select { |item| item.headers[:type] == "log" }
         .flat_map { |item| item.payload[:items] }
     end
 
     def sentry_metrics
+      Sentry.get_current_client&.metric_event_buffer&.flush
+
       sentry_envelopes
         .flat_map(&:items)
         .select { |item| item.headers[:type] == "trace_metric" }

sentry-ruby/lib/sentry/threaded_periodic_worker.rb

@@ -4,41 +4,91 @@ module Sentry
   class ThreadedPeriodicWorker
     include LoggingHelper
 
+    attr_reader :thread
+
     def initialize(sdk_logger, interval)
       @thread = nil
       @exited = false
       @interval = interval
       @sdk_logger = sdk_logger
+
+      @thread_mutex = Mutex.new
+      @wake_condition = ConditionVariable.new
+      @idle_condition = ConditionVariable.new
+
+      @woken = false
+      @running = false
     end
 
     def ensure_thread
-      return false if @exited
-      return true if @thread&.alive?
+      @thread_mutex.synchronize do
+        return false if @exited
+        return true if @thread&.alive?
 
-      @thread = Thread.new do
-        loop do
-          sleep(@interval)
-          run
-        end
-      end
+        @thread = Thread.new { worker_loop }
 
-      true
+        true
+      end
     rescue ThreadError
+      @thread_mutex.synchronize { @exited = true }
       log_debug("[#{self.class.name}] thread creation failed")
-      @exited = true
       false
     end
 
+    def wake
+      @thread_mutex.synchronize do
+        next false if @exited
+
+        @woken = true
+        @wake_condition.signal
+        true
+      end
+    end
+
+    def wait_until_idle
+      @thread_mutex.synchronize do
+        @idle_condition.wait(@thread_mutex) while !@exited && (@running || @woken)
+      end
+    end
+
     def kill
-      @exited = true
+      thread = @thread_mutex.synchronize do
+        @exited = true
+        @woken = false
+        @idle_condition.broadcast
+        @thread
+      end
 
       # Only a started worker has a thread to kill (and to log about).
       # Guarding here keeps a never-started worker's teardown silent, so
       # killing one during test reset can't emit a stray debug line.
-      return unless @thread
+      return unless thread
 
       log_debug("[#{self.class.name}] thread killed")
-      @thread.kill
+      thread.kill
+    end
+
+    private
+
+    def worker_loop
+      loop do
+        @thread_mutex.synchronize do
+          @wake_condition.wait(@thread_mutex, @interval) unless @woken
+          @woken = false
+          @running = true
+        end
+
+        begin
+          run
+        rescue Exception => e
+          log_error("[#{self.class.name}] run failed", e)
+        ensure
+          @thread_mutex.synchronize do
+            @running = false
+            @idle_condition.broadcast
+          end
+        end
+      end
     end
   end
 end

sentry-ruby/spec/isolated/sentry_logger_spec.rb

@@ -7,7 +7,6 @@
     perform_basic_setup do |config|
       config.breadcrumbs_logger = [:sentry_logger]
       config.max_log_events = 1
-      config.enabled_patches = [:logger]
     end
   end
 
@@ -104,22 +103,4 @@
       end
     end
   end
-
-  it "does not conflict with :logger patch" do
-    logger = ::Logger.new(nil)
-
-    logger.info("Hello World")
-
-    expect(sentry_logs).to_not be_empty
-
-    log_event = sentry_logs.last
-
-    expect(log_event[:level]).to eql("info")
-    expect(log_event[:body]).to eql("Hello World")
-
-    breadcrumb = breadcrumbs.peek
-
-    expect(breadcrumb.level).to eq("info")
-    expect(breadcrumb.message).to eq("Hello World")
-  end
 end

sentry-ruby/spec/isolated/std_lib_logger_spec.rb

@@ -2,7 +2,7 @@
 
 SimpleCov.command_name "StdLibLogger"
 
-RSpec.describe Sentry::StdLibLogger do
+RSpec.describe Sentry::StdLibLogger, order: :defined do
   let(:logger) { ::Logger.new($stdout) }
 
   context "when logger patch is not enabled" do
@@ -224,5 +224,23 @@
         end
       end
     end
+
+    it "does not conflict with the Sentry logger patch" do
+      Sentry.configuration.breadcrumbs_logger = [:sentry_logger]
+      logger = ::Logger.new(nil)
+      logger.info("Hello World")
+
+      expect(sentry_logs).to_not be_empty
+
+      log_event = sentry_logs.last
+
+      expect(log_event[:level]).to eql("info")
+      expect(log_event[:body]).to eql("Hello World")
+
+      breadcrumb = Sentry.get_current_scope.breadcrumbs.peek
+
+      expect(breadcrumb.level).to eq("info")
+      expect(breadcrumb.message).to eq("Hello World")
+    end
   end
 end

sentry-ruby/spec/sentry/backpressure_monitor_spec.rb

@@ -30,12 +30,12 @@
 
     it 'spawns new thread' do
       expect { subject.healthy? }.to change { Thread.list.count }.by(1)
-      expect(subject.instance_variable_get(:@thread)).to be_a(Thread)
+      expect(subject.thread).to be_a(Thread)
     end
 
     it 'spawns only one thread' do
       expect { subject.healthy? }.to change { Thread.list.count }.by(1)
-      thread = subject.instance_variable_get(:@thread)
+      thread = subject.thread
       expect(thread).to receive(:alive?).and_return(true)
       expect { subject.healthy? }.to change { Thread.list.count }.by(0)
     end
@@ -109,7 +109,7 @@
   describe '#kill' do
     it 'kills the thread and logs a message' do
       subject.healthy?
-      expect(subject.instance_variable_get(:@thread)).to receive(:kill)
+      expect(subject.thread).to receive(:kill)
       subject.kill
       expect(string_io.string).to include("[#{described_class.name}] thread killed")
     end

sentry-ruby/spec/sentry/session_flusher_spec.rb

@@ -105,15 +105,15 @@
         subject.add_session(session)
       end.to change { Thread.list.count }.by(1)
 
-      expect(subject.instance_variable_get(:@thread)).to be_a(Thread)
+      expect(subject.thread).to be_a(Thread)
     end
 
     it "spawns only one thread" do
       expect do
         subject.add_session(session)
       end.to change { Thread.list.count }.by(1)
 
-      thread = subject.instance_variable_get(:@thread)
+      thread = subject.thread
       expect(thread).to receive(:alive?).and_return(true)
 
       expect do