Introduce log event buffering

sentry-ruby/lib/sentry/client.rb

@@ -2,6 +2,7 @@
 
 require "sentry/transport"
 require "sentry/log_event"
+require "sentry/log_event_buffer"
 
 module Sentry
   class Client
@@ -39,6 +40,8 @@ def initialize(configuration)
       end
 
       @spotlight_transport = SpotlightTransport.new(configuration) if configuration.spotlight
+
+      @log_event_buffer = LogEventBuffer.new(configuration, self).start
     end
 
     # Applies the given scope's data to the event and sends it to Sentry.
@@ -89,6 +92,15 @@ def capture_event(event, scope, hint = {})
       nil
     end
 
+    # Buffer a log event to be sent later with other logs in a single envelope
+    # @param event [LogEvent] the log event to be buffered
+    # @return [LogEvent]
+    def buffer_log_event(event, scope)
+      return unless event.is_a?(LogEvent)
+      @log_event_buffer.add_event(scope.apply_to_event(event))
+      event
+    end
+
     # Capture an envelope directly.
     # @param envelope [Envelope] the envelope to be captured.
     # @return [void]
@@ -101,6 +113,7 @@ def capture_envelope(envelope)
     def flush
       transport.flush if configuration.sending_to_dsn_allowed?
       spotlight_transport.flush if spotlight_transport
+      @log_event_buffer.flush
     end
 
     # Initializes an Event object with the given exception. Returns `nil` if the exception's class is excluded from reporting.

sentry-ruby/lib/sentry/configuration.rb

@@ -12,6 +12,7 @@
 require "sentry/metrics/configuration"
 require "sentry/linecache"
 require "sentry/interfaces/stacktrace_builder"
+require "sentry/log_event_buffer"
 
 module Sentry
   class Configuration
@@ -307,6 +308,10 @@ def capture_exception_frame_locals=(value)
     # @return [Array<Symbol>]
     attr_accessor :enabled_patches
 
+    # Maximum number of log events to buffer before sending
+    # @return [Integer]
+    attr_accessor :max_log_events
+
     # these are not config options
     # @!visibility private
     attr_reader :errors, :gem_specs
@@ -455,6 +460,8 @@ def initialize
       @gem_specs = Hash[Gem::Specification.map { |spec| [spec.name, spec.version.to_s] }] if Gem::Specification.respond_to?(:map)
 
       run_post_initialization_callbacks
+
+      self.max_log_events = LogEventBuffer::DEFAULT_MAX_EVENTS
     end
 
     def validate

sentry-ruby/lib/sentry/hub.rb

@@ -223,7 +223,7 @@ def capture_log_event(message, **options)
 
       return unless event
 
-      capture_event(event, **options)
+      current_client.buffer_log_event(event, current_scope)
     end
 
     def capture_event(event, **options, &block)

sentry-ruby/lib/sentry/log_event_buffer.rb

@@ -0,0 +1,84 @@
+# frozen_string_literal: true
+
+require "sentry/threaded_periodic_worker"
+
+module Sentry
+  class LogEventBuffer < ThreadedPeriodicWorker
+    FLUSH_INTERVAL = 5 # seconds
+    DEFAULT_MAX_EVENTS = 100
+
+    def initialize(configuration, client)
+      super(configuration.logger, FLUSH_INTERVAL)
+      @client = client
+      @pending_events = []
+      @max_events = configuration.max_log_events || DEFAULT_MAX_EVENTS
+      @dsn = configuration.dsn
+      @sdk = Sentry.sdk_meta
+      @mutex = Mutex.new
+
+      log_debug("[LogEvents] Initialized buffer with max_events=#{@max_events}, flush_interval=#{FLUSH_INTERVAL}s")
+    end
+
+    def start
+      ensure_thread
+      self
+    end
+
+    def flush
+      @mutex.synchronize do
+        return unless size >= @max_events
+
+        log_debug("[LogEventBuffer] flushing #{size} log events")
+
+        @client.send_envelope(to_envelope)
+
+        @pending_events.clear
+      end
+
+      log_debug("[LogEventBuffer] flushed #{size} log events")
+
+      self
+    end
+    alias_method :run, :flush
+
+    def add_event(event)
+      raise ArgumentError, "expected a LogEvent, got #{event.class}" unless event.is_a?(LogEvent)
+
+      @mutex.synchronize do
+        @pending_events << event
+      end
+
+      self
+    end
+
+    def empty?
+      @pending_events.empty?
+    end
+
+    def size
+      @pending_events.size
+    end
+
+    private
+
+    def to_envelope
+      envelope = Envelope.new(
+        event_id: SecureRandom.uuid.delete("-"),
+        sent_at: Sentry.utc_now.iso8601,
+        dsn: @dsn,
+        sdk: @sdk
+      )
+
+      envelope.add_item(
+        {
+          type: "log",
+          item_count: size,
+          content_type: "application/vnd.sentry.items.log+json"
+        },
+        { items: @pending_events.map(&:to_hash) }
+      )
+
+      envelope
+    end
+  end
+end

sentry-ruby/lib/sentry/test_helper.rb

@@ -72,6 +72,13 @@ def sentry_envelopes
       sentry_transport.envelopes
     end
 
+    def sentry_logs
+      sentry_envelopes
+        .flat_map(&:items)
+        .select { |item| item.headers[:type] == "log" }
+        .flat_map { |item| item.payload[:items] }
+    end
+
     # Returns the last captured event object.
     # @return [Event, nil]
     def last_sentry_event

sentry-ruby/lib/sentry/threaded_periodic_worker.rb

