Documentation
¶
Index ¶
- Constants
- func RecordAMQPPublishMetrics(ctx context.Context, exchange, routingKey string, duration time.Duration, ...)
- func RecordChannelEvent(eventType string, err error)
- func RecordConnectionEvent(eventType string, err error)
- func RecordConsume(ctx context.Context, attrs ConsumeAttributes, duration time.Duration, ...)
- func RecordPublishRetry(ctx context.Context, exchange, routingKey, reason string)
- func RecordSettlement(lane, successOutcome string, err error)
- func RecordStreamPublish(ctx context.Context, streamName string, duration time.Duration, err error)
- func ResetMeterForTesting()
- type ConsumeAttributes
Constants ¶
const ( // Settlement lanes and outcomes. The attribute set is exactly these two // keys; a settle error records OutcomeFailed instead of the success value. LaneClassic = "classic" LaneStreams = "streams" OutcomeAcked = "acked" OutcomeNacked = "nacked" OutcomeCommitted = "committed" OutcomeFailed = "failed" )
Variables ¶
This section is empty.
Functions ¶
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 ¶
RecordChannelEvent records AMQP channel lifecycle events. eventType should be "create" or "close".
func RecordConnectionEvent ¶
RecordConnectionEvent records AMQP connection lifecycle events. eventType should be "create" or "close".
func RecordConsume ¶ added in v0.60.0
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 ¶
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 RecordSettlement ¶ added in v0.61.0
RecordSettlement records one settle result after the lane's ack, nack, or offset-commit call returns. A non-nil err records outcome=failed; otherwise successOutcome is recorded. Attributes are exactly {lane, outcome}.
The consume path initializes the instruments first, so this does not call getAMQPMeter: a leftover worker settling after a test reset must not race meterOnce. The counter pointer is snapshotted under meterInitMu so a concurrent ResetMeterForTesting cannot tear the read.
func RecordStreamPublish ¶ added in v0.59.0
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.
Types ¶
type ConsumeAttributes ¶ added in v0.60.0
type ConsumeAttributes struct {
// contains filtered or unexported fields
}
ConsumeAttributes identifies one consumed message on the receive instruments. A lane builds it once per message through AMQPConsumeAttributes or StreamConsumeAttributes; the lane's consume path decides when it is recorded.
func AMQPConsumeAttributes ¶ added in v0.60.0
func AMQPConsumeAttributes(exchange, routingKey, queue string) ConsumeAttributes
AMQPConsumeAttributes identifies a classic-lane delivery: the OTel RabbitMQ consumer destination plus the granular fields metric queries filter on.
func StreamConsumeAttributes ¶ added in v0.60.0
func StreamConsumeAttributes(streamName string) ConsumeAttributes
StreamConsumeAttributes identifies a streams-lane delivery. The stream itself is the destination: the stream protocol routes to a stream directly, so there is no exchange and no routing key to attribute.
Source Files