Add component.{id, binary} to monitoring metrics (#3626)

This commit adds `component.id` and `component.binary` to the
monitoring metrics, those fields are necessary to create unique TSDB
entries.

(cherry picked from commit c4462d2)

changelog/fragments/1697735373-Add-component-fields-to-metrics.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add component.{id, binary} to monitoring metrics from Elastic-Agent and Beats
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/3626
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/integrations/issues/7977

internal/pkg/agent/application/monitoring/v1_monitor.go

@@ -594,6 +594,15 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 						"ignore_missing": true,
 					},
 				},
+				map[string]interface{}{
+					"add_fields": map[string]interface{}{
+						"target": "component",
+						"fields": map[string]interface{}{
+							"id":     "elastic-agent",
+							"binary": "elastic-agent",
+						},
+					},
+				},
 			},
 		},
 	}
@@ -655,6 +664,15 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 							},
 						},
 					},
+					map[string]interface{}{
+						"add_fields": map[string]interface{}{
+							"target": "component",
+							"fields": map[string]interface{}{
+								"id":     unit,
+								"binary": binaryName,
+							},
+						},
+					},
 				},
 			})
 		}
@@ -715,6 +733,15 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 						"ignore_missing": true,
 					},
 				},
+				map[string]interface{}{
+					"add_fields": map[string]interface{}{
+						"target": "component",
+						"fields": map[string]interface{}{
+							"id":     unit,
+							"binary": binaryName,
+						},
+					},
+				},
 			},
 		})
 
@@ -777,6 +804,15 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 							"ignore_missing": true,
 						},
 					},
+					map[string]interface{}{
+						"add_fields": map[string]interface{}{
+							"target": "component",
+							"fields": map[string]interface{}{
+								"id":     unit,
+								"binary": binaryName,
+							},
+						},
+					},
 				},
 			})
 		}
@@ -807,7 +843,7 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 				"hosts":      endpoints,
 				"namespace":  "application",
 				"period":     metricsCollectionIntervalString,
-				"processors": createProcessorsForJSONInput(name, monitoringNamespace, b.agentInfo),
+				"processors": createProcessorsForJSONInput(name, comp.ID, monitoringNamespace, b.agentInfo),
 			},
 				map[string]interface{}{
 					idKey: "metrics-monitoring-shipper-stats",
@@ -821,7 +857,7 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 					"hosts":      endpoints,
 					"namespace":  "agent",
 					"period":     metricsCollectionIntervalString,
-					"processors": createProcessorsForJSONInput(name, monitoringNamespace, b.agentInfo),
+					"processors": createProcessorsForJSONInput(name, comp.ID, monitoringNamespace, b.agentInfo),
 				})
 		}
 	}
@@ -878,7 +914,7 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 	return nil
 }
 
