Add more metrics to Filebeat harvesters

CHANGELOG.next.asciidoc

@@ -465,6 +465,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add dashboards for the ActiveMQ Filebeat module. {pull}14880[14880]
 - Add STAN Metricbeat module. {pull}14839[14839]
 - Add new fileset googlecloud/audit for ingesting Google Cloud Audit logs. {pull}15200[15200]
+- Expose more metrics of harvesters (e.g. `read_offset`, `start_time`). {pull}13395[13395]
 
 *Heartbeat*
 - Add non-privileged icmp on linux and darwin(mac). {pull}13795[13795] {issue}11498[11498]

filebeat/input/log/harvester.go

@@ -58,6 +58,7 @@ import (
 
 var (
 	harvesterMetrics = monitoring.Default.NewRegistry("filebeat.harvester")
+	filesMetrics     = harvesterMetrics.NewRegistry("files")
 
 	harvesterStarted   = monitoring.NewInt(harvesterMetrics, "started")
 	harvesterClosed    = monitoring.NewInt(harvesterMetrics, "closed")
@@ -100,9 +101,22 @@ type Harvester struct {
 	outletFactory OutletFactory
 	publishState  func(file.State) bool
 
+	metrics *harvesterProgressMetrics
+
 	onTerminate func()
 }
 
+// stores the metrics of the harvester
+type harvesterProgressMetrics struct {
+	metricsRegistry             *monitoring.Registry
+	filename                    *monitoring.String
+	started                     *monitoring.String
+	lastPublished               *monitoring.Timestamp
+	lastPublishedEventTimestamp *monitoring.Timestamp
+	currentSize                 *monitoring.Int
+	readOffset                  *monitoring.Int
+}
+
 // NewHarvester creates a new harvester
 func NewHarvester(
 	config *common.Config,
@@ -179,11 +193,43 @@ func (h *Harvester) Setup() error {
 		return fmt.Errorf("Harvester setup failed. Unexpected encoding line reader error: %s", err)
 	}
 
+	h.metrics = newHarvesterProgressMetrics(h.id.String())
+	h.metrics.filename.Set(h.source.Name())
+	h.metrics.started.Set(common.Time(time.Now()).String())
+	h.metrics.readOffset.Set(h.state.Offset)
+	err = h.updateCurrentSize()
+	if err != nil {
+		return err
+	}
+
 	logp.Debug("harvester", "Harvester setup successful. Line terminator: %d", h.config.LineTerminator)
 
 	return nil
 }
 
+func newHarvesterProgressMetrics(id string) *harvesterProgressMetrics {
+	r := filesMetrics.NewRegistry(id)
+	return &harvesterProgressMetrics{
+		metricsRegistry:             r,
+		filename:                    monitoring.NewString(r, "name"),
+		started:                     monitoring.NewString(r, "start_time"),
+		lastPublished:               monitoring.NewTimestamp(r, "last_event_published_time"),
+		lastPublishedEventTimestamp: monitoring.NewTimestamp(r, "last_event_timestamp"),
+		currentSize:                 monitoring.NewInt(r, "size"),
+		readOffset:                  monitoring.NewInt(r, "read_offset"),
+	}
+}
+
+func (h *Harvester) updateCurrentSize() error {
+	fInfo, err := h.source.Stat()
+	if err != nil {
+		return err
+	}
+
+	h.metrics.currentSize.Set(fInfo.Size())
+	return nil
+}
+
 // Run start the harvester and reads files line by line and sends events to the defined output
 func (h *Harvester) Run() error {
 	// Allow for some cleanup on termination
@@ -250,6 +296,8 @@ func (h *Harvester) Run() error {
 
 	logp.Info("Harvester started for file: %s", h.state.Source)
 
+	go h.monitorFileSize()
+
 	for {
 		select {
 		case <-h.done:
@@ -298,13 +346,37 @@ func (h *Harvester) Run() error {
 
 		// Update state of harvester as successfully sent
 		h.state = state
+
+		// Update metics of harvester as event was sent
+		h.metrics.readOffset.Set(state.Offset)
+		h.metrics.lastPublished.Set(time.Now())
+		h.metrics.lastPublishedEventTimestamp.Set(message.Ts)
+	}
+}
+
+func (h *Harvester) monitorFileSize() {
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-h.done:
+			return
+		case <-ticker.C:
+			err := h.updateCurrentSize()
+			if err != nil {
+				logp.Err("Error updating file size: %v; File: %v", err, h.state.Source)
+			}
+		}
 	}
 }
 
 // stop is intended for internal use and closed the done channel to stop execution
 func (h *Harvester) stop() {
 	h.stopOnce.Do(func() {
 		close(h.done)
+
+		filesMetrics.Remove(h.id.String())
 	})
 }
 

libbeat/monitoring/metrics.go

@@ -23,7 +23,9 @@ import (
 	"math"
 	"strconv"
 	"sync"
+	"time"
 
+	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/atomic"
 )
 
@@ -232,3 +234,57 @@ func fullName(r *Registry, name string) string {
 	}
 	return r.name + "." + name
 }
+
+// Timestamp is a timestamp variable satisfying the Var interface.
+type Timestamp struct {
+	mu     sync.RWMutex
+	ts     time.Time
+	cached string
+}
+
+// NewTimestamp creates and registeres a new timestamp variable.
+func NewTimestamp(r *Registry, name string, opts ...Option) *Timestamp {
+	if r == nil {
+		r = Default
+	}
+
+	v := &Timestamp{}
+	addVar(r, name, opts, v, makeExpvar(func() string {
+		return v.toString()
+
+	}))
+	return v
+}
+
+func (v *Timestamp) Set(t time.Time) {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+
+	v.ts = t
+	v.cached = ""
+}
+
+func (v *Timestamp) Get() time.Time {
+	v.mu.RLock()
+	defer v.mu.RUnlock()
+
+	return v.ts
+}
+
+func (v *Timestamp) Visit(_ Mode, vs Visitor) {
+	vs.OnString(v.toString())
+}
+
+func (v *Timestamp) toString() string {
+	v.mu.RLock()
+	defer v.mu.RUnlock()
+
+	if v.ts.IsZero() {
+		return ""
+	}
+
+	if v.cached == "" {
+		v.cached = v.ts.Format(common.TsLayout)
+	}
+	return v.cached
+}