tracking

func RecordAMQPConsumeCompletion ¶ added in v0.58.1

func RecordAMQPConsumeCompletion(ctx context.Context, delivery *amqp.Delivery, queueName string, duration time.Duration, err error)

RecordAMQPConsumeCompletion records the receive duration histogram for a delivery whose handling has finished.

It deliberately does NOT touch messaging.client.consumed.messages. That counter has a single owner — StartConsumeSpan, which increments it once per delivery received — so a delivery is counted exactly once regardless of how its handler ended. Incrementing here too would double count every message.

func RecordAMQPConsumeMetrics ¶

func RecordAMQPConsumeMetrics(ctx context.Context, delivery *amqp.Delivery, queueName string, duration time.Duration, err error)

RecordAMQPConsumeMetrics records OpenTelemetry metrics for an AMQP consume operation. This function is called automatically when a message is consumed to emit metrics.

Metrics recorded: - messaging.client.operation.duration: Histogram of operation durations in seconds - messaging.client.consumed.messages: Counter of messages consumed

The function is non-blocking and handles errors gracefully.

func RecordAMQPPublishMetrics ¶

func RecordAMQPPublishMetrics(ctx context.Context, exchange, routingKey string, duration time.Duration, err error)

RecordAMQPPublishMetrics records OpenTelemetry metrics for an AMQP publish operation. This function is called after a publish attempt to emit metrics about the operation.

Parameters:

• ctx: Context for metrics recording
• exchange: The AMQP exchange name (empty string for default exchange)
• routingKey: The routing key used for message delivery
• duration: Time taken for the publish operation
• err: Error if the operation failed, nil if successful

Metrics recorded: - messaging.client.operation.duration: Histogram of operation durations in seconds - messaging.client.sent.messages: Counter of messages sent (incremented on success)

The function is non-blocking and handles errors gracefully - metric recording failures will not impact messaging operation execution.

func RecordChannelEvent ¶

func RecordChannelEvent(eventType string, err error)

RecordChannelEvent records AMQP channel lifecycle events. eventType should be "create" or "close".

func RecordConnectionEvent ¶

func RecordConnectionEvent(eventType string, err error)

RecordConnectionEvent records AMQP connection lifecycle events. eventType should be "create" or "close".

func RecordPublishRetry ¶

func RecordPublishRetry(ctx context.Context, exchange, routingKey, reason string)

RecordPublishRetry records a publish retry attempt in the retry counter. This is called each time a publish operation is retried due to NACK, timeout, or error.

Parameters:

• ctx: Context for metrics recording
• exchange: The AMQP exchange name (empty string for default exchange)
• routingKey: The routing key used for message delivery
• reason: The reason for the retry (e.g., "nack", "timeout", "publish_error")

func RecordStreamConsume ¶ added in v0.59.0

func RecordStreamConsume(ctx context.Context, streamName string, duration time.Duration, err error)

RecordStreamConsume records the metrics for one native stream-protocol delivery, reusing the AMQP instruments so both messaging lanes report under the same names.

Unlike the AMQP lane — where StartConsumeSpan counts the delivery at receive time and RecordAMQPConsumeCompletion only times it — a stream delivery is recorded exactly once, after its handler returned. The consumed counter therefore increments regardless of the handler outcome (the message WAS consumed from the stream); error.type separates the failures.

func RecordStreamPublish ¶ added in v0.59.0

func RecordStreamPublish(ctx context.Context, streamName string, duration time.Duration, err error)

RecordStreamPublish records the metrics for one native stream-protocol publish, reusing the AMQP publish instruments so both messaging lanes report under the same names.

Like the AMQP lane, the sent counter increments only once the publish succeeded — here that means the broker confirmed it.

func ResetMeterForTesting ¶ added in v0.58.1

func ResetMeterForTesting()

ResetMeterForTesting resets the package-level meter state so a test starts with instruments bound to the currently installed global MeterProvider. Intended for messaging-package tests, which cannot reach the unexported state. Not safe against a concurrent getAMQPMeter: that path runs meterOnce.Do without meterInitMu, so callers must ensure no goroutine is recording metrics when this runs.