Fix the custom BPF filter option (elastic#2671) (elastic#2677)

* Fix the custom BPF filter option

Set the `sniffer.filter` before `setFromConfig` is called. Changed the
factory prototype so it doesn't pass around `filter` and instead just
done it directly via Init.

Fixes elastic#2660.

* Simplified code by removing the factory maker

CHANGELOG.asciidoc

@@ -35,6 +35,8 @@ https://github.com/elastic/beats/compare/v5.0.0-beta1...master[Check the HEAD di
 
 *Packetbeat*
 
+- Fix the `bpf_filter` setting. {issue}2660[2660]
+
 *Topbeat*
 
 *Filebeat*

packetbeat/beater/packetbeat.go

@@ -198,52 +198,50 @@ func (pb *Packetbeat) setupSniffer() error {
 	}
 
 	pb.Sniff = &sniffer.SnifferSetup{}
-	return pb.Sniff.Init(false, pb.makeWorkerFactory(filter), &config.Interfaces)
+	return pb.Sniff.Init(false, filter, pb.createWorker, &config.Interfaces)
 }
 
-func (pb *Packetbeat) makeWorkerFactory(filter string) sniffer.WorkerFactory {
-	return func(dl layers.LinkType) (sniffer.Worker, string, error) {
-		var f *flows.Flows
-		var err error
-		config := &pb.Config
-
-		if config.Flows.IsEnabled() {
-			f, err = flows.NewFlows(pb.Pub, config.Flows)
-			if err != nil {
-				return nil, "", err
-			}
-		}
-
-		var icmp4 icmp.ICMPv4Processor
-		var icmp6 icmp.ICMPv6Processor
-		if cfg := config.Protocols["icmp"]; cfg.Enabled() {
-			icmp, err := icmp.New(false, pb.Pub, cfg)
-			if err != nil {
-				return nil, "", err
-			}
-
-			icmp4 = icmp
-			icmp6 = icmp
-		}
+func (pb *Packetbeat) createWorker(dl layers.LinkType) (sniffer.Worker, error) {
+	var f *flows.Flows
+	var err error
+	config := &pb.Config
 
-		tcp, err := tcp.NewTcp(&protos.Protos)
+	if config.Flows.IsEnabled() {
+		f, err = flows.NewFlows(pb.Pub, config.Flows)
 		if err != nil {
-			return nil, "", err
+			return nil, err
 		}
+	}
 
-		udp, err := udp.NewUdp(&protos.Protos)
+	var icmp4 icmp.ICMPv4Processor
+	var icmp6 icmp.ICMPv6Processor
+	if cfg := config.Protocols["icmp"]; cfg.Enabled() {
+		icmp, err := icmp.New(false, pb.Pub, cfg)
 		if err != nil {
-			return nil, "", err
+			return nil, err
 		}
 
-		worker, err := decoder.NewDecoder(f, dl, icmp4, icmp6, tcp, udp)
-		if err != nil {
-			return nil, "", err
-		}
+		icmp4 = icmp
+		icmp6 = icmp
+	}
 
-		if f != nil {
-			pb.services = append(pb.services, f)
-		}
-		return worker, filter, nil
+	tcp, err := tcp.NewTcp(&protos.Protos)
+	if err != nil {
+		return nil, err
+	}
+
+	udp, err := udp.NewUdp(&protos.Protos)
+	if err != nil {
+		return nil, err
+	}
+
+	worker, err := decoder.NewDecoder(f, dl, icmp4, icmp6, tcp, udp)
+	if err != nil {
+		return nil, err
+	}
+
+	if f != nil {
+		pb.services = append(pb.services, f)
 	}
+	return worker, nil
 }

packetbeat/sniffer/sniffer.go

@@ -37,7 +37,7 @@ type Worker interface {
 	OnPacket(data []byte, ci *gopacket.CaptureInfo)
 }
 
-type WorkerFactory func(layers.LinkType) (Worker, string, error)
+type WorkerFactory func(layers.LinkType) (Worker, error)
 
 // Computes the block_size and the num_blocks in such a way that the
 // allocated mmap buffer is close to but smaller than target_size_mb.
@@ -261,21 +261,22 @@ func (sniffer *SnifferSetup) Datalink() layers.LinkType {
 	return layers.LinkTypeEthernet
 }
 
-func (sniffer *SnifferSetup) Init(test_mode bool, factory WorkerFactory, interfaces *config.InterfacesConfig) error {
+func (sniffer *SnifferSetup) Init(test_mode bool, filter string, factory WorkerFactory, interfaces *config.InterfacesConfig) error {
 	var err error
 
 	if !test_mode {
+		sniffer.filter = filter
+		logp.Debug("sniffer", "BPF filter: '%s'", sniffer.filter)
 		err = sniffer.setFromConfig(interfaces)
 		if err != nil {
 			return fmt.Errorf("Error creating sniffer: %v", err)
 		}
 	}
 
-	sniffer.worker, sniffer.filter, err = factory(sniffer.Datalink())
+	sniffer.worker, err = factory(sniffer.Datalink())
 	if err != nil {
 		return fmt.Errorf("Error creating decoder: %v", err)
 	}
-	logp.Debug("sniffer", "BPF filter: '%s'", sniffer.filter)
 
 	if sniffer.config.Dumpfile != "" {
 		p, err := pcap.OpenDead(sniffer.Datalink(), 65535)