Periodically send status to systemd

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