wip

libbeat/monitoring/inputmon/httphandler.go

@@ -36,17 +36,18 @@ const (
 )
 
 type handler struct {
-	registry *monitoring.Registry
+	registryDataset  *monitoring.Registry
+	registryInternal *monitoring.Registry
 }
 
 // AttachHandler attaches an HTTP handler to the given mux.Router to handle
 // requests to /inputs.
 func AttachHandler(r *mux.Router) error {
-	return attachHandler(r, globalRegistry())
+	return attachHandler(r, globalDatasetRegistry(), globalInternalRegistry())
 }
 
-func attachHandler(r *mux.Router, registry *monitoring.Registry) error {
-	h := &handler{registry: registry}
+func attachHandler(r *mux.Router, datasetReg, internalReg *monitoring.Registry) error {
+	h := &handler{registryDataset: datasetReg, registryInternal: internalReg}
 	r = r.PathPrefix(route).Subrouter()
 	return r.StrictSlash(true).Handle("/", validationHandler("GET", []string{"pretty", "type"}, h.allInputs)).GetError()
 }
@@ -64,17 +65,15 @@ func (h *handler) allInputs(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	filtered := filteredSnapshot(h.registry, requestedType)
-	// global := filteredSnapshot(globalRegistry(), "")
-	//
-	// logp.L().Info("global:", global)
-	// logp.L().Info("filtered:", filtered)
+	filtered := filteredSnapshot(
+		h.registryDataset, h.registryInternal, requestedType)
+
 	w.Header().Set(contentType, applicationJSON)
 	serveJSON(w, filtered, requestedPretty)
 }
 
-func filteredSnapshot(r *monitoring.Registry, requestedType string) []map[string]any {
-	metrics := monitoring.CollectStructSnapshot(r, monitoring.Full, false)
+func filteredSnapshot(dataset, internal *monitoring.Registry, requestedType string) []map[string]any {
+	metrics := monitoring.CollectStructSnapshot(dataset, monitoring.Full, false)
 
 	filtered := make([]map[string]any, 0, len(metrics))
 	for _, ifc := range metrics {
@@ -84,7 +83,8 @@ func filteredSnapshot(r *monitoring.Registry, requestedType string) []map[string
 		}
 
 		// Require all entries to have an 'input' and 'id' to be accessed through this API.
-		if id, ok := m["id"].(string); !ok || id == "" {
+		id, ok := m["id"].(string)
+		if !ok || id == "" {
 			continue
 		}
 
@@ -93,11 +93,27 @@ func filteredSnapshot(r *monitoring.Registry, requestedType string) []map[string
 			continue
 		}
 
+		// merge the internal namespace if found
+		mergeInternalMetrics(internal, id, m)
+
 		filtered = append(filtered, m)
 	}
 	return filtered
 }
 
+// mergeInternalMetrics looks
+func mergeInternalMetrics(internal *monitoring.Registry, id string, m map[string]any) {
+	intInput := monitoring.CollectStructSnapshot(internal, monitoring.Full, false)
+	if rawInputPublishMetrics, found := intInput[id]; found {
+		inputPublishMetrics, inputMetrincsOk := rawInputPublishMetrics.(map[string]any)
+		if inputMetrincsOk {
+			for k, v := range inputPublishMetrics {
+				m[k] = v
+			}
+		}
+	}
+}
+
 func serveJSON(w http.ResponseWriter, value any, pretty bool) {
 	w.Header().Set(contentType, applicationJSON)
 	enc := json.NewEncoder(w)

libbeat/monitoring/inputmon/httphandler_test.go

@@ -116,7 +116,7 @@ func BenchmarkHandlers(b *testing.B) {
 		monitoring.NewInt(reg, "gauge").Set(int64(i))
 	}
 
-	h := &handler{registry: reg}
+	h := &handler{registryDataset: reg}
 
 	b.Run("allInputs", func(b *testing.B) {
 		req := httptest.NewRequest(http.MethodGet, "/inputs/", nil)

libbeat/monitoring/inputmon/input.go

@@ -23,12 +23,13 @@ import (
 
 	"github.com/gofrs/uuid/v5"
 
+	libbeatmonitoring "github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
 // NewInputRegistry returns a new monitoring.Registry for metrics related to
-// an input instance. The returned registry will be initialized with a static
+// an input instance. The returned registryDataset will be initialized with a static
 // 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
@@ -39,12 +40,12 @@ func NewInputRegistry(inputType, id string, optionalParent *monitoring.Registry)
 	// Use the default registry unless one was provided (this would be for testing).
 	parentRegistry := optionalParent
 	if parentRegistry == nil {
-		parentRegistry = globalRegistry()
+		parentRegistry = globalDatasetRegistry()
 	}
 
 	// 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() {
+	if (id == "" || inputType == "") && parentRegistry == globalDatasetRegistry() {
 		// Null route metrics without ID or input type.
 		parentRegistry = monitoring.NewRegistry()
 	}
@@ -76,12 +77,21 @@ func sanitizeID(id string) string {
 	return strings.ReplaceAll(id, ".", "_")
 }
 
-func globalRegistry() *monitoring.Registry {
-	return monitoring.GetNamespace("dataset").GetRegistry()
+func globalDatasetRegistry() *monitoring.Registry {
+	return globalRegistry("dataset")
+}
+
+func globalInternalRegistry() *monitoring.Registry {
+	return globalRegistry(libbeatmonitoring.RegistryNameInternalInputs)
+}
+
+func globalRegistry(namespace string) *monitoring.Registry {
+	return monitoring.GetNamespace(namespace).GetRegistry()
 }
 
 // MetricSnapshotJSON returns a snapshot of the input metric values from the
-// global 'dataset' monitoring namespace encoded as a JSON array (pretty formatted).
+// global 'dataset' and 'internal' monitoring namespace merged and encoded as a
+// JSON array (pretty formatted).
 func MetricSnapshotJSON() ([]byte, error) {
-	return json.MarshalIndent(filteredSnapshot(globalRegistry(), ""), "", "  ")
+	return json.MarshalIndent(filteredSnapshot(globalDatasetRegistry(), globalInternalRegistry(), ""), "", "  ")
 }

libbeat/monitoring/inputmon/input_test.go

@@ -44,10 +44,10 @@ func TestNewInputMonitor(t *testing.T) {
 		{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: globalDatasetRegistry(), PublicMetrics: true},
+		{Input: "", ID: id, OptionalParent: globalDatasetRegistry(), PublicMetrics: false},
+		{Input: inputType, ID: "", OptionalParent: globalDatasetRegistry(), PublicMetrics: false},
+		{Input: "", ID: "", OptionalParent: globalDatasetRegistry(), PublicMetrics: false},
 
 		{Input: inputType, ID: id, OptionalParent: monitoring.NewRegistry(), PublicMetrics: false},
 		{Input: "", ID: id, OptionalParent: monitoring.NewRegistry(), PublicMetrics: false},
@@ -66,12 +66,12 @@ func TestNewInputMonitor(t *testing.T) {
 			assert.NotNil(t, reg)
 
 			// Verify that metrics are registered when a custom parent registry is given.
-			if tc.OptionalParent != nil && tc.OptionalParent != globalRegistry() {
+			if tc.OptionalParent != nil && tc.OptionalParent != globalDatasetRegistry() {
 				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)
+			parent := globalDatasetRegistry().GetRegistry(tc.ID)
 			if tc.PublicMetrics {
 				assert.NotNil(t, parent)
 			} else {
@@ -82,9 +82,9 @@ func TestNewInputMonitor(t *testing.T) {
 }
 
 func TestMetricSnapshotJSON(t *testing.T) {
-	require.NoError(t, globalRegistry().Clear())
+	require.NoError(t, globalDatasetRegistry().Clear())
 	t.Cleanup(func() {
-		require.NoError(t, globalRegistry().Clear())
+		require.NoError(t, globalDatasetRegistry().Clear())
 	})
 
 	r, cancel := NewInputRegistry("test", "my-id", nil)

libbeat/monitoring/monitoring.go

@@ -0,0 +1,5 @@
+package monitoring
+
+const (
+	RegistryNameInternalInputs = "internal.inputs"
+)

libbeat/publisher/pipeline/monitoring.go

@@ -21,6 +21,7 @@ import (
 	"strings"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
+	libbeatmonitoring "github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
@@ -201,6 +202,7 @@ func (o *metricsObserver) failedPublishEvent(e beat.Event) {
 	input.inputEventsDropped.Inc()
 }
 
+// TODO: remove metricName
 func (o *metricsObserver) ensureInputMetric(e beat.Event, metricName string) *inputVars {
 	// TODO:
 	// - find the right global registry to add the metrics to. dataset.inputID sanitized. See inputmon.NewInputRegistry()
@@ -217,24 +219,7 @@ func (o *metricsObserver) ensureInputMetric(e beat.Event, metricName string) *in
 		return nil
 	}
 
-	// rawFieldInput, err := e.Fields.GetValue(beat.FieldsKeyInput)
-	// if err != nil {
-	// 	return nil // again, nothing we can do about it
-	// }
-	// fieldInput, ok := rawFieldInput.(mapstr.M)
-	// if !ok {
-	// 	// again, nothing we can do about it
-	// 	return nil
-	// }
-	// rawType, err := fieldInput.GetValue("type")
-	// if err != nil {
-	// 	return nil // again, nothing we can do about it
-	// }
-	// fieldType, ok := rawType.(string)
-	// if !ok {
-	// 	return nil // again, nothing we can do about it
-	// }
-
+	// for debug, remove it
 	datasetReg := monitoring.GetNamespace("dataset").GetRegistry()
 	sanatizedID := strings.ReplaceAll(inputID, ".", "_")
 	inputReg := datasetReg.GetRegistry(sanatizedID)
@@ -247,14 +232,25 @@ func (o *metricsObserver) ensureInputMetric(e beat.Event, metricName string) *in
 
 	input, found := o.vars.inputs[inputID]
 	if !found {
-		reg := o.metrics.GetRegistry("pipeline")
+		intReg := monitoring.
+			GetNamespace(libbeatmonitoring.RegistryNameInternalInputs).
+			GetRegistry()
+		reg := intReg.GetRegistry(inputID)
+		if reg == nil {
+			reg = intReg.NewRegistry(inputID)
+		}
+
 		input = inputVars{
-			inputEventsFiltered:  monitoring.NewUint(reg, "inputs."+inputID+".events.filtered"),
-			inputEventsPublished: monitoring.NewUint(reg, "inputs."+inputID+".events.published"),
+			inputEventsDropped:   monitoring.NewUint(reg, "events_dropped_total"),
+			inputEventsFiltered:  monitoring.NewUint(reg, "events_filtered_total"),
+			inputEventsPublished: monitoring.NewUint(reg, "events_published_total"),
 		}
 		o.vars.inputs[inputID] = input
 	}
-
+	reg := monitoring.GetRegistry(libbeatmonitoring.RegistryNameInternalInputs)
+	if reg != nil {
+		_ = reg
+	}
 	return &input
 }