Periodically send status to systemd (#3006)

* Periodically send status to systemd

With systemd it's possible to send a status line. This will show up when
users run systemctl status puma.service.

Most of this code is based on puma-plugin-systemd.

* Rewrite systemd integration as a plugin

The primary motiviation for this is that plugins have native integration
for background threads. This is much cleaner since it allows tracking of
those. For example, it's possible to clean up those threads in the test
suite.

* add test for systemd plugin and tweak message fetch logic

---------

Co-authored-by: Ewoud Kohl van Wijngaarden <ewoud@kohlvanwijngaarden.nl>

lib/puma/launcher.rb

@@ -59,6 +59,10 @@ def initialize(conf, launcher_args={})
 
       @environment = conf.environment
 
+      if ENV["NOTIFY_SOCKET"]
+        @config.plugins.create('systemd')
+      end
+
       if @config.options[:bind_to_activated_sockets]
         @config.options[:binds] = @binder.synthesize_binds_from_activated_fs(
           @config.options[:binds],
@@ -180,7 +184,6 @@ def run
 
       setup_signals
       set_process_title
-      integrate_with_systemd
 
       # This blocks until the server is stopped
       @runner.run
@@ -311,22 +314,6 @@ def reload_worker_directory
       @runner.reload_worker_directory if @runner.respond_to?(:reload_worker_directory)
     end
 
-    # Puma's systemd integration allows Puma to inform systemd:
-    #  1. when it has successfully started
-    #  2. when it is starting shutdown
-    #  3. periodically for a liveness check with a watchdog thread
-    def integrate_with_systemd
-      return unless ENV["NOTIFY_SOCKET"]
-
-      require_relative 'systemd'
-
-      log "* Enabling systemd notification integration"
-
-      systemd = Systemd.new(@log_writer, @events)
-      systemd.hook_events
-      systemd.start_watchdog
-    end
-
     def log(str)
       @log_writer.log(str)
     end

lib/puma/plugin/systemd.rb

@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+require_relative '../plugin'
+
+# Puma's systemd integration allows Puma to inform systemd:
+#  1. when it has successfully started
+#  2. when it is starting shutdown
+#  3. periodically for a liveness check with a watchdog thread
+#  4. periodically set the status
+Puma::Plugin.create do
+  def start(launcher)
+    require_relative '../sd_notify'
+
+    launcher.log_writer.log "* Enabling systemd notification integration"
+
+    # hook_events
+    launcher.events.on_booted { Puma::SdNotify.ready }
+    launcher.events.on_stopped { Puma::SdNotify.stopping }
+    launcher.events.on_restart { Puma::SdNotify.reloading }
+
+    # start watchdog
+    if Puma::SdNotify.watchdog?
+      ping_f = watchdog_sleep_time
+
+      in_background do
+        launcher.log_writer.log "Pinging systemd watchdog every #{ping_f.round(1)} sec"
+        loop do
+          sleep ping_f
+          Puma::SdNotify.watchdog
+        end
+      end
+    end
+
+    # start status loop
+    instance = self
+    sleep_time = 1.0
+    in_background do
+      launcher.log_writer.log "Sending status to systemd every #{sleep_time.round(1)} sec"
+
+      loop do
+        sleep sleep_time
+        # TODO: error handling?
+        Puma::SdNotify.status(instance.status)
+      end
+    end
+  end
+
+  def status
+    if clustered?
+      messages = stats[:worker_status].map do |worker|
+        common_message(worker[:last_status])
+      end.join(',')
+
+      "Puma #{Puma::Const::VERSION}: cluster: #{booted_workers}/#{workers}, worker_status: [#{messages}]"
+    else
+      "Puma #{Puma::Const::VERSION}: worker: #{common_message(stats)}"
+    end
+  end
+
+  private
+
+  def watchdog_sleep_time
+    usec = Integer(ENV["WATCHDOG_USEC"])
+
+    sec_f = usec / 1_000_000.0
+    # "It is recommended that a daemon sends a keep-alive notification message
+    # to the service manager every half of the time returned here."
+    sec_f / 2
+  end
+
+  def stats
+    Puma.stats_hash
+  end
+
+  def clustered?
+    stats.has_key?(:workers)
+  end
+
+  def workers
+    stats.fetch(:workers, 1)
+  end
+
+  def booted_workers
+    stats.fetch(:booted_workers, 1)
+  end
+
+  def common_message(stats)
+    "{ #{stats[:running]}/#{stats[:max_threads]} threads, #{stats[:pool_capacity]} available, #{stats[:backlog]} backlog }"
+  end
+end

test/test_integration_systemd.rb → test/test_plugin_systemd.rb

@@ -1,7 +1,7 @@
 require_relative "helper"
 require_relative "helpers/integration"
 
-class TestIntegrationSystemd < TestIntegration
+class TestPluginSystemd < TestIntegration
   def setup
     skip "Skipped because Systemd support is linux-only" if windows? || osx?
     skip_unless :unix
@@ -55,6 +55,27 @@ def test_systemd_watchdog
     assert_match(socket_message, "STOPPING=1")
   end
 
+  def test_systemd_notify
+    cli_server "test/rackup/hello.ru"
+    assert_equal(socket_message, "READY=1")
+
+    assert_equal(socket_message(70),
+                    "STATUS=Puma #{Puma::Const::VERSION}: worker: { 0/5 threads, 5 available, 0 backlog }")
+
+    stop_server
+    assert_match(socket_message, "STOPPING=1")
+  end
+
+  def test_systemd_cluster_notify
+    cli_server "-w 2 -q test/rackup/hello.ru"
+    assert_equal(socket_message, "READY=1")
+    assert_equal(socket_message(130),
+                    "STATUS=Puma #{Puma::Const::VERSION}: cluster: 2/2, worker_status: [{ 0/5 threads, 5 available, 0 backlog },{ 0/5 threads, 5 available, 0 backlog }]")
+
+    stop_server
+    assert_match(socket_message, "STOPPING=1")
+  end
+
   private
 
   def assert_restarts_with_systemd(signal, workers: 2)
@@ -75,7 +96,7 @@ def assert_restarts_with_systemd(signal, workers: 2)
     assert_equal socket_message, 'STOPPING=1'
   end
 
-  def socket_message
-    @socket.recvfrom(15)[0]
+  def socket_message(len = 15)
+    @socket.recvfrom(len)[0]
   end
 end