faet: add rabbitmq cluster_name

.chloggen/feat_rabbitmq-cluster-name.yaml

@@ -0,0 +1,5 @@
+change_type: enhancement
+component: receiver/rabbitmq
+note: Add the disabled-by-default `rabbitmq.cluster.name` resource attribute.
+issues: [50168]
+change_logs: [user]

receiver/rabbitmqreceiver/client.go

@@ -25,6 +25,9 @@ const (
 
 	// exchangePath is the endpoint for RabbitMQ exchanges.
 	exchangePath = "/api/exchanges"
+
+	// clusterNamePath is the endpoint for the RabbitMQ cluster name.
+	clusterNamePath = "/api/cluster-name"
 )
 
 type client interface {
@@ -34,6 +37,8 @@ type client interface {
 	GetNodes(ctx context.Context) ([]*models.Node, error)
 	// GetExchanges calls "/api/exchanges" endpoint to get list of exchanges for the target node
 	GetExchanges(ctx context.Context) ([]*models.Exchange, error)
+	// GetClusterName calls "/api/cluster-name" endpoint to get the cluster name.
+	GetClusterName(ctx context.Context) (string, error)
 }
 
 var _ client = (*rabbitmqClient)(nil)
@@ -100,6 +105,17 @@ func (c *rabbitmqClient) GetExchanges(ctx context.Context) ([]*models.Exchange,
 	return exchanges, nil
 }
 
+func (c *rabbitmqClient) GetClusterName(ctx context.Context) (string, error) {
+	var clusterName models.ClusterName
+
+	if err := c.get(ctx, clusterNamePath, &clusterName); err != nil {
+		c.logger.Debug("Failed to retrieve cluster name", zap.Error(err))
+		return "", err
+	}
+
+	return clusterName.Name, nil
+}
+
 func (c *rabbitmqClient) get(ctx context.Context, path string, respObj any) error {
 	// Construct endpoint and create request
 	url := c.hostEndpoint + path

receiver/rabbitmqreceiver/client_test.go

@@ -24,9 +24,10 @@ import (
 )
 
 const (
-	queuesAPIResponseFile    = "get_queues_response.json"
-	nodesAPIResponseFile     = "get_nodes_response.json"
-	exchangesAPIResponseFile = "get_exchanges_response.json"
+	queuesAPIResponseFile      = "get_queues_response.json"
+	nodesAPIResponseFile       = "get_nodes_response.json"
+	exchangesAPIResponseFile   = "get_exchanges_response.json"
+	clusterNameAPIResponseFile = "get_cluster_name_response.json"
 )
 
 func TestNewClient(t *testing.T) {
@@ -300,6 +301,71 @@ func TestGetExchangesDetails(t *testing.T) {
 	}
 }
 
+func TestGetClusterNameDetails(t *testing.T) {
+	testCases := []struct {
+		desc     string
+		testFunc func(*testing.T)
+	}{
+		{
+			desc: "Non-200 Response for GetClusterName",
+			testFunc: func(t *testing.T) {
+				ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+					w.WriteHeader(http.StatusForbidden)
+				}))
+				defer ts.Close()
+
+				tc := createTestClient(t, ts.URL)
+
+				clusterName, err := tc.GetClusterName(t.Context())
+				require.Empty(t, clusterName)
+				require.EqualError(t, err, "non 200 code returned 403")
+			},
+		},
+		{
+			desc: "Bad payload returned for GetClusterName",
+			testFunc: func(t *testing.T) {
+				ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+					_, err := w.Write([]byte("{invalid-json}"))
+					assert.NoError(t, err)
+				}))
+				defer ts.Close()
+
+				tc := createTestClient(t, ts.URL)
+
+				clusterName, err := tc.GetClusterName(t.Context())
+				require.Empty(t, clusterName)
+				require.ErrorContains(t, err, "failed to decode response payload")
+			},
+		},
+		{
+			desc: "Successful GetClusterName call",
+			testFunc: func(t *testing.T) {
+				data := loadAPIResponseData(t, clusterNameAPIResponseFile)
+				ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+					assert.Equal(t, clusterNamePath, req.URL.Path)
+					_, err := w.Write(data)
+					assert.NoError(t, err)
+				}))
+				defer ts.Close()
+
+				tc := createTestClient(t, ts.URL)
+
+				var expected models.ClusterName
+				err := json.Unmarshal(data, &expected)
+				require.NoError(t, err)
+
+				clusterName, err := tc.GetClusterName(t.Context())
+				require.NoError(t, err)
+				require.Equal(t, expected.Name, clusterName)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, tc.testFunc)
+	}
+}
+
 func createTestClient(t *testing.T, baseEndpoint string) client {
 	t.Helper()
 	cfg := createDefaultConfig().(*Config)

receiver/rabbitmqreceiver/documentation.md

@@ -688,6 +688,7 @@ Uptime of the node.
 
 | Name | Description | Values | Enabled | Semantic Convention | Stability |
 | ---- | ----------- | ------ | ------- | ------------------- | --------- |
+| rabbitmq.cluster.name | The name of the RabbitMQ cluster. | Any Str | false | - | - |
 | rabbitmq.exchange.name | The name of the RabbitMQ exchange. | Any Str | true | - | - |
 | rabbitmq.exchange.type | The type of the RabbitMQ exchange (direct, fanout, topic, or headers). | Any Str | true | - | - |
 | rabbitmq.node.name | The name of the RabbitMQ node. | Any Str | true | - | - |

receiver/rabbitmqreceiver/internal/metadata/testdata/config.yaml

@@ -167,6 +167,8 @@ all_set:
     rabbitmq.node.uptime:
       enabled: true
   resource_attributes:
+    rabbitmq.cluster.name:
+      enabled: true
     rabbitmq.exchange.name:
       enabled: true
     rabbitmq.exchange.type:
@@ -345,6 +347,8 @@ reaggregate_set:
     rabbitmq.node.uptime:
       enabled: true
   resource_attributes:
+    rabbitmq.cluster.name:
+      enabled: true
     rabbitmq.exchange.name:
       enabled: true
     rabbitmq.exchange.type:
@@ -523,6 +527,8 @@ none_set:
     rabbitmq.node.uptime:
       enabled: false
   resource_attributes:
+    rabbitmq.cluster.name:
+      enabled: false
     rabbitmq.exchange.name:
       enabled: false
     rabbitmq.exchange.type:
@@ -535,6 +541,10 @@ none_set:
       enabled: false
 filter_set_include:
   resource_attributes:
+    rabbitmq.cluster.name:
+      enabled: true
+      metrics_include:
+        - regexp: ".*"
     rabbitmq.exchange.name:
       enabled: true
       metrics_include:
@@ -557,6 +567,10 @@ filter_set_include:
         - regexp: ".*"
 filter_set_exclude:
   resource_attributes:
+    rabbitmq.cluster.name:
+      enabled: true
+      metrics_exclude:
+        - strict: "rabbitmq.cluster.name-val"
     rabbitmq.exchange.name:
       enabled: true
       metrics_exclude:

receiver/rabbitmqreceiver/internal/models/models.go

@@ -3,6 +3,11 @@
 
 package models // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver/internal/models"
 
+// ClusterName represents the RabbitMQ cluster name API response.
+type ClusterName struct {
+	Name string `json:"name"`
+}
+
 // Queue represents a queue in the API response
 type Queue struct {
 	// Identifiers

receiver/rabbitmqreceiver/metadata.yaml

@@ -170,6 +170,10 @@ status:
     emeritus: [cpheps]
 
 resource_attributes:
+  rabbitmq.cluster.name:
+    description: The name of the RabbitMQ cluster.
+    enabled: false
+    type: string
   rabbitmq.exchange.name:
     description: The name of the RabbitMQ exchange.
     enabled: true