Class: AceMQ::AMQP::Telemetry::OpenTelemetry

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/telemetry/open_telemetry.rb

Overview

Emits OpenTelemetry spans for publishes and deliveries.

tracing = AceMQ::AMQP::Telemetry::OpenTelemetry.new
tracing.install(mq)

The point of tracing a message system is the join: the span covering a handler must be a child of the span that published the message, even though the two ran in different processes minutes apart. That is what the traceparent header carries, and why it is written on the way out and read on the way in.

It is an interceptor, not a telemetry observer

AceMQ::AMQP::Telemetry here counts things — three methods, no dependency, and a Registry for when the numbers are all that is wanted. Tracing is a different shape: a span wraps a publish or a handler, so it has to know when one starts and when it ends, and it has to be able to change the message on the way out. Interceptors is exactly that seam, and it is public, so everything below could have been written outside this gem.

Register on both sides, which #install does:

mq.intercept_publish(tracing)
mq.intercept_consume(tracing)

The names on the wire

traceparent and tracestate, which are the W3C names and are deliberately not x-acemq- prefixed: other tooling already knows them, and renaming them would make this library's traces invisible to everything that did not know to look. The Java library writes the same two, so a Ruby consumer joins a Java producer's trace without either side being configured for the other.

Span names, kinds and attributes

Shared with the Java adapter, which follows the OpenTelemetry messaging conventions:

<destination> publish   PRODUCER
<queue> process         CONSUMER
<destination> request   CLIENT

CLIENT for a request rather than PRODUCER because that span waits for an answer, and its duration means something different as a result — a reader who cannot tell the two apart cannot tell a slow broker from a slow responder.

Events rather than spans

A retry, a dead letter, an outbox failure and a finished pipeline run are events on the span that is already open. A zero-length span at the end of a trace adds a row and no information.

The gem

opentelemetry-api is required when one of these is built, not when this file is read, and the gem declares no runtime dependencies at all. A process that publishes messages and traces nothing installs nothing.

Defined Under Namespace

Classes: Scope

Constant Summary collapse

INSTRUMENTATION_NAME =

What the tracer is registered under, in every AceMQ library.

"org.acemq.amqp"
TRACEPARENT =

The W3C names, which are what makes a trace legible to anything else.

"traceparent"
TRACESTATE = 
"tracestate"
SYSTEM =

The attributes, shared with the Java adapter.

"messaging.system"
DESTINATION = 
"messaging.destination.name"
OPERATION = 
"messaging.operation"
MESSAGE_ID = 
"messaging.message.id"
CONVERSATION_ID = 
"messaging.message.conversation_id"
ROUTING_KEY = 
"messaging.rabbitmq.destination.routing_key"
MESSAGE_TYPE = 
"messaging.acemq.message_type"
ATTEMPT = 
"messaging.acemq.attempt"
OUTCOME = 
"messaging.acemq.outcome"
REASON = 
"messaging.acemq.reason"
OUTBOX_LAG = 
"messaging.acemq.outbox_lag_ms"
FAILED =

What a failure is called when nothing more specific was said. The same word the counters use, which is the whole point of having one.

"failed"
FAILING_OUTCOMES =

The outcomes that make a span an error, and the ones that do not.

retried and rejected are deliberately absent. A message that will be tried again has not failed yet, and a message the handler refused on purpose is the system working; marking either as an error is how a trace view fills with red and stops meaning anything.

parked is deliberately absent, for the same reason rejected is: a handler that parks a message meant to, and a decision somebody made is not a failure. Go and Python leave it out too, so a parked span reads the same in all three. The parking queue still has to be looked at -- that is what acemq.consume.total tagged outcome=parked is for.

refused is absent for a different reason: it is not the word that makes a refused publish's span red, the exception is, recorded by Scope#failed with the cause in the description. Python's adapter does the same, and the set stays the same as Java's.

%w[unroutable failed dead_lettered].freeze
ORDER =

Runs before every other interceptor on the way in and after every other one on the way out, so a span covers whatever they do.

-1000

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(tracer_provider: nil, propagator: nil, transport: "rabbitmq") ⇒ OpenTelemetry

Returns a new instance of OpenTelemetry.

