WIP - running CI

CHANGELOG.next.asciidoc

@@ -20,6 +20,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Replace default Ubuntu-based images with UBI-minimal-based ones {pull}42150[42150]
 - Fix templates and docs to use correct `--` version of command line arguments. {issue}42038[42038] {pull}42060[42060]
 - removed support for a single `-` to precede multi-letter command line arguments.  Use `--` instead. {issue}42117[42117] {pull}42209[42209]
+- add per-input metrics to libbeat pipeline client {pull}42618[42618]
 
 *Auditbeat*
 

filebeat/beater/filebeat.go

@@ -134,15 +134,15 @@ func newBeater(b *beat.Beat, plugins PluginFactory, rawConfig *conf.C) (beat.Bea
 	}
 
 	if b.API != nil {
-		if err = inputmon.AttachHandler(b.API.Router()); err != nil {
+		if err = inputmon.AttachHandler(b.Info, b.API.Router()); err != nil {
 			return nil, fmt.Errorf("failed attach inputs api to monitoring endpoint server: %w", err)
 		}
 	}
 
 	if b.Manager != nil {
 		b.Manager.RegisterDiagnosticHook("input_metrics", "Metrics from active inputs.",
 			"input_metrics.json", "application/json", func() []byte {
-				data, err := inputmon.MetricSnapshotJSON()
+				data, err := inputmon.MetricSnapshotJSON(b.Info)
 				if err != nil {
 					logp.L().Warnw("Failed to collect input metric snapshot for Agent diagnostics.", "error", err)
 					return []byte(err.Error())

filebeat/docs/inputs/input-filestream.asciidoc

@@ -187,6 +187,10 @@ observe the activity of the input. Note that metrics from processors are not inc
 | `messages_truncated_total` | Total number of messages truncated.
 | `bytes_processed_total`    | Total number of bytes processed.
 | `events_processed_total`   | Total number of events processed.
+| `events_pipeline_total`   | Total number of events processed by the publishing pipeline.
+| `events_pipeline_failed_total`   | Total number of events failed at the publishing pipeline.
+| `events_pipeline_filtered_total`   | Total number of events filtered at the publishing pipeline.
+| `events_pipeline_published_total`   | Total number of events published at the publishing pipeline.
 | `processing_errors_total`  | Total number of processing errors.
 | `processing_time`          | Histogram of the elapsed time to process messages (expressed in nanoseconds).
 |=======

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

@@ -221,6 +221,7 @@ func startHarvester(
 		defer releaseResource(resource)
 
 		client, err := hg.pipeline.ConnectWith(beat.ClientConfig{
+			InputID:       ctx.ID,
 			EventListener: newInputACKHandler(hg.ackCH),
 		})
 		if err != nil {

filebeat/input/v2/input-cursor/input.go

@@ -147,6 +147,7 @@ func (inp *managedInput) runSource(
 	}()
 
 	client, err := pipeline.ConnectWith(beat.ClientConfig{
+		InputID:       ctx.ID,
 		EventListener: newInputACKHandler(ctx.Logger),
 	})
 	if err != nil {

filebeat/input/v2/input-stateless/stateless.go

@@ -84,6 +84,7 @@ func (si configuredInput) Run(ctx v2.Context, pipeline beat.PipelineConnector) (
 	}()
 
 	client, err := pipeline.ConnectWith(beat.ClientConfig{
+		InputID:     ctx.ID,
 		PublishMode: beat.DefaultGuarantees,
 	})
 	if err != nil {

libbeat/beat/pipeline.go

@@ -45,6 +45,10 @@ type Client interface {
 // ClientConfig defines common configuration options one can pass to
 // Pipeline.ConnectWith to control the clients behavior and provide ACK support.
 type ClientConfig struct {
+	// InputID is the input ID of the input using the client. The InputID is
+	// used to aggregate pipeline metrics per input.
+	InputID string
+
 	PublishMode PublishMode
 
 	Processing ProcessingConfig

libbeat/cmd/instance/beat.go

@@ -588,11 +588,13 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 
 	var publisher *pipeline.Pipeline
 	monitors := pipeline.Monitors{
-		Metrics:   reg,
-		Telemetry: monitoring.GetNamespace("state").GetRegistry(),
-		Logger:    logp.L().Named("publisher"),
-		Tracer:    b.Instrumentation.Tracer(),
+		BeatRegistry: b.Info.Monitoring.Namespace.GetRegistry(),
+		Metrics:      reg,
+		Telemetry:    monitoring.GetNamespace("state").GetRegistry(),
+		Logger:       logp.L().Named("publisher"),
+		Tracer:       b.Instrumentation.Tracer(),
 	}
+	// here
 	outputFactory := b.makeOutputFactory(b.Config.Output)
 	settings := pipeline.Settings{
 		// Since now publisher is closed on Stop, we want to give some

libbeat/monitoring/inputmon/httphandler.go

@@ -26,6 +26,8 @@ import (
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 
+	"github.com/elastic/beats/v7/libbeat/beat"
+	libbeatmonitoring "github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
@@ -36,19 +38,31 @@ const (
 )
 
 type handler struct {
-	registry *monitoring.Registry
+	registryDataset        *monitoring.Registry
+	registryInternalInputs *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())
+func AttachHandler(beatInfo beat.Info, r *mux.Router) error {
+	intInputsReg := beatInfo.Monitoring.Namespace.GetRegistry().
+		GetRegistry(libbeatmonitoring.RegistryNameInternalInputs)
+	if intInputsReg == nil {
+		intInputsReg = beatInfo.Monitoring.Namespace.GetRegistry().
+			NewRegistry(libbeatmonitoring.RegistryNameInternalInputs)
+	}
+
+	return attachHandler(r, globalRegistry(), intInputsReg)
 }
 
-func attachHandler(r *mux.Router, registry *monitoring.Registry) error {
-	h := &handler{registry: registry}
+func attachHandler(r *mux.Router, datasetReg, intInputsReg *monitoring.Registry) error {
 	r = r.PathPrefix(route).Subrouter()
-	return r.StrictSlash(true).Handle("/", validationHandler("GET", []string{"pretty", "type"}, h.allInputs)).GetError()
+
+	h := &handler{
+		registryDataset:        datasetReg,
+		registryInternalInputs: intInputsReg,
+	}
+	return r.StrictSlash(true).Handle("/", validationHandler(http.MethodGet, []string{"pretty", "type"}, h.allInputs)).GetError()
 }
 
 func (h *handler) allInputs(w http.ResponseWriter, req *http.Request) {
@@ -64,14 +78,15 @@ func (h *handler) allInputs(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	filtered := filteredSnapshot(h.registry, requestedType)
+	filtered := filteredSnapshot(
+		h.registryDataset, h.registryInternalInputs, 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, intInputs *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 {
@@ -81,19 +96,39 @@ 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
 		}
 
-		if inputType, ok := m["input"].(string); !ok || (requestedType != "" && !strings.EqualFold(inputType, requestedType)) {
+		if inputType, ok := m["input"].(string); !ok || (requestedType != "" &&
+			!strings.EqualFold(inputType, requestedType)) {
 			continue
 		}
 
+		// merge metrics stored in the internal namespace if any is found
+		mergeInternalMetrics(intInputs, id, m)
+
 		filtered = append(filtered, m)
 	}
 	return filtered
 }
 
+// mergeInternalMetrics looks for a registry identified by id in the internal
+// registry. If found, all the metrics are merged into m, if not, m is not
+// changed.
+func mergeInternalMetrics(internal *monitoring.Registry, id string, m map[string]any) {
+	reg := internal.GetRegistry(id)
+	if reg == nil {
+		return
+	}
+
+	intInput := monitoring.CollectStructSnapshot(reg, monitoring.Full, false)
+	for k, v := range intInput {
+		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

@@ -30,6 +30,7 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
@@ -53,7 +54,9 @@ var testCases = []TestCase{
 }
 
 func TestHandler(t *testing.T) {
+	namespace := monitoring.GetNamespace("TestHandler")
 	parent := monitoring.NewRegistry()
+
 	reg, _ := NewInputRegistry("foo", "123abc", parent)
 	monitoring.NewInt(reg, "gauge").Set(13344)
 
@@ -68,7 +71,9 @@ func TestHandler(t *testing.T) {
 	s := httptest.NewServer(r)
 	defer s.Close()
 
-	if err := attachHandler(r, parent); err != nil {
+	beatInfo := beat.Info{}
+	beatInfo.Monitoring.Namespace = namespace
+	if err := attachHandler(r, parent, monitoring.NewRegistry()); err != nil {
 		t.Fatal(err)
 	}
 
@@ -116,7 +121,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,6 +23,8 @@ import (
 
 	"github.com/gofrs/uuid/v5"
 
+	"github.com/elastic/beats/v7/libbeat/beat"
+	libbeatmonitoring "github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
@@ -81,7 +83,21 @@ func globalRegistry() *monitoring.Registry {
 }
 
 // MetricSnapshotJSON returns a snapshot of the input metric values from the
-// global 'dataset' monitoring namespace encoded as a JSON array (pretty formatted).
-func MetricSnapshotJSON() ([]byte, error) {
-	return json.MarshalIndent(filteredSnapshot(globalRegistry(), ""), "", "  ")
+// global 'dataset' and libbeatmonitoring.RegistryNameInternalInputs monitoring
+// namespace from the beatInfo instance. It returns a pretty formated JSON array
+// as a byte slice.
+func MetricSnapshotJSON(beatInfo beat.Info) ([]byte, error) {
+	intReg := beatInfo.Monitoring.Namespace.GetRegistry().
+		GetRegistry(libbeatmonitoring.RegistryNameInternalInputs)
+	if intReg == nil {
+		intReg = beatInfo.Monitoring.Namespace.GetRegistry().
+			NewRegistry(libbeatmonitoring.RegistryNameInternalInputs)
+	}
+	return json.MarshalIndent(
+		filteredSnapshot(
+			globalRegistry(),
+			intReg,
+			""),
+		"",
+		"  ")
 }

libbeat/monitoring/inputmon/input_test.go

@@ -24,6 +24,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/elastic/beats/v7/libbeat/beat"
+	libbeatmonitoring "github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
@@ -87,17 +89,45 @@ func TestMetricSnapshotJSON(t *testing.T) {
 		require.NoError(t, globalRegistry().Clear())
 	})
 
-	r, cancel := NewInputRegistry("test", "my-id", nil)
-	defer cancel()
-	monitoring.NewInt(r, "foo_total").Set(100)
-
-	jsonBytes, err := MetricSnapshotJSON()
+	inputID := "input-with-pipeline-metrics"
+	r1, cancel1 := NewInputRegistry("test", inputID, nil)
+	defer cancel1()
+	monitoring.NewInt(r1, "foo1_total").Set(100)
+
+	r2, cancel2 := NewInputRegistry(
+		"test", "input-without-pipeline-metrics", nil)
+	defer cancel2()
+	monitoring.NewInt(r2, "foo2_total").Set(100)
+
+	// this metric should not be reported
+	r3 := globalRegistry().NewRegistry("another-registry")
+	monitoring.NewInt(r3, "foo3_total").Set(100)
+
+	// this metric should not be reported
+	r4 := globalRegistry().NewRegistry("yet-another-registry")
+	monitoring.NewString(r4, "id").Set("some-id")
+	monitoring.NewInt(r3, "foo3_total").Set(100)
+
+	bInfo := beat.Info{}
+	bInfo.Monitoring.Namespace = monitoring.GetNamespace("TestMetricSnapshotJSON")
+	intInputReg := bInfo.Monitoring.Namespace.GetRegistry().
+		NewRegistry(libbeatmonitoring.RegistryNameInternalInputs).
+		NewRegistry(inputID)
+	monitoring.NewInt(intInputReg, "events_pipeline_total").Set(100)
+
+	jsonBytes, err := MetricSnapshotJSON(bInfo)
 	require.NoError(t, err)
 
 	const expected = `[
   {
-    "foo_total": 100,
-    "id": "my-id",
+    "events_pipeline_total": 100,
+    "foo1_total": 100,
+    "id": "input-with-pipeline-metrics",
+    "input": "test"
+  },
+  {
+    "foo2_total": 100,
+    "id": "input-without-pipeline-metrics",
     "input": "test"
   }
 ]`

libbeat/monitoring/monitoring.go

@@ -0,0 +1,22 @@
+// 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 monitoring
+
+const (
+	RegistryNameInternalInputs = "internal.inputs"
+)