Include process_id on process instrumentation events when possible

Author: rosa

app/models/solid_queue/process.rb

@@ -12,8 +12,10 @@ class SolidQueue::Process < SolidQueue::Record
   after_destroy -> { claimed_executions.release_all }
 
   def self.register(**attributes)
-    SolidQueue.instrument :register_process, **attributes do
-      create!(attributes.merge(last_heartbeat_at: Time.current))
+    SolidQueue.instrument :register_process, **attributes do |payload|
+      create!(attributes.merge(last_heartbeat_at: Time.current)).tap do |process|
+        payload[:process_id] = process.id
+      end
     end
   rescue Exception => error
     SolidQueue.instrument :register_process, **attributes.merge(error: error)

lib/solid_queue/log_subscriber.rb

@@ -62,7 +62,8 @@ def start_process(event)
 
     attributes = {
       pid: process.pid,
-      hostname: process.hostname
+      hostname: process.hostname,
+      process_id: process.process_id
     }.merge(process.metadata)
 
     info formatted_event(event, action: "Started #{process.kind}", **attributes)
@@ -73,15 +74,16 @@ def shutdown_process(event)
 
     attributes = {
       pid: process.pid,
-      hostname: process.hostname
+      hostname: process.hostname,
+      process_id: process.process_id
     }.merge(process.metadata)
 
     info formatted_event(event, action: "Shutdown #{process.kind}", **attributes)
   end
 
   def register_process(event)
     process_kind = event.payload[:kind]
-    attributes = event.payload.slice(:pid, :hostname)
+    attributes = event.payload.slice(:pid, :hostname, :process_id)
 
     if error = event.payload[:error]
       warn formatted_event(event, action: "Error registering #{process_kind}", **attributes.merge(error: formatted_error(error)))
@@ -118,7 +120,7 @@ def thread_error(event)
   end
 
   def graceful_termination(event)
-    attributes = event.payload.slice(:supervisor_pid, :supervised_processes)
+    attributes = event.payload.slice(:process_id, :supervisor_pid, :supervised_processes)
 
     if event.payload[:shutdown_timeout_exceeded]
       warn formatted_event(event, action: "Supervisor wasn't terminated gracefully - shutdown timeout exceeded", **attributes)
@@ -128,7 +130,7 @@ def graceful_termination(event)
   end
 
   def immediate_termination(event)
-    info formatted_event(event, action: "Supervisor terminated immediately", **event.payload.slice(:supervisor_pid, :supervised_processes))
+    info formatted_event(event, action: "Supervisor terminated immediately", **event.payload.slice(:process_id, :supervisor_pid, :supervised_processes))
   end
 
   def unhandled_signal_error(event)

lib/solid_queue/processes/registrable.rb

@@ -12,7 +12,7 @@ module Registrable
     end
 
     def process_id
-      process.id
+      process&.id
     end
 
     private

lib/solid_queue/supervisor/fork_supervisor.rb

@@ -44,7 +44,7 @@ def start_process(configured_process)
       end
 
       def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do |payload|
+        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do |payload|
           term_forks
 
           Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
@@ -59,7 +59,7 @@ def terminate_gracefully
       end
 
       def terminate_immediately
-        SolidQueue.instrument(:immediate_termination, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do
+        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do
           quit_forks
         end
       end

test/integration/instrumentation_test.rb

@@ -88,7 +88,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     assert_equal 2, events.size
     register_event, deregister_event = events
-    assert_event register_event, "register_process", kind: "Worker", pid: ::Process.pid
+    assert_event register_event, "register_process", kind: "Worker", pid: ::Process.pid, process_id: process.id
     assert_event deregister_event, "deregister_process", process: process, pruned: false, claimed_size: 1
   end