winlogbeat/eventlog: add initial event processing metrics support

CHANGELOG.next.asciidoc

@@ -228,6 +228,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 
 *Winlogbeat*
 
+- Add metrics for log event processing. {pull}33922[33922]
 
 *Elastic Log Driver*
 

winlogbeat/beater/winlogbeat.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/esleg/eslegclient"
+	"github.com/elastic/beats/v7/libbeat/monitoring/inputmon"
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	"github.com/elastic/beats/v7/winlogbeat/module"
 	conf "github.com/elastic/elastic-agent-libs/config"
@@ -154,6 +155,12 @@ func (eb *Winlogbeat) Run(b *beat.Beat) error {
 
 	// Initialize metrics.
 	initMetrics("total")
+	if b.API != nil {
+		err := inputmon.AttachHandler(b.API.Router())
+		if err != nil {
+			return fmt.Errorf("failed attach inputs api to monitoring endpoint server: %w", err)
+		}
+	}
 
 	var wg sync.WaitGroup
 	for _, log := range eb.eventLogs {

winlogbeat/docs/configuring-howto.asciidoc

@@ -53,6 +53,24 @@ include::{libbeat-dir}/loggingconfig.asciidoc[]
 
 include::{libbeat-dir}/http-endpoint.asciidoc[]
 
+[float]
+=== Metrics
+
+{beatname_uc} exposes the following additional metrics under the <<http-endpoint, HTTP monitoring endpoint>>.
+These metrics are exposed under the `/dataset` path.
+
+[options="header"]
+|=======
+| Metric                   | Description
+| `provider`               | The name of the provider being read.
+| `received_events_total`  | Total number of events read by the input.
+| `discarded_events_total` | Total number of events dropped by the input.
+| `errors_total`           | Total number of errors encountered by the input.
+| `batch_read_period`      | A histogram of intervals between non-empty event batch reads.
+| `received_events_count`  | A histogram of the number of events read in each batch.
+| `source_lag_time`        | The difference between the timestamp recorded in each event and the time when it was read.
+|=======
+
 include::{libbeat-dir}/shared-instrumentation.asciidoc[]
 
 include::{libbeat-dir}/reference-yml.asciidoc[]

winlogbeat/eventlog/eventlog.go

@@ -23,11 +23,16 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/rcrowley/go-metrics"
+
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/monitoring/inputmon"
 	"github.com/elastic/beats/v7/winlogbeat/checkpoint"
 	"github.com/elastic/beats/v7/winlogbeat/sys/winevent"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/monitoring"
+	"github.com/elastic/elastic-agent-libs/monitoring/adapter"
 )
 
 // Debug selectors used in this package.
@@ -138,3 +143,98 @@ func incrementMetric(v *expvar.Map, key interface{}) {
 		v.Add(strconv.Itoa(int(t)), 1)
 	}
 }
+
+// inputMetrics handles event log metric reporting.
+type inputMetrics struct {
+	unregister func()
+
+	lastBatch time.Time
+
+	name        *monitoring.String // name of the provider being read
+	events      *monitoring.Uint   // total number of events received
+	dropped     *monitoring.Uint   // total number of discarded events
+	errors      *monitoring.Uint   // total number of errors
+	batchSize   metrics.Sample     // histogram of the number of events in each non-zero batch
+	sourceLag   metrics.Sample     // histogram of the difference between timestamped event's creation and reading
+	batchPeriod metrics.Sample     // histogram of the elapsed time between non-zero batch reads
+}
+
+// newInputMetrics returns an input metric for windows event logs. If id is empty
+// a nil inputMetric is returned.
+func newInputMetrics(name, id string) *inputMetrics {
+	if id == "" {
+		return nil
+	}
+	reg, unreg := inputmon.NewInputRegistry(name, id, nil)
+	out := &inputMetrics{
+		unregister:  unreg,
+		name:        monitoring.NewString(reg, "provider"),
+		events:      monitoring.NewUint(reg, "received_events_total"),
+		dropped:     monitoring.NewUint(reg, "discarded_events_total"),
+		errors:      monitoring.NewUint(reg, "errors_total"),
+		batchSize:   metrics.NewUniformSample(1024),
+		sourceLag:   metrics.NewUniformSample(1024),
+		batchPeriod: metrics.NewUniformSample(1024),
+	}
+	out.name.Set(name)
+	_ = adapter.NewGoMetrics(reg, "received_events_count", adapter.Accept).
+		Register("histogram", metrics.NewHistogram(out.batchSize))
+	_ = adapter.NewGoMetrics(reg, "source_lag_time", adapter.Accept).
+		Register("histogram", metrics.NewHistogram(out.sourceLag))
+	_ = adapter.NewGoMetrics(reg, "batch_read_period", adapter.Accept).
+		Register("histogram", metrics.NewHistogram(out.batchPeriod))
+
+	return out
+}
+
+// log logs metric for the given batch.
+func (m *inputMetrics) log(batch []Record) {
+	if m == nil {
+		return
+	}
+	if len(batch) == 0 {
+		return
+	}
+
+	now := time.Now()
+	if !m.lastBatch.IsZero() {
+		m.batchPeriod.Update(now.Sub(m.lastBatch).Nanoseconds())
+	}
+	m.lastBatch = now
+
+	m.events.Add(uint64(len(batch)))
+	m.batchSize.Update(int64(len(batch)))
+	for _, r := range batch {
+		m.sourceLag.Update(now.Sub(r.TimeCreated.SystemTime).Nanoseconds())
+	}
+}
+
+// logError logs error metrics. Nil errors do not increment the error
+// count but the err value is currently otherwise not used. It is included
+// to allow easier extension of the metrics to include error stratification.
+func (m *inputMetrics) logError(err error) {
+	if m == nil {
+		return
+	}
+	if err == nil {
+		return
+	}
+	m.errors.Inc()
+}
+
+// logDropped logs dropped event metrics. Nil errors *do* increment the dropped
+// count; the value is currently otherwise not used, but is included to allow
+// easier extension of the metrics to include error stratification.
+func (m *inputMetrics) logDropped(_ error) {
+	if m == nil {
+		return
+	}
+	m.dropped.Inc()
+}
+
+func (m *inputMetrics) close() {
+	if m == nil {
+		return
+	}
+	m.unregister()
+}