Class: Chronos::Application::ApmAggregator

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

Overview

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

Examples:

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

Constant Summary collapse

METRIC_TYPES = 
%w(request query job external_http).freeze
QUERY_ERROR_SIGNALS = 
{
  "connection_error" => /Connection|NoDatabase|Adapter/i,
  "deadlock" => /Deadlock/i,
  "query_timeout" => /StatementTimeout|QueryCanceled|QueryTimeout|Timeout/i,
  "pool_timeout" => /ConnectionTimeout|PoolTimeout/i,
  "lock_timeout" => /LockWaitTimeout|LockTimeout/i,
  "constraint_violation" => /RecordNotUnique|NotNullViolation|ForeignKeyViolation|InvalidForeignKey|Constraint/i
}.freeze
SIGNAL_DIAGNOSTICS = 
{
  "slow_query" => ["warning", "performance", "Query duration reached the slow threshold"].freeze,
  "long_transaction" => ["warning", "transaction", "Transaction SQL reached the long threshold"].freeze,
  "connection_error" => ["error", "connection", "Database connection failed"].freeze,
  "deadlock" => ["error", "locking", "The database reported a deadlock"].freeze,
  "query_timeout" => ["error", "timeout", "The database query timed out"].freeze,
  "pool_timeout" => ["error", "connection_pool", "Database connection-pool checkout timed out"].freeze,
  "lock_timeout" => ["error", "locking", "The database lock wait timed out"].freeze,
  "constraint_violation" => ["error", "constraint", "The database rejected a constraint"].freeze
}.freeze

Instance Method Summary collapse

Constructor Details

#initialize(config, options = {}) ⇒ ApmAggregator

Returns a new instance of ApmAggregator.



38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/chronos/application/apm_aggregator.rb', line 38

def initialize(config, options = {})
  @config = config
  @clock = options[:clock] || proc { Time.now.to_f }
  @mutex = Mutex.new
  @groups = {}
  @transactions = {}
  @observations = 0
  @dropped_groups = 0
  @dropped_trace_trackers = 0
  @expired_trace_trackers = 0
  @dropped_query_fingerprints = 0
end

Instance Method Details

#diagnostics ⇒ Object 



74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/chronos/application/apm_aggregator.rb', line 74

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

#flush ⇒ Object 



68
69
70
71
72
 
# File 'lib/chronos/application/apm_aggregator.rb', line 68

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

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



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/chronos/application/apm_aggregator.rb', line 51

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

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