Class: PgEventstore::Web::Metrics::Collectors::SubscriptionsHealth

Inherits:
Base
  • Object
show all
Defined in:
lib/pg_eventstore/web/metrics/collectors/subscriptions_health.rb,
sig/pg_eventstore/web/metrics/collectors/subscriptions_health.rbs

Overview

Liveness and error state of each reported subscription.

The state column alone can not be trusted: a subscription killed without a graceful shutdown keeps state "running" and its lock forever. heartbeat_age_seconds is the discriminator - a subscription is really running only while its heartbeat stays below SubscriptionsLifecycle::HEARTBEAT_INTERVAL.

Constant Summary

Constants inherited from Base

Base::STATEMENT_TIMEOUT

Instance Attribute Summary

Attributes inherited from Base

#connection, #sets

Instance Method Summary collapse

Methods inherited from Base

#initialize, #subscription_labels, #subscriptions_sql_builder, #transaction_queries, #with_safe_conn

Constructor Details

This class inherits a constructor from PgEventstore::Web::Metrics::Collectors::Base

Instance Method Details

#call ⇒ Array<PgEventstore::Web::Metrics::MetricFamily>

Returns:



15
16
17
18
19
20
21
22
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
48
49
50
51
52
 
# File 'lib/pg_eventstore/web/metrics/collectors/subscriptions_health.rb', line 15

def call
  state = MetricFamily.new(
    name: 'pg_eventstore_subscription_state',
    type: 'gauge',
    help: 'Last recorded state of the subscription. May be stale - correlate with ' \
          'pg_eventstore_subscription_heartbeat_age_seconds.'
  )
  locked = MetricFamily.new(
    name: 'pg_eventstore_subscription_locked',
    type: 'gauge',
    help: 'Whether the subscription is locked by a subscriptions set.'
  )
  heartbeat_age = MetricFamily.new(
    name: 'pg_eventstore_subscription_heartbeat_age_seconds',
    type: 'gauge',
    help: 'Seconds since the subscription row was last touched by its runner. A locked subscription ' \
          'with a stale heartbeat is a dead process that did not shut down gracefully.'
  )
  restarts = MetricFamily.new(
    name: 'pg_eventstore_subscription_restarts_total',
    type: 'counter',
    help: 'Number of times the subscription was restarted after a failure.'
  )
  last_error_age = MetricFamily.new(
    name: 'pg_eventstore_subscription_last_error_age_seconds',
    type: 'gauge',
    help: 'Seconds since the last error occurred. Absent when the subscription never failed.'
  )
  subscription_rows.each do |row|
    labels = subscription_labels(row)
    state.add_sample(labels: labels.merge(state: row['state']), value: 1)
    locked.add_sample(labels:, value: row['locked'])
    heartbeat_age.add_sample(labels:, value: row['heartbeat_age_seconds'])
    restarts.add_sample(labels:, value: row['restart_count'])
    last_error_age.add_sample(labels:, value: row['last_error_age_seconds']) if row['last_error_age_seconds']
  end
  [state, locked, heartbeat_age, restarts, last_error_age]
end

#subscription_rows ⇒ Array<Hash>

Returns:



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/pg_eventstore/web/metrics/collectors/subscriptions_health.rb', line 57

def subscription_rows
  builder = subscriptions_sql_builder
  builder.select("    s.set,\n    s.name,\n    s.state,\n    (s.locked_by is not null)::int as locked,\n    extract(epoch from ((now() at time zone 'utc') - s.updated_at))::float8 as heartbeat_age_seconds,\n    s.restart_count,\n    case when s.last_error_occurred_at is not null\n         then extract(epoch from ((now() at time zone 'utc') - s.last_error_occurred_at))::float8\n    end as last_error_age_seconds\n  SQL\n  with_safe_conn do |conn|\n    conn.exec_params(*builder.to_exec_params)\n  end\nend\n")