Add better error checking for backends

since the nfdump backend requires the nfdump command, check to see if it
is available first.

backend/backend.go

@@ -11,6 +11,7 @@ const maxLineLength = 20 * 1024 * 1024
 type Backend interface {
 	ExtractIps(reader io.Reader, ips *ipset.Set) (uint64, error)
 	Filter(reader io.Reader, query string, writer io.Writer) error
+	Check() error
 }
 
 var backends = map[string]Backend{}

backend/bro.go

@@ -56,6 +56,10 @@ func (b BroBackend) Filter(reader io.Reader, query string, writer io.Writer) err
 	return nil
 }
 
+func (b BroBackend) Check() error {
+	return nil
+}
+
 func init() {
 	RegisterBackend("bro", BroBackend{})
 }

backend/bro_json.go

@@ -78,6 +78,9 @@ func (b BroJSONBackend) Filter(reader io.Reader, query string, writer io.Writer)
 	}
 	return nil
 }
+func (b BroJSONBackend) Check() error {
+	return nil
+}
 
 func init() {
 	RegisterBackend("bro_json", BroJSONBackend{})

backend/nfdump.go

@@ -50,7 +50,19 @@ func (b NFDUMPCSVBackend) ExtractIps(reader io.Reader, ips *ipset.Set) (uint64,
 	return lines, err
 }
 func (b NFDUMPCSVBackend) Filter(reader io.Reader, query string, writer io.Writer) error {
-	return nil
+	filter := fmt.Sprintf("ip in [%s]", query)
+	cmd := exec.Command("nfdump", "-qr", "-", filter)
+	cmd.Stdin = reader
+	cmd.Stdout = writer
+
+	err := cmd.Run()
+	return err
+}
+
+func (b NFDUMPCSVBackend) Check() error {
+	cmd := exec.Command("nfdump", "-V")
+	_, err := cmd.CombinedOutput()
+	return err
 }
 
 type NFDUMPBackend struct {
@@ -95,6 +107,12 @@ func (b NFDUMPBackend) Filter(reader io.Reader, query string, writer io.Writer)
 	return err
 }
 
+func (b NFDUMPBackend) Check() error {
+	cmd := exec.Command("nfdump", "-V")
+	_, err := cmd.CombinedOutput()
+	return err
+}
+
 func init() {
 	RegisterBackend("nfdump-csv", NFDUMPCSVBackend{})
 	RegisterBackend("nfdump", NFDUMPBackend{})

backend/pcap.go

@@ -64,6 +64,10 @@ func (b PCAPBackend) Filter(reader io.Reader, query string, writer io.Writer) er
 	return err
 }
 
+func (b PCAPBackend) Check() error {
+	return nil
+}
+
 func init() {
 	RegisterBackend("pcap", PCAPBackend{})
 }

backend/syslog.go

@@ -77,6 +77,9 @@ func (b SyslogBackend) Filter(reader io.Reader, query string, writer io.Writer)
 	}
 	return nil
 }
+func (b SyslogBackend) Check() error {
+	return nil
+}
 
 func init() {
 	RegisterBackend("syslog", SyslogBackend{})

flowindexer/flowindexer.go

@@ -235,7 +235,9 @@ func (i *Indexer) IndexAll() error {
 		//Assume the last file in the list is the most recent one
 		//and check to see if it is still growing before indexing it
 		checkGrowing := idx == len(logs)-1
-		i.IndexOne(l, checkGrowing)
+		if err = i.IndexOne(l, checkGrowing); err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -252,7 +254,9 @@ func (i *Indexer) IndexRecent() error {
 		//Assume the last file in the list is the most recent one
 		//and check to see if it is still growing before indexing it
 		checkGrowing := idx == len(logs)-1
-		i.IndexOne(l, checkGrowing)
+		if err = i.IndexOne(l, checkGrowing); err != nil {
+			return err
+		}
 	}
 
 	return nil

flowindexer/index.go

@@ -1,6 +1,7 @@
 package flowindexer
 
 import (
+	"fmt"
 	"log"
 	"path/filepath"
 	"runtime"
@@ -20,6 +21,11 @@ func Index(s store.IpStore, b backend.Backend, filename string) error {
 		//log.Printf("%s Already indexed\n", filename)
 		return nil
 	}
+
+	if err = b.Check(); err != nil {
+		return fmt.Errorf("Backend is not usable: %v", err)
+	}
+
 	ips := ipset.New()
 	reader, err := backend.OpenDecompress(filename)
 	if err != nil {