Backport: Re-factor to fix high CPU usage on windows (#1562)

CHANGELOG.asciidoc

@@ -31,6 +31,8 @@ https://github.com/elastic/beats/compare/v1.2.2...1.2[Check the HEAD diff]
 
 *Topbeat*
 
+- Fixed high CPU usage when using filtering under Windows. {pull}1562[1562]
+
 *Filebeat*
 
 *Winlogbeat*

topbeat/beat/sigar.go

@@ -244,52 +244,61 @@ func getProcState(b byte) string {
 	return "unknown"
 }
 
-func GetProcess(pid int, cmdline string) (*Process, error) {
+// newProcess creates a new Process object based on the state information.
+func newProcess(pid int) (*Process, error) {
+
 	state := sigar.ProcState{}
 	if err := state.Get(pid); err != nil {
 		return nil, fmt.Errorf("error getting process state for pid=%d: %v", pid, err)
 	}
 
+	proc := Process{
+		Pid:      pid,
+		Ppid:     state.Ppid,
+		Name:     state.Name,
+		State:    getProcState(byte(state.State)),
+		Username: state.Username,
+		ctime:    time.Now(),
+	}
+
+	return &proc, nil
+}
+
+// getDetails fills in CPU, memory, and command line details for the process
+func (proc *Process) getDetails(cmdline string) error {
+
 	mem := sigar.ProcMem{}
-	if err := mem.Get(pid); err != nil {
-		return nil, fmt.Errorf("error getting process mem for pid=%d: %v", pid, err)
+	if err := mem.Get(proc.Pid); err != nil {
+		return fmt.Errorf("error getting process mem for pid=%d: %v", proc.Pid, err)
+	}
+	proc.Mem = &ProcMemStat{
+		Size:  mem.Size,
+		Rss:   mem.Resident,
+		Share: mem.Share,
 	}
 
 	cpu := sigar.ProcTime{}
-	if err := cpu.Get(pid); err != nil {
-		return nil, fmt.Errorf("error getting process cpu time for pid=%d: %v", pid, err)
+	if err := cpu.Get(proc.Pid); err != nil {
+		return fmt.Errorf("error getting process cpu time for pid=%d: %v", proc.Pid, err)
+	}
+	proc.Cpu = &ProcCpuTime{
+		Start:  cpu.FormatStartTime(),
+		Total:  cpu.Total,
+		User:   cpu.User,
+		System: cpu.Sys,
 	}
 
 	if cmdline == "" {
 		args := sigar.ProcArgs{}
-		if err := args.Get(pid); err != nil {
-			return nil, fmt.Errorf("error getting process arguments for pid=%d: %v", pid, err)
+		if err := args.Get(proc.Pid); err != nil {
+			return fmt.Errorf("error getting process arguments for pid=%d: %v", proc.Pid, err)
 		}
-		cmdline = strings.Join(args.List, " ")
-	}
-
-	proc := Process{
-		Pid:      pid,
-		Ppid:     state.Ppid,
-		Name:     state.Name,
-		State:    getProcState(byte(state.State)),
-		Username: state.Username,
-		CmdLine:  cmdline,
-		Mem: &ProcMemStat{
-			Size:  mem.Size,
-			Rss:   mem.Resident,
-			Share: mem.Share,
-		},
-		Cpu: &ProcCpuTime{
-			Start:  cpu.FormatStartTime(),
-			Total:  cpu.Total,
-			User:   cpu.User,
-			System: cpu.Sys,
-		},
-		ctime: time.Now(),
+		proc.CmdLine = strings.Join(args.List, " ")
+	} else {
+		proc.CmdLine = cmdline
 	}
 
-	return &proc, nil
+	return nil
 }
 
 func GetFileSystemList() ([]sigar.FileSystem, error) {

topbeat/beat/sigar_test.go

@@ -83,12 +83,13 @@ func TestGetProcess(t *testing.T) {
 
 	for _, pid := range pids {
 
-		process, err := GetProcess(pid, "")
-
+		process, err := newProcess(pid)
 		if err != nil {
 			continue
 		}
 		assert.NotNil(t, process)
+		err = process.getDetails("")
+		assert.NoError(t, err)
 
 		assert.True(t, (process.Pid > 0))
 		assert.True(t, (process.Ppid >= 0))
@@ -166,7 +167,11 @@ func BenchmarkGetProcess(b *testing.B) {
 			cmdline = p.CmdLine
 		}
 
-		process, err := GetProcess(pid, cmdline)
+		process, err := newProcess(pid)
+		if err != nil {
+			continue
+		}
+		err = process.getDetails(cmdline)
 		if err != nil {
 			continue
 		}

topbeat/beat/topbeat.go

@@ -2,6 +2,7 @@ package beat
 
 import (
 	"errors"
+	"fmt"
 	"math"
 	"regexp"
 	"strconv"
@@ -21,6 +22,7 @@ type ProcsMap map[int]*Process
 type Topbeat struct {
 	period           time.Duration
 	procs            []string
+	regexps          []*regexp.Regexp
 	procsMap         ProcsMap
 	lastCpuTimes     *CpuTimes
 	lastCpuTimesList []CpuTimes
@@ -101,9 +103,10 @@ func (tb *Topbeat) Setup(b *beat.Beat) error {
 }
 
 func (t *Topbeat) Run(b *beat.Beat) error {
-	var err error
-
-	t.initProcStats()
+	err := t.initProcStats()
+	if err != nil {
+		return err
+	}
 
 	ticker := time.NewTicker(t.period)
 	defer ticker.Stop()
@@ -157,12 +160,21 @@ func (t *Topbeat) Stop() {
 	close(t.done)
 }
 
-func (t *Topbeat) initProcStats() {
+func (t *Topbeat) initProcStats() error {
 
 	t.procsMap = make(ProcsMap)
 
 	if len(t.procs) == 0 {
-		return
+		return nil
+	}
+
+	t.regexps = []*regexp.Regexp{}
+	for _, pattern := range t.procs {
+		reg, err := regexp.Compile(pattern)
+		if err != nil {
+			return fmt.Errorf("Failed to compile regexp [%s]: %v", pattern, err)
+		}
+		t.regexps = append(t.regexps, reg)
 	}
 
 	pids, err := Pids()
@@ -171,13 +183,20 @@ func (t *Topbeat) initProcStats() {
 	}
 
 	for _, pid := range pids {
-		process, err := GetProcess(pid, "")
+		process, err := newProcess(pid)
 		if err != nil {
 			logp.Debug("topbeat", "Skip process pid=%d: %v", pid, err)
 			continue
 		}
+		err = process.getDetails("")
+		if err != nil {
+			logp.Err("Error getting process details pid=%d: %v", pid, err)
+			continue
+		}
 		t.procsMap[process.Pid] = process
 	}
+
+	return nil
 }
 
 func (t *Topbeat) exportProcStats() error {
@@ -199,14 +218,20 @@ func (t *Topbeat) exportProcStats() error {
 			cmdline = previousProc.CmdLine
 		}
 
-		process, err := GetProcess(pid, cmdline)
+		process, err := newProcess(pid)
 		if err != nil {
 			logp.Debug("topbeat", "Skip process pid=%d: %v", pid, err)
 			continue
 		}
 
 		if t.MatchProcess(process.Name) {
 
+			err = process.getDetails(cmdline)
+			if err != nil {
+				logp.Err("Error getting process details. pid=%d: %v", process.Pid, err)
+				continue
+			}
+
 			t.addProcCpuPercentage(process)
 			t.addProcMemPercentage(process, 0 /*read total mem usage */)
 
@@ -334,9 +359,8 @@ func collectFileSystemStats(fss []sigar.FileSystem) []common.MapStr {
 
 func (t *Topbeat) MatchProcess(name string) bool {
 
-	for _, reg := range t.procs {
-		matched, _ := regexp.MatchString(reg, name)
-		if matched {
+	for _, reg := range t.regexps {
+		if reg.MatchString(name) {
 			return true
 		}
 	}

topbeat/beat/topbeat_test.go

@@ -12,13 +12,16 @@ func TestMatchProcs(t *testing.T) {
 	var beat = Topbeat{}
 
 	beat.procs = []string{".*"}
+	beat.initProcStats()
 	assert.True(t, beat.MatchProcess("topbeat"))
 
 	beat.procs = []string{"topbeat"}
+	beat.initProcStats()
 	assert.False(t, beat.MatchProcess("burn"))
 
 	// match no processes
 	beat.procs = []string{"$^"}
+	beat.initProcStats()
 	assert.False(t, beat.MatchProcess("burn"))
 }