Rename metrics

To match .NET client metric names. This commit also removes read/written
bytes metrics because they were no-op. In this client, the AMQP 1.0
protocol is handled by a third-party library. This library does not
expose "hooks" to count reliably the number of bytes read/written.

Author: Zerpet

docs/development/METRICS_IMPLEMENTATION_GUIDE.md

@@ -186,7 +186,10 @@ consumeDisposition(disposition: ConsumeDisposition)
 - **DISCARDED**: Consumer discarded/rejected the message (dead letter)
 - **REQUEUED**: Consumer requeued the message for redelivery
 
-#### Network I/O
+#### Network I/O (Optional)
+
+**Note**: These metrics are **optional** and may not be supported by all client implementations.
+
 ```
 writtenBytes(byteCount: int)
   - Called when: Data is written to the network socket
@@ -203,6 +206,8 @@ readBytes(byteCount: int)
   - Note: Cumulative - caller must aggregate
 ```
 
+> **Go Client Implementation Note:** The Go client does **not** implement `writtenBytes` and `readBytes` metrics due to underlying library limitations. These methods are not present in the Go `MetricsCollector` interface.
+
 ---
 
 ## 4. Metric Types and Categories
@@ -237,13 +242,14 @@ These metrics represent **cumulative totals** that only increase:
 | consumed_accepted | Counter | Total messages accepted by consumers | messages |
 | consumed_discarded | Counter | Total messages discarded by consumers | messages |
 | consumed_requeued | Counter | Total messages requeued by consumers | messages |
-| written_bytes | Counter | Total bytes written to network | bytes |
-| read_bytes | Counter | Total bytes read from network | bytes |
+| written_bytes **(optional)** | Counter | Total bytes written to network | bytes |
+| read_bytes **(optional)** | Counter | Total bytes read from network | bytes |
 
 **Implementation Notes**:
 - These should monotonically increase
 - Never reset during the lifetime of the application
 - Use thread-safe increment operations
+- **Note**: The `written_bytes` and `read_bytes` metrics are **optional**. Some client implementations (including the Go client) may not support these metrics due to underlying library limitations.
 - For byte metrics, expect very high frequency updates
 
 ### 4.3 Metric Naming Convention

pkg/rabbitmqamqp/metrics.go

@@ -83,14 +83,6 @@ type MetricsCollector interface {
 	// ConsumeDisposition is called when the consumer settles (acknowledges) a message.
 	// The context contains OTEL semantic convention attributes for the disposition.
 	ConsumeDisposition(disposition ConsumeDisposition, ctx ConsumeContext)
-
-	// WrittenBytes is called when data is written to the network socket.
-	// The count parameter is the number of bytes written in this operation.
-	WrittenBytes(count int64)
-
-	// ReadBytes is called when data is read from the network socket.
-	// The count parameter is the number of bytes read in this operation.
-	ReadBytes(count int64)
 }
 
 // NoOpMetricsCollector is a MetricsCollector implementation that does nothing.
@@ -111,8 +103,6 @@ func (n *NoOpMetricsCollector) Publish(_ PublishContext)
 func (n *NoOpMetricsCollector) PublishDisposition(_ PublishDisposition, _ PublishContext) {}
 func (n *NoOpMetricsCollector) Consume(_ ConsumeContext)                                  {}
 func (n *NoOpMetricsCollector) ConsumeDisposition(_ ConsumeDisposition, _ ConsumeContext) {}
-func (n *NoOpMetricsCollector) WrittenBytes(_ int64)                                      {}
-func (n *NoOpMetricsCollector) ReadBytes(_ int64)                                         {}
 
 // defaultMetricsCollector is the singleton NoOpMetricsCollector instance used as default.
 var defaultMetricsCollector MetricsCollector = &NoOpMetricsCollector{}

pkg/rabbitmqamqp/metrics_otel.go

@@ -30,8 +30,6 @@ type OTELMetricsCollector struct {
 	consumedAccepted  metric.Int64Counter
 	consumedDiscarded metric.Int64Counter
 	consumedRequeued  metric.Int64Counter
-	writtenBytes      metric.Int64Counter
-	readBytes         metric.Int64Counter
 }
 
 // Ensure OTELMetricsCollector implements MetricsCollector
@@ -89,7 +87,7 @@ func NewOTELMetricsCollector(meterProvider metric.MeterProvider, prefix string)
 	}
 
 	collector.publishedAccepted, err = meter.Int64Counter(
-		prefix+".published.accepted",
+		prefix+".published_accepted",
 		metric.WithDescription("Total number of messages accepted by broker"),
 		metric.WithUnit("{message}"),
 	)
