Packetbeat process monitor enhancements (#7135)

This patch extends the functionality of the processes monitor in
packetbeat:

- report the full command-line for all processes (not only those
  configured as `monitored`.
- Add support for Windows process monitoring.
- Disable the monitor when using file input.
- Get rid of refresh delays.

Closes #541

CHANGELOG.asciidoc

@@ -257,6 +257,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Fix high memory usage on HTTP body if body is not published. {pull}6680[6680]
 - Allow to capture the HTTP request or response bodies independently. {pull}6784[6784]
 - HTTP publishes an Error event for unmatched requests or responses. {pull}6794[6794]
+- The process monitor now reports the command-line for all processes, under Linux and Windows. {pull}7135[7135]
 
 *Winlogbeat*
 

libbeat/common/endpoint.go

@@ -8,3 +8,21 @@ type Endpoint struct {
 	Cmdline string
 	Proc    string
 }
+
+// MakeEndpointPair returns source and destination endpoints from a TCP or IP tuple
+// and a command-line tuple.
+func MakeEndpointPair(tuple BaseTuple, cmdlineTuple *CmdlineTuple) (src Endpoint, dst Endpoint) {
+	src = Endpoint{
+		IP:      tuple.SrcIP.String(),
+		Port:    tuple.SrcPort,
+		Proc:    string(cmdlineTuple.Src),
+		Cmdline: string(cmdlineTuple.SrcCommand),
+	}
+	dst = Endpoint{
+		IP:      tuple.DstIP.String(),
+		Port:    tuple.DstPort,
+		Proc:    string(cmdlineTuple.Dst),
+		Cmdline: string(cmdlineTuple.DstCommand),
+	}
+	return src, dst
+}

libbeat/common/tuples.go

@@ -15,10 +15,15 @@ const MaxIPPortTupleRawSize = 16 + 16 + 2 + 2
 
 type HashableIPPortTuple [MaxIPPortTupleRawSize]byte
 
-type IPPortTuple struct {
-	IPLength         int
+type BaseTuple struct {
 	SrcIP, DstIP     net.IP
 	SrcPort, DstPort uint16
+}
+
+type IPPortTuple struct {
+	BaseTuple
+
+	IPLength int
 
 	raw    HashableIPPortTuple // Src_ip:Src_port:Dst_ip:Dst_port
 	revRaw HashableIPPortTuple // Dst_ip:Dst_port:Src_ip:Src_port
@@ -29,10 +34,12 @@ func NewIPPortTuple(ipLength int, srcIP net.IP, srcPort uint16,
 
 	tuple := IPPortTuple{
 		IPLength: ipLength,
-		SrcIP:    srcIP,
-		DstIP:    dstIP,
-		SrcPort:  srcPort,
-		DstPort:  dstPort,
+		BaseTuple: BaseTuple{
+			SrcIP:   srcIP,
+			DstIP:   dstIP,
+			SrcPort: srcPort,
+			DstPort: dstPort,
+		},
 	}
 	tuple.ComputeHashebles()
 
@@ -76,21 +83,23 @@ const MaxTCPTupleRawSize = 16 + 16 + 2 + 2 + 4
 type HashableTCPTuple [MaxTCPTupleRawSize]byte
 
 type TCPTuple struct {
-	IPLength         int
-	SrcIP, DstIP     net.IP
-	SrcPort, DstPort uint16
-	StreamID         uint32
+	BaseTuple
+	IPLength int
+
+	StreamID uint32
 
 	raw HashableTCPTuple // Src_ip:Src_port:Dst_ip:Dst_port:stream_id
 }
 
 func TCPTupleFromIPPort(t *IPPortTuple, streamID uint32) TCPTuple {
 	tuple := TCPTuple{
 		IPLength: t.IPLength,
-		SrcIP:    t.SrcIP,
-		DstIP:    t.DstIP,
-		SrcPort:  t.SrcPort,
-		DstPort:  t.DstPort,
+		BaseTuple: BaseTuple{
+			SrcIP:   t.SrcIP,
+			DstIP:   t.DstIP,
+			SrcPort: t.SrcPort,
+			DstPort: t.DstPort,
+		},
 		StreamID: streamID,
 	}
 	tuple.ComputeHashebles()
@@ -129,7 +138,22 @@ func (t *TCPTuple) Hashable() HashableTCPTuple {
 	return t.raw
 }
 
-// Source and destination process names, as found by the proc module.
+// CmdlineTuple contains the source and destination process names, as found by
+// the proc module.
 type CmdlineTuple struct {
+	// Source and destination processes names as specified in packetbeat.procs.monitored
 	Src, Dst []byte
+	// Source and destination full command lines
+	SrcCommand, DstCommand []byte
+}
+
+// Reverse returns a copy of the receiver with the source and destination fields
+// swapped.
+func (c *CmdlineTuple) Reverse() CmdlineTuple {
+	return CmdlineTuple{
+		Src:        c.Dst,
+		Dst:        c.Src,
+		SrcCommand: c.DstCommand,
+		DstCommand: c.SrcCommand,
+	}
 }

packetbeat/beater/packetbeat.go

@@ -92,11 +92,17 @@ func New(b *beat.Beat, rawConfig *common.Config) (beat.Beater, error) {
 
 // init packetbeat components
 func (pb *packetbeat) init(b *beat.Beat) error {
+	var err error
 	cfg := &pb.config
-	err := procs.ProcWatcher.Init(cfg.Procs)
-	if err != nil {
-		logp.Critical(err.Error())
-		return err
+	// Enable the process watcher only if capturing live traffic
+	if cfg.Interfaces.File == "" {
+		err = procs.ProcWatcher.Init(cfg.Procs)
+		if err != nil {
+			logp.Critical(err.Error())
+			return err
+		}
+	} else {
+		logp.Info("Process watcher disabled when file input is used")
 	}
 
 	pb.pipeline = b.Publisher