Class: Sixty::Exporter

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

Overview

Transport. Ships aggregate windows and exemplar traces to the collector.

Rules this obeys, because an observability agent that harms the host is worse than no agent at all:

- never block the request path (the flush runs on the agent's own thread)
- never grow without bound (the exemplar queue is capped and drops oldest)
- never keep the process alive (the thread is not joined at exit; the
final flush is bounded by its own timeout)
- never raise into user code (every failure is swallowed and counted)

Constant Summary collapse

MAX_QUEUED_EXEMPLARS = 
200
TIMEOUT_SECONDS = 
10

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(endpoint:, api_key:, service:, environment:, release:, repo_url: '', path: '/v1/ingest', on_warn: ->(_msg) {}) ⇒ Exporter

Returns a new instance of Exporter.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/sixty/exporter.rb', line 23

def initialize(endpoint:, api_key:, service:, environment:, release:, repo_url: '',
               path: '/v1/ingest', on_warn: ->(_msg) {})
  @endpoint = endpoint.to_s.sub(%r{/\z}, '')
  # The collector's receive path, appended to the endpoint — and overridable
  # because not every endpoint is the collector. A service reporting through
  # a proxy mounted on its own origin is given the exact URL that proxy
  # lives at, and appending anything to it produces a route the application
  # never registered. That failure is unusually quiet: the request 404s, the
  # agent swallows it like any other transport error, and the only symptom
  # is an absence of data.
  @path = path
  @api_key = api_key
  @resource = {
    service: service,
    environment: environment,
    release: release,
    repoUrl: repo_url.to_s
  }
  @on_warn = on_warn
  @exemplars = []
  @mutex = Mutex.new
  @failures = 0
  @warned_at = 0.0
  @last_attempt = nil
end

Instance Attribute Details

#failures ⇒ Object (readonly)

Returns the value of attribute failures.



21
22
23
 
# File 'lib/sixty/exporter.rb', line 21

def failures
  @failures
end

Instance Method Details

#flush(metrics, sections: {}, timeout: TIMEOUT_SECONDS) ⇒ Object

Parameters:

  • metrics (Hash, nil) —

    a drained aggregator window

  • sections (Hash) (defaults to: {}) —

    payload sections this exporter knows nothing about, so a future adapter can add one without teaching the transport about it

  • timeout (Numeric) (defaults to: TIMEOUT_SECONDS) —

    socket timeout; shortened for the exit flush, where the process is trying to stop and a dead collector must not be able to hold it open



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/sixty/exporter.rb', line 62

def flush(metrics, sections: {}, timeout: TIMEOUT_SECONDS)
  exemplars = @mutex.synchronize do
    taken = @exemplars
    @exemplars = []
    taken
  end

  has_sections = sections.any? { |_k, v| !v.nil? }
  return if metrics.nil? && exemplars.empty? && !has_sections

  # A collector that is down stops being contacted for a while, rather than
  # being dialled on every interval for as long as the outage lasts. The
  # window that would have been sent is dropped here, deliberately: an agent
  # that retained data through an outage would be an agent whose memory
  # footprint is decided by somebody else's uptime.
  return if backing_off?

  body = JSON.generate(
    {
      resource: @resource,
      metrics: metrics,
      exemplars: exemplars,
      sentAt: (Time.now.to_f * 1000).round
    }.merge(sections.reject { |_k, v| v.nil? })
  )

  post(body, timeout)
end

#queue_exemplar(trace) ⇒ Object 



49
50
51
52
53
54
 
# File 'lib/sixty/exporter.rb', line 49

def queue_exemplar(trace)
  @mutex.synchronize do
    @exemplars.shift if @exemplars.length >= MAX_QUEUED_EXEMPLARS
    @exemplars << trace
  end
end