Class: PgEventstore::Web::Metrics::Collectors::SubscriptionsThroughput

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

Overview

Processing volume and speed of each reported subscription.

Two deliberately different numbers:

  • processed_events_total is a counter; rate() over it gives the actual current throughput and correctly drops to 0 when no matching events arrive.
  • capacity_events_per_second derives from the average handler execution time of the last SubscriptionHandlerPerformance::TIMINGS_TO_KEEP processed events - whenever they happened. It answers "how fast can this handler go when fed", is sticky while the subscription is idle and must not be read as current throughput.

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:



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/pg_eventstore/web/metrics/collectors/subscriptions_throughput.rb', line 18

def call
  processed = MetricFamily.new(
    name: 'pg_eventstore_subscription_processed_events_total',
    type: 'counter',
    help: 'Total number of events processed by the subscription. Use rate() for current throughput.'
  )
  capacity = MetricFamily.new(
    name: 'pg_eventstore_subscription_capacity_events_per_second',
    type: 'gauge',
    help: 'Average processing speed over the last few processed events, whenever they happened. ' \
          'Sticky while idle - this is handler capacity, not current throughput.'
  )
  subscription_rows.each do |row|
    labels = subscription_labels(row)
    processed.add_sample(labels:, value: row['total_processed_events'])
    capacity.add_sample(labels:, value: row['capacity_eps']) if row['capacity_eps']
  end
  [processed, capacity]
end

#subscription_rows ⇒ Array<Hash>

Returns:

  • (Array<Hash>) 


41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/pg_eventstore/web/metrics/collectors/subscriptions_throughput.rb', line 41

def subscription_rows
  builder = subscriptions_sql_builder
  builder.select("    s.set,\n    s.name,\n    s.total_processed_events,\n    case when s.average_event_processing_time > 0\n         then (1.0 / s.average_event_processing_time)::float8\n    end as capacity_eps\n  SQL\n  with_safe_conn do |conn|\n    conn.exec_params(*builder.to_exec_params)\n  end\nend\n")