Parameters:

  • tracer_provider (#tracer, nil) (defaults to: nil) —

    defaults to the process's

  • propagator (#inject, nil) (defaults to: nil) —

    defaults to the process's, which is tracecontext once the SDK has been configured

  • transport (String) (defaults to: "rabbitmq") —

    the messaging.system attribute

Raises:



136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 136

def initialize(tracer_provider: nil, propagator: nil, transport: "rabbitmq")
  self.class.load_api!
  @tracer = (tracer_provider || ::OpenTelemetry.tracer_provider)
            .tracer(INSTRUMENTATION_NAME, AceMQ::AMQP::VERSION)
  @propagator = propagator || ::OpenTelemetry.propagation
  @transport = transport
  # Per instance and per fiber, so two adapters on one connection do not
  # pop each other's spans and a handler running on another thread has
  # its own.
  @publishing = :"acemq_otel_publish_#{object_id}"
  @consuming = :"acemq_otel_consume_#{object_id}"
end

Instance Attribute Details

#tracer ⇒ Object (readonly)

The tracer this emits through, for a caller that wants to open a span of its own inside one of these.



151
152
153
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 151

def tracer
  @tracer
end

Class Method Details

.clean(attributes) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Attributes OpenTelemetry will take: nothing nil, nothing empty.

An empty string is an attribute that says nothing and costs a column in every backend that stores one. The Java adapter writes them because its API will not take a null; Ruby's will not take a nil either, and leaving the key out entirely is the better answer.



443
444
445
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 443

def self.clean(attributes)
  attributes.reject { |_, value| value.nil? || value == "" }
end

.load_api! ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Reaches for the gem, and says which one when it is not there.

The gemspec declares no runtime dependencies on purpose, so this is the first moment anybody finds out. "cannot load such file -- opentelemetry-api" does not say which library wanted it or what to do about it.

Raises:



456
457
458
459
460
461
462
463
464
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 456

def self.load_api!
  require "opentelemetry-api"
rescue LoadError => e
  raise DependencyMissing,
        "the AceMQ OpenTelemetry adapter needs the opentelemetry-api gem, which is " \
        "not installed. Add `gem \"opentelemetry-api\", \"~> 1.8\"` to your Gemfile " \
        "— this gem declares no runtime dependencies, so a process that traces " \
        "nothing installs nothing. (#{e.message})"
end

.outcome_of(ack) ⇒ String

What an ack alone is called on a span.

The same words the Java adapter writes, plus parked for the action Java's vocabulary does not have. dead_lettered rather than retried for a retry marked fatal, because that is what the consumer will actually do with it — honouring the mark rather than the request is the entire point of having it.

This is the fallback. What an ack cannot know is whether a retry has any attempts left, so a Settlement on the context wins over it; see #after_handle.

Parameters:

Returns:

  • (String) 


425
426
427
428
429
430
431
432
433
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 425

def self.outcome_of(ack)
  return "acked" if ack.accept?
  return "rejected" if ack.reject?
  return "parked" if ack.park?
  return "in_progress" if ack.in_progress?
  return "dead_lettered" if ack.error.is_a?(FatalError)

  "retried"
end

.timed_out?(failure) ⇒ Boolean

Whether a failure is a request that went unanswered.

Asked with defined? rather than named in a rescue clause, because Patterns::RequestTimedOut lives in the patterns file and a process that traces without ever asking anybody a question need not have loaded it — and a rescue naming a constant that is not there raises a NameError in place of the exception the caller threw.

Parameters:

  • failure (Exception)

Returns:

  • (Boolean) 


405
406
407
408
409
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 405

def self.timed_out?(failure)
  return false unless defined?(Patterns::RequestTimedOut)

  failure.is_a?(Patterns::RequestTimedOut)
end

Instance Method Details

#after_confirm(_context) ⇒ Object

The broker has the message.



335
336
337
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 335

def after_confirm(_context)
  finish(pop(@publishing), "confirmed")
end

#after_handle(context, ack) ⇒ Object

The handler has returned or raised, and the delivery has not been settled yet — which is the last moment the outcome and the span are both in hand.

The outcome comes off the context's Settlement when there is one, because that is what the consumer is about to do; the ack is only what the handler asked for. A retry on the last attempt is a dead letter, and reading it off the ack is how a dead-lettered message ends up with outcome=retried on its span and nothing in the trace to say it was dropped. The ack is still the fallback, for anything driving this interceptor without a consumer behind it.



356
357
358
359
360
361
362
363
364
365
366
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 356

def after_handle(context, ack)
  scope = pop(@consuming)
  return context if scope.nil?

  settlement = settlement_of(context)
  outcome = settlement&.outcome || self.class.outcome_of(ack)
  note(context, ack, settlement, outcome)
  scope.failed(ack.error) if ack.error.is_a?(Exception)
  finish(scope, outcome)
  context
end

#before_handle(context) ⇒ Object

Opens the handler span, joined to whoever published the message.



340
341
342
343
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 340

def before_handle(context)
  push(@consuming, consume_started(queue: context.queue, envelope: context.envelope))
  context
end

#before_publish(context) ⇒ Object

Opens the publish span and writes the trace into the message.

The headers are set through the context rather than on the envelope directly, so what goes on the wire is what every other interceptor sees too — and so an interceptor registered after this one can read the trace it is part of.



326
327
328
329
330
331
332
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 326

def before_publish(context)
  scope = publish_started(exchange: context.exchange, routing_key: context.routing_key,
                          envelope: context.envelope)
  push(@publishing, scope)
  propagation_headers.each { |name, value| context.set_header(name, value) }
  context
end

#consume_started(queue:, envelope:) ⇒ Scope

Opens the span covering a handler, joined to the publish that caused it.

The parent comes out of the message's own headers rather than out of whatever this thread happened to be doing, which is the whole point: the publish happened in another process, possibly minutes ago, and nothing in this one remembers it. Ambient context is the fallback and not the source, the same way round as the Java adapter.

Parameters:

Returns:



195
196
197
198
199
200
201
202
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 195

def consume_started(queue:, envelope:)
  parent = @propagator.extract(carrier_from(envelope.headers),
                               context: ::OpenTelemetry::Context.current)
  start("#{queue} process", :consumer,
        { SYSTEM => @transport, DESTINATION => queue.to_s, OPERATION => "process",
          ATTEMPT => envelope.attempt.to_i, **envelope_attributes(envelope) },
        parent: parent)
end

#install(connection) ⇒ OpenTelemetry

Registers on both sides of a connection.

Parameters:

Returns:



157
158
159
160
161
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 157

def install(connection)
  connection.intercept_publish(self)
  connection.intercept_consume(self)
  self
end

#message_dead_lettered(queue:, envelope:, reason:) ⇒ Object

Parameters:

  • queue (String)
  • envelope (Envelope)
  • reason (String) —

    unbounded text, which a span tolerates and a metric does not



312
313
314
315
316
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 312

def message_dead_lettered(queue:, envelope:, reason:)
  record("message.dead_lettered",
         { DESTINATION => queue.to_s, ATTEMPT => envelope.attempt.to_i,
           REASON => reason.to_s })
end

#message_retried(queue:, envelope:, delay: nil) ⇒ Object

Parameters:

  • queue (String)
  • envelope (Envelope)
  • delay (Numeric, nil) (defaults to: nil) —

    seconds this message will wait



302
303
304
305
306
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 302

def message_retried(queue:, envelope:, delay: nil)
  attributes = { DESTINATION => queue.to_s, ATTEMPT => envelope.attempt.to_i }
  attributes["messaging.acemq.retry_delay_ms"] = (delay * 1000).to_i if delay
  record("message.retried", attributes)
end

#on_error(context, failure) ⇒ Object

A publish that never reached the broker or reached no queue, or a handler that raised.

On the publish side this is the only hook that runs, so the span is closed here. On the consume side after_handle still runs afterwards and closes it, so this records the exception and leaves it open.

A message the broker took and could not route is unroutable rather than failed, and carries the broker's own words as the reason. Both are errors — a publish that went nowhere is not a success — but they send whoever is reading the trace to two different places, and Go splits its publish span the same way for the same reason.



380
381
382
383
384
385
386
387
388
389
390
391
392
393
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 380

def on_error(context, failure)
  if context.is_a?(PublishContext)
    scope = pop(@publishing)
    outcome = Telemetry::Outcome.of_publish_failure(failure)
    if scope && outcome == Telemetry::Outcome::UNROUTABLE
      scope.span.set_attribute(REASON, failure.message.to_s)
    end
    scope&.failed(failure)
    finish(scope, outcome)
  else
    peek(@consuming)&.failed(failure)
  end
  context
end

#order ⇒ Object 



163
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 163

def order = ORDER

#outbox_publish_failed(exchange:, reason:) ⇒ Object

Parameters:

  • exchange (String)
  • reason (String) 


266
267
268
269
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 266

def outbox_publish_failed(exchange:, reason:)
  record("outbox.publish_failed",
         { DESTINATION => exchange.to_s, REASON => reason.to_s })
end

#outbox_published(lag:) ⇒ Object

Records how far behind the outbox relay is running.

An attribute on the current span rather than an event, and the Java, Go and Python adapters write the same one: it measures the publish that is happening, not something that happened during it. Nothing when no span is open, for the same reason #outbox_publish_failed records nothing then.

Parameters:

  • lag (Numeric) —

    seconds the record sat before it was published



280
281
282
283
284
285
286
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 280

def outbox_published(lag:)
  span = ::OpenTelemetry::Trace.current_span
  return unless span.recording?

  span.set_attribute(OUTBOX_LAG, (lag * 1000).to_i)
  nil
end

#pipeline_run_finished(pipeline:, step:, outcome:, age: nil) ⇒ Object

Parameters:

  • pipeline (String)
  • step (String)
  • outcome (String)
  • age (Numeric, nil) (defaults to: nil) —

    seconds since the run started



292
293
294
295
296
297
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 292

def pipeline_run_finished(pipeline:, step:, outcome:, age: nil)
  attributes = { "pipeline" => pipeline.to_s, "step" => step.to_s,
                 "outcome" => outcome.to_s }
  attributes["messaging.acemq.run_age_ms"] = (age * 1000).to_i if age
  record("pipeline.run_finished", attributes)
end

#propagation_headers ⇒ Hash{String=>String}

The current trace context, as headers.

For anything publishing outside this library — an outbox relay writing rows, a scheduled job — that still wants its messages to join the trace it is running in.

Returns:

  • (Hash{String=>String}) 


256
257
258
259
260
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 256

def propagation_headers
  carrier = {}
  @propagator.inject(carrier, context: ::OpenTelemetry::Context.current)
  carrier
end

#publish_started(exchange:, routing_key:, envelope:) ⇒ Scope

Opens the span covering a publish, and makes it current.

Parameters:

  • exchange (String) —

    empty for the default exchange

  • routing_key (String)
  • envelope (Envelope)

Returns:



173
174
175
176
177
178
179
180
181
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 173

def publish_started(exchange:, routing_key:, envelope:)
  # The exchange is the destination when there is one. Publishing
  # through the default exchange has no exchange to name, and the
  # routing key is the queue, which is what a reader wants to see.
  destination = exchange.to_s.empty? ? routing_key.to_s : exchange.to_s
  start("#{destination} publish", :producer,
        { SYSTEM => @transport, DESTINATION => destination, OPERATION => "publish",
          ROUTING_KEY => routing_key.to_s, **envelope_attributes(envelope) })
end

#request(destination, envelope: nil) ⇒ Object

#request_started with the closing done for you.

Parameters:

  • destination (String)
  • envelope (Envelope, nil) (defaults to: nil)

Yield Returns:

  • (Object) —

    whatever the block returns



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 225

def request(destination, envelope: nil)
  scope = request_started(destination: destination, envelope: envelope)
  answer = yield
  scope.outcome("answered")
  answer
rescue StandardError => e
  # +timed_out+ rather than +failed+ for a request nobody answered in
  # time, and deliberately not an error: that is the absence of a reply
  # rather than a failure of this process — usually a responder's queue
  # being long — and the Java and Go adapters both write it without
  # marking the span red, because a trace view that fills with red for
  # slow responders stops meaning anything. The exception still reaches
  # the caller, who is free to decide it was one.
  if self.class.timed_out?(e)
    scope&.outcome("timed_out")
  else
    scope&.outcome("failed")
    scope&.failed(e)
  end
  raise
ensure
  scope&.close
end

#request_started(destination:, envelope: nil) ⇒ Scope

Opens the span covering a request that waits for an answer.

tracing.request("pricing.quote") { requester.call(order) }

CLIENT rather than PRODUCER, because a reader has to know that this span's duration includes somebody else's work.

Parameters:

  • destination (String)
  • envelope (Envelope, nil) (defaults to: nil)

Returns:



214
215
216
217
218
 
# File 'lib/acemq/amqp/telemetry/open_telemetry.rb', line 214

def request_started(destination:, envelope: nil)
  start("#{destination} request", :client,
        { SYSTEM => @transport, DESTINATION => destination.to_s,
          OPERATION => "request", **envelope_attributes(envelope) })
end