{,x-pack/}filebeat: remove dead code an unused fields, vars and consts

This removes the majority of unused identified by running:

staticcheck -checks 'U*' ./{,x-pack/}filebeat/...

Care has been taken to not remove platform-specific code and mage-required code.

filebeat/beater/signalwait.go

@@ -18,7 +18,6 @@
 package beater
 
 import (
-	"sync"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/logp"
@@ -71,10 +70,6 @@ func (s *signalWait) Signal() {
 	s.Add(func() {})
 }
 
-func waitGroup(wg *sync.WaitGroup) signaler {
-	return wg.Wait
-}
-
 func waitChannel(c <-chan struct{}) signaler {
 	return func() { <-c }
 }

filebeat/generator/fields/fields.go

@@ -30,9 +30,6 @@ import (
 )
 
 const (
-	pipelinePath  = "%s/module/%s/%s/ingest/pipeline.json"
-	fieldsYmlPath = "%s/module/%s/%s/_meta/fields.yml"
-
 	typeIdx     = 0
 	elementsIdx = 1
 	hintIdx     = 2

filebeat/input/filestream/copytruncate_prospector.go

@@ -167,12 +167,6 @@ func (r rotatedFilestreams) isOriginalAdded(path string) bool {
 	return ok
 }
 
-// originalSrc returns the original Source information of a given
-// original file path.
-func (r rotatedFilestreams) originalSrc(path string) loginp.Source {
-	return r.table[path].originalSrc
-}
-
 // addRotatedFile adds a new rotated file to the list and returns its index.
 // if a file is already added, the source is updated and the index is returned.
 func (r rotatedFilestreams) addRotatedFile(original, rotated string, src loginp.Source) int {
@@ -195,17 +189,6 @@ func (r rotatedFilestreams) addRotatedFile(original, rotated string, src loginp.
 	return -1
 }
 
-// addRotatedFile adds a new rotated file to the list and returns its index.
-// if a file is already added, the source is updated and the index is returned.
-func (r rotatedFilestreams) removeRotatedFile(original, rotated string) {
-	for idx, fi := range r.table[original].rotated {
-		if fi.path == rotated {
-			r.table[original].rotated = append(r.table[original].rotated[:idx], r.table[original].rotated[idx+1:]...)
-			return
-		}
-	}
-}
-
 type copyTruncateFileProspector struct {
 	fileProspector
 	rotatedSuffix *regexp.Regexp

filebeat/input/filestream/environment_test.go

@@ -15,6 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build integration
+// +build integration
+
 package filestream
 
 import (

filebeat/input/filestream/fswatch_test.go

@@ -241,11 +241,3 @@ func (t testFileInfo) Mode() os.FileMode  { return 0 }
 func (t testFileInfo) ModTime() time.Time { return t.time }
 func (t testFileInfo) IsDir() bool        { return false }
 func (t testFileInfo) Sys() interface{}   { return t.sys }
-
-func mustDuration(durStr string) time.Duration {
-	dur, err := time.ParseDuration(durStr)
-	if err != nil {
-		panic(err)
-	}
-	return dur
-}

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

@@ -81,10 +81,7 @@ type Source interface {
 	Name() string
 }
 
-var (
-	errNoSourceConfigured = errors.New("no source has been configured")
-	errNoInputRunner      = errors.New("no input runner available")
-)
+var errNoInputRunner = errors.New("no input runner available")
 
 const globalInputID = ".global"
 

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

@@ -54,7 +54,6 @@ type updateOp struct {
 
 	// state updates to persist
 	timestamp time.Time
-	ttl       time.Duration
 	delta     interface{}
 }
 

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

@@ -554,15 +554,3 @@ func checkEqualStoreState(t *testing.T, want, got map[string]state) bool {
 	}
 	return true
 }
-
-// requireEqualStoreState compares 2 store snapshot tables for equality. The test
-// fails with Fatalf if the state differ.
-//
-// Note: testify is too strict when comparing timestamp, better use checkEqualStoreState.
-func requireEqualStoreState(t *testing.T, want, got map[string]state) bool {
-	if d := cmp.Diff(want, got); d != "" {
-		t.Fatalf("store state mismatch (-want +got):\n%s", d)
-		return false
-	}
-	return true
-}

filebeat/input/filestream/prospector_test.go

@@ -505,7 +505,6 @@ func (u *mockUnpackValue) UnpackCursorMeta(to interface{}) error {
 type mockProspectorCleaner struct {
 	available   map[string]loginp.Value
 	cleanedKeys []string
-	newEntries  map[string]fileMeta
 	updatedKeys map[string]string
 }
 

filebeat/input/journald/pkg/journalread/mode.go

@@ -17,10 +17,7 @@
 
 package journalread
 
-import (
-	"errors"
-	"fmt"
-)
+import "fmt"
 
 // SeekMode is used by (*Reader).Seek to decide where to advance the read pointer to.
 type SeekMode uint
@@ -42,8 +39,6 @@ var seekModes = map[string]SeekMode{
 	"cursor": SeekCursor,
 }
 
-var errInvalidSeekFallback = errors.New("invalid setting for cursor_seek_fallback")
-
 // Unpack validates and unpack "seek" config options. It returns an error if
 // the string is no valid seek mode.
 func (m *SeekMode) Unpack(value string) error {

filebeat/input/kafka/input.go

@@ -279,7 +279,6 @@ type groupHandler struct {
 	// ex. in this case are the azure fielsets where the events are found under the json object "records"
 	expandEventListFromField string // TODO
 	log                      *logp.Logger
-	reader                   reader.Reader
 }
 
 func (h *groupHandler) Setup(session sarama.ConsumerGroupSession) error {

filebeat/input/mqtt/client_mocked.go

@@ -35,7 +35,6 @@ type mockedMessage struct {
 	retained  bool
 	topic     string
 	payload   []byte
-	ack       func()
 }
 
 var _ libmqtt.Message = new(mockedMessage)

filebeat/input/v2/error.go

@@ -58,15 +58,6 @@ var ErrPluginWithoutName = errors.New("the plugin has no name")
 // error because there is no existing plugin that can create the input.
 func IsUnknownInputError(err error) bool { return errors.Is(err, ErrUnknownInput) }
 
-func failedInputName(err error) string {
-	switch e := err.(type) {
-	case *LoadError:
-		return e.Name
-	default:
-		return ""
-	}
-}
-
 // Unwrap returns the reason if present
 func (e *LoadError) Unwrap() error { return e.Reason }
 

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

@@ -55,7 +55,6 @@ type updateOp struct {
 
 	// state updates to persist
 	timestamp time.Time
-	ttl       time.Duration
 	delta     interface{}
 }
 

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

@@ -336,15 +336,3 @@ func checkEqualStoreState(t *testing.T, want, got map[string]state) bool {
 	}
 	return true
 }
-
-// requireEqualStoreState compares 2 store snapshot tables for equality. The test
-// fails with Fatalf if the state differ.
-//
-// Note: testify is too strict when comparing timestamp, better use checkEqualStoreState.
-func requireEqualStoreState(t *testing.T, want, got map[string]state) bool {
-	if d := cmp.Diff(want, got); d != "" {
-		t.Fatalf("store state mismatch (-want +got):\n%s", d)
-		return false
-	}
-	return true
-}

filebeat/input/v2/internal/inputest/inputest.go

@@ -19,7 +19,6 @@ package inputest
 
 import (
 	"errors"
-	"testing"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -47,13 +46,6 @@ type MockInput struct {
 	OnRun  func(v2.Context, beat.PipelineConnector) error
 }
 
-func makeConfigFakeInput(prototype MockInput) func(*common.Config) (v2.Input, error) {
-	return func(cfg *common.Config) (v2.Input, error) {
-		tmp := prototype
-		return &tmp, nil
-	}
-}
-
 // Init returns nil if OnInit is not set. Otherwise the return value of OnInit is returned.
 func (m *MockInputManager) Init(_ unison.Group, mode v2.Mode) error {
 	if m.OnInit != nil {
@@ -112,15 +104,3 @@ func SinglePlugin(name string, manager v2.InputManager) []v2.Plugin {
 		Manager:   manager,
 	}}
 }
-
-func expectError(t *testing.T, err error) {
-	if err == nil {
-		t.Errorf("expected error")
-	}
-}
-
-func expectNoError(t *testing.T, err error) {
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-}

filebeat/registrar/migrate.go

@@ -249,21 +249,6 @@ func (m *Migrator) updateToVersion1(regHome string) error {
 	return nil
 }
 
-func writeMeta(path string, version string, perm os.FileMode) error {
-	logp.Info("Write registry meta file with version: %v", version)
-	doc := struct{ Version string }{version}
-	body, err := json.Marshal(doc)
-	if err != nil {
-		panic(err) // must not fail
-	}
-
-	if err = safeWriteFile(path+".tmp", body, perm); err != nil {
-		return errors.Wrap(err, "failed writing registry meta.json")
-	}
-
-	return helper.SafeFileRotate(path, path+".tmp")
-}
-
 func readVersion(regHome, migrateFile string) (registryVersion, error) {
 	if isFile(migrateFile) {
 		return legacyVersion, nil

x-pack/filebeat/input/awscloudwatch/cloudwatch.go

@@ -28,7 +28,6 @@ type cloudwatchPoller struct {
 	logStreamPrefix      string
 	startTime            int64
 	endTime              int64
-	prevEndTime          int64
 	workerSem            *awscommon.Sem
 	log                  *logp.Logger
 	metrics              *inputMetrics

x-pack/filebeat/input/awscloudwatch/input_test.go

@@ -126,8 +126,6 @@ type MockCloudwatchlogsClient struct {
 	cloudwatchlogsiface.ClientAPI
 }
 
-var mockSvc = &MockCloudwatchlogsClient{}
-
 func (m *MockCloudwatchlogsClient) FilterLogEventsRequest(input *cloudwatchlogs.FilterLogEventsInput) cloudwatchlogs.FilterLogEventsRequest {
 	events := []cloudwatchlogs.FilteredLogEvent{
 		{

x-pack/filebeat/input/azureeventhub/input.go

@@ -44,7 +44,6 @@ type azureInput struct {
 	workerWg     sync.WaitGroup          // waits on worker goroutine.
 	processor    *eph.EventProcessorHost // eph will be assigned if users have enabled the option
 	hub          *eventhub.Hub           // hub will be assigned
-	ackChannel   chan int
 }
 
 const (

x-pack/filebeat/input/netflow/convert.go

@@ -23,21 +23,6 @@ import (
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/record"
 )
 
-var (
-	// RFC 1918
-	privateIPv4 = []net.IPNet{
-		{IP: net.IPv4(10, 0, 0, 0), Mask: net.IPv4Mask(255, 0, 0, 0)},
-		{IP: net.IPv4(172, 16, 0, 0), Mask: net.IPv4Mask(255, 240, 0, 0)},
-		{IP: net.IPv4(192, 168, 0, 0), Mask: net.IPv4Mask(255, 255, 0, 0)},
-	}
-
-	// RFC 4193
-	privateIPv6 = net.IPNet{
-		IP:   net.IP{0xfd, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
-		Mask: net.IPMask{0xff, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
-	}
-)
-
 func toBeatEvent(flow record.Record, internalNetworks []string) (event beat.Event) {
 	switch flow.Type {
 	case record.Flow:

x-pack/filebeat/input/netflow/decoder/decoder.go

@@ -27,7 +27,6 @@ type Decoder struct {
 	protos  map[uint16]protocol.Protocol
 	started bool
 	logger  log.Logger
-	config  config.Config
 }
 
 // NewDecoder returns a new NetFlow decoder configured using the passed