Class: SolidQueue::LogSubscriber

Inherits:
ActiveSupport::LogSubscriber
  • Object
show all
Defined in:
lib/solid_queue/log_subscriber.rb

Instance Method Summary collapse

Instance Method Details

#batch_progress_error(event) ⇒ Object 



53
54
55
 
# File 'lib/solid_queue/log_subscriber.rb', line 53

def batch_progress_error(event)
  error formatted_event(event, action: "Error updating batch progress", **event.payload.slice(:batch_id, :job_id), error: formatted_error(event.payload[:error]))
end

#claim(event) ⇒ Object 



10
11
12
 
# File 'lib/solid_queue/log_subscriber.rb', line 10

def claim(event)
  debug formatted_event(event, action: "Claim jobs", **event.payload.slice(:process_id, :job_ids, :claimed_job_ids, :size))
end

#deregister_process(event) ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/solid_queue/log_subscriber.rb', line 116

def deregister_process(event)
  process = event.payload[:process]

  attributes = {
    process_id: process.id,
    pid: process.pid,
    hostname: process.hostname,
    name: process.name,
    last_heartbeat_at: process.last_heartbeat_at.iso8601,
    claimed_size: event.payload[:claimed_size],
    pruned: event.payload[:pruned]
  }

  if error = event.payload[:error]
    warn formatted_event(event, action: "Error deregistering #{process.kind}", **attributes.merge(error: formatted_error(error)))
  else
    debug formatted_event(event, action: "Deregister #{process.kind}", **attributes)
  end
end

#discard(event) ⇒ Object 



41
42
43
 
# File 'lib/solid_queue/log_subscriber.rb', line 41

def discard(event)
  debug formatted_event(event, action: "Discard job", **event.payload.slice(:job_id, :status))
end

#discard_all(event) ⇒ Object 



37
38
39
 
# File 'lib/solid_queue/log_subscriber.rb', line 37

def discard_all(event)
  debug formatted_event(event, action: "Discard jobs", **event.payload.slice(:jobs_size, :size, :status))
end

#dispatch_scheduled(event) ⇒ Object 



6
7
8
 
# File 'lib/solid_queue/log_subscriber.rb', line 6

def dispatch_scheduled(event)
  debug formatted_event(event, action: "Dispatch scheduled jobs", **event.payload.slice(:batch_size, :size))
end

#enqueue_recurring_task(event) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/solid_queue/log_subscriber.rb', line 65

def enqueue_recurring_task(event)
  attributes = event.payload.slice(:task, :active_job_id, :enqueue_error)
  attributes[:at] = event.payload[:at]&.iso8601

  if attributes[:active_job_id].nil? && event.payload[:skipped].nil?
    error formatted_event(event, action: "Error enqueuing recurring task", **attributes)
  elsif event.payload[:other_adapter]
    debug formatted_event(event, action: "Enqueued recurring task outside Solid Queue", **attributes)
  else
    action = event.payload[:skipped].present? ? "Skipped recurring task – already dispatched" : "Enqueued recurring task"
    debug formatted_event(event, action: action, **attributes)
  end
end

#fail_many_claimed(event) ⇒ Object 



18
19
20
21
22
23
 
# File 'lib/solid_queue/log_subscriber.rb', line 18

def fail_many_claimed(event)
  attributes = event.payload.slice(:job_ids, :process_ids)
  attributes[:error] = formatted_error(event.payload[:error]) if event.payload[:error]

  warn formatted_event(event, action: "Fail claimed jobs", **attributes)
end

#finish_batch(event) ⇒ Object 



45
46
47
 
# File 'lib/solid_queue/log_subscriber.rb', line 45

def finish_batch(event)
  info formatted_event(event, action: "Finish batch", **event.payload.slice(:batch_id, :total_jobs, :completed_jobs, :failed_jobs))
end

#fork_boot_timeout(event) ⇒ Object 



179
180
181
182
 
# File 'lib/solid_queue/log_subscriber.rb', line 179

def fork_boot_timeout(event)
  process = event.payload[:process]
  warn formatted_event(event, action: "Terminate #{process.kind} that failed to boot in time", **event.payload.slice(:pid).merge(hostname: process.hostname, name: process.name))
end

#graceful_termination(event) ⇒ Object 



144
145
146
147
148
149
150
151
152
 
# File 'lib/solid_queue/log_subscriber.rb', line 144

def graceful_termination(event)
  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)
  else
    info formatted_event(event, action: "Supervisor terminated gracefully", **attributes)
  end
end

#immediate_termination(event) ⇒ Object 



154
155
156
 
# File 'lib/solid_queue/log_subscriber.rb', line 154

def immediate_termination(event)
  info formatted_event(event, action: "Supervisor terminated immediately", **event.payload.slice(:process_id, :supervisor_pid, :supervised_processes))
