Class: Sixty::Aggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/sixty/aggregator.rb

Overview

In-process rollup.

Instead of shipping one record per call, the agent keeps sketches keyed by operation and flushes them on an interval. A service doing 50k req/s across 400 distinct operations sends 400 rows per flush, not 50k per second.

Cardinality is capped here as well as on the collector. Once MAX_OPERATIONS distinct operations are seen in a window, further new ones collapse into a single __overflow__ bucket and a warning is emitted once. Silently dropping them would make the numbers quietly wrong; unbounded growth would take the host down. Overflow is visible and bounded.

── Why this is locked and the JavaScript one is not ──────────────────────

Node's agent runs on one thread. A Rails app under Puma runs the request path on many, and every one of them records into this object — so a mutex is not defensive programming here, it is the difference between a correct counter and a torn one. It is held for the duration of a hash update and a sketch insertion, both of which are microseconds; the flush swaps the maps under the same lock and serializes outside it.

Constant Summary collapse

MAX_OPERATIONS = 
2000
MAX_EDGES = 
5000

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(alpha: 0.01, on_warn: ->(_msg) {}) ⇒ Aggregator

Returns a new instance of Aggregator.



31
32
33
34
35
36
37
38
39
 
# File 'lib/sixty/aggregator.rb', line 31

def initialize(alpha: 0.01, on_warn: ->(_msg) {})
  @alpha = alpha
  @on_warn = on_warn
  @ops = {}
  @edges = {}
  @overflowed = false
  @window_start = now_ms
  @mutex = Mutex.new
end

Class Method Details

.span_key(span) ⇒ Object

Stable identity for an operation within one agent window.

A db operation is identified by its statement, never by its label: the label is presentation and will keep improving, and folding it into the key would re-identify the operation — orphaning its history — every time somebody improves the naming.



72
73
74
75
76
77
78
 
# File 'lib/sixty/aggregator.rb', line 72

def self.span_key(span)
  if span.kind == Tracer::KIND_DB
    "db\x01#{span.attrs[:normalized_sql] || span.name}"
  else
    "#{span.kind}\x01#{span.name}"
  end
end

Instance Method Details

#drain ⇒ Object

Serialize and reset. Returns nil when there is nothing to send.



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/sixty/aggregator.rb', line 46

def drain
  ops, edges, window_start = @mutex.synchronize do
    return nil if @ops.empty?

    taken = [@ops, @edges, @window_start]
    @ops = {}
    @edges = {}
    @overflowed = false
    @window_start = now_ms
    taken
  end

  {
    windowStart: window_start,
    windowEnd: now_ms,
    operations: ops.map { |key, op| serialize_operation(key, op) },
    edges: edges.map { |key, edge| serialize_edge(key, edge) }
  }
end

#record(span) ⇒ Object 



41
42
43
 
# File 'lib/sixty/aggregator.rb', line 41

def record(span)
  @mutex.synchronize { record_locked(span) }
end