Module: Postburner::Instrumentation

Defined in:
lib/postburner/instrumentation.rb

Overview

Instrumentation helpers for ActiveSupport::Notifications events.

Provides standardized payload builders for all Postburner instrumentation events. Event naming follows the pattern: verb.noun.postburner

Event Categories

Job Events

  • perform_start.job.postburner - Before job execution
  • perform.job.postburner - Around job execution (includes duration)
  • retry.job.postburner - Default job retried
  • discard.job.postburner - Default job exhausts retries
  • enqueue.job.postburner - Job queued (immediate)
  • enqueue_at.job.postburner - Job queued with delay
  • retry_stopped.job.postburner - Tracked job buried after failures

Schedule Events

  • create.schedule.postburner - Schedule created
  • update.schedule.postburner - Schedule updated
  • audit.schedule.postburner - Scheduler audits a schedule
  • reconcile.schedule.postburner - Schedule reconciled (executions converged to config)

Schedule Execution Events

  • create.schedule_execution.postburner - Execution created
  • enqueue.schedule_execution.postburner - Execution enqueued to Beanstalkd
  • skip.schedule_execution.postburner - Execution skipped (operator cancels one occurrence)
  • supersede.schedule_execution.postburner - Execution superseded (replaced by a recreate)

Watchdog Events (Worker Level)

  • perform_start.watchdog.postburner - Watchdog job execution begins
  • perform.watchdog.postburner - Around watchdog job execution

Scheduler Events (Scheduler Level)

  • perform_start.scheduler.postburner - Scheduler processing begins
  • perform.scheduler.postburner - Around scheduler processing (summary)

Examples:

Subscribing to job events

ActiveSupport::Notifications.subscribe('perform.job.postburner') do |name, start, finish, id, payload|
  duration = (finish - start) * 1000
  Rails.logger.info "[Postburner] #{payload[:job][:class]} completed in #{duration.round(2)}ms"
end

Subscribing to schedule events

ActiveSupport::Notifications.subscribe('create.schedule.postburner') do |*args|
  payload = args.last
  schedule = payload[:schedule]
  Rails.logger.info "[Postburner] Schedule '#{schedule[:name]}' created"
end

Class Method Summary collapse

Class Method Details

.changes_payload(model, exclude: []) ⇒ Hash

Build changes hash from ActiveRecord model, excluding specified attributes.

Parameters:

  • model (ActiveRecord::Base) —

    Model with changes

  • exclude (Array<String, Symbol>) (defaults to: []) —

    Attributes to exclude

Returns:

  • (Hash) —

    Changes hash in { attribute: [old, new] } format



186
187
188
189
 
# File 'lib/postburner/instrumentation.rb', line 186

def changes_payload(model, exclude: [])
  exclude = exclude.map(&:to_s)
  model.saved_changes.except(*exclude, 'updated_at', 'created_at')
end

.execution_payload(execution) ⇒ Hash

Build a schedule execution payload hash for instrumentation.

Parameters:

Returns:

  • (Hash) —

    Standardized execution payload



168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/postburner/instrumentation.rb', line 168

def execution_payload(execution)
  {
    id: execution.id,
    schedule_id: execution.schedule_id,
    run_at: execution.run_at,
    next_run_at: execution.next_run_at,
    status: execution.status,
    beanstalk_job_id: execution.beanstalk_job_id,
    job_id: execution.job_id
  }
end

.instrument(event, payload = {}) { ... } ⇒ Object

Instrument an event with ActiveSupport::Notifications.

Parameters:

  • event (String) —

    Event name (e.g., 'perform.job.postburner')

  • payload (Hash) (defaults to: {}) —

    Event payload

Yields:

  • Block to execute within instrumentation (for timing)

Returns:

  • (Object) —

    Result of block if given



198
199
200
 
# File 'lib/postburner/instrumentation.rb', line 198

def instrument(event, payload = {}, &block)
  ActiveSupport::Notifications.instrument(event, payload, &block)
end

.job_payload(job, beanstalk_job_id: nil) ⇒ Hash

Build a job payload hash for instrumentation.

Parameters:

  • job (Postburner::Job, Hash) —

    Job instance or parsed payload hash

  • beanstalk_job_id (Integer, nil) (defaults to: nil) —

    Beanstalkd job ID

Returns:

  • (Hash) —

    Standardized job payload