end

#prune_processes(event) ⇒ Object 



136
137
138
 
# File 'lib/solid_queue/log_subscriber.rb', line 136

def prune_processes(event)
  debug formatted_event(event, action: "Prune dead processes", **event.payload.slice(:size))
end

#register_process(event) ⇒ Object 



105
106
107
108
109
110
111
112
113
114
 
# File 'lib/solid_queue/log_subscriber.rb', line 105

def register_process(event)
  process_kind = event.payload[:kind]
  attributes = event.payload.slice(:pid, :hostname, :process_id, :name)

  if error = event.payload[:error]
    warn formatted_event(event, action: "Error registering #{process_kind}", **attributes.merge(error: formatted_error(error)))
  else
    debug formatted_event(event, action: "Register #{process_kind}", **attributes)
  end
end

#release_blocked(event) ⇒ Object 



61
62
63
 
# File 'lib/solid_queue/log_subscriber.rb', line 61

def release_blocked(event)
  debug formatted_event(event, action: "Release blocked job", **event.payload.slice(:job_id, :concurrency_key, :released))
end

#release_claimed(event) ⇒ Object 



25
26
27
 
# File 'lib/solid_queue/log_subscriber.rb', line 25

def release_claimed(event)
  info formatted_event(event, action: "Release claimed job", **event.payload.slice(:job_id, :process_id))
end

#release_many_blocked(event) ⇒ Object 



57
58
59
 
# File 'lib/solid_queue/log_subscriber.rb', line 57

def release_many_blocked(event)
  debug formatted_event(event, action: "Unblock jobs", **event.payload.slice(:limit, :size))
end

#release_many_claimed(event) ⇒ Object 



14
15
16
 
# File 'lib/solid_queue/log_subscriber.rb', line 14

def release_many_claimed(event)
  info formatted_event(event, action: "Release claimed jobs", **event.payload.slice(:size))
end

#replace_fork(event) ⇒ Object 



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/solid_queue/log_subscriber.rb', line 162

def replace_fork(event)
  supervisor_pid = event.payload[:supervisor_pid]
  status = event.payload[:status]
  attributes = event.payload.slice(:pid).merge \
    status: (status.exitstatus || "no exit status set"),
    pid_from_status: status.pid,
    signaled: status.signaled?,
    stopsig: status.stopsig,
    termsig: status.termsig

  if replaced_fork = event.payload[:fork]
    info formatted_event(event, action: "Replaced terminated #{replaced_fork.kind}", **attributes.merge(hostname: replaced_fork.hostname, name: replaced_fork.name))
  elsif supervisor_pid != 1 # Running Docker, possibly having some processes that have been reparented
    warn formatted_event(event, action: "Tried to replace forked process but it had already died", **attributes)
  end
end

#retry(event) ⇒ Object 



33
34
35
 
# File 'lib/solid_queue/log_subscriber.rb', line 33

def retry(event)
  debug formatted_event(event, action: "Retry failed job", **event.payload.slice(:job_id))
end

#retry_all(event) ⇒ Object 



29
30
31
 
# File 'lib/solid_queue/log_subscriber.rb', line 29

def retry_all(event)
  debug formatted_event(event, action: "Retry failed jobs", **event.payload.slice(:jobs_size, :size))
end

#shutdown_process(event) ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/solid_queue/log_subscriber.rb', line 92

def shutdown_process(event)
  process = event.payload[:process]

  attributes = {
    pid: process.pid,
    hostname: process.hostname,
    process_id: process.process_id,
    name: process.name
  }.merge(process.metadata)

  info formatted_event(event, action: "Shutdown #{process.kind}", **attributes)
end

#start_process(event) ⇒ Object 



79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/solid_queue/log_subscriber.rb', line 79

def start_process(event)
  process = event.payload[:process]

  attributes = {
    pid: process.pid,
    hostname: process.hostname,
    process_id: process.process_id,
    name: process.name
  }.merge(process.metadata)

  info formatted_event(event, action: "Started #{process.kind}", **attributes)
end

#sweep_stalled_batches(event) ⇒ Object 



49
50
51
 
# File 'lib/solid_queue/log_subscriber.rb', line 49

def sweep_stalled_batches(event)
  debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :started_batches))
end

#thread_error(event) ⇒ Object 



140
141
142
 
# File 'lib/solid_queue/log_subscriber.rb', line 140

def thread_error(event)
  error formatted_event(event, action: "Error in thread", error: formatted_error(event.payload[:error]))
end

#unhandled_signal_error(event) ⇒ Object 



158
159
160
 
# File 'lib/solid_queue/log_subscriber.rb', line 158

def unhandled_signal_error(event)
  error formatted_event(event, action: "Received unhandled signal", **event.payload.slice(:signal))
end