tracking

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 RecordConsume ¶ added in v0.60.0

func RecordConsume(ctx context.Context, attrs ConsumeAttributes, duration time.Duration, err error)

RecordConsume records one finished delivery on the receive instruments both messaging lanes share: the duration histogram, and the consumed counter, which increments regardless of the outcome — the message WAS consumed — with error.type separating the failures.

Each lane's consume path calls this exactly once per message, at completion.

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 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.