@@ -98,7 +96,7 @@ func NewOTELMetricsCollector(meterProvider metric.MeterProvider, prefix string)
 	}
 
 	collector.publishedRejected, err = meter.Int64Counter(
-		prefix+".published.rejected",
+		prefix+".published_rejected",
 		metric.WithDescription("Total number of messages rejected by broker"),
 		metric.WithUnit("{message}"),
 	)
@@ -107,7 +105,7 @@ func NewOTELMetricsCollector(meterProvider metric.MeterProvider, prefix string)
 	}
 
 	collector.publishedReleased, err = meter.Int64Counter(
-		prefix+".published.released",
+		prefix+".published_released",
 		metric.WithDescription("Total number of messages released by broker"),
 		metric.WithUnit("{message}"),
 	)
@@ -126,7 +124,7 @@ func NewOTELMetricsCollector(meterProvider metric.MeterProvider, prefix string)
 	}
 
 	collector.consumedAccepted, err = meter.Int64Counter(
-		prefix+".consumed.accepted",
+		prefix+".consumed_accepted",
 		metric.WithDescription("Total number of messages accepted by consumer"),
 		metric.WithUnit("{message}"),
 	)
@@ -135,7 +133,7 @@ func NewOTELMetricsCollector(meterProvider metric.MeterProvider, prefix string)
 	}
 
 	collector.consumedDiscarded, err = meter.Int64Counter(
-		prefix+".consumed.discarded",
+		prefix+".consumed_discarded",
 		metric.WithDescription("Total number of messages discarded by consumer"),
 		metric.WithUnit("{message}"),
 	)
@@ -144,33 +142,14 @@ func NewOTELMetricsCollector(meterProvider metric.MeterProvider, prefix string)
 	}
 
 	collector.consumedRequeued, err = meter.Int64Counter(
-		prefix+".consumed.requeued",
+		prefix+".consumed_requeued",
 		metric.WithDescription("Total number of messages requeued by consumer"),
 		metric.WithUnit("{message}"),
 	)
 	if err != nil {
 		return nil, err
 	}
 
-	// Create counter metrics for network I/O
-	collector.writtenBytes, err = meter.Int64Counter(
-		prefix+".written_bytes",
-		metric.WithDescription("Total bytes written to network"),
-		metric.WithUnit("By"),
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	collector.readBytes, err = meter.Int64Counter(
-		prefix+".read_bytes",
-		metric.WithDescription("Total bytes read from network"),
-		metric.WithUnit("By"),
-	)
-	if err != nil {
-		return nil, err
-	}
-
 	return collector, nil
 }
 
@@ -204,16 +183,6 @@ func (o *OTELMetricsCollector) CloseConsumer() {
 	o.consumers.Add(context.Background(), -1)
 }
 
-// WrittenBytes increments the written bytes counter.
-func (o *OTELMetricsCollector) WrittenBytes(count int64) {
-	o.writtenBytes.Add(context.Background(), count)
-}
-
-// ReadBytes increments the read bytes counter.
-func (o *OTELMetricsCollector) ReadBytes(count int64) {
-	o.readBytes.Add(context.Background(), count)
-}
-
 // buildBaseAttributes creates the common OTEL semantic convention attributes
 // for RabbitMQ messaging operations.
 func buildBaseAttributes(serverAddress string, serverPort int, destinationName string) []attribute.KeyValue {

tests/otelmetrics/metrics_e2e_test.go

@@ -148,16 +148,16 @@ func TestMetricsEndToEndBasicFlow(t *testing.T) {
 		assertMetric(t, metrics, "test.e2e.published", int64(messageCount), "should have published %d messages", messageCount)
 	})
 
