Class: LittleGhost::Instrumentation::Bus

Inherits:
Object
  • Object
show all
Defined in:
lib/little_ghost/instrumentation.rb

Overview

Thread-safe notification bus used by the process-wide Instrumentation API.

Instance Method Summary collapse

Constructor Details

#initialize(subscribers: [], content_capture: Support::ContentCapture.disabled) ⇒ Bus

Starts an independent bus with ordered subscribers and a content policy.



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

def initialize(subscribers: [], content_capture: Support::ContentCapture.disabled)
  @subscribers = []
  @content_capture = content_capture
  @handles = {}
  @children = Hash.new(0)
  @finishing = {}
  @mutex = Mutex.new
  @reported_failures = {}
  @shutdown = false
  Array(subscribers).each { |subscriber| subscribe(subscriber) }
end

Instance Method Details

#active?(handle = nil) ⇒ Boolean

With a handle, tests whether that exact handle is active. Without one, reports whether the bus owns any active operations.

Returns:

  • (Boolean) 


226
227
228
229
230
 
# File 'lib/little_ghost/instrumentation.rb', line 226

def active?(handle = nil)
  @mutex.synchronize do
    handle ? @handles[handle.operation_id].equal?(handle) : !@handles.empty?
  end
end

#capture_content(policy) ⇒ Object

Selects the diagnostic content policy used for future notifications.

Raises:

  • (ArgumentError) 


125
126
127
128
129
130
 
# File 'lib/little_ghost/instrumentation.rb', line 125

def capture_content(policy)
  raise ArgumentError, "content capture policy must respond to capture" unless policy.respond_to?(:capture)

  @mutex.synchronize { @content_capture = policy }
  policy
end

#context ⇒ Object

Copies the attributes active in the current execution.



215
216
217
 
# File 'lib/little_ghost/instrumentation.rb', line 215

def context
  deep_copy(ExecutionState[context_key] || {})
end

#current ⇒ Object

Finds the current non-detached Handle for this fiber, if any.



220
221
222
 
# File 'lib/little_ghost/instrumentation.rb', line 220

def current
  ExecutionState[current_key]
end

#finish(handle, diagnostic: nil, **attributes) ⇒ Object

Finishes an active handle and returns the final attribute hash.



181
182
183
184
185
186
187
188
189
190
 
# File 'lib/little_ghost/instrumentation.rb', line 181

def finish(handle, diagnostic: nil, **attributes)
  validate_finish!(handle)
  validated = true
  values = handle.payload.merge(attributes).merge(duration_ms: elapsed_ms(handle.started_at))
  values = prepare_attributes(values.compact, diagnostic:)
  notify(:finish, handle.name, values)
  values
ensure
  complete(handle) if validated && active_handle?(handle)
end

#flush(timeout: nil) ⇒ Object

Flushes subscribers in registration order within an optional total timeout budget.



234
235
236
237
238
239
240
 
# File 'lib/little_ghost/instrumentation.rb', line 234

def flush(timeout: nil)
  deadline = monotonic_time + Float(timeout) if timeout
  subscribers.each do |subscriber|
    remaining = deadline && [deadline - monotonic_time, 0].max
    notify_subscriber(subscriber, :flush, timeout: remaining)
  end
end

#instrument(name, payload = {}) ⇒ Object

Measures a block and records raised errors before re-raising them.



193
194
195
196
197
198
199
200
201
202
203
204
205
206
 
# File 'lib/little_ghost/instrumentation.rb', line 193

def instrument(name, payload = {})
  values = payload.dup
  handle = start(name, **values)
  result = yield values if block_given?
  handle.finish(**values)
  result
rescue => error
  values ||= payload.dup
  values[:outcome] ||= :error
  values[:error_type] ||= error.class.name
  values[:diagnostic] ||= {exception: diagnostic_exception(error)}
  handle.finish(**values) if handle && active_handle?(handle)
  raise
end

#publish(name, diagnostic: nil, **attributes) ⇒ Object

Publishes a point event with the current context and operation ID.



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

def publish(name, diagnostic: nil, **attributes)
  current_handle = current
  values = context.merge(attributes)
  values[:operation_id] ||= current_handle&.operation_id
  values = prepare_attributes(values.compact, diagnostic:)
  notify(:emit, name.to_sym, values)
  values
