Refactor procstat input (#2540)

fixes #1636 
fixes #2315

plugins/inputs/procstat/pgrep.go

@@ -0,0 +1,91 @@
+package procstat
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os/exec"
+	"strconv"
+	"strings"
+)
+
+type PIDFinder interface {
+	PidFile(path string) ([]PID, error)
+	Pattern(pattern string) ([]PID, error)
+	Uid(user string) ([]PID, error)
+	FullPattern(path string) ([]PID, error)
+}
+
+// Implemention of PIDGatherer that execs pgrep to find processes
+type Pgrep struct {
+	path string
+}
+
+func NewPgrep() (PIDFinder, error) {
+	path, err := exec.LookPath("pgrep")
+	if err != nil {
+		return nil, fmt.Errorf("Could not find pgrep binary: %s", err)
+	}
+	return &Pgrep{path}, nil
+}
+
+func (pg *Pgrep) PidFile(path string) ([]PID, error) {
+	var pids []PID
+	pidString, err := ioutil.ReadFile(path)
+	if err != nil {
+		return pids, fmt.Errorf("Failed to read pidfile '%s'. Error: '%s'",
+			path, err)
+	}
+	pid, err := strconv.Atoi(strings.TrimSpace(string(pidString)))
+	if err != nil {
+		return pids, err
+	}
+	pids = append(pids, PID(pid))
+	return pids, nil
+}
+
+func (pg *Pgrep) Pattern(pattern string) ([]PID, error) {
+	args := []string{pattern}
+	return find(pg.path, args)
+}
+
+func (pg *Pgrep) Uid(user string) ([]PID, error) {
+	args := []string{"-u", user}
+	return find(pg.path, args)
+}
+
+func (pg *Pgrep) FullPattern(pattern string) ([]PID, error) {
+	args := []string{"-f", pattern}
+	return find(pg.path, args)
+}
+
+func find(path string, args []string) ([]PID, error) {
+	out, err := run(path, args)
+	if err != nil {
+		return nil, err
+	}
+
+	return parseOutput(out)
+}
+
+func run(path string, args []string) (string, error) {
+	out, err := exec.Command(path, args...).Output()
+	if err != nil {
+		return "", fmt.Errorf("Error running %s: %s", path, err)
+	}
+	return string(out), err
+}
+
+func parseOutput(out string) ([]PID, error) {
+	pids := []PID{}
+	fields := strings.Fields(out)
+	for _, field := range fields {
+		pid, err := strconv.Atoi(field)
+		if err != nil {
+			return nil, err
+		}
+		if err == nil {
+			pids = append(pids, PID(pid))
+		}
+	}
+	return pids, nil
+}

plugins/inputs/procstat/process.go

@@ -0,0 +1,60 @@
+package procstat
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/shirou/gopsutil/cpu"
+	"github.com/shirou/gopsutil/process"
+)
+
+type Process interface {
+	PID() PID
+	Tags() map[string]string
+
+	IOCounters() (*process.IOCountersStat, error)
+	MemoryInfo() (*process.MemoryInfoStat, error)
+	Name() (string, error)
+	NumCtxSwitches() (*process.NumCtxSwitchesStat, error)
+	NumFDs() (int32, error)
+	NumThreads() (int32, error)
+	Percent(interval time.Duration) (float64, error)
+	Times() (*cpu.TimesStat, error)
+}
+
+type Proc struct {
+	hasCPUTimes bool
+	tags        map[string]string
+	*process.Process
+}
+
+func NewProc(pid PID) (Process, error) {
+	process, err := process.NewProcess(int32(pid))
+	if err != nil {
+		return nil, err
+	}
+
+	proc := &Proc{
+		Process:     process,
+		hasCPUTimes: false,
+		tags:        make(map[string]string),
+	}
+	return proc, nil
+}
+
+func (p *Proc) Tags() map[string]string {
+	return p.tags
+}
+
+func (p *Proc) PID() PID {
+	return PID(p.Process.Pid)
+}
+
+func (p *Proc) Percent(interval time.Duration) (float64, error) {
+	cpu_perc, err := p.Process.Percent(time.Duration(0))
+	if !p.hasCPUTimes && err == nil {
+		p.hasCPUTimes = true
+		return 0, fmt.Errorf("Must call Percent twice to compute percent cpu.")
+	}
+	return cpu_perc, err
+}

