Class: OpenTelemetry::Instrumentation::ActiveJob::Handlers::Step

Inherits:
Default
  • Object
show all
Defined in:
lib/opentelemetry/instrumentation/active_job/handlers/step.rb

Overview

Handles step.active_job to generate child spans for continuable job steps

Instance Method Summary collapse

Methods inherited from Default

#finish_span, #initialize, #on_exception, #start, #tracer

Constructor Details

This class inherits a constructor from OpenTelemetry::Instrumentation::ActiveJob::Handlers::Default

Instance Method Details

#finish(_name, _id, payload) ⇒ Object

Overrides the Default#finish method to record step-specific attributes before closing the span

Parameters:

  • _name (String) —

    of the Event (unused)

  • _id (String) —

    of the event (unused)

  • payload (Hash) —

    containing job run information



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/opentelemetry/instrumentation/active_job/handlers/step.rb', line 43

def finish(_name, _id, payload)
  otel = payload.delete(:__otel)
  span = otel&.fetch(:span)
  token = otel&.fetch(:ctx_token)

  step = payload.fetch(:step)
  span&.set_attribute('messaging.active_job.step.result', 'interrupted') if payload[:interrupted]
  span&.set_attribute('messaging.active_job.step.cursor', step.cursor.to_s) if step.cursor

  # Continuation::Interrupt is control flow, not a real error — skip recording it
  on_exception(payload[:error] || payload[:exception_object], span) unless payload[:interrupted]
rescue StandardError => e
  OpenTelemetry.handle_error(exception: e)
ensure
  finish_span(span, token)
end

#start_span(name, _id, payload) ⇒ Hash

Overrides the Default#start_span method to create a child span for a continuable job step

Parameters:

  • name (String) —

    of the Event

  • id (String) —

    of the event

  • payload (Hash) —

    containing job run information

Returns:

  • (Hash) —

    with the span and generated context tokens



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/opentelemetry/instrumentation/active_job/handlers/step.rb', line 20

def start_span(name, _id, payload)
  job = payload.fetch(:job)
  job_name = job.class.name
  step = payload.fetch(:step)
  step_name = step.name.to_s

  attributes = @mapper.call(payload).merge(
    'messaging.active_job.step.name' => step_name,
    'messaging.active_job.step.state' => step.resumed? ? 'resumed' : 'started'
  )

  span = tracer.start_span("#{step_name} #{job_name}", attributes: attributes)
  token = OpenTelemetry::Context.attach(OpenTelemetry::Trace.context_with_span(span))

  { span: span, ctx_token: token }
end