end

#shutdown(timeout: nil) ⇒ Object

Permanently shuts down this bus after all operations have finished.



243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/little_ghost/instrumentation.rb', line 243

def shutdown(timeout: nil)
  should_shutdown = @mutex.synchronize do
    return if @shutdown
    raise Error, "cannot shut down instrumentation with active operations" unless @handles.empty?

    @shutdown = true
  end
  return unless should_shutdown

  deadline = monotonic_time + Float(timeout) if timeout
  subscribers.reverse_each do |subscriber|
    remaining = deadline && [deadline - monotonic_time, 0].max
    notify_subscriber(subscriber, :shutdown, timeout: remaining)
  end
end

#start(name, parent: current, operation_id: SecureRandom.uuid, detached: false, diagnostic: nil, **attributes) ⇒ Object

Starts an operation. parent may be a local Handle, a remote operation ID, or nil. Set detached for work that will not finish in stack order.



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/little_ghost/instrumentation.rb', line 144

def start(name, parent: current, operation_id: SecureRandom.uuid, detached: false, diagnostic: nil, **attributes)
  previous = current unless detached
  validate_parent!(parent)
  parent_operation_id = parent.is_a?(Handle) ? parent.operation_id : parent
  payload = context.merge(attributes).merge(operation_id:, parent_operation_id:).compact
  payload = prepare_attributes(payload, diagnostic:)
  handle = Handle.new(
    self,
    name,
    operation_id:,
    parent_operation_id:,
    local_parent: parent.is_a?(Handle),
    previous:,
    payload:,
    started_at: monotonic_time,
    detached:
  )
  @mutex.synchronize do
    raise Error, "instrumentation is shut down" if @shutdown
    raise ArgumentError, "instrumentation operation is already active" if @handles.key?(operation_id)
    if parent.is_a?(Handle)
      raise Error, "instrumentation parent is not active" unless @handles[parent.operation_id].equal?(parent)
      raise Error, "instrumentation parent is finishing" if @finishing.key?(parent.operation_id)
    end

    @handles[operation_id] = handle
    @children[parent_operation_id] += 1 if parent.is_a?(Handle)
  end
  set_current(handle) unless detached
  notify(:start, handle.name, handle.payload)
  handle
rescue
  abandon(handle) if handle
  raise
end

#subscribe(subscriber, prepend: false) ⇒ Object

Subscribes a backend once. prepend controls notification order.



106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/little_ghost/instrumentation.rb', line 106

def subscribe(subscriber, prepend: false)
  unless subscriber.is_a?(Subscriber)
    raise ArgumentError, "instrumentation subscriber must be a LittleGhost::Instrumentation::Subscriber"
  end

  @mutex.synchronize do
    @subscribers.reject! { |listener| listener.equal?(subscriber) }
    prepend ? @subscribers.unshift(subscriber) : @subscribers << subscriber
  end
  subscriber
end

#trace_context(**attributes) ⇒ Object

Uses the first non-empty downstream trace context supplied by a subscriber.



260
261
262
263
264
265
266
267
268
 
# File 'lib/little_ghost/instrumentation.rb', line 260

def trace_context(**attributes)
  subscribers.each do |subscriber|
    value = subscriber.trace_context(**attributes)
    return value unless value.nil? || value.empty?
  rescue => error
    warn_failure(error, component: :subscriber)
  end
  {}
end

#unsubscribe(subscriber) ⇒ Object

Unsubscribes a backend by identity.



119
120
121
122
 
# File 'lib/little_ghost/instrumentation.rb', line 119

def unsubscribe(subscriber)
  @mutex.synchronize { @subscribers.reject! { |listener| listener.equal?(subscriber) } }
  subscriber
end

#with_context(attributes) ⇒ Object

Adds copied attributes to notifications emitted while the block runs.



209
210
211
212
 
# File 'lib/little_ghost/instrumentation.rb', line 209

def with_context(attributes)
  values = deep_copy(context).merge(deep_copy(attributes.compact))
  ExecutionState.with(context_key => values) { yield }
end