plugins/inputs/procstat/procstat.go

@@ -2,18 +2,20 @@ package procstat
 
 import (
 	"fmt"
-	"io/ioutil"
-	"log"
-	"os/exec"
 	"strconv"
-	"strings"
-
-	"github.com/shirou/gopsutil/process"
+	"time"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
+var (
+	defaultPIDFinder = NewPgrep
+	defaultProcess   = NewProc
+)
+
+type PID int32
+
 type Procstat struct {
 	PidFile     string `toml:"pid_file"`
 	Exe         string
@@ -23,17 +25,10 @@ type Procstat struct {
 	User        string
 	PidTag      bool
 
-	// pidmap maps a pid to a process object, so we don't recreate every gather
-	pidmap map[int32]*process.Process
-	// tagmap maps a pid to a map of tags for that pid
-	tagmap map[int32]map[string]string
-}
-
-func NewProcstat() *Procstat {
-	return &Procstat{
-		pidmap: make(map[int32]*process.Process),
-		tagmap: make(map[int32]map[string]string),
-	}
+	pidFinder       PIDFinder
+	createPIDFinder func() (PIDFinder, error)
+	procs           map[PID]Process
+	createProcess   func(PID) (Process, error)
 }
 
 var sampleConfig = `
@@ -67,174 +62,179 @@ func (_ *Procstat) Description() string {
 }
 
 func (p *Procstat) Gather(acc telegraf.Accumulator) error {
-	err := p.createProcesses()
+	procs, err := p.updateProcesses(p.procs)
 	if err != nil {
-		log.Printf("E! Error: procstat getting process, exe: [%s] pidfile: [%s] pattern: [%s] user: [%s] %s",
+		return fmt.Errorf(
+			"E! Error: procstat getting process, exe: [%s] pidfile: [%s] pattern: [%s] user: [%s] %s",
 			p.Exe, p.PidFile, p.Pattern, p.User, err.Error())
-	} else {
-		for pid, proc := range p.pidmap {
-			if p.PidTag {
-				p.tagmap[pid]["pid"] = fmt.Sprint(pid)
-			}
-			p := NewSpecProcessor(p.ProcessName, p.Prefix, pid, acc, proc, p.tagmap[pid])
-			p.pushMetrics()
 	}
+	p.procs = procs
+
+	for _, proc := range p.procs {
+		p.addMetrics(proc, acc)
 	}
 
 	return nil
 }
 
-func (p *Procstat) createProcesses() error {
-	var errstring string
-	var outerr error
-
-	pids, err := p.getAllPids()
-	if err != nil {
-		errstring += err.Error() + " "
+// Add metrics a single Process
+func (p *Procstat) addMetrics(proc Process, acc telegraf.Accumulator) {
+	var prefix string
+	if p.Prefix != "" {
+		prefix = p.Prefix + "_"
 	}
 
+	fields := map[string]interface{}{}
+
+	//If process_name tag is not already set, set to actual name
+	if _, nameInTags := proc.Tags()["process_name"]; !nameInTags {
+		name, err := proc.Name()
+		if err == nil {
+			proc.Tags()["process_name"] = name
+		}
+	}
+
+	//If pid is not present as a tag, include it as a field.
+	if _, pidInTags := proc.Tags()["pid"]; !pidInTags {
+		fields["pid"] = int32(proc.PID())
+	}
+
+	numThreads, err := proc.NumThreads()
+	if err == nil {
+		fields[prefix+"num_threads"] = numThreads
+	}
+
+	fds, err := proc.NumFDs()
+	if err == nil {
+		fields[prefix+"num_fds"] = fds
+	}
+
+	ctx, err := proc.NumCtxSwitches()
+	if err == nil {
+		fields[prefix+"voluntary_context_switches"] = ctx.Voluntary
+		fields[prefix+"involuntary_context_switches"] = ctx.Involuntary
+	}
+
+	io, err := proc.IOCounters()
+	if err == nil {
+		fields[prefix+"read_count"] = io.ReadCount
+		fields[prefix+"write_count"] = io.WriteCount
+		fields[prefix+"read_bytes"] = io.ReadBytes
+		fields[prefix+"write_bytes"] = io.WriteBytes
+	}
+
+	cpu_time, err := proc.Times()
+	if err == nil {
+		fields[prefix+"cpu_time_user"] = cpu_time.User
+		fields[prefix+"cpu_time_system"] = cpu_time.System
+		fields[prefix+"cpu_time_idle"] = cpu_time.Idle
+		fields[prefix+"cpu_time_nice"] = cpu_time.Nice
+		fields[prefix+"cpu_time_iowait"] = cpu_time.Iowait
+		fields[prefix+"cpu_time_irq"] = cpu_time.Irq
+		fields[prefix+"cpu_time_soft_irq"] = cpu_time.Softirq
+		fields[prefix+"cpu_time_steal"] = cpu_time.Steal
+		fields[prefix+"cpu_time_stolen"] = cpu_time.Stolen
+		fields[prefix+"cpu_time_guest"] = cpu_time.Guest
+		fields[prefix+"cpu_time_guest_nice"] = cpu_time.GuestNice
+	}
+
+	cpu_perc, err := proc.Percent(time.Duration(0))
+	if err == nil {
+		fields[prefix+"cpu_usage"] = cpu_perc
+	}
+
+	mem, err := proc.MemoryInfo()
+	if err == nil {
+		fields[prefix+"memory_rss"] = mem.RSS
+		fields[prefix+"memory_vms"] = mem.VMS
+		fields[prefix+"memory_swap"] = mem.Swap
+	}
+
+	acc.AddFields("procstat", fields, proc.Tags())
+}
+
+// Update monitored Processes
+func (p *Procstat) updateProcesses(prevInfo map[PID]Process) (map[PID]Process, error) {
+	pids, tags, err := p.findPids()
+	if err != nil {
+		return nil, err
+	}
+
+	procs := make(map[PID]Process, len(prevInfo))
+
 	for _, pid := range pids {
-		_, ok := p.pidmap[pid]
-		if !ok {
-			proc, err := process.NewProcess(pid)
-			if err == nil {
-				p.pidmap[pid] = proc
+		info, ok := prevInfo[pid]
+		if ok {
+			procs[pid] = info
 		} else {
-				errstring += err.Error() + " "
-			}
+			proc, err := p.createProcess(pid)
+			if err != nil {
+				// No problem; process may have ended after we found it
+				continue
 			}
+			procs[pid] = proc
+
+			// Add initial tags
+			for k, v := range tags {
+				proc.Tags()[k] = v
 			}
 
-	if errstring != "" {
-		outerr = fmt.Errorf("%s", errstring)
+			// Add pid tag if needed
+			if p.PidTag {
+				proc.Tags()["pid"] = strconv.Itoa(int(pid))
+			}
+			if p.ProcessName != "" {
+				proc.Tags()["process_name"] = p.ProcessName
+			}
+		}
+	}
+	return procs, nil
 }
 
-	return outerr
+// Create and return PIDGatherer lazily
+func (p *Procstat) getPIDFinder() (PIDFinder, error) {
+	if p.pidFinder == nil {
+		f, err := p.createPIDFinder()
+		if err != nil {
+			return nil, err
+		}
+		p.pidFinder = f
+	}
+	return p.pidFinder, nil
 }
 
-func (p *Procstat) getAllPids() ([]int32, error) {
-	var pids []int32
+// Get matching PIDs and their initial tags
+func (p *Procstat) findPids() ([]PID, map[string]string, error) {
+	var pids []PID
+	var tags map[string]string
 	var err error
 
+	f, err := p.getPIDFinder()
+	if err != nil {
+		return nil, nil, err
+	}
+
 	if p.PidFile != "" {
-		pids, err = p.pidsFromFile()
+		pids, err = f.PidFile(p.PidFile)
+		tags = map[string]string{"pidfile": p.PidFile}
 	} else if p.Exe != "" {
-		pids, err = p.pidsFromExe()
+		pids, err = f.Pattern(p.Exe)
+		tags = map[string]string{"exe": p.Exe}
 	} else if p.Pattern != "" {
-		pids, err = p.pidsFromPattern()
+		pids, err = f.FullPattern(p.Pattern)
+		tags = map[string]string{"pattern": p.Pattern}
 	} else if p.User != "" {
-		pids, err = p.pidsFromUser()
+		pids, err = f.Uid(p.User)
+		tags = map[string]string{"user": p.User}
 	} else {
 		err = fmt.Errorf("Either exe, pid_file, user, or pattern has to be specified")
 	}
 
-	return pids, err
-}
-
-func (p *Procstat) pidsFromFile() ([]int32, error) {
-	var out []int32
-	var outerr error
-	pidString, err := ioutil.ReadFile(p.PidFile)
-	if err != nil {
-		outerr = fmt.Errorf("Failed to read pidfile '%s'. Error: '%s'",
-			p.PidFile, err)
-	} else {
-		pid, err := strconv.Atoi(strings.TrimSpace(string(pidString)))
-		if err != nil {
-			outerr = err
-		} else {
-			out = append(out, int32(pid))
-			p.tagmap[int32(pid)] = map[string]string{
-				"pidfile": p.PidFile,
-			}
-		}
-	}
-	return out, outerr
-}
-
-func (p *Procstat) pidsFromExe() ([]int32, error) {
-	var out []int32
-	var outerr error
-	bin, err := exec.LookPath("pgrep")
-	if err != nil {
-		return out, fmt.Errorf("Couldn't find pgrep binary: %s", err)
-	}
-	pgrep, err := exec.Command(bin, p.Exe).Output()
-	if err != nil {
-		return out, fmt.Errorf("Failed to execute %s. Error: '%s'", bin, err)
-	} else {
-		pids := strings.Fields(string(pgrep))
-		for _, pid := range pids {
-			ipid, err := strconv.Atoi(pid)
-			if err == nil {
-				out = append(out, int32(ipid))
-				p.tagmap[int32(ipid)] = map[string]string{
-					"exe": p.Exe,
-				}
-			} else {
-				outerr = err
-			}
-		}
-	}
-	return out, outerr
-}
-
-func (p *Procstat) pidsFromPattern() ([]int32, error) {
-	var out []int32
-	var outerr error
-	bin, err := exec.LookPath("pgrep")
-	if err != nil {
-		return out, fmt.Errorf("Couldn't find pgrep binary: %s", err)
-	}
-	pgrep, err := exec.Command(bin, "-f", p.Pattern).Output()
-	if err != nil {
-		return out, fmt.Errorf("Failed to execute %s. Error: '%s'", bin, err)
-	} else {
-		pids := strings.Fields(string(pgrep))
-		for _, pid := range pids {
-			ipid, err := strconv.Atoi(pid)
-			if err == nil {
-				out = append(out, int32(ipid))
-				p.tagmap[int32(ipid)] = map[string]string{
-					"pattern": p.Pattern,
-				}
-			} else {
-				outerr = err
-			}
-		}
-	}
-	return out, outerr
-}
-
-func (p *Procstat) pidsFromUser() ([]int32, error) {
-	var out []int32
-	var outerr error
-	bin, err := exec.LookPath("pgrep")
-	if err != nil {
-		return out, fmt.Errorf("Couldn't find pgrep binary: %s", err)
-	}
-	pgrep, err := exec.Command(bin, "-u", p.User).Output()
-	if err != nil {
-		return out, fmt.Errorf("Failed to execute %s. Error: '%s'", bin, err)
-	} else {
-		pids := strings.Fields(string(pgrep))
-		for _, pid := range pids {
-			ipid, err := strconv.Atoi(pid)
-			if err == nil {
-				out = append(out, int32(ipid))
-				p.tagmap[int32(ipid)] = map[string]string{
-					"user": p.User,
-				}
-			} else {
-				outerr = err
-			}
-		}
-	}
-	return out, outerr
+	return pids, tags, err
 }
 
 func init() {
 	inputs.Add("procstat", func() telegraf.Input {
-		return NewProcstat()
+		return &Procstat{}
 	})
 }

plugins/inputs/procstat/procstat_test.go

@@ -1,33 +1,290 @@
 package procstat
 
 import (
-	"io/ioutil"
+	"fmt"
 	"os"
-	"strconv"
 	"testing"
+	"time"
 
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/shirou/gopsutil/cpu"
 	"github.com/shirou/gopsutil/process"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
-	"github.com/influxdata/telegraf/testutil"
 )
 
-func TestGather(t *testing.T) {
+type testPgrep struct {
+	pids []PID
+	err  error
+}
+
+func pidFinder(pids []PID, err error) func() (PIDFinder, error) {
+	return func() (PIDFinder, error) {
+		return &testPgrep{
+			pids: pids,
+			err:  err,
+		}, nil
+	}
+}
+
+func (pg *testPgrep) PidFile(path string) ([]PID, error) {
+	return pg.pids, pg.err
+}
+
+func (pg *testPgrep) Pattern(pattern string) ([]PID, error) {
+	return pg.pids, pg.err
+}
+
+func (pg *testPgrep) Uid(user string) ([]PID, error) {
+	return pg.pids, pg.err
+}
+
+func (pg *testPgrep) FullPattern(pattern string) ([]PID, error) {
+	return pg.pids, pg.err
+}
+
+type testProc struct {
+	pid  PID
+	tags map[string]string
+}
+
+func newTestProc(pid PID) (Process, error) {
+	proc := &testProc{
+		tags: make(map[string]string),
+	}
+	return proc, nil
+}
+
+func (p *testProc) PID() PID {
+	return p.pid
+}
+
+func (p *testProc) Tags() map[string]string {
+	return p.tags
+}
+
+func (p *testProc) IOCounters() (*process.IOCountersStat, error) {
+	return &process.IOCountersStat{}, nil
+}
+
+func (p *testProc) MemoryInfo() (*process.MemoryInfoStat, error) {
+	return &process.MemoryInfoStat{}, nil
+}
+
+func (p *testProc) Name() (string, error) {
+	return "test_proc", nil
+}
+
+func (p *testProc) NumCtxSwitches() (*process.NumCtxSwitchesStat, error) {
+	return &process.NumCtxSwitchesStat{}, nil
+}
+
+func (p *testProc) NumFDs() (int32, error) {
+	return 0, nil
+}
+
+func (p *testProc) NumThreads() (int32, error) {
+	return 0, nil
+}
+
+func (p *testProc) Percent(interval time.Duration) (float64, error) {
+	return 0, nil
+}
+
+func (p *testProc) Times() (*cpu.TimesStat, error) {
+	return &cpu.TimesStat{}, nil
+}
+
+var pid PID = PID(42)
+var exe string = "foo"
+
+func TestGather_CreateProcessErrorOk(t *testing.T) {
 	var acc testutil.Accumulator
-	pid := os.Getpid()
-	file, err := ioutil.TempFile(os.TempDir(), "telegraf")
-	require.NoError(t, err)
-	file.Write([]byte(strconv.Itoa(pid)))
-	file.Close()
-	defer os.Remove(file.Name())
+
 	p := Procstat{
-		PidFile: file.Name(),
-		Prefix:  "foo",
-		pidmap:  make(map[int32]*process.Process),
-		tagmap:  make(map[int32]map[string]string),
+		Exe:             exe,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess: func(PID) (Process, error) {
+			return nil, fmt.Errorf("createProcess error")
+		},
 	}
-	p.Gather(&acc)
-	assert.True(t, acc.HasFloatField("procstat", "foo_cpu_time_user"))
-	assert.True(t, acc.HasUIntField("procstat", "foo_memory_vms"))
+	require.NoError(t, p.Gather(&acc))
+}
+
+func TestGather_CreatePIDFinderError(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		createPIDFinder: func() (PIDFinder, error) {
+			return nil, fmt.Errorf("createPIDFinder error")
+		},
+		createProcess: newTestProc,
+	}
+	require.Error(t, p.Gather(&acc))
+}
+
+func TestGather_ProcessName(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		Exe:             exe,
+		ProcessName:     "custom_name",
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.Equal(t, "custom_name", acc.TagValue("procstat", "process_name"))
+}
+
+func TestGather_NoProcessNameUsesReal(t *testing.T) {
+	var acc testutil.Accumulator
+	pid := PID(os.Getpid())
+
+	p := Procstat{
+		Exe:             exe,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.True(t, acc.HasTag("procstat", "process_name"))
+}
+
+func TestGather_NoPidTag(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		Exe:             exe,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+	assert.True(t, acc.HasInt32Field("procstat", "pid"))
+	assert.False(t, acc.HasTag("procstat", "pid"))
+}
+
+func TestGather_PidTag(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		Exe:             exe,
+		PidTag:          true,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+	assert.Equal(t, "42", acc.TagValue("procstat", "pid"))
+	assert.False(t, acc.HasInt32Field("procstat", "pid"))
+}
+
+func TestGather_Prefix(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		Exe:             exe,
+		Prefix:          "custom_prefix",
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+	assert.True(t, acc.HasInt32Field("procstat", "custom_prefix_num_fds"))
+}
+
+func TestGather_Exe(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		Exe:             exe,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.Equal(t, exe, acc.TagValue("procstat", "exe"))
+}
+
+func TestGather_User(t *testing.T) {
+	var acc testutil.Accumulator
+	user := "ada"
+
+	p := Procstat{
+		User:            user,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.Equal(t, user, acc.TagValue("procstat", "user"))
+}
+
+func TestGather_Pattern(t *testing.T) {
+	var acc testutil.Accumulator
+	pattern := "foo"
+
+	p := Procstat{
+		Pattern:         pattern,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.Equal(t, pattern, acc.TagValue("procstat", "pattern"))
+}
+
+func TestGather_MissingPidMethod(t *testing.T) {
+	var acc testutil.Accumulator
+
+	p := Procstat{
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.Error(t, p.Gather(&acc))
+}
+
+func TestGather_PidFile(t *testing.T) {
+	var acc testutil.Accumulator
+	pidfile := "/path/to/pidfile"
+
+	p := Procstat{
+		PidFile:         pidfile,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   newTestProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.Equal(t, pidfile, acc.TagValue("procstat", "pidfile"))
+}
+
+func TestGather_PercentFirstPass(t *testing.T) {
+	var acc testutil.Accumulator
+	pid := PID(os.Getpid())
+
+	p := Procstat{
+		Pattern:         "foo",
+		PidTag:          true,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   NewProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+
+	assert.True(t, acc.HasFloatField("procstat", "cpu_time_user"))
+	assert.False(t, acc.HasFloatField("procstat", "cpu_usage"))
+}
+
+func TestGather_PercentSecondPass(t *testing.T) {
+	var acc testutil.Accumulator
+	pid := PID(os.Getpid())
+
+	p := Procstat{
+		Pattern:         "foo",
+		PidTag:          true,
+		createPIDFinder: pidFinder([]PID{pid}, nil),
+		createProcess:   NewProc,
+	}
+	require.NoError(t, p.Gather(&acc))
+	require.NoError(t, p.Gather(&acc))
+
+	assert.True(t, acc.HasFloatField("procstat", "cpu_time_user"))
+	assert.True(t, acc.HasFloatField("procstat", "cpu_usage"))
 }

plugins/inputs/procstat/spec_processor.go

@@ -1,110 +0,0 @@
-package procstat
-
-import (
-	"time"
-
-	"github.com/shirou/gopsutil/process"
-
-	"github.com/influxdata/telegraf"
-)
-
-type SpecProcessor struct {
-	Prefix string
-	pid    int32
-	tags   map[string]string
-	fields map[string]interface{}
-	acc    telegraf.Accumulator
-	proc   *process.Process
-}
-
-func NewSpecProcessor(
-	processName string,
-	prefix string,
-	pid int32,
-	acc telegraf.Accumulator,
-	p *process.Process,
-	tags map[string]string,
-) *SpecProcessor {
-	if processName != "" {
-		tags["process_name"] = processName
-	} else {
-		name, err := p.Name()
-		if err == nil {
-			tags["process_name"] = name
-		}
-	}
-	return &SpecProcessor{
-		Prefix: prefix,
-		pid:    pid,
-		tags:   tags,
-		fields: make(map[string]interface{}),
-		acc:    acc,
-		proc:   p,
-	}
-}
-
-func (p *SpecProcessor) pushMetrics() {
-	var prefix string
-	if p.Prefix != "" {
-		prefix = p.Prefix + "_"
-	}
-	fields := map[string]interface{}{}
-
-	//If pid is not present as a tag, include it as a field.
-	if _, pidInTags := p.tags["pid"]; !pidInTags {
-		fields["pid"] = p.pid
-	}
-
-	numThreads, err := p.proc.NumThreads()
-	if err == nil {
-		fields[prefix+"num_threads"] = numThreads
-	}
-
-	fds, err := p.proc.NumFDs()
-	if err == nil {
-		fields[prefix+"num_fds"] = fds
-	}
-
-	ctx, err := p.proc.NumCtxSwitches()
-	if err == nil {
-		fields[prefix+"voluntary_context_switches"] = ctx.Voluntary
-		fields[prefix+"involuntary_context_switches"] = ctx.Involuntary
-	}
-
-	io, err := p.proc.IOCounters()
-	if err == nil {
-		fields[prefix+"read_count"] = io.ReadCount
-		fields[prefix+"write_count"] = io.WriteCount
-		fields[prefix+"read_bytes"] = io.ReadBytes
-		fields[prefix+"write_bytes"] = io.WriteBytes
-	}
-
-	cpu_time, err := p.proc.Times()
-	if err == nil {
-		fields[prefix+"cpu_time_user"] = cpu_time.User
-		fields[prefix+"cpu_time_system"] = cpu_time.System
-		fields[prefix+"cpu_time_idle"] = cpu_time.Idle
-		fields[prefix+"cpu_time_nice"] = cpu_time.Nice
-		fields[prefix+"cpu_time_iowait"] = cpu_time.Iowait
-		fields[prefix+"cpu_time_irq"] = cpu_time.Irq
-		fields[prefix+"cpu_time_soft_irq"] = cpu_time.Softirq
-		fields[prefix+"cpu_time_steal"] = cpu_time.Steal
-		fields[prefix+"cpu_time_stolen"] = cpu_time.Stolen
-		fields[prefix+"cpu_time_guest"] = cpu_time.Guest
-		fields[prefix+"cpu_time_guest_nice"] = cpu_time.GuestNice
-	}
-
-	cpu_perc, err := p.proc.Percent(time.Duration(0))
-	if err == nil && cpu_perc != 0 {
-		fields[prefix+"cpu_usage"] = cpu_perc
-	}
-
-	mem, err := p.proc.MemoryInfo()
-	if err == nil {
-		fields[prefix+"memory_rss"] = mem.RSS
-		fields[prefix+"memory_vms"] = mem.VMS
-		fields[prefix+"memory_swap"] = mem.Swap
-	}
-
-	p.acc.AddFields("procstat", fields, p.tags)
-}

testutil/accumulator.go

@@ -161,6 +161,29 @@ func (a *Accumulator) Get(measurement string) (*Metric, bool) {
 	return nil, false
 }
 
+func (a *Accumulator) HasTag(measurement string, key string) bool {
+	for _, p := range a.Metrics {
+		if p.Measurement == measurement {
+			_, ok := p.Tags[key]
+			return ok
+		}
+	}
+	return false
+}
+
+func (a *Accumulator) TagValue(measurement string, key string) string {
+	for _, p := range a.Metrics {
+		if p.Measurement == measurement {
+			v, ok := p.Tags[key]
+			if !ok {
+				return ""
+			}
+			return v
+		}
+	}
+	return ""
+}
+
 // NFields returns the total number of fields in the accumulator, across all
 // measurements
 func (a *Accumulator) NFields() int {