Unify LogEventBuffer and MetricEventBuffer logic (#2830)

* Unify LogEventBuffer and MetricEventBuffer logic

* Apply suggestions from code review

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

* log_error

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

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

---------

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

Author: sl0thentr0py

CHANGELOG.md

@@ -59,6 +59,7 @@
 ### Internal
 
 - 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))
 
 ## 6.2.0
 

sentry-ruby/lib/sentry/client.rb

@@ -107,7 +107,7 @@ def capture_event(event, scope, hint = {})
     # @return [LogEvent]
     def buffer_log_event(event, scope)
       return unless event.is_a?(LogEvent)
-      @log_event_buffer.add_event(scope.apply_to_telemetry(event))
+      @log_event_buffer.add_item(scope.apply_to_telemetry(event))
       event
     end
 
@@ -117,7 +117,7 @@ def buffer_log_event(event, scope)
     def buffer_metric_event(event, scope)
       return unless event.is_a?(MetricEvent)
       event = scope.apply_to_telemetry(event)
-      @metric_event_buffer.add_metric(event)
+      @metric_event_buffer.add_item(event)
       event
     end
 
@@ -295,104 +295,6 @@ def send_event(event, hint = nil)
       raise
     end
 
-    # Send an envelope with batched logs
-    # @param log_events [Array<LogEvent>] the log events to be sent
-    # @api private
-    # @return [void]
-    def send_logs(log_events)
-      envelope = Envelope.new(
-        event_id: Sentry::Utils.uuid,
-        sent_at: Sentry.utc_now.iso8601,
-        dsn: configuration.dsn,
-        sdk: Sentry.sdk_meta
-      )
-
-      discarded_count = 0
-      envelope_items = []
-
-      if configuration.before_send_log
-        log_events.each do |log_event|
-          processed_log_event = configuration.before_send_log.call(log_event)
-
-          if processed_log_event
-            envelope_items << processed_log_event.to_h
-          else
-            discarded_count += 1
-          end
-        end
-
-        envelope_items
-      else
-        envelope_items = log_events.map(&:to_h)
-      end
-
-      envelope.add_item(
-        {
-          type: "log",
-          item_count: envelope_items.size,
-          content_type: "application/vnd.sentry.items.log+json"
-        },
-        { items: envelope_items }
-      )
-
-      send_envelope(envelope)
-
-      unless discarded_count.zero?
-        transport.record_lost_event(:before_send, "log_item", num: discarded_count)
-      end
-    end
-
-    # Send an envelope with batched metrics
-    # @param metrics [Array<MetricEvent>] the metrics to send
-    # @api private
-    # @return [void]
-    def send_metrics(metrics)
-      return if metrics.nil? || metrics.empty?
-
-      envelope = Envelope.new(
-        event_id: Sentry::Utils.uuid,
-        sent_at: Sentry.utc_now.iso8601,
-        dsn: configuration.dsn,
-        sdk: Sentry.sdk_meta
-      )
-
-      discarded_count = 0
-      envelope_items = []
-
-      if configuration.before_send_metric
-        metrics.each do |metric|
-          processed_metric = configuration.before_send_metric.call(metric)
-
-          if processed_metric
-            envelope_items << processed_metric.to_h
-          else
-            discarded_count += 1
-          end
-        end
-
-        envelope_items
-      else
-        envelope_items = metrics.map(&:to_h)
-      end
-
-      return if envelope_items.empty?
-
-      envelope.add_item(
-        {
-          type: "trace_metric",
-          item_count: envelope_items.size,
-          content_type: "application/vnd.sentry.items.trace-metric+json"
-        },
-        { items: envelope_items }
-      )
-
-      send_envelope(envelope)
-
-      unless discarded_count.zero?
-        transport.record_lost_event(:before_send, "metric", num: discarded_count)
-      end
-    end
-
     # Send an envelope directly to Sentry.
     # @param envelope [Envelope] the envelope to be sent.
     # @return [void]

sentry-ruby/lib/sentry/envelope/item.rb

@@ -15,7 +15,8 @@ class Envelope::Item
     # rate limits and client reports use the data_category rather than envelope item type
     def self.data_category(type)
       case type
-      when "session", "attachment", "transaction", "profile", "span", "log", "trace_metric" then type
+      when "session", "attachment", "transaction", "profile", "span", "trace_metric" then type
+      when "log" then "log_item"
       when "sessions" then "session"
       when "check_in" then "monitor"
       when "event" then "error"

sentry-ruby/lib/sentry/log_event_buffer.rb

