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.



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

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) 


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

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) 


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

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.



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

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

#current ⇒ Object

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



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

def current
  ExecutionState[current_key]
end

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

Finishes an active handle and returns the final attribute hash.



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

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.



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

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.



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

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.



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

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.



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

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.



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
179
 
# File 'lib/little_ghost/instrumentation.rb', line 145

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.



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

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.



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

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.



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

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.



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

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