@@ -4,10 +4,10 @@ module Sentry
   class ThreadedPeriodicWorker
     include LoggingHelper
 
-    def initialize(logger, internal)
+    def initialize(logger, interval)
       @thread = nil
       @exited = false
-      @interval = internal
+      @interval = interval
       @logger = logger
     end
 

sentry-ruby/spec/sentry/log_event_buffer_spec.rb

@@ -0,0 +1,81 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+RSpec.describe Sentry::LogEventBuffer do
+  subject(:log_event_buffer) { described_class.new(Sentry.configuration, client) }
+
+  let(:string_io) { StringIO.new }
+  let(:logger) { ::Logger.new(string_io) }
+
+  before do
+    perform_basic_setup do |config|
+      config.logger = logger
+      config.background_worker_threads = 0
+      config.max_log_events = 3
+    end
+
+    Sentry.background_worker = Sentry::BackgroundWorker.new(Sentry.configuration)
+  end
+
+  after do
+    Sentry.background_worker = Class.new { def shutdown; end; }.new
+  end
+
+  let(:client) { Sentry.get_current_client }
+  let(:transport) { client.transport }
+
+  describe "#add_event" do
+    let(:log_event) do
+      Sentry::LogEvent.new(
+        configuration: Sentry.configuration,
+        level: :info,
+        body: "Test message"
+      )
+    end
+
+    it "does nothing when there are no pending events" do
+      expect(client).not_to receive(:capture_envelope)
+
+      log_event_buffer.flush
+
+      expect(sentry_envelopes.size).to be(0)
+    end
+
+    it "does nothing when the number of events is less than max_events " do
+      2.times { log_event_buffer.add_event(log_event) }
+
+      log_event_buffer.flush
+
+      expect(sentry_envelopes.size).to be(0)
+    end
+
+    it "sends pending events to the client" do
+      3.times { log_event_buffer.add_event(log_event) }
+
+      log_event_buffer.flush
+
+      expect(sentry_envelopes.size).to be(1)
+
+      expect(log_event_buffer).to be_empty
+    end
+
+    it "thread-safely handles concurrent access", skip: RUBY_ENGINE == "jruby" do
+      expect(client).to receive(:send_envelope) do |_envelope|
+        sleep 0.1
+      end
+
+      threads = 100.times.map do
+        (1..50).to_a.sample.times { log_event_buffer.add_event(log_event) }
+
+        Thread.new do
+          log_event_buffer.flush
+        end
+      end
+
+      threads.each(&:join)
+
+      expect(log_event_buffer).to be_empty
+    end
+  end
+end

sentry-ruby/spec/sentry/session_flusher_spec.rb

@@ -105,15 +105,15 @@
     it "spawns new thread" do
       expect do
         subject.add_session(session)
-      end.to change { Thread.list.count }.by(1)
+      end.to change { Thread.list.count }.by(2)
 
       expect(subject.instance_variable_get(:@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)
+      end.to change { Thread.list.count }.by(2)
 
       thread = subject.instance_variable_get(:@thread)
       expect(thread).to receive(:alive?).and_return(true)
@@ -132,7 +132,7 @@
 
     context "when thread creation fails" do
       before do
-        expect(Thread).to receive(:new).and_raise(ThreadError)
+        allow(Thread).to receive(:new).and_raise(ThreadError)
       end
 
       it "doesn't create new thread" do

sentry-ruby/spec/sentry_spec.rb

@@ -356,6 +356,7 @@
     before do
       perform_basic_setup do |config|
         config.traces_sample_rate = 1.0
+        config.max_log_events = 1
       end
     end
 
@@ -364,17 +365,17 @@
         Sentry.with_scope do |scope|
           described_class.capture_log("Test", level: :info, tags: { foo: "baz" })
         end
-      end.to change { sentry_events.count }.by(1)
+      end.to_not change { sentry_events.count }
+
+      Sentry.get_current_client.flush
 
-      log_event = sentry_events.first
+      expect(sentry_envelopes.count).to be(1)
 
-      expect(log_event.type).to eql("log")
-      expect(log_event.level).to eq(:info)
-      expect(log_event.attributes).to eql({ tags: { foo: "baz" } })
+      log_event = sentry_logs.first
 
-      hash = log_event.to_hash
-      expect(hash[:trace_id]).to_not be(nil)
-      expect(hash[:attributes]).to_not have_key("sentry.trace.parent_span_id")
+      expect(log_event[:level]).to eq("info")
+      expect(log_event[:trace_id]).to_not be(nil)
+      expect(log_event[:attributes]).to_not have_key("sentry.trace.parent_span_id")
     end
 
     it "sends a log event with parent_span_id" do
@@ -387,15 +388,15 @@
 
       transaction.finish
 
-      log_event = sentry_events.first
+      Sentry.get_current_client.flush
+
+      expect(sentry_envelopes.size).to be(2)
 
-      expect(log_event.type).to eql("log")
-      expect(log_event.level).to eq(:info)
-      expect(log_event.attributes).to eql({ tags: { foo: "baz" } })
+      log_event = sentry_logs.first
 
-      hash = log_event.to_hash
-      expect(hash[:trace_id]).to eq(transaction.trace_id)
-      expect(hash[:attributes]["sentry.trace.parent_span_id"]).to eql({ value: transaction.span_id, type: "string" })
+      expect(log_event[:level]).to eq("info")
+      expect(log_event[:trace_id]).to_not be(nil)
+      expect(log_event[:attributes]).to have_key("sentry.trace.parent_span_id")
     end
   end