Class: Karafka::Pro::Swarm::LivenessListener

Inherits:
Swarm::LivenessListener show all
Defined in:
lib/karafka/pro/swarm/liveness_listener.rb

Overview

Note:

This listener should not break anything if subscribed in the supervisor prior to forking as it relies on server events for operations.

Pro listener that monitors RSS usage and other heartbeat metrics (if configured) to ensure that everything operates.

It can:

- monitor poll frequency to make sure things are not polled not often enough
- monitor consumption to make sure we do not process data for too long
- monitor RSS to make sure that we do not use too much memory

By default it does not monitor memory and consuming and polling is configured in such a way to align with max.poll.interval.ms and other defaults.

Failure statuses reported are as follows:

- 1 - polling ttl exceeded
- 2 - consuming ttl exceeded
- 3 - memory limit exceeded
- 4 - stability ttl exceeded (stuck in non-steady librdkafka join state)

Instance Method Summary collapse

Constructor Details

#initialize(memory_limit: Float::INFINITY, consuming_ttl: DEFAULT_CONSUMING_TTL, polling_ttl: DEFAULT_POLLING_TTL, stability_ttl: nil) ⇒ LivenessListener

Returns a new instance of LivenessListener.

Parameters:

  • memory_limit (Integer) (defaults to: Float::INFINITY) —

    max memory in MB for this process to be considered healthy

  • consuming_ttl (Integer) (defaults to: DEFAULT_CONSUMING_TTL) —

    see Kubernetes::LivenessListener for full documentation.

  • polling_ttl (Integer) (defaults to: DEFAULT_POLLING_TTL) —

    see Kubernetes::LivenessListener for full documentation.

  • stability_ttl (Integer, nil) (defaults to: nil) —

    see Kubernetes::LivenessListener for full documentation, including the derivation rules (max max.poll.interval.ms across all active subscription groups, times 2, resolved lazily on first health evaluation). Same semantics apply here; the node is reported unhealthy with status code 4 instead of returning HTTP 500.



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 76

def initialize(
  memory_limit: Float::INFINITY,
  consuming_ttl: DEFAULT_CONSUMING_TTL,
  polling_ttl: DEFAULT_POLLING_TTL,
  stability_ttl: nil
)
  @polling_ttl = polling_ttl
  @consuming_ttl = consuming_ttl
  # When nil, resolved lazily via #stability_ttl on first health evaluation because
  # the default derives from the routing subscription groups, which may not be drawn
  # yet when the listener is constructed in karafka.rb
  @stability_ttl = stability_ttl
  # We cast it just in case someone would provide '10MB' or something similar
  @memory_limit = memory_limit.is_a?(String) ? memory_limit.to_i : memory_limit
  @pollings = {}
  @consumptions = {}
  @join_states = {}
  @instabilities = {}

  super()
end

Instance Method Details

#on_client_events_poll(_event) ⇒ Object

Report liveness during events poll so it works during long processing without statistics. This event fires periodically during wait even when the listener is blocked on consumer jobs, preventing the supervisor from killing the node.

Parameters:

  • _event (Karafka::Core::Monitoring::Event) 


117
118
119
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 117

def on_client_events_poll(_event)
  report_status
end

#on_connection_listener_after_fetch_loop(event) ⇒ Object

Parameters:

  • event (Karafka::Core::Monitoring::Event)

See Also:



192
193
194
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 192

def on_connection_listener_after_fetch_loop(event)
  clear_instability_tracking(event[:subscription_group])
end

#on_connection_listener_before_fetch_loop(_event) ⇒ Object

Report status before the first fetch loop iteration so the supervisor gets an initial healthy report even if the first consumption takes longer than the report timeout.

Parameters:

  • _event (Karafka::Core::Monitoring::Event) 


101
102
103
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 101

def on_connection_listener_before_fetch_loop(_event)
  report_status
end

#on_connection_listener_fetch_loop(_event) ⇒ Object

Tick on each fetch and report liveness so it works even when statistics are disabled

Parameters:

  • _event (Karafka::Core::Monitoring::Event) 


108
109
110
111
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 108

def on_connection_listener_fetch_loop(_event)
  mark_polling_tick
  report_status
end

#on_connection_listener_stopped(_event) ⇒ Object

Deregister the polling tracker for given listener

Parameters:

  • _event (Karafka::Core::Monitoring::Event) 


183
184
185
186
187
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 183

def on_connection_listener_stopped(_event)
  return if Karafka::App.done?

  clear_polling_tick
end

#on_connection_listener_stopping(_event) ⇒ Object

Deregister the polling tracker for given listener

Parameters:

  • _event (Karafka::Core::Monitoring::Event) 


171
172
173
174
175
176
177
178
179
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 171

def on_connection_listener_stopping(_event)
  # We are interested in disabling tracking for given listener only if it was requested
  # when karafka was running. If we would always clear, it would not catch the shutdown
  # polling requirements. The "running" listener shutdown operations happen only when
  # the manager requests it for downscaling.
  return if Karafka::App.done?

  clear_polling_tick
end

#on_error_occurred(_event) ⇒ Object

Parameters:

  • _event (Karafka::Core::Monitoring::Event) 


164
165
166
167
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 164

def on_error_occurred(_event)
  clear_consumption_tick
  clear_polling_tick
end

#on_statistics_emitted(event) ⇒ Object

Parameters:

  • event (Karafka::Core::Monitoring::Event)

See Also:



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/karafka/pro/swarm/liveness_listener.rb', line 144

def on_statistics_emitted(event)
  cgrp = event[:statistics]&.dig("cgrp")
  return unless cgrp

  sg_id = event[:subscription_group_id]
  join_state = cgrp["join_state"]
  return unless join_state

  synchronize do
    if join_state == "steady" || join_state == "init" || join_state == "wait-metadata"
      @instabilities.delete(sg_id)
      @join_states.delete(sg_id)
    elsif @join_states[sg_id] != join_state
      @join_states[sg_id] = join_state
      @instabilities[sg_id] = monotonic_now
    end
  end
end