Class: ActiveJob::Temporal::Observability::Prometheus

Inherits:
Adapter
  • Object
show all
Defined in:
lib/activejob/temporal/observability/prometheus.rb

Overview

rubocop:disable Metrics/ClassLength

Constant Summary collapse

DURATION_BUCKETS = 
[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120].freeze
PAYLOAD_SIZE_BUCKETS = 
[512, 1024, 2_048, 4_096, 8_192, 16_384, 32_768, 65_536, 131_072, 262_144,
524_288, 1_048_576].freeze

Instance Attribute Summary collapse

Attributes inherited from Adapter

#name

Instance Method Summary collapse

Methods inherited from Adapter

#started?, #validate!

Constructor Details

#initialize(registry: nil, monotonic_clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }) ⇒ Prometheus

Returns a new instance of Prometheus.



61
62
63
64
65
66
67
68
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 61

def initialize(registry: nil, monotonic_clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) })
  super(:prometheus)
  @registry = registry
  @monotonic_clock = monotonic_clock
  @metrics_server = MetricsServerConfiguration.new
  @server = nil
  @registered = false
end

Instance Attribute Details

#metrics_server ⇒ Object (readonly)

Returns the value of attribute metrics_server.



59
60
61
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 59

def metrics_server
  @metrics_server
end

#registry ⇒ Object (readonly)

Returns the value of attribute registry.



59
60
61
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 59

def registry
  @registry
end

Instance Method Details

#instrument(event_name, payload) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 94

def instrument(event_name, payload)
  return yield unless event_name == :perform

  ensure_metrics_registered
  started_at = monotonic_time

  result = yield
  @jobs_completed.increment(labels: { class: label(payload[:job_class]), queue: label(payload[:queue]) })
  result
rescue StandardError => e
  @jobs_failed.increment(labels: {
                           class: label(payload[:job_class]),
                           queue: label(payload[:queue]),
                           error: PrometheusErrorLabels.for(e)
                         })
  raise
ensure
  if started_at
    @job_duration.observe(monotonic_time - started_at, labels: { class: label(payload[:job_class]) })
  end
end

#record(event_name, payload) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 81

def record(event_name, payload)
  ensure_metrics_registered

  case event_name
  when :enqueue then record_enqueue(payload)
  when :payload_serialize then observe_payload_size(payload)
  when :retry then record_retry(payload)
  when :worker_start then record_worker_started
  when :worker_stop then record_worker_stopped
  when :active_tasks then record_active_tasks(payload[:count])
  end
end

#render ⇒ Object 



116
117
118
119
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 116

def render
  ensure_metrics_registered
  ::Prometheus::Client::Formats::Text.marshal(registry)
end

#start! ⇒ Object 



70
71
72
73
74
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 70

def start!
  super
  ensure_metrics_registered
  self
end

#start_metrics_server(port: metrics_server.port, bind_address: metrics_server.bind, allow_public_bind: metrics_server.allow_public_bind) ⇒ Object

Raises:

  • (ArgumentError) 


121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 121

def start_metrics_server(port: metrics_server.port,
                         bind_address: metrics_server.bind,
                         allow_public_bind: metrics_server.allow_public_bind)
  raise ArgumentError, "Prometheus metrics server port is required" unless port

  @server = MetricsServer.new(
    port: port,
    bind_address: bind_address,
    allow_public_bind: allow_public_bind,
    provider: self
  ).start
end

#stop! ⇒ Object 



76
77
78
79
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 76

def stop!
  stop_metrics_server
  super
end

#stop_metrics_server ⇒ Object 



134
135
136
137
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 134

def stop_metrics_server
  @server&.stop
  @server = nil
end

#validate_dependencies! ⇒ Object 



139
140
141
142
143
144
 
# File 'lib/activejob/temporal/observability/prometheus.rb', line 139

def validate_dependencies!
  require_dependency("prometheus-client", "prometheus/client", "Prometheus")
  require_dependency("prometheus-client", "prometheus/client/formats/text", "Prometheus")
  @registry ||= ::Prometheus::Client::Registry.new
  self
end