[receiver/rabbitmq] Enhancement/rabbitmq Added node level metrics

.chloggen/rabbitmqreceiver_node_metrics_enhancement.yaml

@@ -0,0 +1,11 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: rabbitmqreceiver
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Enhance the RabbitMQ receiver to collect and report node-level metrics (`rabbitmq.node.disk_free`, `rabbitmq.node.fd_used`, `rabbitmq.node.mem_limit`, and `rabbitmq.node.mem_used`). This provides additional observability into the state and resource usage of RabbitMQ nodes."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36925]

exporter/zipkinexporter/zipkin_test.go

@@ -86,7 +86,8 @@ func TestZipkinExporter_roundtripJSON(t *testing.T) {
 	require.NoError(t, mzr.Flush())
 
 	// We expect back the exact JSON that was received
-	wants := []string{`
+	wants := []string{
+		`
 		[{
 		  "traceId": "4d1e00c0db9010db86154a4ba6e91385","parentId": "86154a4ba6e91385","id": "4d1e00c0db9010db",
 		  "kind": "CLIENT","name": "get",
@@ -120,7 +121,8 @@ func TestZipkinExporter_roundtripJSON(t *testing.T) {
 		  "timestamp": 1472470996199000,
 		  "duration": 207000
 		}]
-		`}
+		`,
+	}
 	for i, s := range wants {
 		want := unmarshalZipkinSpanArrayToMap(t, s)
 		gotBytes := buf.Next(int(sizes[i]))

receiver/googlecloudspannerreceiver/internal/statsreader/statsreaders_mockedspanner_test.go

@@ -138,12 +138,14 @@ func TestStatsReaders_Read(t *testing.T) {
 
 	op, err := databaseAdminClient.UpdateDatabaseDdl(ctx, &databasepb.UpdateDatabaseDdlRequest{
 		Database: databaseName,
-		Statements: []string{`CREATE TABLE STATS (
+		Statements: []string{
+			`CREATE TABLE STATS (
 			INTERVAL_END TIMESTAMP,
 			METRIC_LABEL STRING(MAX),
 			METRIC_VALUE INT64
 		) PRIMARY KEY (METRIC_LABEL)
-		`},
+		`,
+		},
 	})
 	require.NoError(t, err)
 

receiver/rabbitmqreceiver/README.md

@@ -34,7 +34,9 @@ The following settings are optional:
 
 - `endpoint` (default: `http://localhost:15672`): The URL of the node to be monitored.
 - `collection_interval` (default = `10s`): This receiver collects metrics on an interval. Valid time units are `ns`, `us` (or `µs`), `ms`, `s`, `m`, `h`.
+
 - `tls`: TLS control. [By default, insecure settings are rejected and certificate verification is on](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md).
+- `enable_node_metrics` (default = `true`): Enables collection of RabbitMQ node-level metrics such as memory usage, file descriptors, and disk space.
 
 ### Example Configuration
 
@@ -45,6 +47,7 @@ receivers:
     username: otelu
     password: ${env:RABBITMQ_PASSWORD}
     collection_interval: 10s
+    enable_node_metrics: true
 ```
 
 The full list of settings exposed for this receiver are documented in [config.go](./config.go) with detailed sample configurations in [testdata/config.yaml](./testdata/config.yaml). TLS config is documented further under the [opentelemetry collector's configtls package](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md).

receiver/rabbitmqreceiver/client.go

@@ -16,12 +16,19 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver/internal/models"
 )
 
-// queuePath is the path to queues endpoint
-const queuePath = "/api/queues"
+const (
+	// queuePath is the endpoint for RabbitMQ queues.
+	queuePath = "/api/queues"
+
+	// nodePath is the endpoint for RabbitMQ nodes.
+	nodePath = "/api/nodes"
+)
 
 type client interface {
 	// GetQueues calls "/api/queues" endpoint to get list of queues for the target node
 	GetQueues(ctx context.Context) ([]*models.Queue, error)
+	// GetQueues calls "/api/nodes" endpoint to get list of nodes for the target node
+	GetNodes(ctx context.Context) ([]*models.Node, error)
 }
 
 var _ client = (*rabbitmqClient)(nil)
@@ -66,6 +73,17 @@ func (c *rabbitmqClient) GetQueues(ctx context.Context) ([]*models.Queue, error)
 	return queues, nil
 }
 
+func (c *rabbitmqClient) GetNodes(ctx context.Context) ([]*models.Node, error) {
+	var nodes []*models.Node
+
+	if err := c.get(ctx, nodePath, &nodes); err != nil {
+		c.logger.Error("Failed to retrieve nodes", zap.Error(err))
+		return nil, err
+	}
+
+	return nodes, nil
+}
+
 func (c *rabbitmqClient) get(ctx context.Context, path string, respObj any) error {
 	// Construct endpoint and create request
 	url := c.hostEndpoint + path
@@ -94,15 +112,11 @@ func (c *rabbitmqClient) get(ctx context.Context, path string, respObj any) erro
 	if resp.StatusCode != http.StatusOK {
 		c.logger.Debug("rabbitMQ API non-200", zap.Error(err), zap.Int("status_code", resp.StatusCode))
 
-		// Attempt to extract the error payload
-		payloadData, err := io.ReadAll(resp.Body)
-		if err != nil {
-			c.logger.Debug("failed to read payload error message", zap.Error(err))
-		} else {
-			c.logger.Debug("rabbitMQ API Error", zap.ByteString("api_error", payloadData))
+		payload, readErr := io.ReadAll(resp.Body)
+		if readErr == nil {
+			c.logger.Debug("Error response payload", zap.ByteString("error_payload", payload))
 		}
-
-		return fmt.Errorf("non 200 code returned %d", resp.StatusCode)
+		return fmt.Errorf("non-200 response code: %d", resp.StatusCode)
 	}
 
 	// Decode the payload into the passed in response object

receiver/rabbitmqreceiver/client_test.go

@@ -6,7 +6,6 @@ package rabbitmqreceiver // import "github.com/open-telemetry/opentelemetry-coll
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -26,12 +25,13 @@ import (
 
 const (
 	queuesAPIResponseFile = "get_queues_response.json"
+	nodesAPIResponseFile  = "get_nodes_response.json"
 )
 
 func TestNewClient(t *testing.T) {
-	clientConfigNonExistandCA := confighttp.NewDefaultClientConfig()
-	clientConfigNonExistandCA.Endpoint = defaultEndpoint
-	clientConfigNonExistandCA.TLSSetting = configtls.ClientConfig{
+	clientConfigInvalid := confighttp.NewDefaultClientConfig()
+	clientConfigInvalid.Endpoint = defaultEndpoint
+	clientConfigInvalid.TLSSetting = configtls.ClientConfig{
 		Config: configtls.Config{
 			CAFile: "/non/existent",
 		},
@@ -40,42 +40,45 @@ func TestNewClient(t *testing.T) {
 	clientConfig := confighttp.NewDefaultClientConfig()
 	clientConfig.Endpoint = defaultEndpoint
 
-	testCase := []struct {
+	testCases := []struct {
 		desc        string
 		cfg         *Config
 		host        component.Host
 		settings    component.TelemetrySettings
 		logger      *zap.Logger
-		expectError error
+		expectError string // Updated to string to match substrings in the error
 	}{
 		{
 			desc: "Invalid HTTP config",
 			cfg: &Config{
-				ClientConfig: clientConfigNonExistandCA,
+				ClientConfig: clientConfigInvalid,
 			},
 			host:        componenttest.NewNopHost(),
 			settings:    componenttest.NewNopTelemetrySettings(),
 			logger:      zap.NewNop(),
-			expectError: errors.New("failed to create HTTP Client"),
+			expectError: "failed to create HTTP Client: failed to load TLS config",
 		},
 		{
 			desc: "Valid Configuration",
 			cfg: &Config{
 				ClientConfig: clientConfig,
+				Username:     "valid_user",
+				Password:     "valid_password",
 			},
 			host:        componenttest.NewNopHost(),
 			settings:    componenttest.NewNopTelemetrySettings(),
 			logger:      zap.NewNop(),
-			expectError: nil,
+			expectError: "",
 		},
 	}
 
-	for _, tc := range testCase {
+	for _, tc := range testCases {
 		t.Run(tc.desc, func(t *testing.T) {
 			ac, err := newClient(context.Background(), tc.cfg, tc.host, tc.settings, tc.logger)
-			if tc.expectError != nil {
+			if tc.expectError != "" {
 				require.Nil(t, ac)
-				require.ErrorContains(t, err, tc.expectError.Error())
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.expectError) // Check for the substring
 			} else {
 				require.NoError(t, err)
 
@@ -108,9 +111,9 @@ func TestGetQueuesDetails(t *testing.T) {
 
 				tc := createTestClient(t, ts.URL)
 
-				clusters, err := tc.GetQueues(context.Background())
-				require.Nil(t, clusters)
-				require.EqualError(t, err, "non 200 code returned 401")
+				queues, err := tc.GetQueues(context.Background())
+				require.Nil(t, queues)
+				require.EqualError(t, err, "non-200 response code: 401")
 			},
 		},
 		{
@@ -125,13 +128,13 @@ func TestGetQueuesDetails(t *testing.T) {
 
 				tc := createTestClient(t, ts.URL)
 
-				clusters, err := tc.GetQueues(context.Background())
-				require.Nil(t, clusters)
-				require.ErrorContains(t, err, "failed to decode response payload")
+				queues, err := tc.GetQueues(context.Background())
+				require.Nil(t, queues)
+				require.ErrorContains(t, err, "failed to decode response payload: json: cannot unmarshal object into Go value of type []*models.Queue")
 			},
 		},
 		{
-			desc: "Successful call",
+			desc: "Successful Queue API call",
 			testFunc: func(t *testing.T) {
 				data := loadAPIResponseData(t, queuesAPIResponseFile)
 
@@ -149,9 +152,9 @@ func TestGetQueuesDetails(t *testing.T) {
 				err := json.Unmarshal(data, &expected)
 				require.NoError(t, err)
 
-				clusters, err := tc.GetQueues(context.Background())
+				queues, err := tc.GetQueues(context.Background())
 				require.NoError(t, err)
-				require.Equal(t, expected, clusters)
+				require.Equal(t, expected, queues)
 			},
 		},
 	}

receiver/rabbitmqreceiver/config.go

@@ -31,20 +31,25 @@ type Config struct {
 	confighttp.ClientConfig        `mapstructure:",squash"`
 	Username                       string              `mapstructure:"username"`
 	Password                       configopaque.String `mapstructure:"password"`
+	EnableNodeMetrics              bool                `mapstructure:"enable_node_metrics"` // Added flag for node metrics
 	metadata.MetricsBuilderConfig  `mapstructure:",squash"`
 }
 
 // Validate validates the configuration by checking for missing or invalid fields
 func (cfg *Config) Validate() error {
 	var err []error
+
+	// Validate username
 	if cfg.Username == "" {
 		err = append(err, errMissingUsername)
 	}
 
+	// Validate password
 	if cfg.Password == "" {
 		err = append(err, errMissingPassword)
 	}
 
+	// Validate endpoint
 	_, parseErr := url.Parse(cfg.Endpoint)
 	if parseErr != nil {
 		wrappedErr := fmt.Errorf("%s: %w", errInvalidEndpoint.Error(), parseErr)

receiver/rabbitmqreceiver/config_test.go

@@ -20,7 +20,7 @@ import (
 
 func TestValidate(t *testing.T) {
 	clientConfigInvalid := confighttp.NewDefaultClientConfig()
-	clientConfigInvalid.Endpoint = "invalid://endpoint:  12efg"
+	clientConfigInvalid.Endpoint = "invalid://endpoint: 12efg"
 
 	clientConfig := confighttp.NewDefaultClientConfig()
 	clientConfig.Endpoint = defaultEndpoint
@@ -38,7 +38,7 @@ func TestValidate(t *testing.T) {
 			expectedErr: errors.Join(
 				errMissingUsername,
 				errMissingPassword,
-				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint:  12efg": invalid port ":  12efg" after host`)),
+				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint: 12efg": invalid port ": 12efg" after host`)),
 		},
 		{
 			desc: "missing password and invalid endpoint",
@@ -48,7 +48,7 @@ func TestValidate(t *testing.T) {
 			},
 			expectedErr: errors.Join(
 				errMissingPassword,
-				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint:  12efg": invalid port ":  12efg" after host`),
+				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint: 12efg": invalid port ": 12efg" after host`),
 			),
 		},
 		{
@@ -59,7 +59,7 @@ func TestValidate(t *testing.T) {
 			},
 			expectedErr: errors.Join(
 				errMissingUsername,
-				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint:  12efg": invalid port ":  12efg" after host`),
+				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint: 12efg": invalid port ": 12efg" after host`),
 			),
 		},
 		{
@@ -70,15 +70,26 @@ func TestValidate(t *testing.T) {
 				ClientConfig: clientConfigInvalid,
 			},
 			expectedErr: errors.Join(
-				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint:  12efg": invalid port ":  12efg" after host`),
+				fmt.Errorf("%w: %s", errInvalidEndpoint, `parse "invalid://endpoint: 12efg": invalid port ": 12efg" after host`),
 			),
 		},
 		{
-			desc: "valid config",
+			desc: "valid config with node metrics enabled",
 			cfg: &Config{
-				Username:     "otelu",
-				Password:     "otelp",
-				ClientConfig: clientConfig,
+				Username:          "otelu",
+				Password:          "otelp",
+				ClientConfig:      clientConfig,
+				EnableNodeMetrics: true,
+			},
+			expectedErr: nil,
+		},
+		{
+			desc: "valid config with node metrics disabled",
+			cfg: &Config{
+				Username:          "otelu",
+				Password:          "otelp",
+				ClientConfig:      clientConfig,
+				EnableNodeMetrics: false,
 			},
 			expectedErr: nil,
 		},
@@ -112,6 +123,7 @@ func TestLoadConfig(t *testing.T) {
 	expected.Username = "otelu"
 	expected.Password = "${env:RABBITMQ_PASSWORD}"
 	expected.CollectionInterval = 10 * time.Second
+	expected.EnableNodeMetrics = true
 
 	require.Equal(t, expected, cfg)
 }

receiver/rabbitmqreceiver/documentation.md

@@ -66,6 +66,38 @@ The number of messages published to a queue.
 | ---- | ----------- | ---------- | ----------------------- | --------- |
 | {messages} | Sum | Int | Cumulative | true |
 
+### rabbitmq.node.disk_free
+
+Free disk space on the node.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {bytes} | Sum | Int | Cumulative | false |
+
+### rabbitmq.node.fd_used
+
+The number of file descriptors used on the node.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {fd} | Sum | Int | Cumulative | false |
+
+### rabbitmq.node.mem_limit
+
+The memory limit on the node.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {bytes} | Sum | Int | Cumulative | false |
+
+### rabbitmq.node.mem_used
+
+The memory used on the node.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {bytes} | Sum | Int | Cumulative | false |
+
 ## Resource Attributes
 
 | Name | Description | Values | Enabled |