Class: NewRelic::Agent::InfiniteTracing::StreamingBuffer

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Enumerable, Constants
Defined in:
lib/infinite_tracing/streaming_buffer.rb

Constant Summary collapse

DEFAULT_QUEUE_SIZE = 
10_000
FLUSH_DELAY = 
0.005
MAX_FLUSH_WAIT =

three seconds

3

Constants included from Constants

Constants::GRPC_ERROR_NAME_METRIC, Constants::GRPC_OTHER_ERROR_METRIC, Constants::QUEUE_DUMPED_METRIC, Constants::RESPONSE_ERROR_METRIC, Constants::SPANS_SEEN_METRIC, Constants::SPANS_SENT_METRIC, Constants::SUPPORTABILITY_PREFIX

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_size = DEFAULT_QUEUE_SIZE) ⇒ StreamingBuffer

Returns a new instance of StreamingBuffer.



24
25
26
27
28
29
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 24

def initialize(max_size = DEFAULT_QUEUE_SIZE)
  @max_size = max_size
  @lock = Mutex.new
  @queue = Queue.new
  @batch = Array.new
end

Instance Attribute Details

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



22
23
24
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 22

def queue
  @queue
end

Instance Method Details

#<<(segment) ⇒ Object

Pushes the segment given onto the queue.

If the queue is at capacity, it is dumped and a supportability metric is recorded for the event.

When a restart signal is received, the queue is locked with a mutex, blocking the push until the queue has restarted.



48
49
50
51
52
53
54
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 48

def <<(segment)
  @lock.synchronize do
    clear_queue if @queue.size >= @max_size
    NewRelic::Agent.increment_metric(SPANS_SEEN_METRIC)
    @queue.push(segment)
  end
end

#batch_enumerator ⇒ Object

Returns the blocking enumerator that will pop items off the queue while any items are present

yielding is deferred until batch_size spans is reached.

If nil is popped, the queue is closing. A final yield on non-empty batch is fired.

The segment is transformed into a serializable span here so processing is taking place within the gRPC call's thread rather than in the main application thread.



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 119

def batch_enumerator
  return enum_for(:enumerator) unless block_given?
  loop do
    if proc_or_segment = @queue.pop(false)
      NewRelic::Agent.increment_metric(SPANS_SENT_METRIC)
      @batch << transform(proc_or_segment)
      if @batch.size >= BATCH_SIZE
        yield(SpanBatch.new(spans: @batch))
        @batch.clear
      end

    else
      yield(SpanBatch.new(spans: @batch)) unless @batch.empty?
      raise ClosedQueueError
    end
  end
end

#clear_queue ⇒ Object

Drops all segments from the queue and records a supportability metric for the event.



58
59
60
61
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 58

def clear_queue
  @queue.clear
  NewRelic::Agent.increment_metric(QUEUE_DUMPED_METRIC)
end

#close_queue ⇒ Object 



81
82
83
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 81

def close_queue
  @lock.synchronize { @queue.close }
end

#enumerator ⇒ Object

Returns the blocking enumerator that will pop items off the queue while any items are present If nil is popped, the queue is closing.

The segment is transformed into a serializable span here so processing is taking place within the gRPC call's thread rather than in the main application thread.



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 93

def enumerator
  return enum_for(:enumerator) unless block_given?
  loop do
    if segment = @queue.pop(false)
      NewRelic::Agent.increment_metric(SPANS_SENT_METRIC)
      yield(transform(segment))

    else
      raise ClosedQueueError
    end
  end
end

#flush_queue ⇒ Object

Waits for the queue to be fully consumed or for the waiting consumers to release.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 65

def flush_queue
  @queue.num_waiting.times { @queue.push(nil) }
  close_queue

  # Logs if we're throwing away spans because nothing's
  # waiting to take them off the queue.
  if @queue.num_waiting == 0 && !@queue.empty?
    NewRelic::Agent.logger.warn("Discarding #{@queue.size} segments on Streaming Buffer")
    return
  end

  # Only wait a short while for queue to flush
  cutoff = Process.clock_gettime(Process::CLOCK_MONOTONIC) + MAX_FLUSH_WAIT
  until @queue.empty? || Process.clock_gettime(Process::CLOCK_MONOTONIC) >= cutoff do sleep(FLUSH_DELAY) end
end

#transfer(new_buffer) ⇒ Object

Dumps the contents of this streaming buffer onto the given buffer and closes the queue



33
34
35
36
37
38
 
# File 'lib/infinite_tracing/streaming_buffer.rb', line 33

def transfer(new_buffer)
  @lock.synchronize do
    until @queue.empty? do new_buffer.push(@queue.pop) end
    @queue.close
  end
end