Class: Karafka::Instrumentation::Vendors::Kubernetes::LivenessListener

Inherits:
BaseListener
  • Object
show all
Defined in:
lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb

Overview

Note:

This listener will bind itself only when Karafka will actually attempt to start and moves from initializing to running. Before that, the TCP server will NOT be active. This is done on purpose to mitigate a case where users would subscribe this listener in ‘karafka.rb` without checking the recommendations of conditional assignment.

Note:

In case of usage within an embedding with Puma, you need to select different port then the one used by Puma itself.

Note:

Please use ‘Kubernetes::SwarmLivenessListener` when operating in the swarm mode

Kubernetes HTTP listener that does not only reply when process is not fully hanging, but also allows to define max time of processing and looping.

Processes like Karafka server can hang while still being reachable. For example, in case something would hang inside of the user code, Karafka could stop polling and no new data would be processed, but process itself would still be active. This listener allows for defining of a ttl that gets bumped on each poll loop and before and after processing of a given messages batch.

Instance Method Summary collapse

Constructor Details

#initialize(hostname: nil, port: 3000, consuming_ttl: 5 * 60 * 1_000, polling_ttl: 5 * 60 * 1_000) ⇒ LivenessListener

Note:

The default TTL matches the default ‘max.poll.interval.ms`

Returns a new instance of LivenessListener.

Parameters:

  • hostname (String, nil) (defaults to: nil) —

    hostname or nil to bind on all

  • port (Integer) (defaults to: 3000) —

    TCP port on which we want to run our HTTP status server

  • consuming_ttl (Integer) (defaults to: 5 * 60 * 1_000) —

    time in ms after which we consider consumption hanging. It allows us to define max consumption time after which k8s should consider given process as hanging

  • polling_ttl (Integer) (defaults to: 5 * 60 * 1_000) —

    max time in ms for polling. If polling (any) does not happen that often, process should be considered dead.



37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 37

def initialize(
  hostname: nil,
  port: 3000,
  consuming_ttl: 5 * 60 * 1_000,
  polling_ttl: 5 * 60 * 1_000
)
  @polling_ttl = polling_ttl
  @consuming_ttl = consuming_ttl
  @mutex = Mutex.new
  @pollings = {}
  @consumptions = {}
  super(hostname: hostname, port: port)
end

Instance Method Details

#healthy? ⇒ String

Did we exceed any of the ttls

Returns:

  • (String) —

    204 string if ok, 500 otherwise



117
118
119
120
121
122
123
124
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 117

def healthy?
  time = monotonic_now

  return false if @pollings.values.any? { |tick| (time - tick) > @polling_ttl }
  return false if @consumptions.values.any? { |tick| (time - tick) > @consuming_ttl }

  true
end

#on_app_running(_event) ⇒ Object

Parameters:

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


52
53
54
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 52

def on_app_running(_event)
  start
end

#on_app_stopped(_event) ⇒ Object

Stop the http server when we stop the process

Parameters:

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


58
59
60
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 58

def on_app_stopped(_event)
  stop
end

#on_connection_listener_fetch_loop(_event) ⇒ Object

Tick on each fetch

Parameters:

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


64
65
66
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 64

def on_connection_listener_fetch_loop(_event)
  mark_polling_tick
end

#on_connection_listener_stopped(_event) ⇒ Object

Deregister the polling tracker for given listener

Parameters:

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


109
110
111
112
113
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 109

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) 


97
98
99
100
101
102
103
104
105
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 97

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) 


90
91
92
93
 
# File 'lib/karafka/instrumentation/vendors/kubernetes/liveness_listener.rb', line 90

def on_error_occurred(_event)
  clear_consumption_tick
  clear_polling_tick
end