From a962e958ebf64118cdd48fd3d6ff1583a56c7702 Mon Sep 17 00:00:00 2001
From: Daniel Nelson <danielnelson@users.noreply.github.com>
Date: Fri, 17 Mar 2017 16:49:11 -0700
Subject: [PATCH] Refactor procstat input (#2540)
fixes #1636
fixes #2315
---
plugins/inputs/procstat/pgrep.go | 91 +++++++
plugins/inputs/procstat/process.go | 60 +++++
plugins/inputs/procstat/procstat.go | 290 ++++++++++-----------
plugins/inputs/procstat/procstat_test.go | 293 ++++++++++++++++++++--
plugins/inputs/procstat/spec_processor.go | 110 --------
testutil/accumulator.go | 23 ++
6 files changed, 594 insertions(+), 273 deletions(-)
create mode 100644 plugins/inputs/procstat/pgrep.go
create mode 100644 plugins/inputs/procstat/process.go
delete mode 100644 plugins/inputs/procstat/spec_processor.go
diff --git a/plugins/inputs/procstat/pgrep.go b/plugins/inputs/procstat/pgrep.go
new file mode 100644
index 00000000..bae5161e
--- /dev/null
+++ b/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
+}
diff --git a/plugins/inputs/procstat/process.go b/plugins/inputs/procstat/process.go
new file mode 100644
index 00000000..ec2363f6
--- /dev/null
+++ b/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
+}
diff --git a/plugins/inputs/procstat/procstat.go b/plugins/inputs/procstat/procstat.go
index 929490e4..46b88fbc 100644
--- a/plugins/inputs/procstat/procstat.go
+++ b/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 + "_"
}
- for _, pid := range pids {
- _, ok := p.pidmap[pid]
- if !ok {
- proc, err := process.NewProcess(pid)
- if err == nil {
- p.pidmap[pid] = proc
- } else {
- errstring += err.Error() + " "
- }
+ 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 errstring != "" {
- outerr = fmt.Errorf("%s", errstring)
+ //If pid is not present as a tag, include it as a field.
+ if _, pidInTags := proc.Tags()["pid"]; !pidInTags {
+ fields["pid"] = int32(proc.PID())
}
- return outerr
-}
+ numThreads, err := proc.NumThreads()
+ if err == nil {
+ fields[prefix+"num_threads"] = numThreads
+ }
-func (p *Procstat) getAllPids() ([]int32, error) {
- var pids []int32
- var err error
+ fds, err := proc.NumFDs()
+ if err == nil {
+ fields[prefix+"num_fds"] = fds
+ }
- if p.PidFile != "" {
- pids, err = p.pidsFromFile()
- } else if p.Exe != "" {
- pids, err = p.pidsFromExe()
- } else if p.Pattern != "" {
- pids, err = p.pidsFromPattern()
- } else if p.User != "" {
- pids, err = p.pidsFromUser()
- } else {
- err = fmt.Errorf("Either exe, pid_file, user, or pattern has to be specified")
+ ctx, err := proc.NumCtxSwitches()
+ if err == nil {
+ fields[prefix+"voluntary_context_switches"] = ctx.Voluntary
+ fields[prefix+"involuntary_context_switches"] = ctx.Involuntary
}
- return pids, err
-}
+ 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
+ }
-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,
- }
- }
+ 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
}
- return out, outerr
+
+ 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())
}
-func (p *Procstat) pidsFromExe() ([]int32, error) {
- var out []int32
- var outerr error
- bin, err := exec.LookPath("pgrep")
+// Update monitored Processes
+func (p *Procstat) updateProcesses(prevInfo map[PID]Process) (map[PID]Process, error) {
+ pids, tags, err := p.findPids()
if err != nil {
- return out, fmt.Errorf("Couldn't find pgrep binary: %s", err)
+ return nil, 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
+
+ procs := make(map[PID]Process, len(prevInfo))
+
+ for _, pid := range pids {
+ info, ok := prevInfo[pid]
+ if ok {
+ procs[pid] = info
+ } else {
+ 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
+ }
+
+ // 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 out, outerr
+ return procs, nil
}
-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
- }
+// 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 out, outerr
+ return p.pidFinder, nil
}
-func (p *Procstat) pidsFromUser() ([]int32, error) {
- var out []int32
- var outerr error
- bin, err := exec.LookPath("pgrep")
+// 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 out, fmt.Errorf("Couldn't find pgrep binary: %s", err)
+ return nil, nil, 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)
+
+ if p.PidFile != "" {
+ pids, err = f.PidFile(p.PidFile)
+ tags = map[string]string{"pidfile": p.PidFile}
+ } else if p.Exe != "" {
+ pids, err = f.Pattern(p.Exe)
+ tags = map[string]string{"exe": p.Exe}
+ } else if p.Pattern != "" {
+ pids, err = f.FullPattern(p.Pattern)
+ tags = map[string]string{"pattern": p.Pattern}
+ } else if p.User != "" {
+ pids, err = f.Uid(p.User)
+ tags = map[string]string{"user": p.User}
} 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
- }
- }
+ err = fmt.Errorf("Either exe, pid_file, user, or pattern has to be specified")
}
- return out, outerr
+
+ return pids, tags, err
}
func init() {
inputs.Add("procstat", func() telegraf.Input {
- return NewProcstat()
+ return &Procstat{}
})
}
diff --git a/plugins/inputs/procstat/procstat_test.go b/plugins/inputs/procstat/procstat_test.go
index ccc72bdb..1f6f2764 100644
--- a/plugins/inputs/procstat/procstat_test.go
+++ b/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
+
+ p := Procstat{
+ Exe: exe,
+ createPIDFinder: pidFinder([]PID{pid}, nil),
+ createProcess: func(PID) (Process, error) {
+ return nil, fmt.Errorf("createProcess error")
+ },
+ }
+ 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 := 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())
+ pid := PID(os.Getpid())
+
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: newTestProc,
}
- 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))
+
+ 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"))
}
diff --git a/plugins/inputs/procstat/spec_processor.go b/plugins/inputs/procstat/spec_processor.go
deleted file mode 100644
index 3b56fbc3..00000000
--- a/plugins/inputs/procstat/spec_processor.go
+++ /dev/null
@@ -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)
-}
diff --git a/testutil/accumulator.go b/testutil/accumulator.go
index 25e60920..63dfddd7 100644
--- a/testutil/accumulator.go
+++ b/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 {
--
GitLab