Class: Chronos::Application::ApmAggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/chronos/application/apm_aggregator.rb

Overview

Aggregates bounded request, query, and job observations into metric batches.

Examples:

aggregator.record("request", payload, context)
batches = aggregator.flush

Constant Summary collapse

METRIC_TYPES = 
%w(request query job).freeze

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ ApmAggregator

Returns a new instance of ApmAggregator.



18
19
20
21
22
23
24
25
 
# File 'lib/chronos/application/apm_aggregator.rb', line 18

def initialize(config)
  @config = config
  @mutex = Mutex.new
  @groups = {}
  @transactions = {}
  @observations = 0
  @dropped_groups = 0
end

Instance Method Details

#diagnostics ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/chronos/application/apm_aggregator.rb', line 49

def diagnostics
  @mutex.synchronize do
    {
      "groups" => @groups.length, "dropped_groups" => @dropped_groups,
      "transactions" => @transactions.length,
      "tracked_queries" => @transactions.values.inject(0) do |total, transaction|
        total + transaction["queries"].length
      end
    }
  end
end

#flush ⇒ Object 



43
44
45
46
47
 
# File 'lib/chronos/application/apm_aggregator.rb', line 43

def flush
  @mutex.synchronize { drain_locked }
rescue StandardError
  []
end

#record(event_type, payload = {}, context = {}) ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/chronos/application/apm_aggregator.rb', line 27

def record(event_type, payload = {}, context = {})
  return [] unless @config.apm_enabled

  @mutex.synchronize do
    type = event_type.to_s
    data = hash(payload)
    execution = hash(context)
    observe_component(type, data, execution)
    add_metric(type, data, execution) if aggregate_metric?(type, data)
    @observations += 1 if aggregate_metric?(type, data)
    @observations >= @config.apm_flush_count ? drain_locked : []
  end
rescue StandardError
  []
end