-func createProcessorsForJSONInput(name string, monitoringNamespace string, agentInfo *info.AgentInfo) []interface{} {
+func createProcessorsForJSONInput(name string, compID, monitoringNamespace string, agentInfo *info.AgentInfo) []interface{} {
 	return []interface{}{
 		map[string]interface{}{
 			"add_fields": map[string]interface{}{
@@ -932,6 +968,15 @@ func createProcessorsForJSONInput(name string, monitoringNamespace string, agent
 				"ignore_missing": true,
 			},
 		},
+		map[string]interface{}{
+			"add_fields": map[string]interface{}{
+				"target": "component",
+				"fields": map[string]interface{}{
+					"id":     compID,
+					"binary": name,
+				},
+			},
+		},
 	}
 }
 

internal/pkg/agent/application/monitoring/v1_monitor_test.go

@@ -6,16 +6,20 @@ package monitoring
 
 import (
 	"context"
+	"encoding/json"
 	"runtime"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v2"
 
+	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
 	monitoringcfg "github.com/elastic/elastic-agent/internal/pkg/core/monitoring/config"
+	"github.com/elastic/elastic-agent/pkg/component"
 )
 
 func TestMonitoringConfigMetricsInterval(t *testing.T) {
@@ -94,3 +98,114 @@ func TestMonitoringConfigMetricsInterval(t *testing.T) {
 
 	}
 }
+
+func TestMonitoringConfigComponentFields(t *testing.T) {
+	agentInfo, err := info.NewAgentInfo(context.Background(), false)
+	require.NoError(t, err, "Error creating agent info")
+
+	cfg := &monitoringConfig{
+		C: &monitoringcfg.MonitoringConfig{
+			Enabled:        true,
+			MonitorMetrics: true,
+			Namespace:      "tiaog",
+			HTTP: &monitoringcfg.MonitoringHTTPConfig{
+				Enabled: false,
+			},
+		},
+	}
+
+	policy := map[string]any{
+		"agent": map[string]any{
+			"monitoring": map[string]any{
+				"metrics": true,
+				"logs":    false,
+			},
+		},
+		"outputs": map[string]any{
+			"default": map[string]any{},
+		},
+	}
+
+	b := &BeatsMonitor{
+		enabled:   true,
+		config:    cfg,
+		agentInfo: agentInfo,
+	}
+
+	components := []component.Component{
+		{
+			ID: "filestream-default",
+			InputSpec: &component.InputRuntimeSpec{
+				Spec: component.InputSpec{
+					Service: &component.ServiceSpec{
+						Log: &component.ServiceLogSpec{
+							Path: "/tmp/foo",
+						},
+					},
+				},
+			},
+		},
+	}
+	monitoringConfig, err := b.MonitoringConfig(policy, components, map[string]string{"filestream-default": "filebeat"})
+	if err != nil {
+		t.Fatalf("cannot render monitoring configuration: %s", err)
+	}
+
+	// This is a test and the structure of `monitoringConfig` is well know,
+	// so we coerce everything to the correct type. If something does not match
+	// the test will panic.
+	inputsSlice := monitoringConfig["inputs"].([]any)
+	for _, input := range inputsSlice {
+		inpMap := input.(map[string]any)
+		for _, rawStream := range inpMap["streams"].([]any) {
+			streamID := rawStream.(map[string]any)["id"].(string)
+			processors := rawStream.(map[string]any)["processors"].([]any)
+			for _, rawProcessor := range processors {
+				processor := rawProcessor.(map[string]any)
+				if _, exists := processor["add_fields"]; !exists {
+					continue
+				}
+				p := Processor{}
+				if err := json.Unmarshal([]byte(mapstr.M(processor).String()), &p); err != nil {
+					t.Errorf("could not decode processor config: %q, err: %s", "foo", err)
+				}
+				if p.AddFields.Target != "component" {
+					continue
+				}
+
+				binary := p.AddFields.Fields.Binary
+				componentID := p.AddFields.Fields.ID
+
+				// The elastic-Agent is a special case, handle it first
+				if strings.Contains(streamID, "monitoring-agent") {
+					if binary != "elastic-agent" {
+						t.Errorf("expecting fields['binary'] = 'elastic-agent', got %q", binary)
+					}
+					if componentID != "elastic-agent" {
+						t.Errorf("expecting fields['id'] = 'elastic-agent', got %q", componentID)
+					}
+					continue
+				}
+				if binary != "filebeat" {
+					t.Errorf("expecting fields['binary'] = 'filebeat', got %q", binary)
+				}
+				if componentID != "filestream-default" {
+					t.Errorf("expecting fields['id'] = 'filestream-default', got %q", componentID)
+				}
+
+			}
+		}
+	}
+}
+
+type Processor struct {
+	AddFields AddFields `json:"add_fields"`
+}
+type Fields struct {
+	Binary string `json:"binary"`
+	ID     string `json:"id"`
+}
+type AddFields struct {
+	Fields Fields `json:"fields"`
+	Target string `json:"target"`
+}