Module: Pgbus::Metrics::Subscriber

Defined in:
lib/pgbus/metrics/subscriber.rb

Overview

Consumes Pgbus::Instrumentation events and forwards them to the configured metrics backend. The event→metric mapping (names, tags, counter vs histogram) intentionally mirrors the AppSignal subscriber so a team can switch backends and keep the same dashboards.

Every handler runs inside safely: a raising backend (Prometheus registry bug, StatsD socket down, custom backend) is logged and swallowed — a metrics failure must never propagate into the producer thread that emitted the event.

Class Method Summary collapse

Class Method Details

.install!(backend:) ⇒ Object 



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

def install!(backend:)
  return false if @installed

  @backend = backend
  @subscriptions = [
    subscribe("pgbus.executor.execute") { |event| on_executor_execute(event) },
    subscribe("pgbus.job_completed") { |event| on_job_completed(event) },
    subscribe("pgbus.job_failed") { |event| on_job_failed(event) },
    subscribe("pgbus.job_dead_lettered") { |event| on_job_dead_lettered(event) },
    subscribe("pgbus.event_processed") { |event| on_event_processed(event) },
    subscribe("pgbus.event_failed") { |event| on_event_failed(event) },
    subscribe("pgbus.client.send_message") { |event| on_send_message(event) },
    subscribe("pgbus.client.send_batch") { |event| on_send_batch(event) },
    subscribe("pgbus.client.read_batch") { |event| on_read_batch(event) },
    subscribe("pgbus.stream.broadcast") { |event| on_stream_broadcast(event) },
    subscribe("pgbus.outbox.publish") { |event| on_outbox_publish(event) },
    subscribe("pgbus.recurring.enqueue") { |event| on_recurring_enqueue(event) },
    subscribe("pgbus.worker.recycle") { |event| on_worker_recycle(event) },
    subscribe("pgbus.consumer.recycle") { |event| on_consumer_recycle(event) },
    subscribe("pgbus.client.pool") { |event| on_client_pool(event) }
  ]
  @installed = true
end

.installed? ⇒ Boolean

Returns:

  • (Boolean) 


48
49
50
 
# File 'lib/pgbus/metrics/subscriber.rb', line 48

def installed?
  @installed == true
end

.reset! ⇒ Object 



52
53
54
55
56
57
 
# File 'lib/pgbus/metrics/subscriber.rb', line 52

def reset!
  @subscriptions&.each { |s| ActiveSupport::Notifications.unsubscribe(s) }
  @subscriptions = []
  @backend = nil
  @installed = false
end