[filebeat] Check for duplicate ID for filestream metrics (#35972)

- If a duplicate ID was used for filestream, set the ID for metrics
to an empty string to prevent a panic.
- A valid metrics instance will still be generated, but collection of
metrics will not occur.

CHANGELOG.next.asciidoc

@@ -142,6 +142,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix CEL input JSON marshalling of nested objects. {issue}35763[35763] {pull}35774[35774]
 - Fix metric collection in GCPPubSub input. {pull}35773[35773]
 - Fix end point deregistration in http_endpoint input. {issue}35899[35899] {pull}35903[35903]
+- Fix duplicate ID panic in filestream metrics. {issue}35964[35964] {pull}35972[35972]
 
 *Heartbeat*
 

filebeat/input/filestream/internal/input-logfile/input.go

@@ -30,6 +30,7 @@ import (
 
 type managedInput struct {
 	userID           string
+	metricsID        string
 	manager          *InputManager
 	ackCH            *updateChan
 	sourceIdentifier *sourceIdentifier
@@ -61,7 +62,7 @@ func (inp *managedInput) Run(
 	defer cancel()
 	ctx.Cancelation = cancelCtx
 
-	metrics := NewMetrics(ctx.ID)
+	metrics := NewMetrics(inp.metricsID)
 	defer metrics.Close()
 
 	hg := &defaultHarvesterGroup{

filebeat/input/filestream/internal/input-logfile/manager.go

@@ -177,10 +177,13 @@ func (cim *InputManager) Create(config *conf.C) (v2.Input, error) {
 			" duplication, please add an ID and restart Filebeat")
 	}
 
+	metricsID := settings.ID
 	cim.idsMux.Lock()
 	if _, exists := cim.ids[settings.ID]; exists {
 		cim.Logger.Errorf("filestream input with ID '%s' already exists, this "+
-			"will lead to data duplication, please use a different ID", settings.ID)
+			"will lead to data duplication, please use a different ID. Metrics "+
+			"collection has been disabled on this input.", settings.ID)
+		metricsID = ""
 	}
 
 	// TODO: improve how inputs with empty IDs are tracked.
@@ -223,6 +226,7 @@ func (cim *InputManager) Create(config *conf.C) (v2.Input, error) {
 		manager:          cim,
 		ackCH:            cim.ackCH,
 		userID:           settings.ID,
+		metricsID:        metricsID,
 		prospector:       prospector,
 		harvester:        harvester,
 		sourceIdentifier: sourceIdentifier,