Class: Mammoth::ObservabilitySnapshot

Inherits:
Object
  • Object
show all
Includes:
ObservabilityMetrics
Defined in:
lib/mammoth/observability_snapshot.rb

Overview

Builds health, readiness, and metrics snapshots from Mammoth's operational state and optional PostgreSQL slot-health provider.

ObservabilitySnapshot is intentionally read-only. It does not start the relay, mutate checkpoints, replay dead letters, or change PostgreSQL slot state. An injected provider may perform read-only slot inspection. The endpoints expose process, operational-state, and source status predictably.

Constant Summary

Constants included from ObservabilityMetrics

Mammoth::ObservabilityMetrics::DISPATCH_METRIC_NAMES

Constants included from PostgresObservabilityMetrics

PostgresObservabilityMetrics::POSTGRES_METRICS

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, state_adapter: nil, clock: -> { Time.now.utc }, dispatch_metrics: DispatchMetrics::INSTANCE, slot_health_provider: nil) ⇒ ObservabilitySnapshot

Returns a new instance of ObservabilitySnapshot.

Parameters:

  • config (Mammoth::Configuration) —

    loaded configuration

  • state_adapter (Mammoth::OperationalState::Adapter, nil) (defaults to: nil) —

    operational state dependency

  • clock (#call) (defaults to: -> { Time.now.utc }) —

    time source returning a Time-like object

  • dispatch_metrics (Mammoth::DispatchMetrics) (defaults to: DispatchMetrics::INSTANCE) —

    dispatch counter registry

  • slot_health_provider (#slot_health, nil) (defaults to: nil) —

    PostgreSQL slot health dependency



23
24
25
26
27
28
29
30
 
# File 'lib/mammoth/observability_snapshot.rb', line 23

def initialize(config, state_adapter: nil, clock: -> { Time.now.utc }, dispatch_metrics: DispatchMetrics::INSTANCE,
               slot_health_provider: nil)
  @config = config
  @state_adapter = state_adapter || OperationalState::Registry.build_configured(config)
  @clock = clock
  @dispatch_metrics = dispatch_metrics
  @slot_health_provider = slot_health_provider
end

Instance Attribute Details

#clock ⇒ Object (readonly)

Returns the value of attribute clock.



16
17
18
 
# File 'lib/mammoth/observability_snapshot.rb', line 16

def clock
  @clock
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



16
17
18
 
# File 'lib/mammoth/observability_snapshot.rb', line 16

def config
  @config
end

#dispatch_metrics ⇒ Object (readonly)

Returns the value of attribute dispatch_metrics.



16
17
18
 
# File 'lib/mammoth/observability_snapshot.rb', line 16

def dispatch_metrics
  @dispatch_metrics
end

#slot_health_provider ⇒ Object (readonly)

Returns the value of attribute slot_health_provider.



16
17
18
 
# File 'lib/mammoth/observability_snapshot.rb', line 16

def slot_health_provider
  @slot_health_provider
end

#state_adapter ⇒ Object (readonly)

Returns the value of attribute state_adapter.



16
17
18
 
# File 'lib/mammoth/observability_snapshot.rb', line 16

def state_adapter
  @state_adapter
end

Instance Method Details

#health ⇒ Hash

Build a liveness response.

Returns:

  • (Hash) —

    health payload



35
36
37
38
39
40
41
42
43
 
# File 'lib/mammoth/observability_snapshot.rb', line 35

def health
  {
    status: "ok",
    service: "mammoth",
    name: mammoth_name,
    version: Mammoth::VERSION,
    checked_at: checked_at
  }
end

#prometheus ⇒ String

Build a Prometheus text exposition document.

Returns:

  • (String) —

    Prometheus metrics text



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/mammoth/observability_snapshot.rb', line 75

def prometheus
  return down_metrics unless state_adapter.ready?

  lines = metric_headers + aggregate_metric_lines(
    checkpoint_store: state_adapter.checkpoint_store,
    dead_letter_store: state_adapter.dead_letter_store,
    delivered_store: state_adapter.delivered_envelope_store
  ) + destination_metric_lines(
    dead_letter_store: state_adapter.dead_letter_store,
    delivered_store: state_adapter.delivered_envelope_store
  ) + dispatch_metric_lines + postgres_slot_metric_lines
  "#{lines.join("\n")}\n"
rescue Mammoth::Error
  down_metrics
end

#readiness ⇒ Hash

Build a readiness response.

Returns:

  • (Hash) —

    readiness payload



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/mammoth/observability_snapshot.rb', line 48

def readiness
  return unready_payload unless state_adapter.ready?

  payload = {
    status: "ready",
    service: "mammoth",
    name: mammoth_name,
    operational_state: "ok",
    adapter: state_summary.fetch(:adapter),
    summary: state_summary,
    checked_at: checked_at
  }
  return payload unless slot_health_provider

  health = postgres_slot_health
  return postgres_unready_payload(health) unless health.ready?

  payload.merge(postgres_slot: health.summary)
rescue ReplicationError => e
  postgres_error_payload(e)
rescue Mammoth::Error => e
  adapter_error_payload(e)
end