Config management improvements in filebeat

* Code cleanup
* Add humanize for buffersize

filebeat/beater/filebeat.go

@@ -22,7 +22,10 @@ type Filebeat struct {
 
 // New creates a new Filebeat pointer instance.
 func New() *Filebeat {
-	return &Filebeat{}
+	config := cfg.DefaultConfig
+	return &Filebeat{
+		config: &config,
+	}
 }
 
 // Config setups up the filebeat configuration by fetch all additional config files

filebeat/config/config.go

@@ -14,10 +14,17 @@ import (
 
 // Defaults for config variables which are not set
 const (
-	DefaultRegistryFile string        = "registry"
-	DefaultSpoolSize    uint64        = 2048
-	DefaultIdleTimeout  time.Duration = 5 * time.Second
-	DefaultInputType                  = "log"
+	DefaultInputType = "log"
+)
+
+var (
+	DefaultConfig = Config{
+		FilebeatConfig{
+			RegistryFile: "registry",
+			SpoolSize:    2048,
+			IdleTimeout:  5 * time.Second,
+		},
+	}
 )
 
 type Config struct {
@@ -26,9 +33,9 @@ type Config struct {
 
 type FilebeatConfig struct {
 	Prospectors  []*common.Config `config:"prospectors"`
-	SpoolSize    uint64           `config:"spool_size"`
+	SpoolSize    uint64           `config:"spool_size" validate:"min=1"`
 	PublishAsync bool             `config:"publish_async"`
-	IdleTimeout  time.Duration    `config:"idle_timeout"`
+	IdleTimeout  time.Duration    `config:"idle_timeout" validate:"nonzero,min=0s"`
 	RegistryFile string           `config:"registry_file"`
 	ConfigDir    string           `config:"config_dir"`
 }

filebeat/harvester/config.go

@@ -8,11 +8,13 @@ import (
 	cfg "github.com/elastic/beats/filebeat/config"
 	"github.com/elastic/beats/filebeat/harvester/processor"
 	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/dustin/go-humanize"
 )
 
 var (
 	defaultConfig = harvesterConfig{
-		BufferSize:      16 << 10, // 16384
+		BufferSize:      16 * humanize.KiByte,
 		DocumentType:    "log",
 		InputType:       cfg.DefaultInputType,
 		TailFiles:       false,

filebeat/harvester/log.go

@@ -162,7 +162,7 @@ func (h *Harvester) openFile() (encoding.Encoding, error) {
 	f, err := file.ReadOpen(h.Path)
 	if err == nil {
 		// Check we are not following a rabbit hole (symlinks, etc.)
-		if !file.IsRegularFile(f) {
+		if !file.IsRegular(f) {
 			return nil, errors.New("Given file is not a regular file.")
 		}
 

filebeat/input/file/file.go

@@ -14,7 +14,7 @@ type File struct {
 }
 
 // Check that the file isn't a symlink, mode is regular or file is nil
-func (f *File) IsRegularFile() bool {
+func (f *File) IsRegular() bool {
 	if f.File == nil {
 		logp.Critical("Harvester: BUG: f arg is nil")
 		return false
@@ -50,7 +50,7 @@ func IsSameFile(path string, info os.FileInfo) bool {
 	return os.SameFile(fileInfo, info)
 }
 
-func IsRegularFile(file *os.File) bool {
+func IsRegular(file *os.File) bool {
 	f := &File{File: file}
-	return f.IsRegularFile()
+	return f.IsRegular()
 }

filebeat/registrar/registrar.go

@@ -41,11 +41,6 @@ func New(registryFile string) (*Registrar, error) {
 // Init sets up the Registrar and make sure the registry file is setup correctly
 func (r *Registrar) Init() error {
 
-	// Set to default in case it is not set
-	if r.registryFile == "" {
-		r.registryFile = cfg.DefaultRegistryFile
-	}
-
 	// The registry file is opened in the data path
 	r.registryFile = paths.Resolve(paths.Data, r.registryFile)
 

filebeat/spooler/spooler.go

@@ -36,23 +36,13 @@ func New(
 	publisher chan<- []*input.FileEvent,
 ) (*Spooler, error) {
 	spoolSize := config.SpoolSize
-	if spoolSize <= 0 {
-		spoolSize = cfg.DefaultSpoolSize
-		debugf("Spooler will use the default spool_size of %d", spoolSize)
-	}
-
-	idleTimeout := config.IdleTimeout
-	if idleTimeout <= 0 {
-		idleTimeout = cfg.DefaultIdleTimeout
-		debugf("Spooler will use the default idle_timeout of %s", idleTimeout)
-	}
 
 	return &Spooler{
 		Channel:       make(chan *input.FileEvent, channelSize),
-		idleTimeout:   idleTimeout,
+		idleTimeout:   config.IdleTimeout,
 		spoolSize:     spoolSize,
 		exit:          make(chan struct{}),
-		nextFlushTime: time.Now().Add(idleTimeout),
+		nextFlushTime: time.Now().Add(config.IdleTimeout),
 		publisher:     publisher,
 		spool:         make([]*input.FileEvent, 0, spoolSize),
 	}, nil

filebeat/spooler/spooler_test.go

@@ -17,13 +17,13 @@ func load(t *testing.T, in string) cfg.FilebeatConfig {
 		t.Fatalf("Failed to parse config input: %v", err)
 	}
 
-	var config cfg.FilebeatConfig
+	config := cfg.DefaultConfig
 	err = yaml.Unpack(&config)
 	if err != nil {
 		t.Fatalf("Failed to unpack config: %v", err)
 	}
 
-	return config
+	return config.Filebeat
 }
 
 func TestNewSpoolerDefaultConfig(t *testing.T) {
@@ -33,8 +33,8 @@ func TestNewSpoolerDefaultConfig(t *testing.T) {
 	spooler, err := New(config, nil)
 
 	assert.NoError(t, err)
-	assert.Equal(t, cfg.DefaultSpoolSize, spooler.spoolSize)
-	assert.Equal(t, cfg.DefaultIdleTimeout, spooler.idleTimeout)
+	assert.Equal(t, cfg.DefaultConfig.Filebeat.SpoolSize, spooler.spoolSize)
+	assert.Equal(t, cfg.DefaultConfig.Filebeat.IdleTimeout, spooler.idleTimeout)
 }
 
 func TestNewSpoolerSpoolSize(t *testing.T) {
@@ -47,7 +47,7 @@ func TestNewSpoolerSpoolSize(t *testing.T) {
 }
 
 func TestNewSpoolerIdleTimeout(t *testing.T) {
-	config := load(t, "idle_timeout: 10s")
+	config := load(t, "filebeat.idle_timeout: 10s")
 	spooler, err := New(config, nil)
 
 	assert.NoError(t, err)