Rewrite Registrar to have a configurable registryFile

Makefile

@@ -2,7 +2,7 @@ ARCH?=$(shell uname -m)
 GODEP=$(GOPATH)/bin/godep
 
 
-filebeat:
+filebeat: $(GOFILES)
 	# first make sure we have godep
 	go get github.com/tools/godep
 	$(GODEP) go build

beat/filebeat.go

@@ -80,10 +80,21 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	fb.SpoolChan = make(chan *FileEvent, 16)
 	fb.publisherChan = make(chan []*FileEvent, 1)
 	fb.RegistrarChan = make(chan []*FileEvent, 1)
+
 	persist := make(map[string]*FileState)
 
-	restart := &ProspectorResume{}
-	restart.LoadState()
+	registrar := &Registrar{
+		registryFile: fb.FbConfig.Filebeat.RegistryFile,
+	}
+	registrar.Init()
+
+	restart := &ProspectorResume{
+		Persist: make(chan *FileState),
+		// Load the previous log file locations now, for use in prospector
+		Files: make(map[string]*FileState),
+	}
+
+	registrar.LoadState(restart.Files)
 	restart.Scan(fb.FbConfig.Filebeat.Files, persist, fb.SpoolChan)
 
 	// Start spooler: Harvesters dump events into the spooler.
@@ -93,7 +104,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	go Publish(b, fb)
 
 	// registrar records last acknowledged positions in all files.
-	Registrar(persist, fb.RegistrarChan)
+	registrar.WriteState(persist, fb.RegistrarChan)
 
 	return nil
 }

beat/registrar.go

@@ -8,7 +8,38 @@ import (
 	"github.com/elastic/libbeat/logp"
 )
 
-func Registrar(state map[string]*FileState, input chan []*FileEvent) {
+type Registrar struct {
+	registryFile string
+}
+
+func (r *Registrar) Init() {
+	// Set to default in case it is not set
+	if r.registryFile == "" {
+		r.registryFile = ".filebeat"
+	}
+
+	logp.Debug("registrar", "Registry file set to: %s", r.registryFile)
+
+}
+
+// loadState fetches the previous reading state from the configure registryFile file
+// The default file is .filebeat file which is stored in the same path as the binary is running
+func (r *Registrar) LoadState(files map[string]*FileState) {
+
+	if existing, e := os.Open(r.registryFile); e == nil {
+		defer existing.Close()
+		wd := ""
+		if wd, e = os.Getwd(); e != nil {
+			logp.Warn("WARNING: os.Getwd retuned unexpected error %s -- ignoring", e.Error())
+		}
+		logp.Info("Loading registrar data from %s/%s", wd, r.registryFile)
+
+		decoder := json.NewDecoder(existing)
+		decoder.Decode(&files)
+	}
+}
+
+func (r *Registrar) WriteState(state map[string]*FileState, input chan []*FileEvent) {
 	logp.Debug("registrar", "Starting Registrar")
 	for events := range input {
 		logp.Debug("registrar", "Registrar: processing %d events", len(events))
@@ -22,7 +53,7 @@ func Registrar(state map[string]*FileState, input chan []*FileEvent) {
 			state[*event.Source] = event.GetState()
 		}
 
-		if e := writeRegistry(state, ".filebeat"); e != nil {
+		if e := r.writeRegistry(state); e != nil {
 			// REVU: but we should panic, or something, right?
 			logp.Warn("WARNING: (continuing) update of registry returned error: %s", e)
 		}
@@ -31,8 +62,10 @@ func Registrar(state map[string]*FileState, input chan []*FileEvent) {
 }
 
 // writeRegistry Writes the new json registry file  to disk
-func writeRegistry(state map[string]*FileState, path string) error {
-	tempfile := path + ".new"
+func (r *Registrar) writeRegistry(state map[string]*FileState) error {
+	logp.Debug("registrar", "Write registry file:", r.registryFile)
+
+	tempfile := r.registryFile + ".new"
 	file, e := os.Create(tempfile)
 	if e != nil {
 		logp.Err("Failed to create tempfile (%s) for writing: %s", tempfile, e)
@@ -43,5 +76,5 @@ func writeRegistry(state map[string]*FileState, path string) error {
 	encoder := json.NewEncoder(file)
 	encoder.Encode(state)
 
-	return SafeFileRotate(path, tempfile)
+	return SafeFileRotate(r.registryFile, tempfile)
 }

config/config.go

@@ -21,6 +21,7 @@ type FilebeatConfig struct {
 	IdleTimeout         time.Duration
 	TailOnRotate        bool
 	Quiet               bool
+	RegistryFile        string
 }
 
 type FileConfig struct {
@@ -104,7 +105,7 @@ func mergeConfigFiles(configFiles []string, config *Config) error {
 }
 
 // Fetches and merges all config files given by Options.configArgs. All are put into one config object
-func (config *Config) FetchConfigs(path string)  {
+func (config *Config) FetchConfigs(path string) {
 
 	configFiles, err := getConfigFiles(path)
 

crawler/prospector.go

@@ -1,7 +1,6 @@
 package crawler
 
 import (
-	"encoding/json"
 	"os"
 	"path/filepath"
 	"time"
@@ -30,28 +29,6 @@ type ProspectorInfo struct {
 	Last_seen uint32      /* int number of the last iterations in which we saw this file */
 }
 
-// loadState fetches the previes reading state from the .filebeat file
-// The .filebeat file is stored in the same path as the binary is running
-func (restart *ProspectorResume) LoadState() {
-	restart.Persist = make(chan *FileState)
-
-	// Load the previous log file locations now, for use in prospector
-	restart.Files = make(map[string]*FileState)
-
-	// TODO: Should the location and path of .filebeat be configurable?
-	if existing, e := os.Open(".filebeat"); e == nil {
-		defer existing.Close()
-		wd := ""
-		if wd, e = os.Getwd(); e != nil {
-			logp.Warn("WARNING: os.Getwd retuned unexpected error %s -- ignoring", e.Error())
-		}
-		logp.Info("Loading registrar data from %s/.filebeat", wd)
-
-		decoder := json.NewDecoder(existing)
-		decoder.Decode(&restart.Files)
-	}
-}
-
 func (restart *ProspectorResume) Scan(files []cfg.FileConfig, persist map[string]*FileState, eventChan chan *FileEvent) {
 	pendingProspectorCnt := 0
 

etc/filebeat.dev.yml

@@ -7,7 +7,7 @@ filebeat:
       # Paths that should be crawled and fetched
       paths:
         #- /var/log/app*.log
-        - /var/log/*.log
+        - /var/log/*
       # Type of the files. Annotated in every documented
       type: syslog
       # Optional additional fields
@@ -27,12 +27,13 @@ filebeat:
       type: stdin
       paths:
         - "-"
-  spoolSize:
-  harvesterBufferSize:
-  cpuProfileFile:
-  idleTimeout:
-  tailOnRotate:
+  spoolsize:
+  harvesterbuffersize:
+  cpuprofilefile:
+  idletimeout:
+  tailonRotate:
   quiet:
+  registryfile: .filebeat
 
 ############################# Shipper ############################################
 shipper:

etc/filebeat.yml

@@ -33,6 +33,8 @@ filebeat:
   idleTimeout:
   tailOnRotate:
   quiet:
+  registryfile: .filebeat
+
 
 ############################# Shipper ############################################
 shipper:

main_test.go

@@ -3,8 +3,8 @@ package main
 // This file is mandatory as otherwise the packetbeat.test binary is not generated correctly. Reason???
 
 import (
-	"testing"
 	"flag"
+	"testing"
 )
 
 var integration *bool