libbeat/monitoring/inputmon - Add defensive check for empty ID or inp…

…ut type (#35427)

Add a defensive check to prevent metrics from being registered into the global
'dataset' namespace when the ID or input type are empty.

This is not in response to any known problem. There was already a similar check
on HTTP /inputs/ endpoint that would have prevented metrics from inputs with
an ID or input type from being exposed. This pushes that check earlier and in
effect "null routes" those metrics an a manner that is transparent to the callers.

I would consider an empty input type to be a developer mistake whereas an the
ID is user controllable. So you could conceive of having a hard error (like a panic)
if input type were empty. At the moment I'm taking a softer approach in the defense.

Relates #35354

libbeat/monitoring/inputmon/input.go

@@ -28,23 +28,32 @@ import (
 // string values for the input and id. When the input stops it should invoke
 // the returned cancel function to unregister the metrics. For testing purposes
 // an optional monitoring.Registry may be provided as an alternative to using
-// the global 'dataset' monitoring namespace.
+// the global 'dataset' monitoring namespace. The inputType and id must be
+// non-empty for the metrics to be published to the global 'dataset' monitoring
+// namespace.
 func NewInputRegistry(inputType, id string, optionalParent *monitoring.Registry) (reg *monitoring.Registry, cancel func()) {
 	// Use the default registry unless one was provided (this would be for testing).
-	rootRegistry := optionalParent
-	if rootRegistry == nil {
-		rootRegistry = globalRegistry()
+	parentRegistry := optionalParent
+	if parentRegistry == nil {
+		parentRegistry = globalRegistry()
+	}
+
+	// If an ID has not been assigned to an input then metrics cannot be exposed
+	// in the global metric registry. The returned registry still behaves the same.
+	if (id == "" || inputType == "") && parentRegistry == globalRegistry() {
+		// Null route metrics without ID or input type.
+		parentRegistry = monitoring.NewRegistry()
 	}
 
 	// Sanitize dots from the id because they created nested objects within
 	// the monitoring registry, and we want a consistent flat level of nesting
 	key := sanitizeID(id)
 
-	reg = rootRegistry.NewRegistry(key)
+	reg = parentRegistry.NewRegistry(key)
 	monitoring.NewString(reg, "input").Set(inputType)
 	monitoring.NewString(reg, "id").Set(id)
 
-	return reg, func() { rootRegistry.Remove(key) }
+	return reg, func() { parentRegistry.Remove(key) }
 }
 
 func sanitizeID(id string) string {

libbeat/monitoring/inputmon/input_test.go

@@ -0,0 +1,81 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package inputmon
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/elastic-agent-libs/monitoring"
+)
+
+func TestNewInputMonitor(t *testing.T) {
+	const (
+		inputType = "foo-input"
+		id        = "my-id"
+	)
+
+	testCases := []struct {
+		Input          string
+		ID             string
+		OptionalParent *monitoring.Registry
+		PublicMetrics  bool // Are the metrics registered in the global metric namespace making them public?
+	}{
+		{Input: inputType, ID: id, PublicMetrics: true},
+		{Input: "", ID: id, PublicMetrics: false},
+		{Input: inputType, ID: "", PublicMetrics: false},
+		{Input: "", ID: "", PublicMetrics: false},
+
+		{Input: inputType, ID: id, OptionalParent: globalRegistry(), PublicMetrics: true},
+		{Input: "", ID: id, OptionalParent: globalRegistry(), PublicMetrics: false},
+		{Input: inputType, ID: "", OptionalParent: globalRegistry(), PublicMetrics: false},
+		{Input: "", ID: "", OptionalParent: globalRegistry(), PublicMetrics: false},
+
+		{Input: inputType, ID: id, OptionalParent: monitoring.NewRegistry(), PublicMetrics: false},
+		{Input: "", ID: id, OptionalParent: monitoring.NewRegistry(), PublicMetrics: false},
+		{Input: inputType, ID: "", OptionalParent: monitoring.NewRegistry(), PublicMetrics: false},
+		{Input: "", ID: "", OptionalParent: monitoring.NewRegistry(), PublicMetrics: false},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+		testName := fmt.Sprintf("with_id=%v/with_input=%v/custom_parent=%v/public_metrics=%v",
+			tc.ID != "", tc.Input != "", tc.OptionalParent != nil, tc.PublicMetrics)
+
+		t.Run(testName, func(t *testing.T) {
+			reg, unreg := NewInputRegistry(tc.Input, tc.ID, tc.OptionalParent)
+			defer unreg()
+			assert.NotNil(t, reg)
+
+			// Verify that metrics are registered when a custom parent registry is given.
+			if tc.OptionalParent != nil && tc.OptionalParent != globalRegistry() {
+				assert.NotNil(t, tc.OptionalParent.Get(tc.ID))
+			}
+
+			// Verify whether the metrics are exposed in the global registry which makes the public.
+			parent := globalRegistry().GetRegistry(tc.ID)
+			if tc.PublicMetrics {
+				assert.NotNil(t, parent)
+			} else {
+				assert.Nil(t, parent)
+			}
+		})
+	}
+}