Documentation
¶
Overview ¶
Package otel adds optional OpenTelemetry tracing and metrics to a babelqueue producer or consumer. It lives in its own module so the zero-dependency core never imports OpenTelemetry — wiring a TracerProvider (and a MeterProvider) is opt-in.
Span shape (ADR-0038, semantic conventions v1.44.0): a PRODUCER span "send <queue>" and a CONSUMER span "process <queue>", with messaging.system set by WithSystem (the broker, e.g. "kafka"), messaging.operation.type / .name, the babelqueue.* identity attributes, and error.type on a failure. The URN is an attribute, never part of the span name.
Metrics (ADR-0039): pass a meter with WithMeter and the wrappers record messaging.process.duration, messaging.client.sent.messages, messaging.client.consumed.messages and babelqueue.retries; NewDeadLetterObserver adds babelqueue.dlq.messages. Without a meter nothing is recorded.
Cross-hop trace propagation works at two levels, layered:
- trace_id ↔ TraceID (ADR-0025, v0.1): the envelope's trace_id maps 1:1 to an OTel TraceID, so every hop that shares a trace_id shares one trace — correlation and per-hop timing with zero wire/transport change.
- W3C traceparent (ADR-0028, v0.2): the producer also injects the active span context as a traceparent transport header (beside the frozen envelope, never in it), so the consumer starts its span as a true child of the producer span — real cross-hop parent-child linkage. This rides the out-of-band HeaderPublisher / ReceivedMessage.Headers seam (ADR-0027) and is available on any transport that carries headers. When no traceparent header is present it falls back to the v0.1 trace_id behaviour — a strict, backward- compatible upgrade.
Index ¶
- Constants
- func NewDeadLetterObserver(opts ...Option) babelqueue.DeadLetterObserver
- func Publish(ctx context.Context, tracer trace.Tracer, app *babelqueue.App, urn string, ...) (string, error)
- func TraceIDOf(traceID string) trace.TraceID
- func UUIDOf(t trace.TraceID) string
- func WrapHandler(tracer trace.Tracer, handler babelqueue.Handler, opts ...Option) babelqueue.Handler
- type Extracted
- type Option
- type Publisher
Constants ¶
const ( // MetricProcessDuration is the handler's wall time in seconds, recorded once per invocation, // success or failure (histogram, unit "s"). MetricProcessDuration = "messaging.process.duration" // MetricSentMessages counts publish attempts, success or failure (counter, unit "{message}"). MetricSentMessages = "messaging.client.sent.messages" // MetricConsumedMessages counts the deliveries that enter the consumer wrapper (counter). MetricConsumedMessages = "messaging.client.consumed.messages" // MetricRetries counts the deliveries that enter the wrapper with top-level attempts > 0 (counter). MetricRetries = "babelqueue.retries" // MetricDeadLetters counts the dead-letters the SDK carried out (counter), fed by the // observer from [NewDeadLetterObserver]. MetricDeadLetters = "babelqueue.dlq.messages" )
The standard metric set (ADR-0039). Names, units and kinds are identical in every BabelQueue SDK.
const ( HeaderTraceparent = "traceparent" HeaderTracestate = "tracestate" )
HeaderTraceparent / HeaderTracestate are the out-of-band transport headers that carry W3C Trace Context across a hop (ADR-0028, implementing ADR-0025 Option 2). They ride beside the frozen envelope on the transport's per-message metadata channel — the same out-of-band seam as the replay-bypass marker (ADR-0027) — so a consumer can start its span as a true child of the producer's span, not merely share the trace_id-derived trace. The envelope is never touched (GR-1).
const ( // SemconvVersion is the OpenTelemetry semantic-conventions release this module targets // (ADR-0038 section 1). SemconvVersion = "1.44.0" // SchemaURL is the schema URL of [SemconvVersion]. This module does not create the tracer or // the meter, so create them with it: // // tp.Tracer("my-app", trace.WithSchemaURL(otel.SchemaURL)) // mp.Meter("my-app", metric.WithSchemaURL(otel.SchemaURL)) SchemaURL = "https://opentelemetry.io/schemas/" + SemconvVersion )
Variables ¶
This section is empty.
Functions ¶
func NewDeadLetterObserver ¶ added in v0.4.0
func NewDeadLetterObserver(opts ...Option) babelqueue.DeadLetterObserver
NewDeadLetterObserver returns the babelqueue.DeadLetterObserver that feeds babelqueue.dlq.messages (ADR-0039 section 5): +1 per dead-letter the App carried out, with messaging.system from WithSystem, messaging.operation.name "process" and messaging.destination.name the source queue when known. Register it on the runtime:
app := babelqueue.NewApp(transport,
babelqueue.WithDeadLetterObserver(otel.NewDeadLetterObserver(
otel.WithSystem("kafka"), otel.WithMeter(meter))))
Without WithMeter the observer records nothing.
func Publish ¶
func Publish( ctx context.Context, tracer trace.Tracer, app *babelqueue.App, urn string, data map[string]any, opts ...babelqueue.Option, ) (string, error)
Publish is Publisher.Publish with the default options: the deprecated messaging.system "babelqueue" and no metrics. Use NewPublisher to pass WithSystem or WithMeter.
func TraceIDOf ¶
TraceIDOf maps an envelope trace_id to a deterministic OTel TraceID: a UUID maps to its 16 raw bytes; any other string is hashed (SHA-256, first 16 bytes). The inverse of UUIDOf for the UUID case.
func UUIDOf ¶
UUIDOf formats an OTel TraceID (16 bytes) as a canonical UUID string — the form a producer stamps into the message's trace_id so a consumer can recover the same TraceID.
func WrapHandler ¶
func WrapHandler(tracer trace.Tracer, handler babelqueue.Handler, opts ...Option) babelqueue.Handler
WrapHandler returns handler decorated to emit a CONSUMER span "process <queue>" per message (ADR-0038), recording the handler's error/status, and, when WithMeter is given, the consumer metrics (ADR-0039). Register it like any handler: app.Handle(urn, otel.WrapHandler(tracer, handler, otel.WithSystem("kafka"))).
The span name and messaging.destination.name come from the delivered envelope's meta.queue; when it is absent or blank the span is just "process" and carries no destination. The options are additive: WrapHandler(tracer, handler) keeps working and writes the deprecated default messaging.system "babelqueue".
Parent selection (ADR-0028 / ADR-0025 Option 2): when the producer carried a W3C traceparent on the transport (surfaced by the runtime via babelqueue.HeadersFromContext), the span is started as a true child of the producer's span — real cross-hop parent→child linkage with per-hop span timing. When no traceparent header is present, it falls back to the v0.1 behaviour: a remote parent derived from the envelope's trace_id (ADR-0025 Option 1), which shares the trace but not exact span linkage. So enabling traceparent propagation is a strict, backward-compatible upgrade — no regression for messages produced without it. A trace_id that is not a UUID is preserved untouched and mapped to a TraceID by hashing; that is logged as a warning, once per distinct value.
Types ¶
type Extracted ¶ added in v0.3.0
type Extracted struct {
// Source is "traceparent" when a structurally valid header was delivered, else
// "trace_id" (the v0.1 fallback of ADR-0025).
Source string
// TraceIDHex is the header's 32-hex trace id, or the envelope trace_id without hyphens.
TraceIDHex string
// ParentSpanID (16 hex) and Flags (2 hex) are set only when Source is "traceparent".
ParentSpanID string
Flags string
}
Extracted describes where a consumer's trace identity came from and its raw hex parts. It is the pure, header-level view of the ADR-0028 extraction contract that the cross-SDK conformance section `traceparent` pins; WrapHandler performs the same decision with the OTel propagator and live span contexts.
func ExtractTraceparent ¶ added in v0.3.0
ExtractTraceparent resolves the trace identity for a delivered message from the out-of-band traceparent header value (empty when none was delivered) and the envelope trace_id. It is a thin wrapper over [remoteParentFromHeaders], the exact extraction WrapHandler performs at runtime, so what it reports is what a consumer span really gets: a header the W3C propagator rejects (malformed, all-zero ids, uppercase hex, ...) is ignored — it never fails — and behaves exactly like no header. The envelope trace_id is never modified (GR-4).
type Option ¶ added in v0.4.0
type Option func(*config)
Option configures WrapHandler, NewPublisher and NewDeadLetterObserver.
func WithConsumerGroup ¶ added in v0.4.0
WithConsumerGroup sets messaging.consumer.group.name on the consumer span (ADR-0038 section 4), intended for Kafka. It is written whenever given and non-empty; it is never a metric attribute.
func WithMeter ¶ added in v0.4.0
WithMeter turns on the standard metrics (ADR-0039): the instruments are created from m once per wrapper. Without it the wrappers record no metric. Create the meter with metric.WithSchemaURL(SchemaURL).
func WithSystem ¶ added in v0.4.0
WithSystem sets messaging.system to the broker name (ADR-0038 section 3): "redis", "rabbitmq", "aws_sqs", "servicebus", "pulsar", "kafka" or "activemq". Any other value is written as given, unvalidated. An empty value counts as absent, which keeps the deprecated default "babelqueue".
type Publisher ¶ added in v0.4.0
type Publisher struct {
// contains filtered or unexported fields
}
Publisher publishes through a babelqueue.App under a PRODUCER span "send <queue>" (ADR-0038) and, when WithMeter is given, counts the attempt (ADR-0039). Build it once and reuse it: the metric instruments are created per Publisher, not per message.
func NewPublisher ¶ added in v0.4.0
NewPublisher returns a Publisher over app. The options are those of WrapHandler; the consumer-only WithConsumerGroup is accepted and ignored.
func (*Publisher) Publish ¶ added in v0.4.0
func (p *Publisher) Publish(ctx context.Context, urn string, data map[string]any, opts ...babelqueue.Option) (string, error)
Publish starts a PRODUCER span "send <queue>" and publishes via app, propagating the trace to the downstream consumer two ways:
- It injects the active span context as a W3C traceparent (and tracestate) onto the outgoing transport headers, so a consumer can start its span as a true child of this producer span — real cross-hop parent→child linkage (ADR-0028 / ADR-0025 Option 2). The header rides beside the frozen envelope, never in it (GR-1), via babelqueue.App.PublishEnvelopeWithHeaders; if the transport can't carry headers it degrades to a plain publish (the traceparent is simply not propagated — no error).
- It also carries the active trace's TraceID into the message's trace_id (the v0.1 behaviour, ADR-0025 Option 1), so even a consumer that ignores the header — or a transport that drops it — still recovers the same trace (correlation without exact span linkage).
The span starts as the bare "send": the queue is only known from the envelope the App builds (a per-call babelqueue.WithQueue or the app's default queue), so once the envelope has been handed to the transport the span is renamed "send <queue>" and gets messaging.destination.name, messaging.message.id and messaging.message.conversation_id. When no envelope is ever handed to the transport (the build or the encode failed) the span stays "send" and carries none of those, never a guessed destination.
It otherwise behaves like babelqueue.App.Publish, returning the message id.
Source Files