Class: LogStruct::Integrations::ActiveJob::LogSubscriber

Inherits:
ActiveSupport::LogSubscriber
  • Object
show all
Extended by:
T::Sig
Defined in:
lib/log_struct/integrations/active_job/log_subscriber.rb

Overview

Structured logging for ActiveJob

Instance Method Summary collapse

Instance Method Details

#discard(event) ⇒ void

This method returns an undefined value.

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


121
122
123
124
125
126
127
128
129
130
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 121

def discard(event)
  job = T.cast(event.payload[:job], ::ActiveJob::Base)
  error = T.cast(event.payload[:error], Exception)
  logger.error(Log::ActiveJob::Discard.new(
    **build_base_fields(job).to_kwargs,
    error_class: error.class,
    message: error.message,
    timestamp: EventTime.coerce_event_time(event.time)
  ))
end

#enqueue(event) ⇒ void

This method returns an undefined value.

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


19
20
21
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 19

def enqueue(event)
  log_enqueue(event)
end

#enqueue_all(event) ⇒ void

This method returns an undefined value.

perform_all_later: a line for each job, as perform_later writes

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 30

def enqueue_all(event)
  ts = EventTime.coerce_event_time(event.time)
  jobs = Array(event.payload[:jobs]).grep(::ActiveJob::Base)
  error = event.payload[:exception_object]
  if error
    log_enqueue_all_error(error, jobs, ts)
    return
  end

  # Adapters with their own enqueue_all (Sidekiq's) say only how many jobs they enqueued
  if jobs.none? { |job| job.successfully_enqueued? || job.enqueue_error }
    log_enqueued_count(jobs, event.payload[:enqueued_count], ts)
    return
  end

  jobs.each do |job|
    if job.successfully_enqueued?
      log_enqueued(job, ts)
    elsif job.enqueue_error
      log_exception(job.enqueue_error, job)
    end
  end
end

#enqueue_at(event) ⇒ void

This method returns an undefined value.

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


24
25
26
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 24

def enqueue_at(event)
  log_enqueue(event)
end

#enqueue_retry(event) ⇒ void

This method returns an undefined value.

retry_on caught the exception and enqueued the job again

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


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

def enqueue_retry(event)
  job = T.cast(event.payload[:job], ::ActiveJob::Base)
  # retry_job can be called without an exception or a wait
  error = event.payload[:error]
  wait = event.payload[:wait]
  logger.info(Log::ActiveJob::Retry.new(
    **build_base_fields(job).to_kwargs,
    error_class: error&.class,
    message: error&.message,
    wait_ms: wait.nil? ? nil : wait.to_f * 1000,
    timestamp: EventTime.coerce_event_time(event.time)
  ))
end

#perform(event) ⇒ void

This method returns an undefined value.

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 55

def perform(event)
  job = T.cast(event.payload[:job], ::ActiveJob::Base)
  exception = event.payload[:exception_object]

  if exception
    # Log the exception with the job context
    log_exception(exception, job)
  # An aborted job didn't run: a before_perform callback stopped it
  elsif !event.payload[:aborted]
    finished_at = EventTime.coerce_event_time(event.end)
    base_fields = build_base_fields(job)
    logger.info(Log::ActiveJob::Finish.new(
      **base_fields.to_kwargs,
      duration_ms: event.duration.to_f,
      finished_at: finished_at,
      timestamp: finished_at
    ))
  end
end

#perform_start(event) ⇒ void

This method returns an undefined value.

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 76

def perform_start(event)
  job = T.cast(event.payload[:job], ::ActiveJob::Base)
  ts = EventTime.coerce_event_time(event.time)
  started_at = ts
  attempt = job.executions
  base_fields = build_base_fields(job)
  logger.info(Log::ActiveJob::Start.new(
    **base_fields.to_kwargs,
    started_at: started_at,
    attempt: attempt,
    timestamp: ts
  ))
end

#retry_stopped(event) ⇒ void

This method returns an undefined value.

retry_on ran out of attempts. The job then raises the exception (an error line), unless retry_on was given a block.

Parameters:

  • event (::ActiveSupport::Notifications::Event) 


109
110
111
112
113
114
115
116
117
118
 
# File 'lib/log_struct/integrations/active_job/log_subscriber.rb', line 109

def retry_stopped(event)
  job = T.cast(event.payload[:job], ::ActiveJob::Base)
  error = T.cast(event.payload[:error], Exception)
  logger.error(Log::ActiveJob::RetryStopped.new(
    **build_base_fields(job).to_kwargs,
    error_class: error.class,
    message: error.message,
    timestamp: EventTime.coerce_event_time(event.time)
  ))
end