fix panic when running 2 filestream inputs with same ID

libbeat/monitoring/inputmon/input.go

@@ -116,6 +116,9 @@ func MetricSnapshotJSON(reg *monitoring.Registry) ([]byte, error) {
 // any '.' is replaced by '_'. The new registry is initialized with
 // 'id: inputID' and 'input: inputType'.
 //
+// If there is already a registry with the same name on parent, a new registry
+// not associated with parent will be returned.
+//
 // Call CancelMetricsRegistry to remove it from the parent registry and free up
 // the associated resources.
 func NewMetricsRegistry(
@@ -129,8 +132,10 @@ func NewMetricsRegistry(
 	if reg == nil {
 		reg = parent.NewRegistry(registryID)
 	} else {
+		// Null route metrics for duplicated ID.
+		reg = monitoring.NewRegistry()
 		log.Warnw(fmt.Sprintf(
-			"parent metrics registry already contains a %q registry, reusing it",
+			"parent metrics registry already contains a %q registry, returning an unregistered registry. Metrics won't be available for this input instance",
 			registryID),
 			"registry_id", registryID,
 			"input_type", inputType,

libbeat/monitoring/inputmon/input_test.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/monitoring"
+	"github.com/elastic/elastic-agent-libs/monitoring/adapter"
 )
 
 func TestNewInputMonitor(t *testing.T) {
@@ -220,6 +221,42 @@ func TestNewMetricsRegistry(t *testing.T) {
 	assert.Equal(t, inputType, vals.Strings["input"])
 }
 
+func TestNewMetricsRegistry_duplicatedInputID(t *testing.T) {
+	parent := monitoring.NewRegistry()
+	inputID := "input-inputID"
+	inputType := "input-type"
+	metricName := "foo_total"
+	goMetricsRegistryName := "bar_registry"
+
+	// 1st call, create the registry
+	got := NewMetricsRegistry(
+		inputID,
+		inputType,
+		parent,
+		logptest.NewTestingLogger(t, "test"))
+
+	require.NotNil(t, got, "new metrics registry should not be nil")
+	assert.Equal(t, parent.GetRegistry(inputID), got)
+	// register a metric to the registry
+	monitoring.NewInt(got, metricName)
+	adapter.NewGoMetrics(got, goMetricsRegistryName, adapter.Accept)
+
+	// 2nd call, return an unregistered registry
+	got = NewMetricsRegistry(
+		inputID,
+		inputType,
+		parent,
+		logptest.NewTestingLogger(t, "test"))
+	require.NotNil(t, got, "new metrics registry should not be nil")
+	assert.NotEqual(t, parent.GetRegistry(inputID), got,
+		"should get an unregistered registry, but found the registry on parent")
+	assert.NotPanics(t, func() {
+		// register the same metric again
+		monitoring.NewInt(got, metricName)
+		adapter.NewGoMetrics(got, goMetricsRegistryName, adapter.Accept)
+	}, "the registry should be a new and empty registry")
+}
+
 func TestCancelMetricsRegistry(t *testing.T) {
 	parent := monitoring.NewRegistry()
 	inputID := "input-ID"