@@ -1,82 +1,26 @@
 # frozen_string_literal: true
 
-require "sentry/threaded_periodic_worker"
+require "sentry/telemetry_event_buffer"
 
 module Sentry
   # LogEventBuffer buffers log events and sends them to Sentry in a single envelope.
   #
   # This is used internally by the `Sentry::Client`.
   #
   # @!visibility private
-  class LogEventBuffer < ThreadedPeriodicWorker
-    FLUSH_INTERVAL = 5 # seconds
+  class LogEventBuffer < TelemetryEventBuffer
     DEFAULT_MAX_EVENTS = 100
 
-    # @!visibility private
-    attr_reader :pending_events
-
     def initialize(configuration, client)
-      super(configuration.sdk_logger, FLUSH_INTERVAL)
-
-      @client = client
-      @pending_events = []
-      @max_events = configuration.max_log_events || DEFAULT_MAX_EVENTS
-      @mutex = Mutex.new
-
-      log_debug("[Logging] 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 if empty?
-
-        log_debug("[LogEventBuffer] flushing #{size} log events")
-
-        send_events
-      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
-        send_events if size >= @max_events
-      end
-
-      self
-    end
-
-    def empty?
-      @pending_events.empty?
-    end
-
-    def size
-      @pending_events.size
-    end
-
-    def clear!
-      @pending_events.clear
-    end
-
-    private
-
-    def send_events
-      @client.send_logs(@pending_events)
-    rescue => e
-      log_debug("[LogEventBuffer] Failed to send logs: #{e.message}")
-    ensure
-      clear!
+      super(
+        configuration,
+        client,
+        event_class: LogEvent,
+        max_items: configuration.max_log_events || DEFAULT_MAX_EVENTS,
+        envelope_type: "log",
+        envelope_content_type: "application/vnd.sentry.items.log+json",
+        before_send: configuration.before_send_log
+      )
     end
   end
 end

sentry-ruby/lib/sentry/metric_event_buffer.rb

@@ -1,78 +1,26 @@
 # frozen_string_literal: true
 
-require "sentry/threaded_periodic_worker"
+require "sentry/telemetry_event_buffer"
 
 module Sentry
   # MetricEventBuffer buffers metric events and sends them to Sentry in a single envelope.
   #
   # This is used internally by the `Sentry::Client`.
   #
   # @!visibility private
-  class MetricEventBuffer < ThreadedPeriodicWorker
-    FLUSH_INTERVAL = 5 # seconds
+  class MetricEventBuffer < TelemetryEventBuffer
     DEFAULT_MAX_METRICS = 100
 
-    # @!visibility private
-    attr_reader :pending_metrics
-
     def initialize(configuration, client)
-      super(configuration.sdk_logger, FLUSH_INTERVAL)
-
-      @client = client
-      @pending_metrics = []
-      @max_metrics = configuration.max_metric_events || DEFAULT_MAX_METRICS
-      @mutex = Mutex.new
-
-      log_debug("[Metrics] Initialized buffer with max_metrics=#{@max_metrics}, flush_interval=#{FLUSH_INTERVAL}s")
-    end
-
-    def start
-      ensure_thread
-      self
-    end
-
-    def flush
-      @mutex.synchronize do
-        return if empty?
-
-        log_debug("[MetricEventBuffer] flushing #{size} metrics")
-
-        send_metrics
-      end
-    end
-    alias_method :run, :flush
-
-    def add_metric(metric)
-      raise ArgumentError, "expected a MetricEvent, got #{metric.class}" unless metric.is_a?(MetricEvent)
-
-      @mutex.synchronize do
-        @pending_metrics << metric
-        send_metrics if size >= @max_metrics
-      end
-
-      self
-    end
-
-    def empty?
-      @pending_metrics.empty?
-    end
-
-    def size
-      @pending_metrics.size
-    end
-
-    def clear!
-      @pending_metrics.clear
-    end
-
-    private
-
-    def send_metrics
-      @client.send_metrics(@pending_metrics)
-    rescue => e
-      log_debug("[MetricEventBuffer] Failed to send metrics: #{e.message}")
-    ensure
-      clear!
+      super(
+        configuration,
+        client,
+        event_class: MetricEvent,
+        max_items: configuration.max_metric_events || DEFAULT_MAX_METRICS,
+        envelope_type: "trace_metric",
+        envelope_content_type: "application/vnd.sentry.items.trace-metric+json",
+        before_send: configuration.before_send_metric
+      )
     end
   end
 end