-	t.Run("published.accepted counter", func(t *testing.T) {
-		assertMetric(t, metrics, "test.e2e.published.accepted", int64(messageCount), "all messages should be accepted by broker")
+	t.Run("published_accepted counter", func(t *testing.T) {
+		assertMetric(t, metrics, "test.e2e.published_accepted", int64(messageCount), "all messages should be accepted by broker")
 	})
 
 	t.Run("consumed counter", func(t *testing.T) {
 		assertMetric(t, metrics, "test.e2e.consumed", int64(messageCount), "should have consumed %d messages", messageCount)
 	})
 
-	t.Run("consumed.accepted counter", func(t *testing.T) {
-		assertMetric(t, metrics, "test.e2e.consumed.accepted", int64(messageCount), "all messages should be accepted by consumer")
+	t.Run("consumed_accepted counter", func(t *testing.T) {
+		assertMetric(t, metrics, "test.e2e.consumed_accepted", int64(messageCount), "all messages should be accepted by consumer")
 	})
 }
 
@@ -487,11 +487,11 @@ func TestMetricsEndToEndWithSemanticConventions(t *testing.T) {
 		}
 	})
 
-	// Verify consumed.accepted has settle operation type
-	t.Run("consumed.accepted has settle operation attributes", func(t *testing.T) {
-		dps, ok := metricsWithAttrs["test.semconv.consumed.accepted"]
+	// Verify consumed_accepted has settle operation type
+	t.Run("consumed_accepted has settle operation attributes", func(t *testing.T) {
+		dps, ok := metricsWithAttrs["test.semconv.consumed_accepted"]
 		if !ok {
-			t.Fatal("test.semconv.consumed.accepted metric not found")
+			t.Fatal("test.semconv.consumed_accepted metric not found")
 		}
 
 		found := false
@@ -503,7 +503,7 @@ func TestMetricsEndToEndWithSemanticConventions(t *testing.T) {
 			}
 		}
 		if !found {
-			t.Error("consumed.accepted metric with settle operation attributes not found")
+			t.Error("consumed_accepted metric with settle operation attributes not found")
 		}
 	})
 }

tests/otelmetrics/metrics_test.go

@@ -64,8 +64,6 @@ func TestNoOpMetricsCollector(t *testing.T) {
 	collector.ConsumeDisposition(rmq.ConsumeAccepted, consumeCtx)
 	collector.ConsumeDisposition(rmq.ConsumeDiscarded, consumeCtx)
 	collector.ConsumeDisposition(rmq.ConsumeRequeued, consumeCtx)
-	collector.WrittenBytes(1024)
-	collector.ReadBytes(2048)
 }
 
 // TestOTELMetricsCollectorCreation verifies that OTEL collector can be created with various options.
@@ -139,11 +137,6 @@ func TestOTELMetricsCollectorRecordsMetrics(t *testing.T) {
 	collector.ConsumeDisposition(rmq.ConsumeRequeued, consumeCtx)
 	collector.CloseConsumer()
 
-	// Simulate byte tracking
-	collector.WrittenBytes(1000)
-	collector.WrittenBytes(500)
-	collector.ReadBytes(2000)
-
 	// Collect metrics
 	var rm metricdata.ResourceMetrics
 	err = reader.Collect(context.Background(), &rm)
@@ -189,32 +182,24 @@ func TestOTELMetricsCollectorRecordsMetrics(t *testing.T) {
 		t.Errorf("expected published counter to be 3, got %d (found: %v)", val, ok)
 	}
 
-	if val, ok := metrics["test.rabbitmq.published.accepted"]; !ok || val != 2 {
-		t.Errorf("expected published.accepted counter to be 2, got %d (found: %v)", val, ok)
+	if val, ok := metrics["test.rabbitmq.published_accepted"]; !ok || val != 2 {
+		t.Errorf("expected published_accepted counter to be 2, got %d (found: %v)", val, ok)
 	}
 
-	if val, ok := metrics["test.rabbitmq.published.rejected"]; !ok || val != 1 {
-		t.Errorf("expected published.rejected counter to be 1, got %d (found: %v)", val, ok)
+	if val, ok := metrics["test.rabbitmq.published_rejected"]; !ok || val != 1 {
+		t.Errorf("expected published_rejected counter to be 1, got %d (found: %v)", val, ok)
 	}
 
 	if val, ok := metrics["test.rabbitmq.consumed"]; !ok || val != 2 {
 		t.Errorf("expected consumed counter to be 2, got %d (found: %v)", val, ok)
 	}
 
-	if val, ok := metrics["test.rabbitmq.consumed.accepted"]; !ok || val != 1 {
-		t.Errorf("expected consumed.accepted counter to be 1, got %d (found: %v)", val, ok)
-	}
-
-	if val, ok := metrics["test.rabbitmq.consumed.requeued"]; !ok || val != 1 {
-		t.Errorf("expected consumed.requeued counter to be 1, got %d (found: %v)", val, ok)
-	}
-
-	if val, ok := metrics["test.rabbitmq.written_bytes"]; !ok || val != 1500 {
-		t.Errorf("expected written_bytes counter to be 1500, got %d (found: %v)", val, ok)
+	if val, ok := metrics["test.rabbitmq.consumed_accepted"]; !ok || val != 1 {
+		t.Errorf("expected consumed_accepted counter to be 1, got %d (found: %v)", val, ok)
 	}
 
-	if val, ok := metrics["test.rabbitmq.read_bytes"]; !ok || val != 2000 {
-		t.Errorf("expected read_bytes counter to be 2000, got %d (found: %v)", val, ok)
+	if val, ok := metrics["test.rabbitmq.consumed_requeued"]; !ok || val != 1 {
+		t.Errorf("expected consumed_requeued counter to be 1, got %d (found: %v)", val, ok)
 	}
 }
 
@@ -331,8 +316,6 @@ func TestMetricsCollectorThreadSafety(t *testing.T) {
 				collector.PublishDisposition(rmq.PublishAccepted, publishCtx)
 				collector.Consume(consumeCtx)
 				collector.ConsumeDisposition(rmq.ConsumeAccepted, consumeCtx)
-				collector.WrittenBytes(100)
-				collector.ReadBytes(100)
 				collector.CloseConsumer()
 				collector.ClosePublisher()
 				collector.CloseConnection()
@@ -469,10 +452,10 @@ func TestOTELMetricsCollectorWithContextMethods(t *testing.T) {
 
 	// Verify consume disposition metrics have attributes
 	t.Run("consume dispositions with context have attributes", func(t *testing.T) {
-		// Check consumed.accepted
-		dps, ok := metricsWithAttrs["test.ctx.consumed.accepted"]
+		// Check consumed_accepted
+		dps, ok := metricsWithAttrs["test.ctx.consumed_accepted"]
 		if !ok {
-			t.Error("test.ctx.consumed.accepted metric not found")
+			t.Error("test.ctx.consumed_accepted metric not found")
 		} else {
 			found := false
 			for _, dp := range dps {
@@ -482,14 +465,14 @@ func TestOTELMetricsCollectorWithContextMethods(t *testing.T) {
 				}
 			}
 			if !found {
-				t.Error("consumed.accepted with ack operation name not found")
+				t.Error("consumed_accepted with ack operation name not found")
 			}
 		}
 
-		// Check consumed.discarded
-		dps, ok = metricsWithAttrs["test.ctx.consumed.discarded"]
+		// Check consumed_discarded
+		dps, ok = metricsWithAttrs["test.ctx.consumed_discarded"]
 		if !ok {
-			t.Error("test.ctx.consumed.discarded metric not found")
+			t.Error("test.ctx.consumed_discarded metric not found")
 		} else {
 			found := false
 			for _, dp := range dps {
@@ -499,14 +482,14 @@ func TestOTELMetricsCollectorWithContextMethods(t *testing.T) {
 				}
 			}
 			if !found {
-				t.Error("consumed.discarded with nack operation name not found")
+				t.Error("consumed_discarded with nack operation name not found")
 			}
 		}
 
-		// Check consumed.requeued
-		dps, ok = metricsWithAttrs["test.ctx.consumed.requeued"]
+		// Check consumed_requeued
+		dps, ok = metricsWithAttrs["test.ctx.consumed_requeued"]
 		if !ok {
-			t.Error("test.ctx.consumed.requeued metric not found")
+			t.Error("test.ctx.consumed_requeued metric not found")
 		} else {
 			found := false
 			for _, dp := range dps {
@@ -516,7 +499,7 @@ func TestOTELMetricsCollectorWithContextMethods(t *testing.T) {
 				}
 			}
 			if !found {
-				t.Error("consumed.requeued with requeue operation name not found")
+				t.Error("consumed_requeued with requeue operation name not found")
 			}
 		}
 	})
@@ -603,9 +586,9 @@ func TestDoubleCountingBug(t *testing.T) {
 		}
 	})
 
-	t.Run("consumed.accepted should count once per disposition", func(t *testing.T) {
-		if total := metricTotals["test.double.consumed.accepted"]; total != 1 {
-			t.Errorf("BUG DETECTED: consumed.accepted metric should be 1 for one disposition, got %d (double-counting!)", total)
+	t.Run("consumed_accepted should count once per disposition", func(t *testing.T) {
+		if total := metricTotals["test.double.consumed_accepted"]; total != 1 {
+			t.Errorf("BUG DETECTED: consumed_accepted metric should be 1 for one disposition, got %d (double-counting!)", total)
 		}
 	})
 }