Class: ActiveAgents::Telemetry::BatchingReporter

Inherits:
Reporter
  • Object
show all
Defined in:
lib/activeagents/telemetry/batching_reporter.rb

Overview

A Reporter that buffers traces and delivers them in batches — fewer HTTP requests under sustained traffic, at the cost of traces arriving up to flush_interval seconds late.

A full buffer (configuration.batch_size) flushes immediately; a background thread flushes whatever accumulated every configuration.flush_interval seconds. Call #shutdown before process exit or the tail of the buffer is lost.

Sampling happens on enqueue, so a dropped trace never occupies buffer space; enabled?/configured? are also checked on enqueue AND inherited from Reporter#report at delivery time.

Constant Summary

Constants inherited from Reporter

Reporter::SDK_NAME

Instance Attribute Summary

Attributes inherited from Reporter

#configuration

Instance Method Summary collapse

Methods inherited from Reporter

#report_now

Constructor Details

#initialize(configuration, **options) ⇒ BatchingReporter

Returns a new instance of BatchingReporter.



18
19
20
21
22
23
24
25
 
# File 'lib/activeagents/telemetry/batching_reporter.rb', line 18

def initialize(configuration, **options)
  super
  @buffer = []
  @mutex = Mutex.new
  @flusher = nil
  @send_threads = []
  @shutdown = false
end

Instance Method Details

#flush ⇒ Object

Delivers everything buffered, blocking until done.



46
47
48
49
50
 
# File 'lib/activeagents/telemetry/batching_reporter.rb', line 46

def flush
  batch = @mutex.synchronize { @buffer.slice!(0..) }
  deliver_batch(batch, blocking: true) unless batch.empty?
  nil
end

#report(traces) ⇒ Object

Enqueues a trace, flushing if the batch is full.



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/activeagents/telemetry/batching_reporter.rb', line 28

def report(traces)
  return if @shutdown

  accepted = normalize(traces).select { sample_trace? }
  return if accepted.empty?
  return unless configuration.enabled? && configuration.configured?

  batch = nil
  @mutex.synchronize do
    @buffer.concat(accepted)
    batch = @buffer.slice!(0..) if @buffer.size >= configuration.batch_size
    start_flusher
  end
  deliver_batch(batch) if batch
  nil
end

#shutdown ⇒ Object

Flushes, waits out in-flight sends, and stops the background thread — traces reported just before process exit are delivered, not dropped. Idempotent.



55
56
57
58
59
60
61
 
# File 'lib/activeagents/telemetry/batching_reporter.rb', line 55

def shutdown
  @shutdown = true
  flush
  @mutex.synchronize { @send_threads.dup }.each { |thread| thread.join(configuration.timeout) }
  @flusher&.kill
  @flusher = nil
end