62
63
64
65
66
67
68
69
70
 
# File 'lib/postburner/instrumentation.rb', line 62

def job_payload(job, beanstalk_job_id: nil)
  if job.is_a?(Postburner::Job)
    job_payload_from_model(job, beanstalk_job_id: beanstalk_job_id)
  elsif job.is_a?(Hash)
    job_payload_from_hash(job, beanstalk_job_id: beanstalk_job_id)
  else
    raise ArgumentError, "Expected Postburner::Job or Hash, got #{job.class}"
  end
end

.job_payload_from_activejob(job, tracked:, postburner_job_id: nil, beanstalk_job_id: nil) ⇒ Hash

Build job payload from an ActiveJob instance.

Parameters:

  • job (ActiveJob::Base) —

    ActiveJob instance

  • tracked (Boolean) —

    Whether job is tracked in PostgreSQL

  • postburner_job_id (Integer, nil) (defaults to: nil) —

    Postburner::TrackedJob ID if tracked

  • beanstalk_job_id (Integer, nil) (defaults to: nil) —

    Beanstalkd job ID

Returns:

  • (Hash) —

    Standardized job payload



133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/postburner/instrumentation.rb', line 133

def job_payload_from_activejob(job, tracked:, postburner_job_id: nil, beanstalk_job_id: nil)
  {
    class: job.class.name,
    id: postburner_job_id,
    job_id: job.job_id,
    arguments: job.arguments,
    queue_name: job.queue_name,
    beanstalk_job_id: beanstalk_job_id,
    tracked: tracked
  }
end

.job_payload_from_hash(payload, beanstalk_job_id: nil) ⇒ Hash

Build job payload from a parsed Beanstalkd payload hash.

Handles both ActiveJob format and native Postburner::Job format.

Parameters:

  • payload (Hash) —

    Parsed JSON payload from Beanstalkd

  • beanstalk_job_id (Integer, nil) (defaults to: nil) —

    Beanstalkd job ID

Returns:

  • (Hash) —

    Standardized job payload



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/postburner/instrumentation.rb', line 98

def job_payload_from_hash(payload, beanstalk_job_id: nil)
  if Postburner::ActiveJob::Payload.native_format?(payload)
    # Native Postburner::Job format: { "class" => "JobClass", "args" => [id] }
    {
      class: payload['class'],
      id: payload['args']&.first,
      job_id: nil,
      arguments: payload['args'],
      queue_name: nil,
      beanstalk_job_id: beanstalk_job_id,
      tracked: true
    }
  else
    # ActiveJob format
    tracked = payload['tracked'] == true
    {
      class: payload['job_class'],
      id: tracked ? payload['postburner_job_id'] : nil,
      job_id: payload['job_id'],
      arguments: payload['arguments'],
      queue_name: payload['queue_name'],
      beanstalk_job_id: beanstalk_job_id,
      tracked: tracked
    }
  end
end

.job_payload_from_model(job, beanstalk_job_id: nil) ⇒ Hash

Build job payload from a Postburner::Job model.

Parameters:

  • job (Postburner::Job) —

    Job model instance

  • beanstalk_job_id (Integer, nil) (defaults to: nil) —

    Beanstalkd job ID

Returns:

  • (Hash) —

    Standardized job payload



78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/postburner/instrumentation.rb', line 78

def job_payload_from_model(job, beanstalk_job_id: nil)
  {
    class: job.class.name,
    id: job.id,
    job_id: job.respond_to?(:args) && job.args.is_a?(Hash) ? job.args['job_id'] : nil,
    arguments: job.args,
    queue_name: job.respond_to?(:queue_name) ? job.queue_name : job.class.try(:postburner_queue),
    beanstalk_job_id: beanstalk_job_id || job.bkid,
    tracked: true
  }
end

.schedule_payload(schedule) ⇒ Hash

Build a schedule payload hash for instrumentation.

Parameters:

Returns:

  • (Hash) —

    Standardized schedule payload



150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/postburner/instrumentation.rb', line 150

def schedule_payload(schedule)
  {
    id: schedule.id,
    name: schedule.name,
    job_class: schedule.job_class,
    enabled: schedule.enabled,
    interval: schedule.interval,
    interval_unit: schedule.interval_unit,
    cron: schedule.cron,
    timezone: schedule.timezone
  }
end