Initialize support for ignore_inactive in filestream input

filebeat/_meta/config/filebeat.inputs.reference.yml.tmpl

@@ -320,6 +320,11 @@ filebeat.inputs:
   # Time strings like 2h (2 hours), 5m (5 minutes) can be used.
   #ignore_older: 0
 
+  # Ignore files that have not been updated since the selected event.
+  # ignore_inactive is disabled by default, so no files are ignored by setting it to "".
+  # Available options: since_first_start, since_last_start.
+  #ignore_inactive: ""
+
   # Defines the buffer size every harvester uses when fetching the file
   #harvester_buffer_size: 16384
 

filebeat/filebeat.reference.yml

@@ -727,6 +727,11 @@ filebeat.inputs:
   # Time strings like 2h (2 hours), 5m (5 minutes) can be used.
   #ignore_older: 0
 
+  # Ignore files that have not been updated since the selected event.
+  # ignore_inactive is disabled by default, so no files are ignored by setting it to "".
+  # Available options: since_first_start, since_last_start.
+  #ignore_inactive: ""
+
   # Defines the buffer size every harvester uses when fetching the file
   #harvester_buffer_size: 16384
 

filebeat/input/default-inputs/inputs.go

@@ -28,14 +28,14 @@ import (
 
 func Init(info beat.Info, log *logp.Logger, components beater.StateStore) []v2.Plugin {
 	return append(
-		genericInputs(log, components),
+		genericInputs(info, log, components),
 		osInputs(info, log, components)...,
 	)
 }
 
-func genericInputs(log *logp.Logger, components beater.StateStore) []v2.Plugin {
+func genericInputs(info beat.Info, log *logp.Logger, components beater.StateStore) []v2.Plugin {
 	return []v2.Plugin{
-		filestream.Plugin(log, components),
+		filestream.Plugin(info, log, components),
 		unix.Plugin(),
 	}
 }

filebeat/input/filestream/config.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/dustin/go-humanize"
 
+	loginp "github.com/elastic/beats/v7/filebeat/input/filestream/internal/input-logfile"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/match"
 	"github.com/elastic/beats/v7/libbeat/reader/readfile"
@@ -32,14 +33,15 @@ import (
 type config struct {
 	Reader readerConfig `config:",inline"`
 
-	Paths          []string                `config:"paths"`
-	Close          closerConfig            `config:"close"`
-	FileWatcher    *common.ConfigNamespace `config:"prospector"`
-	FileIdentity   *common.ConfigNamespace `config:"file_identity"`
-	CleanInactive  time.Duration           `config:"clean_inactive" validate:"min=0"`
-	CleanRemoved   bool                    `config:"clean_removed"`
-	HarvesterLimit uint32                  `config:"harvester_limit" validate:"min=0"`
-	IgnoreOlder    time.Duration           `config:"ignore_older"`
+	Paths          []string                  `config:"paths"`
+	Close          closerConfig              `config:"close"`
+	FileWatcher    *common.ConfigNamespace   `config:"prospector"`
+	FileIdentity   *common.ConfigNamespace   `config:"file_identity"`
+	CleanInactive  time.Duration             `config:"clean_inactive" validate:"min=0"`
+	CleanRemoved   bool                      `config:"clean_removed"`
+	HarvesterLimit uint32                    `config:"harvester_limit" validate:"min=0"`
+	IgnoreOlder    time.Duration             `config:"ignore_older"`
+	IgnoreInactive loginp.IgnoreInactiveType `config:"ignore_inactive"`
 }
 
 type closerConfig struct {

filebeat/input/filestream/environment_test.go

@@ -81,7 +81,7 @@ func (e *inputTestingEnvironment) mustCreateInput(config map[string]interface{})
 
 func (e *inputTestingEnvironment) getManager() v2.InputManager {
 	e.pluginInitOnce.Do(func() {
-		e.plugin = Plugin(logp.L(), e.stateStore)
+		e.plugin = Plugin(beat.Info{FirstStart: time.Now(), StartTime: time.Now()}, logp.L(), e.stateStore)
 	})
 	return e.plugin.Manager
 }

filebeat/input/filestream/input.go

@@ -60,7 +60,7 @@ type filestream struct {
 }
 
 // Plugin creates a new filestream input plugin for creating a stateful input.
-func Plugin(log *logp.Logger, store loginp.StateStore) input.Plugin {
+func Plugin(info beat.Info, log *logp.Logger, store loginp.StateStore) input.Plugin {
 	return input.Plugin{
 		Name:       pluginName,
 		Stability:  feature.Experimental,
@@ -69,6 +69,8 @@ func Plugin(log *logp.Logger, store loginp.StateStore) input.Plugin {
 		Doc:        "The filestream input collects logs from the local filestream service",
 		Manager: &loginp.InputManager{
 			Logger:     log,
+			FirstStart: info.FirstStart,
+			StartTime:  info.StartTime,
 			StateStore: store,
 			Type:       pluginName,
 			Configure:  configure,

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

@@ -51,6 +51,11 @@ import (
 type InputManager struct {
 	Logger *logp.Logger
 
+	// FirstStart is the time when Filebeat was first started on a given host.
+	FirstStart time.Time
+	// StartTime is the last time Filebeat was started on a given host.
+	StartTime time.Time
+
 	// StateStore gives the InputManager access to the persitent key value store.
 	StateStore StateStore
 
@@ -155,9 +160,10 @@ func (cim *InputManager) Create(config *common.Config) (input.Input, error) {
 	}
 
 	settings := struct {
-		ID             string        `config:"id"`
-		CleanTimeout   time.Duration `config:"clean_timeout"`
-		HarvesterLimit uint64        `config:"harvester_limit"`
+		ID             string             `config:"id"`
+		CleanTimeout   time.Duration      `config:"clean_timeout"`
+		HarvesterLimit uint64             `config:"harvester_limit"`
+		IgnoreSince    IgnoreInactiveType `config:"ignore_since"`
 	}{ID: "", CleanTimeout: cim.DefaultCleanTimeout, HarvesterLimit: 0}
 	if err := config.Unpack(&settings); err != nil {
 		return nil, err
@@ -179,7 +185,7 @@ func (cim *InputManager) Create(config *common.Config) (input.Input, error) {
 	pStore := cim.getRetainedStore()
 	defer pStore.Release()
 	prospectorStore := newSourceStore(pStore, sourceIdentifier)
-	err = prospector.Init(prospectorStore)
+	err = prospector.Init(prospectorStore, cim.getIgnoreSince(settings.IgnoreSince))
 	if err != nil {
 		return nil, err
 	}
@@ -195,6 +201,17 @@ func (cim *InputManager) Create(config *common.Config) (input.Input, error) {
 	}, nil
 }
 
+func (cim *InputManager) getIgnoreSince(t IgnoreInactiveType) time.Time {
+	switch t {
+	case IgnoreInactiveSinceLastStart:
+		return cim.StartTime
+	case IgnoreInactiveSinceFirstStart:
+		return cim.FirstStart
+	default:
+		return time.Time{}
+	}
+}
+
 func (cim *InputManager) getRetainedStore() *store {
 	store := cim.store
 	store.Retain()

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

@@ -18,6 +18,9 @@
 package input_logfile
 
 import (
+	"fmt"
+	"time"
+
 	input "github.com/elastic/beats/v7/filebeat/input/v2"
 )
 
@@ -26,7 +29,7 @@ import (
 // It also updates the statestore with the meta data of the running harvesters.
 type Prospector interface {
 	// Init runs the cleanup processes before starting the prospector.
-	Init(c ProspectorCleaner) error
+	Init(c ProspectorCleaner, ignoreSince time.Time) error
 	// Run starts the event loop and handles the incoming events
 	// either by starting/stopping a harvester, or updating the statestore.
 	Run(input.Context, StateMetadataUpdater, HarvesterGroup)
@@ -63,3 +66,30 @@ type Value interface {
 	// UnpackCursorMeta returns the cursor metadata required by the prospector.
 	UnpackCursorMeta(to interface{}) error
 }
+
+type IgnoreInactiveType uint8
+
+const (
+	InvalidIgnoreInactive = iota
+	IgnoreInactiveSinceLastStart
+	IgnoreInactiveSinceFirstStart
+
+	ignoreInactiveSinceLastStartStr  = "since_last_start"
+	ignoreInactiveSinceFirstStartStr = "since_first_start"
+)
+
+var (
+	ignoreInactiveSettings = map[string]IgnoreInactiveType{
+		ignoreInactiveSinceLastStartStr:  IgnoreInactiveSinceLastStart,
+		ignoreInactiveSinceFirstStartStr: IgnoreInactiveSinceFirstStart,
+	}
+)
+
+func (t *IgnoreInactiveType) Unpack(v string) error {
+	val, ok := ignoreInactiveSettings[v]
+	if !ok {
+		return fmt.Errorf("invalid ignore_inactive setting: %s", v)
+	}
+	*t = val
+	return nil
+}

filebeat/input/filestream/prospector.go

@@ -37,14 +37,15 @@ const (
 // The FS events then trigger either new Harvester runs or updates
 // the statestore.
 type fileProspector struct {
-	filewatcher       loginp.FSWatcher
-	identifier        fileIdentifier
-	ignoreOlder       time.Duration
-	cleanRemoved      bool
-	stateChangeCloser stateChangeCloserConfig
+	filewatcher         loginp.FSWatcher
+	identifier          fileIdentifier
+	ignoreOlder         time.Duration
+	ignoreInactiveSince time.Time
+	cleanRemoved        bool
+	stateChangeCloser   stateChangeCloserConfig
 }
 
-func (p *fileProspector) Init(cleaner loginp.ProspectorCleaner) error {
+func (p *fileProspector) Init(cleaner loginp.ProspectorCleaner, ignoreSince time.Time) error {
 	files := p.filewatcher.GetFiles()
 
 	if p.cleanRemoved {
@@ -82,6 +83,8 @@ func (p *fileProspector) Init(cleaner loginp.ProspectorCleaner) error {
 		return "", fm
 	})
 
+	p.ignoreInactiveSince = ignoreSince
+
 	return nil
 }
 
@@ -130,6 +133,12 @@ func (p *fileProspector) Run(ctx input.Context, s loginp.StateMetadataUpdater, h
 						break
 					}
 				}
+
+				if !p.ignoreInactiveSince.IsZero() && fe.Info.ModTime().Sub(p.ignoreInactiveSince) <= 0 {
+					log.Debugf("Ignore file because ignore_since.* reached time %v. File %s", p.ignoreInactiveSince, fe.NewPath)
+					break
+				}
+
 				hg.Start(ctx, src)
 
 			case loginp.OpTruncate:

filebeat/input/filestream/prospector_test.go

@@ -85,7 +85,7 @@ func TestProspector_InitCleanIfRemoved(t *testing.T) {
 				cleanRemoved: testCase.cleanRemoved,
 				filewatcher:  &mockFileWatcher{filesOnDisk: testCase.filesOnDisk},
 			}
-			p.Init(testStore)
+			p.Init(testStore, time.Time{})
 
 			assert.ElementsMatch(t, testCase.expectedCleanedKeys, testStore.cleanedKeys)
 		})
@@ -152,7 +152,7 @@ func TestProspector_InitUpdateIdentifiers(t *testing.T) {
 				identifier:  mustPathIdentifier(false),
 				filewatcher: &mockFileWatcher{filesOnDisk: testCase.filesOnDisk},
 			}
-			p.Init(testStore)
+			p.Init(testStore, time.Time{})
 
 			assert.EqualValues(t, testCase.expectedUpdatedKeys, testStore.updatedKeys)
 		})

libbeat/beat/info.go

@@ -17,7 +17,11 @@
 
 package beat
 
-import "github.com/gofrs/uuid"
+import (
+	"time"
+
+	"github.com/gofrs/uuid"
+)
 
 // Info stores a beats instance meta data.
 type Info struct {
@@ -29,6 +33,8 @@ type Info struct {
 	Hostname        string    // hostname
 	ID              uuid.UUID // ID assigned to beat machine
 	EphemeralID     uuid.UUID // ID assigned to beat process invocation (PID)
+	FirstStart      time.Time // The time of the first start of the Beat.
+	StartTime       time.Time // The time of last start of the Beat. Updated when the Beat is started or restarted.
 
 	// Monitoring-related fields
 	Monitoring struct {

libbeat/cmd/instance/beat.go

@@ -239,6 +239,8 @@ func NewBeat(name, indexPrefix, v string, elasticLicensed bool) (*Beat, error) {
 			Name:            hostname,
 			Hostname:        hostname,
 			ID:              id,
+			FirstStart:      time.Now(),
+			StartTime:       time.Now(),
 			EphemeralID:     metrics.EphemeralID(),
 		},
 		Fields: fields,
@@ -695,7 +697,8 @@ func (b *Beat) configure(settings Settings) error {
 
 func (b *Beat) loadMeta(metaPath string) error {
 	type meta struct {
-		UUID uuid.UUID `json:"uuid"`
+		UUID       uuid.UUID `json:"uuid"`
+		FirstStart time.Time `json:"first_start"`
 	}
 
 	logp.Debug("beat", "Beat metadata path: %v", metaPath)
@@ -713,14 +716,21 @@ func (b *Beat) loadMeta(metaPath string) error {
 		}
 
 		f.Close()
+
+		if !m.FirstStart.IsZero() {
+			b.Info.FirstStart = m.FirstStart
+		}
 		valid := m.UUID != uuid.Nil
 		if valid {
 			b.Info.ID = m.UUID
+		}
+
+		if valid && !m.FirstStart.IsZero() {
 			return nil
 		}
 	}
 
-	// file does not exist or ID is invalid, let's create a new one
+	// file does not exist or ID is invalid or first start time is not defined, let's create a new one
 
 	// write temporary file first
 	tempFile := metaPath + ".new"
@@ -729,7 +739,7 @@ func (b *Beat) loadMeta(metaPath string) error {
 		return fmt.Errorf("Failed to create Beat meta file: %s", err)
 	}
 
-	encodeErr := json.NewEncoder(f).Encode(meta{UUID: b.Info.ID})
+	encodeErr := json.NewEncoder(f).Encode(meta{UUID: b.Info.ID, FirstStart: b.Info.FirstStart})
 	err = f.Sync()
 	if err != nil {
 		return fmt.Errorf("Beat meta file failed to write: %s", err)

x-pack/filebeat/filebeat.reference.yml

@@ -2578,6 +2578,11 @@ filebeat.inputs:
   # Time strings like 2h (2 hours), 5m (5 minutes) can be used.
   #ignore_older: 0
 
+  # Ignore files that have not been updated since the selected event.
+  # ignore_inactive is disabled by default, so no files are ignored by setting it to "".
+  # Available options: since_first_start, since_last_start.
+  #ignore_inactive: ""
+
   # Defines the buffer size every harvester uses when fetching the file
   #harvester_buffer_size: 16384