Class: JobWorkflow::Instrumentation::OpenTelemetrySubscriber

Inherits:
Object
  • Object
show all
Defined in:
lib/job_workflow/instrumentation/opentelemetry_subscriber.rb,
sig/generated/job_workflow/instrumentation/opentelemetry_subscriber.rbs

Overview

Note:

This subscriber requires the opentelemetry-api gem to be installed. If not available, subscription will be silently skipped.

OpenTelemetrySubscriber provides OpenTelemetry tracing integration for JobWorkflow events. It subscribes to ActiveSupport::Notifications and creates OpenTelemetry spans.

Examples:

Enable OpenTelemetry integration

```ruby
# Ensure OpenTelemetry is configured first
OpenTelemetry::SDK.configure do |c|
  c.service_name = "my-app"
end

# Then subscribe JobWorkflow events
JobWorkflow::Instrumentation::OpenTelemetrySubscriber.subscribe!
```

Defined Under Namespace

Modules: Attributes

Constant Summary collapse

SUBSCRIBED_EVENTS =

Returns:

  • (Object) 
[
  Events::WORKFLOW,
  Events::TASK,
  Events::TASK_SKIP,
  Events::TASK_ENQUEUE,
  Events::TASK_RETRY,
  Events::THROTTLE_ACQUIRE,
  Events::DEPENDENT_WAIT
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.opentelemetry_available? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


74
75
76
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 74

def opentelemetry_available?
  !!defined?(::OpenTelemetry::Trace)
end

.reset! ⇒ void

This method returns an undefined value.

: () -> void



69
70
71
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 69

def reset!
  unsubscribe!
end

.subscribe! ⇒ Array[untyped]?

: () -> Array?

Returns:

  • (Array[untyped], nil) 


53
54
55
56
57
58
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 53

def subscribe!
  return unless opentelemetry_available?
  return subscriptions unless subscriptions.empty?

  self.subscriptions = SUBSCRIBED_EVENTS.map { |event| ActiveSupport::Notifications.subscribe(event, new) }
end

.subscriptions ⇒ Array[untyped]

Returns:

  • (Array[untyped]) 


46
 
# File 'sig/generated/job_workflow/instrumentation/opentelemetry_subscriber.rbs', line 46

def self.subscriptions: () -> Array[untyped]

.subscriptions= ⇒ void

This method returns an undefined value.

Parameters:

  • (Array[untyped]) 


48
 
# File 'sig/generated/job_workflow/instrumentation/opentelemetry_subscriber.rbs', line 48

def self.subscriptions=: (Array[untyped]) -> void

.unsubscribe! ⇒ void

This method returns an undefined value.

: () -> void



61
62
63
64
65
66
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 61

def unsubscribe!
  return if subscriptions.empty?

  subscriptions.each { |sub| ActiveSupport::Notifications.unsubscribe(sub) }
  self.subscriptions = []
end

Instance Method Details

#add_error_attributes(attrs, payload) ⇒ void

This method returns an undefined value.

: (Hash[String, untyped], Hash[Symbol, untyped]) -> void

Parameters:

  • (Hash[String, untyped])
  • (Hash[Symbol, untyped]) 


191
192
193
194
195
196
197
198
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 191

def add_error_attributes(attrs, payload)
  return unless payload[:error]

  attrs.merge!(
    Attributes::ERROR_CLASS => payload[:error_class] || payload[:error].class.name,
    Attributes::ERROR_MESSAGE => payload[:error_message] || payload[:error].message
  )
end

#attach_span_context(span) ⇒ Object

: (untyped) -> untyped

Parameters:

  • (Object)

Returns:

  • (Object) 


124
125
126
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 124

def attach_span_context(span)
  OpenTelemetry::Context.attach(OpenTelemetry::Trace.context_with_span(span))
end

#build_attributes(payload) ⇒ Hash[String, untyped]

: (Hash[Symbol, untyped]) -> Hash[String, untyped]

Parameters:

  • (Hash[Symbol, untyped])

Returns:

  • (Hash[String, untyped]) 


176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 176

def build_attributes(payload)
  attrs = {
    Attributes::JOB_NAME => payload[:job_name],
    Attributes::JOB_ID => payload[:job_id],
    Attributes::TASK_NAME => payload[:task_name],
    Attributes::TASK_EACH_INDEX => payload[:each_index],
    Attributes::TASK_RETRY_COUNT => payload[:retry_count],
    Attributes::CONCURRENCY_KEY => payload[:concurrency_key],
    Attributes::CONCURRENCY_LIMIT => payload[:concurrency_limit]
  }.compact
  add_error_attributes(attrs, payload)
  attrs
end

#build_span_name(event_name, payload) ⇒ String

: (String, Hash[Symbol, untyped]) -> String

Parameters:

  • (String)
  • (Hash[Symbol, untyped])

Returns:

  • (String) 


166
167
168
169
170
171
172
173
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 166

def build_span_name(event_name, payload)
  base_name = event_name.delete_suffix(".#{Instrumentation::NAMESPACE}")

  return "#{payload[:job_name]}.#{payload[:task_name]} #{base_name}" if payload[:task_name]
  return "#{payload[:job_name]} #{base_name}" if payload[:job_name]

  "JobWorkflow #{base_name}"
end

#detach_context_safe(token) ⇒ void

This method returns an undefined value.

: (untyped) -> void

Parameters:

  • (Object) 


159
160
161
162
163
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 159

def detach_context_safe(token)
  OpenTelemetry::Context.detach(token) if token
rescue StandardError => e
  handle_error(e)
end

#determine_span_kind(event_name) ⇒ Symbol

: (String) -> Symbol

Parameters:

  • (String)

Returns:

  • (Symbol) 


201
202
203
204
205
206
207
208
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 201

def determine_span_kind(event_name)
  case event_name
  when Events::TASK_ENQUEUE
    :producer
  else
    :internal
  end
end

#extract_otel_info(payload) ⇒ Array[untyped?]

: (Hash[Symbol, untyped]) -> Array

Parameters:

  • (Hash[Symbol, untyped])

Returns:

  • (Array[untyped?]) 


107
108
109
110
111
112
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 107

def extract_otel_info(payload)
  otel = payload.delete(:__otel)
  span = otel&.fetch(:span)
  token = otel&.fetch(:ctx_token)
  [span, token]
end

#finish(_name, _id, payload) ⇒ void

This method returns an undefined value.

: (String, String, Hash[Symbol, untyped]) -> void

Parameters:

  • (String)
  • (String)
  • (Hash[Symbol, untyped]) 


91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 91

def finish(_name, _id, payload)
  return unless self.class.opentelemetry_available?

  span, token = extract_otel_info(payload)
  return if span.nil? || token.nil?

  handle_exception(payload, span)
rescue StandardError => e
  handle_error(e)
ensure
  finish_span(span, token) if span || token
end

#finish_span(span, token) ⇒ void

This method returns an undefined value.

: (untyped, untyped) -> void

Parameters:

  • (Object)
  • (Object) 


143
144
145
146
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 143

def finish_span(span, token)
  finish_span_safe(span)
  detach_context_safe(token)
end

#finish_span_safe(span) ⇒ void

This method returns an undefined value.

: (untyped) -> void

Parameters:

  • (Object) 


149
150
151
152
153
154
155
156
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 149

def finish_span_safe(span)
  return unless span&.recording?

  span.status = OpenTelemetry::Trace::Status.ok if span.status.code == OpenTelemetry::Trace::Status::UNSET
  span.finish
rescue StandardError => e
  handle_error(e)
end

#handle_error(error) ⇒ void

This method returns an undefined value.

: (StandardError) -> void

Parameters:

  • (StandardError) 


216
217
218
219
220
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 216

def handle_error(error)
  return unless defined?(OpenTelemetry) && OpenTelemetry.respond_to?(:handle_error)

  OpenTelemetry.handle_error(exception: error, message: "JobWorkflow OpenTelemetry subscriber error")
end

#handle_exception(payload, span) ⇒ void

This method returns an undefined value.

: (Hash[Symbol, untyped], untyped) -> void

Parameters:

  • (Hash[Symbol, untyped])
  • (Object) 


134
135
136
137
138
139
140
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 134

def handle_exception(payload, span)
  error = payload[:error] || payload[:exception_object]
  return unless error

  span.record_exception(error)
  span.status = OpenTelemetry::Trace::Status.error("Unhandled exception: #{error.class}")
end

#start(name, _id, payload) ⇒ void

This method returns an undefined value.

: (String, String, Hash[Symbol, untyped]) -> void

Parameters:

  • (String)
  • (String)
  • (Hash[Symbol, untyped]) 


80
81
82
83
84
85
86
87
88
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 80

def start(name, _id, payload)
  return unless self.class.opentelemetry_available?

  span = start_span(name, payload)
  token = attach_span_context(span)
  store_span_info(payload, span, token)
rescue StandardError => e
  handle_error(e)
end

#start_span(name, payload) ⇒ Object

: (String, Hash[Symbol, untyped]) -> untyped

Parameters:

  • (String)
  • (Hash[Symbol, untyped])

Returns:

  • (Object) 


115
116
117
118
119
120
121
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 115

def start_span(name, payload)
  span_name = build_span_name(name, payload)
  attributes = build_attributes(payload)
  kind = determine_span_kind(name)

  tracer.start_span(span_name, kind:, attributes:)
end

#store_span_info(payload, span, token) ⇒ void

This method returns an undefined value.

: (Hash[Symbol, untyped], untyped, untyped) -> void

Parameters:

  • (Hash[Symbol, untyped])
  • (Object)
  • (Object) 


129
130
131
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 129

def store_span_info(payload, span, token)
  payload[:__otel] = { span: span, ctx_token: token }
end

#tracer ⇒ Object

: () -> untyped

Returns:

  • (Object) 


211
212
213
 
# File 'lib/job_workflow/instrumentation/opentelemetry_subscriber.rb', line 211

def tracer
  OpenTelemetry.tracer_provider.tracer(NAMESPACE, JobWorkflow::VERSION)
end