Class: NewRelic::Agent::EventAggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/new_relic/agent/event_aggregator.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(events) ⇒ EventAggregator

Returns a new instance of EventAggregator.



51
52
53
54
55
56
57
58
59
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 51

def initialize(events)
  @lock = Mutex.new
  @buffer = self.class.buffer_class.new(NewRelic::Agent.config[self.class.capacity_key])
  @enabled = self.class.enabled_fn ? self.class.enabled_fn.call : false
  @notified_full = false
  register_capacity_callback
  register_enabled_callback(events)
  after_initialize
end

Class Method Details

.buffer_class(klass = nil) ⇒ Object 



42
43
44
45
46
47
48
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 42

def buffer_class(klass = nil)
  if klass
    @buffer_class = klass
  else
    @buffer_class ||= PrioritySampledBuffer
  end
end

.capacity_key(key = nil) ⇒ Object 



15
16
17
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 15

def capacity_key(key = nil)
  key ? @capacity_key = key : @capacity_key
end

.enabled_fn(fn = nil) ⇒ Object

This can be used instead of ‘enabled_key(s)` for more fine grained control over whether an aggregator should be enabled. The enabled fn will be reevaluated after configuration changes



38
39
40
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 38

def enabled_fn(fn = nil)
  fn ? @enabled_fn = fn : @enabled_fn
end

.enabled_keys(*keys) ⇒ Object Also known as: enabled_key

An aggregator can specify one or more keys to check to see if it is enabled. Multiple keys will be &&‘d and the enabled status of the aggregator will be reset when agent configuration changes.



23
24
25
26
27
28
29
30
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 23

def enabled_keys(*keys)
  if keys.empty?
    @enabled_keys ||= []
  else
    @enabled_keys = Array(keys)
    @enabled_fn = ->() { @enabled_keys.all? { |k| Agent.config[k] } }
  end
end

.named(named = nil) ⇒ Object 



11
12
13
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 11

def named(named = nil)
  named ? @named = named.to_s.freeze : @named
end

Instance Method Details

#after_harvest(metadata) ⇒ Object

interface method for subclasses to override to provide post harvest functionality



66
67
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 66

def after_harvest(metadata)
end

#after_initialize ⇒ Object

interface method for subclasses to override to provide post-initialization setup



62
63
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 62

def after_initialize
end

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


69
70
71
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 69

def enabled?
  @enabled
end

#harvest! ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 77

def harvest!
  metadata = nil
  samples = []
  @lock.synchronize do
    samples.concat(@buffer.to_a)
    metadata = @buffer.metadata
    reset_buffer!
  end
  after_harvest(metadata)
  [reservoir_metadata(metadata), samples]
end

#has_metadata? ⇒ Boolean

Returns:

  • (Boolean) 


73
74
75
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 73

def has_metadata?
  true
end

#merge!(payload, adjust_count = true) ⇒ Object

Merges samples from payload back into buffer and optionally adjusts the count of the buffer to ensure accuracy of buffer of metadata. We want to make sure not to double count samples being merged back in from a failed harvest, yet we do not want to under-count samples being merged from the PipeService.



93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 93

def merge!(payload, adjust_count = true)
  @lock.synchronize do
    _, samples = payload

    if adjust_count
      @buffer.decrement_lifetime_counts_by(samples.count)
    end

    samples.each { |s| @buffer.append(event: s) }
  end
end

#reset! ⇒ Object 



105
106
107
108
109
 
# File 'lib/new_relic/agent/event_aggregator.rb', line 105

def reset!
  @lock.synchronize do
    reset_buffer!
  end
end