CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(inputs.procstat): Cleanup code (#14326)

This commit is contained in:
Sven Rebhan 2023-11-27 21:50:30 +01:00 committed by GitHub
parent 1c2c03d778
commit f2cc928178
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 773 additions and 833 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
plugins/inputs/procstat/dev/telegraf.conf
View File

@ -1,9 +0,0 @@
[agent]
interval="1s"
flush_interval="1s"
[[inputs.procstat]]
exe = "telegraf"
[[outputs.file]]
files = ["stdout"]

5
plugins/inputs/procstat/memmap_notlinux.go
View File

@ -1,5 +0,0 @@
//go:build !linux
package procstat
func collectMemmap(Process, string, map[string]any) {}

71
plugins/inputs/procstat/native_finder.go
View File

@ -11,13 +11,7 @@ import (
)
// NativeFinder uses gopsutil to find processes
type NativeFinder struct {
}
// NewNativeFinder ...
func NewNativeFinder() (PIDFinder, error) {
return &NativeFinder{}, nil
}
type NativeFinder struct{}
// Uid will return all pids for the given user
func (pg *NativeFinder) UID(user string) ([]PID, error) {
@ -80,38 +74,24 @@ func (pg *NativeFinder) FullPattern(pattern string) ([]PID, error) {
return pids, err
}
// ChildPattern matches children pids on the command line when the process was executed
func (pg *NativeFinder) ChildPattern(pattern string) ([]PID, error) {
regxPattern, err := regexp.Compile(pattern)
// Children matches children pids on the command line when the process was executed
func (pg *NativeFinder) Children(pid PID) ([]PID, error) {
// Get all running processes
p, err := process.NewProcess(int32(pid))
if err != nil {
return nil, fmt.Errorf("compiling regexp failed: %w", err)
return nil, fmt.Errorf("getting process %d failed: %w", pid, err)
}
procs, err := process.Processes()
// Get all children of the current process
children, err := p.Children()
if err != nil {
return nil, fmt.Errorf("getting processes failed: %w", err)
return nil, fmt.Errorf("unable to get children of process %d: %w", p.Pid, err)
}
pids := make([]PID, 0, len(children))
for _, child := range children {
pids = append(pids, PID(child.Pid))
}
var pids []PID
for _, p := range procs {
cmd, err := p.Cmdline()
if err != nil || !regxPattern.MatchString(cmd) {
continue
}
parent, err := process.NewProcess(p.Pid)
if err != nil {
return nil, fmt.Errorf("unable to get process %d: %w", p.Pid, err)
}
children, err := parent.Children()
if err != nil {
return nil, fmt.Errorf("unable to get children of process %d: %w", p.Pid, err)
}
for _, child := range children {
pids = append(pids, PID(child.Pid))
}
}
return pids, err
}
@ -127,3 +107,28 @@ func (pg *NativeFinder) FastProcessList() ([]*process.Process, error) {
}
return result, nil
}
// Pattern matches on the process name
func (pg *NativeFinder) Pattern(pattern string) ([]PID, error) {
var pids []PID
regxPattern, err := regexp.Compile(pattern)
if err != nil {
return pids, err
}
procs, err := pg.FastProcessList()
if err != nil {
return pids, err
}
for _, p := range procs {
name, err := processName(p)
if err != nil {
//skip, this can be caused by the pid no longer existing
//or you having no permissions to access it
continue
}
if regxPattern.MatchString(name) {
pids = append(pids, PID(p.Pid))
}
}
return pids, err
}

32
plugins/inputs/procstat/native_finder_notwindows.go
View File

@ -1,32 +0,0 @@
//go:build !windows
package procstat
import (
"regexp"
)
// Pattern matches on the process name
func (pg *NativeFinder) Pattern(pattern string) ([]PID, error) {
var pids []PID
regxPattern, err := regexp.Compile(pattern)
if err != nil {
return pids, err
}
procs, err := pg.FastProcessList()
if err != nil {
return pids, err
}
for _, p := range procs {
name, err := p.Exe()
if err != nil {
//skip, this can be caused by the pid no longer existing
//or you having no permissions to access it
continue
}
if regxPattern.MatchString(name) {
pids = append(pids, PID(p.Pid))
}
}
return pids, err
}

52
plugins/inputs/procstat/native_finder_test.go
View File

@ -4,6 +4,7 @@ import (
"context"
"os"
"os/exec"
"os/user"
"runtime"
"testing"
@ -11,17 +12,15 @@ import (
)
func BenchmarkPattern(b *testing.B) {
finder, err := NewNativeFinder()
require.NoError(b, err)
finder := &NativeFinder{}
for n := 0; n < b.N; n++ {
_, err = finder.Pattern(".*")
_, err := finder.Pattern(".*")
require.NoError(b, err)
}
}
func BenchmarkFullPattern(b *testing.B) {
finder, err := NewNativeFinder()
require.NoError(b, err)
finder := &NativeFinder{}
for n := 0; n < b.N; n++ {
_, err := finder.FullPattern(".*")
require.NoError(b, err)
@ -53,10 +52,47 @@ func TestChildPattern(t *testing.T) {
expected = append(expected, PID(cmd2.Process.Pid))
// Use the plugin to find the children
finder, err := NewNativeFinder()
finder := &NativeFinder{}
parent, err := finder.Pattern(parentName)
require.NoError(t, err)
childs, err := finder.ChildPattern(parentName)
require.Len(t, parent, 1)
childs, err := finder.Children(parent[0])
require.NoError(t, err)
require.ElementsMatch(t, expected, childs)
}
func TestGather_RealPatternIntegration(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
pg := &NativeFinder{}
pids, err := pg.Pattern(`procstat`)
require.NoError(t, err)
require.NotEmpty(t, pids)
}
func TestGather_RealFullPatternIntegration(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
if runtime.GOOS != "windows" {
t.Skip("Skipping integration test on Non-Windows OS")
}
pg := &NativeFinder{}
pids, err := pg.FullPattern(`%procstat%`)
require.NoError(t, err)
require.NotEmpty(t, pids)
}
func TestGather_RealUserIntegration(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
currentUser, err := user.Current()
require.NoError(t, err)
pg := &NativeFinder{}
pids, err := pg.UID(currentUser.Username)
require.NoError(t, err)
require.NotEmpty(t, pids)
}

30
plugins/inputs/procstat/native_finder_windows.go
View File

@ -1,30 +0,0 @@
package procstat
import (
"regexp"
)
// Pattern matches on the process name
func (pg *NativeFinder) Pattern(pattern string) ([]PID, error) {
var pids []PID
regxPattern, err := regexp.Compile(pattern)
if err != nil {
return pids, err
}
procs, err := pg.FastProcessList()
if err != nil {
return pids, err
}
for _, p := range procs {
name, err := p.Name()
if err != nil {
//skip, this can be caused by the pid no longer existing
//or you having no permissions to access it
continue
}
if regxPattern.MatchString(name) {
pids = append(pids, PID(p.Pid))
}
}
return pids, err
}

44
plugins/inputs/procstat/native_finder_windows_test.go
View File

@ -1,44 +0,0 @@
package procstat
import (
"os/user"
"testing"
"github.com/stretchr/testify/require"
)
func TestGather_RealPatternIntegration(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
pg, err := NewNativeFinder()
require.NoError(t, err)
pids, err := pg.Pattern(`procstat`)
require.NoError(t, err)
require.NotEmpty(t, pids)
}
func TestGather_RealFullPatternIntegration(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
pg, err := NewNativeFinder()
require.NoError(t, err)
pids, err := pg.FullPattern(`%procstat%`)
require.NoError(t, err)
require.NotEmpty(t, pids)
}
func TestGather_RealUserIntegration(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test in short mode")
}
currentUser, err := user.Current()
require.NoError(t, err)
pg, err := NewNativeFinder()
require.NoError(t, err)
pids, err := pg.UID(currentUser.Username)
require.NoError(t, err)
require.NotEmpty(t, pids)
}

14
plugins/inputs/procstat/memmap.go → plugins/inputs/procstat/os_linux.go
View File

@ -2,6 +2,20 @@
package procstat
import (
"errors"
"github.com/shirou/gopsutil/v3/process"
)
func processName(p *process.Process) (string, error) {
return p.Exe()
}
func queryPidWithWinServiceName(_ string) (uint32, error) {
return 0, errors.New("os not supporting win_service option")
}
func collectMemmap(proc Process, prefix string, fields map[string]any) {
memMapStats, err := proc.MemoryMaps(true)
if err == nil && len(*memMapStats) == 1 {

19
plugins/inputs/procstat/os_others.go Normal file
View File

@ -0,0 +1,19 @@
//go:build !linux && !windows
package procstat
import (
"errors"
"github.com/shirou/gopsutil/v3/process"
)
func processName(p *process.Process) (string, error) {
return p.Exe()
}
func queryPidWithWinServiceName(_ string) (uint32, error) {
return 0, errors.New("os not supporting win_service option")
}
func collectMemmap(Process, string, map[string]any) {}

7
plugins/inputs/procstat/win_service_windows.go → plugins/inputs/procstat/os_windows.go
View File

@ -6,10 +6,15 @@ import (
"errors"
"unsafe"
"github.com/shirou/gopsutil/v3/process"
"golang.org/x/sys/windows"
"golang.org/x/sys/windows/svc/mgr"
)
func processName(p *process.Process) (string, error) {
return p.Name()
}
func getService(name string) (*mgr.Service, error) {
m, err := mgr.Connect()
if err != nil {
@ -48,3 +53,5 @@ func queryPidWithWinServiceName(winServiceName string) (uint32, error) {
return p.ProcessId, nil
}
func collectMemmap(Process, string, map[string]any) {}

67
plugins/inputs/procstat/pgrep.go
View File

@ -15,7 +15,7 @@ type Pgrep struct {
path string
}
func NewPgrep() (PIDFinder, error) {
func newPgrepFinder() (PIDFinder, error) {
path, err := exec.LookPath("pgrep")
if err != nil {
return nil, fmt.Errorf("could not find pgrep binary: %w", err)
@ -40,69 +40,38 @@ func (pg *Pgrep) PidFile(path string) ([]PID, error) {
func (pg *Pgrep) Pattern(pattern string) ([]PID, error) {
args := []string{pattern}
return find(pg.path, args)
return pg.find(args)
}
func (pg *Pgrep) UID(user string) ([]PID, error) {
args := []string{"-u", user}
return find(pg.path, args)
return pg.find(args)
}
func (pg *Pgrep) FullPattern(pattern string) ([]PID, error) {
args := []string{"-f", pattern}
return find(pg.path, args)
return pg.find(args)
}
func (pg *Pgrep) ChildPattern(pattern string) ([]PID, error) {
args := []string{"-P", pattern}
out, err := run(pg.path, args)
if err != nil {
return nil, err
}
func (pg *Pgrep) Children(pid PID) ([]PID, error) {
args := []string{"-P", strconv.FormatInt(int64(pid), 10)}
return pg.find(args)
}
pids := []PID{}
pid, err := strconv.ParseInt(pattern, 10, 32)
func (pg *Pgrep) find(args []string) ([]PID, error) {
// Execute pgrep with the given arguments
buf, err := exec.Command(pg.path, args...).Output()
if err != nil {
return nil, err
}
pids = append(pids, PID(pid))
fields := strings.Fields(out)
for _, field := range fields {
pid, err := strconv.ParseInt(field, 10, 32)
if err != nil {
return pids, err
// Exit code 1 means "no processes found" so we should not return
// an error in this case.
if status, _ := internal.ExitStatus(err); status == 1 {
return nil, nil
}
pids = append(pids, PID(pid))
return nil, fmt.Errorf("error running %q: %w", pg.path, err)
}
out := string(buf)
return pids, nil
}
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 exit code 1, ie no processes found, do not return error
if i, _ := internal.ExitStatus(err); i == 1 {
return "", nil
}
if err != nil {
return "", fmt.Errorf("error running %q: %w", path, err)
}
return string(out), err
}
func parseOutput(out string) ([]PID, error) {
// Parse the command output to extract the PIDs
pids := []PID{}
fields := strings.Fields(out)
for _, field := range fields {

196
plugins/inputs/procstat/process.go
View File

@ -2,34 +2,20 @@ package procstat
import (
"fmt"
"runtime"
"time"
"github.com/shirou/gopsutil/v3/cpu"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/metric"
"github.com/shirou/gopsutil/v3/process"
)
//nolint:interfacebloat // conditionally allow to contain more methods
type Process interface {
PID() PID
Tags() map[string]string
PageFaults() (*process.PageFaultsStat, error)
IOCounters() (*process.IOCountersStat, error)
MemoryInfo() (*process.MemoryInfoStat, error)
MemoryMaps(bool) (*[]process.MemoryMapsStat, error)
Name() (string, error)
Cmdline() (string, error)
NumCtxSwitches() (*process.NumCtxSwitchesStat, error)
NumFDs() (int32, error)
NumThreads() (int32, error)
Percent(interval time.Duration) (float64, error)
MemoryPercent() (float32, error)
Times() (*cpu.TimesStat, error)
RlimitUsage(bool) ([]process.RlimitStat, error)
Username() (string, error)
CreateTime() (int64, error)
Ppid() (int32, error)
Status() ([]string, error)
SetTag(string, string)
MemoryMaps(bool) (*[]process.MemoryMapsStat, error)
Metric(prefix string, cmdLineTag, solarisMode bool) telegraf.Metric
}
type PIDFinder interface {
@ -37,7 +23,7 @@ type PIDFinder interface {
Pattern(pattern string) ([]PID, error)
UID(user string) ([]PID, error)
FullPattern(path string) ([]PID, error)
ChildPattern(path string) ([]PID, error)
Children(pid PID) ([]PID, error)
}
type Proc struct {
@ -46,7 +32,7 @@ type Proc struct {
*process.Process
}
func NewProc(pid PID) (Process, error) {
func newProc(pid PID) (Process, error) {
p, err := process.NewProcess(int32(pid))
if err != nil {
return nil, err
@ -60,19 +46,15 @@ func NewProc(pid PID) (Process, error) {
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) Username() (string, error) {
return p.Process.Username()
func (p *Proc) SetTag(k, v string) {
p.tags[k] = v
}
func (p *Proc) Percent(_ time.Duration) (float64, error) {
func (p *Proc) percent(_ time.Duration) (float64, error) {
cpuPerc, err := p.Process.Percent(time.Duration(0))
if !p.hasCPUTimes && err == nil {
p.hasCPUTimes = true
@ -80,3 +62,159 @@ func (p *Proc) Percent(_ time.Duration) (float64, error) {
}
return cpuPerc, err
}
// Add metrics a single Process
func (p *Proc) Metric(prefix string, cmdLineTag, solarisMode bool) telegraf.Metric {
if prefix != "" {
prefix += "_"
}
fields := make(map[string]interface{})
if _, nameInTags := p.tags["process_name"]; !nameInTags {
name, err := p.Name()
if err == nil {
p.tags["process_name"] = name
}
}
if _, ok := p.tags["user"]; !ok {
user, err := p.Username()
if err == nil {
p.tags["user"] = user
}
}
// If pid is not present as a tag, include it as a field.
if _, pidInTags := p.tags["pid"]; !pidInTags {
fields["pid"] = p.Pid
}
// Add the command line as a tag if the option is set
if cmdLineTag {
if _, ok := p.tags["cmdline"]; !ok {
cmdline, err := p.Cmdline()
if err == nil {
p.tags["cmdline"] = cmdline
}
}
}
numThreads, err := p.NumThreads()
if err == nil {
fields[prefix+"num_threads"] = numThreads
}
fds, err := p.NumFDs()
if err == nil {
fields[prefix+"num_fds"] = fds
}
ctx, err := p.NumCtxSwitches()
if err == nil {
fields[prefix+"voluntary_context_switches"] = ctx.Voluntary
fields[prefix+"involuntary_context_switches"] = ctx.Involuntary
}
faults, err := p.PageFaults()
if err == nil {
fields[prefix+"minor_faults"] = faults.MinorFaults
fields[prefix+"major_faults"] = faults.MajorFaults
fields[prefix+"child_minor_faults"] = faults.ChildMinorFaults
fields[prefix+"child_major_faults"] = faults.ChildMajorFaults
}
io, err := p.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
}
createdAt, err := p.CreateTime() // returns epoch in ms
if err == nil {
fields[prefix+"created_at"] = createdAt * 1000000 // ms to ns
}
cpuTime, err := p.Times()
if err == nil {
fields[prefix+"cpu_time_user"] = cpuTime.User
fields[prefix+"cpu_time_system"] = cpuTime.System
fields[prefix+"cpu_time_iowait"] = cpuTime.Iowait // only reported on Linux
}
cpuPerc, err := p.percent(time.Duration(0))
if err == nil {
if solarisMode {
fields[prefix+"cpu_usage"] = cpuPerc / float64(runtime.NumCPU())
} else {
fields[prefix+"cpu_usage"] = cpuPerc
}
}
// This only returns values for RSS and VMS
mem, err := p.MemoryInfo()
if err == nil {
fields[prefix+"memory_rss"] = mem.RSS
fields[prefix+"memory_vms"] = mem.VMS
}
collectMemmap(p, prefix, fields)
memPerc, err := p.MemoryPercent()
if err == nil {
fields[prefix+"memory_usage"] = memPerc
}
rlims, err := p.RlimitUsage(true)
if err == nil {
for _, rlim := range rlims {
var name string
switch rlim.Resource {
case process.RLIMIT_CPU:
name = "cpu_time"
case process.RLIMIT_DATA:
name = "memory_data"
case process.RLIMIT_STACK:
name = "memory_stack"
case process.RLIMIT_RSS:
name = "memory_rss"
case process.RLIMIT_NOFILE:
name = "num_fds"
case process.RLIMIT_MEMLOCK:
name = "memory_locked"
case process.RLIMIT_AS:
name = "memory_vms"
case process.RLIMIT_LOCKS:
name = "file_locks"
case process.RLIMIT_SIGPENDING:
name = "signals_pending"
case process.RLIMIT_NICE:
name = "nice_priority"
case process.RLIMIT_RTPRIO:
name = "realtime_priority"
default:
continue
}
fields[prefix+"rlimit_"+name+"_soft"] = rlim.Soft
fields[prefix+"rlimit_"+name+"_hard"] = rlim.Hard
if name != "file_locks" { // gopsutil doesn't currently track the used file locks count
fields[prefix+name] = rlim.Used
}
}
}
ppid, err := p.Ppid()
if err == nil {
fields[prefix+"ppid"] = ppid
}
status, err := p.Status()
if err == nil {
fields[prefix+"status"] = status[0]
}
return metric.New("procstat", p.tags, fields, time.Time{})
}

587
plugins/inputs/procstat/procstat.go
View File

@ -14,8 +14,6 @@ import (
"strings"
"time"
"github.com/shirou/gopsutil/v3/process"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs"
)
@ -23,38 +21,39 @@ import (
//go:embed sample.conf
var sampleConfig string
// execCommand is so tests can mock out exec.Command usage.
var execCommand = exec.Command
type PID int32
type Procstat struct {
PidFinder string `toml:"pid_finder"`
PidFile string `toml:"pid_file"`
Exe string
Pattern string
Prefix string
CmdLineTag bool `toml:"cmdline_tag"`
ProcessName string
User string
SystemdUnits string `toml:"systemd_units"`
SupervisorUnit []string `toml:"supervisor_unit"`
IncludeSystemdChildren bool `toml:"include_systemd_children"`
CGroup string `toml:"cgroup"`
PidTag bool
WinService string `toml:"win_service"`
Mode string
PidFinder string `toml:"pid_finder"`
PidFile string `toml:"pid_file"`
Exe string `toml:"exe"`
Pattern string `toml:"pattern"`
Prefix string `toml:"prefix"`
CmdLineTag bool `toml:"cmdline_tag"`
ProcessName string `toml:"process_name"`
User string `toml:"user"`
SystemdUnits string `toml:"systemd_units"`
SupervisorUnit []string `toml:"supervisor_unit"`
IncludeSystemdChildren bool `toml:"include_systemd_children"`
CGroup string `toml:"cgroup"`
PidTag bool `toml:"pid_tag"`
WinService string `toml:"win_service"`
Mode string `toml:"mode"`
Log telegraf.Logger `toml:"-"`
solarisMode bool
finder PIDFinder
processes map[PID]Process
finder PIDFinder
createPIDFinder func() (PIDFinder, error)
procs map[PID]Process
createProcess func(PID) (Process, error)
createProcess func(PID) (Process, error)
}
type PidsTags struct {
PIDS []PID
PIDs []PID
Tags map[string]string
Err error
}
func (*Procstat) SampleConfig() string {
@ -62,76 +61,128 @@ func (*Procstat) SampleConfig() string {
}
func (p *Procstat) Init() error {
if strings.ToLower(p.Mode) == "solaris" {
p.solarisMode = true
// Check solaris mode
p.solarisMode = strings.ToLower(p.Mode) == "solaris"
// Check filtering
switch {
case len(p.SupervisorUnit) > 0, p.SystemdUnits != "", p.WinService != "",
p.CGroup != "", p.PidFile != "", p.Exe != "", p.Pattern != "",
p.User != "":
// Do nothing as those are valid settings
default:
return errors.New("require filter option but none set")
}
// Instantiate the finder
switch p.PidFinder {
case "":
case "", "pgrep":
p.PidFinder = "pgrep"
p.createPIDFinder = NewPgrep
finder, err := newPgrepFinder()
if err != nil {
return fmt.Errorf("creating pgrep finder failed: %w", err)
}
p.finder = finder
case "native":
p.createPIDFinder = NewNativeFinder
case "pgrep":
p.createPIDFinder = NewPgrep
// gopsutil relies on pgrep when looking up children on darwin
// see https://github.com/shirou/gopsutil/blob/v3.23.10/process/process_darwin.go#L235
requiresChildren := len(p.SupervisorUnit) > 0 && p.Pattern != ""
if requiresChildren && runtime.GOOS == "darwin" {
return errors.New("configuration requires the 'pgrep' finder on you OS")
}
p.finder = &NativeFinder{}
case "test":
p.Log.Warn("running in test mode")
default:
return fmt.Errorf("unknown pid_finder %q", p.PidFinder)
}
// gopsutil relies on pgrep when looking up children on darwin
// see https://github.com/shirou/gopsutil/blob/v3.23.10/process/process_darwin.go#L235
requiresChildren := len(p.SupervisorUnit) > 0 && p.Pattern != ""
if requiresChildren && p.PidFinder == "native" && runtime.GOOS == "darwin" {
return errors.New("configuration requires the 'pgrep' finder on you OS")
}
// Initialize the running process cache
p.processes = make(map[PID]Process)
return nil
}
func (p *Procstat) Gather(acc telegraf.Accumulator) error {
pidCount := 0
now := time.Now()
newProcs := make(map[PID]Process, len(p.procs))
tags := make(map[string]string)
pidTags := p.findPids()
for _, pidTag := range pidTags {
if len(pidTag.PIDS) < 1 && len(p.SupervisorUnit) > 0 {
results, err := p.findPids()
if err != nil {
// Add lookup error-metric
fields := map[string]interface{}{
"pid_count": 0,
"running": 0,
"result_code": 1,
}
tags := map[string]string{
"pid_finder": p.PidFinder,
"result": "lookup_error",
}
acc.AddFields("procstat_lookup", fields, tags, now)
return err
}
var count int
running := make(map[PID]bool)
for _, r := range results {
if len(r.PIDs) < 1 && len(p.SupervisorUnit) > 0 {
continue
}
pids := pidTag.PIDS
err := pidTag.Err
pidCount += len(pids)
for key, value := range pidTag.Tags {
tags[key] = value
}
if err != nil {
fields := map[string]interface{}{
"pid_count": 0,
"running": 0,
"result_code": 1,
count += len(r.PIDs)
for _, pid := range r.PIDs {
// Use the cached processes as we need the existing instances
// to compute delta-metrics (e.g. cpu-usage).
proc, found := p.processes[pid]
if !found {
// We've found a process that was not recorded before so add it
// to the list of processes
proc, err = p.createProcess(pid)
if err != nil {
// No problem; process may have ended after we found it
continue
}
// Assumption: if a process has no name, it probably does not exist
if name, _ := proc.Name(); name == "" {
continue
}
// Add initial tags
for k, v := range r.Tags {
proc.SetTag(k, v)
}
// Add pid tag if needed
if p.PidTag {
proc.SetTag("pid", strconv.Itoa(int(pid)))
}
if p.ProcessName != "" {
proc.SetTag("process_name", p.ProcessName)
}
p.processes[pid] = proc
}
tags["pid_finder"] = p.PidFinder
tags["result"] = "lookup_error"
acc.AddFields("procstat_lookup", fields, tags, now)
return err
running[pid] = true
m := proc.Metric(p.Prefix, p.CmdLineTag, p.solarisMode)
m.SetTime(now)
acc.AddMetric(m)
}
p.updateProcesses(pids, pidTag.Tags, p.procs, newProcs)
}
p.procs = newProcs
for _, proc := range p.procs {
p.addMetric(proc, acc, now)
// Cleanup processes that are not running anymore
for pid := range p.processes {
if !running[pid] {
delete(p.processes, pid)
}
}
// Add lookup statistics-metric
fields := map[string]interface{}{
"pid_count": pidCount,
"running": len(p.procs),
"pid_count": count,
"running": len(running),
"result_code": 0,
}
tags["pid_finder"] = p.PidFinder
tags["result"] = "success"
tags := map[string]string{
"pid_finder": p.PidFinder,
"result": "success",
}
if len(p.SupervisorUnit) > 0 {
tags["supervisor_unit"] = strings.Join(p.SupervisorUnit, ";")
}
@ -140,315 +191,99 @@ func (p *Procstat) Gather(acc telegraf.Accumulator) error {
return nil
}
// Add metrics a single Process
func (p *Procstat) addMetric(proc Process, acc telegraf.Accumulator, t time.Time) {
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 user tag is not already set, set to actual name
if _, ok := proc.Tags()["user"]; !ok {
user, err := proc.Username()
if err == nil {
proc.Tags()["user"] = user
}
}
//If pid is not present as a tag, include it as a field.
if _, pidInTags := proc.Tags()["pid"]; !pidInTags {
fields["pid"] = int32(proc.PID())
}
//If cmd_line tag is true and it is not already set add cmdline as a tag
if p.CmdLineTag {
if _, ok := proc.Tags()["cmdline"]; !ok {
cmdline, err := proc.Cmdline()
if err == nil {
proc.Tags()["cmdline"] = cmdline
}
}
}
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
}
faults, err := proc.PageFaults()
if err == nil {
fields[prefix+"minor_faults"] = faults.MinorFaults
fields[prefix+"major_faults"] = faults.MajorFaults
fields[prefix+"child_minor_faults"] = faults.ChildMinorFaults
fields[prefix+"child_major_faults"] = faults.ChildMajorFaults
}
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
}
createdAt, err := proc.CreateTime() // returns epoch in ms
if err == nil {
fields[prefix+"created_at"] = createdAt * 1000000 // ms to ns
}
cpuTime, err := proc.Times()
if err == nil {
fields[prefix+"cpu_time_user"] = cpuTime.User
fields[prefix+"cpu_time_system"] = cpuTime.System
fields[prefix+"cpu_time_iowait"] = cpuTime.Iowait // only reported on Linux
}
cpuPerc, err := proc.Percent(time.Duration(0))
if err == nil {
if p.solarisMode {
fields[prefix+"cpu_usage"] = cpuPerc / float64(runtime.NumCPU())
} else {
fields[prefix+"cpu_usage"] = cpuPerc
}
}
// This only returns values for RSS and VMS
mem, err := proc.MemoryInfo()
if err == nil {
fields[prefix+"memory_rss"] = mem.RSS
fields[prefix+"memory_vms"] = mem.VMS
}
collectMemmap(proc, prefix, fields)
memPerc, err := proc.MemoryPercent()
if err == nil {
fields[prefix+"memory_usage"] = memPerc
}
rlims, err := proc.RlimitUsage(true)
if err == nil {
for _, rlim := range rlims {
var name string
switch rlim.Resource {
case process.RLIMIT_CPU:
name = "cpu_time"
case process.RLIMIT_DATA:
name = "memory_data"
case process.RLIMIT_STACK:
name = "memory_stack"
case process.RLIMIT_RSS:
name = "memory_rss"
case process.RLIMIT_NOFILE:
name = "num_fds"
case process.RLIMIT_MEMLOCK:
name = "memory_locked"
case process.RLIMIT_AS:
name = "memory_vms"
case process.RLIMIT_LOCKS:
name = "file_locks"
case process.RLIMIT_SIGPENDING:
name = "signals_pending"
case process.RLIMIT_NICE:
name = "nice_priority"
case process.RLIMIT_RTPRIO:
name = "realtime_priority"
default:
continue
}
fields[prefix+"rlimit_"+name+"_soft"] = rlim.Soft
fields[prefix+"rlimit_"+name+"_hard"] = rlim.Hard
if name != "file_locks" { // gopsutil doesn't currently track the used file locks count
fields[prefix+name] = rlim.Used
}
}
}
ppid, err := proc.Ppid()
if err == nil {
fields[prefix+"ppid"] = ppid
}
status, err := proc.Status()
if err == nil {
fields[prefix+"status"] = status[0]
}
acc.AddFields("procstat", fields, proc.Tags(), t)
}
// Update monitored Processes
func (p *Procstat) updateProcesses(pids []PID, tags map[string]string, prevInfo map[PID]Process, procs map[PID]Process) {
for _, pid := range pids {
info, ok := prevInfo[pid]
if ok {
// Assumption: if a process has no name, it probably does not exist
if name, _ := info.Name(); name == "" {
continue
}
procs[pid] = info
} else {
proc, err := p.createProcess(pid)
if err != nil {
// No problem; process may have ended after we found it
continue
}
// Assumption: if a process has no name, it probably does not exist
if name, _ := proc.Name(); name == "" {
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
}
}
}
}
// Create and return PIDGatherer lazily
func (p *Procstat) getPIDFinder() (PIDFinder, error) {
if p.finder == nil {
f, err := p.createPIDFinder()
// Get matching PIDs and their initial tags
func (p *Procstat) findPids() ([]PidsTags, error) {
switch {
case len(p.SupervisorUnit) > 0:
return p.findSupervisorUnits()
case p.SystemdUnits != "":
return p.systemdUnitPIDs()
case p.WinService != "":
pids, err := p.winServicePIDs()
if err != nil {
return nil, err
}
p.finder = f
}
return p.finder, nil
}
// Get matching PIDs and their initial tags
func (p *Procstat) findPids() []PidsTags {
var pidTags []PidsTags
if len(p.SupervisorUnit) > 0 {
groups, groupsTags, err := p.supervisorPIDs()
tags := map[string]string{"win_service": p.WinService}
return []PidsTags{{pids, tags}}, nil
case p.CGroup != "":
return p.cgroupPIDs()
case p.PidFile != "":
pids, err := p.finder.PidFile(p.PidFile)
if err != nil {
pidTags = append(pidTags, PidsTags{nil, nil, err})
return pidTags
return nil, err
}
// According to the PID, find the system process number and use pgrep to filter to get the number of child processes
for _, group := range groups {
f, err := p.getPIDFinder()
if err != nil {
pidTags = append(pidTags, PidsTags{nil, nil, err})
return pidTags
}
p.Pattern = groupsTags[group]["pid"]
if p.Pattern == "" {
pidTags = append(pidTags, PidsTags{nil, groupsTags[group], err})
return pidTags
}
pids, tags, err := p.SimpleFindPids(f)
if err != nil {
pidTags = append(pidTags, PidsTags{nil, nil, err})
return pidTags
}
// Handle situations where the PID does not exist
if len(pids) == 0 {
pidTags = append(pidTags, PidsTags{nil, groupsTags[group], err})
continue
}
stats := groupsTags[group]
// Merge tags map
for k, v := range stats {
_, ok := tags[k]
if !ok {
tags[k] = v
}
}
// Remove duplicate pid tags
delete(tags, "pid")
pidTags = append(pidTags, PidsTags{pids, tags, err})
tags := map[string]string{"pidfile": p.PidFile}
return []PidsTags{{pids, tags}}, nil
case p.Exe != "":
pids, err := p.finder.Pattern(p.Exe)
if err != nil {
return nil, err
}
return pidTags
} else if p.SystemdUnits != "" {
groups := p.systemdUnitPIDs()
return groups
} else if p.CGroup != "" {
groups := p.cgroupPIDs()
return groups
tags := map[string]string{"exe": p.Exe}
return []PidsTags{{pids, tags}}, nil
case p.Pattern != "":
pids, err := p.finder.FullPattern(p.Pattern)
if err != nil {
return nil, err
}
tags := map[string]string{"pattern": p.Pattern}
return []PidsTags{{pids, tags}}, nil
case p.User != "":
pids, err := p.finder.UID(p.User)
if err != nil {
return nil, err
}
tags := map[string]string{"user": p.User}
return []PidsTags{{pids, tags}}, nil
}
return nil, errors.New("no filter option set")
}
f, err := p.getPIDFinder()
func (p *Procstat) findSupervisorUnits() ([]PidsTags, error) {
groups, groupsTags, err := p.supervisorPIDs()
if err != nil {
pidTags = append(pidTags, PidsTags{nil, nil, err})
return pidTags
}
pids, tags, err := p.SimpleFindPids(f)
pidTags = append(pidTags, PidsTags{pids, tags, err})
return pidTags
}
// Get matching PIDs and their initial tags
func (p *Procstat) SimpleFindPids(f PIDFinder) ([]PID, map[string]string, error) {
var pids []PID
tags := make(map[string]string)
var err error
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 len(p.SupervisorUnit) > 0 && p.Pattern != "" {
pids, err = f.ChildPattern(p.Pattern)
tags = map[string]string{"pattern": p.Pattern, "parent_pid": p.Pattern}
} 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 if p.WinService != "" {
pids, err = p.winServicePIDs()
tags = map[string]string{"win_service": p.WinService}
} else {
err = fmt.Errorf("either exe, pid_file, user, pattern, systemd_unit, cgroup, or win_service must be specified")
return nil, fmt.Errorf("getting supervisor PIDs failed: %w", err)
}
return pids, tags, err
}
// According to the PID, find the system process number and get the child processes
pidTags := make([]PidsTags, 0, len(groups))
for _, group := range groups {
grppid := groupsTags[group]["pid"]
if grppid == "" {
pidTags = append(pidTags, PidsTags{nil, groupsTags[group]})
continue
}
// execCommand is so tests can mock out exec.Command usage.
var execCommand = exec.Command
pid, err := strconv.ParseInt(grppid, 10, 32)
if err != nil {
return nil, fmt.Errorf("converting PID %q failed: %w", grppid, err)
}
// Get all children of the supervisor unit
pids, err := p.finder.Children(PID(pid))
if err != nil {
return nil, fmt.Errorf("getting children for %d failed: %w", pid, err)
}
tags := map[string]string{"pattern": p.Pattern, "parent_pid": p.Pattern}
// Handle situations where the PID does not exist
if len(pids) == 0 {
continue
}
// Merge tags map
for k, v := range groupsTags[group] {
_, ok := tags[k]
if !ok {
tags[k] = v
}
}
// Remove duplicate pid tags
delete(tags, "pid")
pidTags = append(pidTags, PidsTags{pids, tags})
}
return pidTags, nil
}
func (p *Procstat) supervisorPIDs() ([]string, map[string]map[string]string, error) {
out, err := execCommand("supervisorctl", "status", strings.Join(p.SupervisorUnit, " ")).Output()
@ -494,7 +329,7 @@ func (p *Procstat) supervisorPIDs() ([]string, map[string]map[string]string, err
return p.SupervisorUnit, mainPids, nil
}
func (p *Procstat) systemdUnitPIDs() []PidsTags {
func (p *Procstat) systemdUnitPIDs() ([]PidsTags, error) {
if p.IncludeSystemdChildren {
p.CGroup = fmt.Sprintf("systemd/system.slice/%s", p.SystemdUnits)
return p.cgroupPIDs()
@ -502,9 +337,12 @@ func (p *Procstat) systemdUnitPIDs() []PidsTags {
var pidTags []PidsTags
pids, err := p.simpleSystemdUnitPIDs()
if err != nil {
return nil, err
}
tags := map[string]string{"systemd_unit": p.SystemdUnits}
pidTags = append(pidTags, PidsTags{pids, tags, err})
return pidTags
pidTags = append(pidTags, PidsTags{pids, tags})
return pidTags, nil
}
func (p *Procstat) simpleSystemdUnitPIDs() ([]PID, error) {
@ -536,7 +374,7 @@ func (p *Procstat) simpleSystemdUnitPIDs() ([]PID, error) {
return pids, nil
}
func (p *Procstat) cgroupPIDs() []PidsTags {
func (p *Procstat) cgroupPIDs() ([]PidsTags, error) {
procsPath := p.CGroup
if procsPath[0] != '/' {
procsPath = "/sys/fs/cgroup/" + procsPath
@ -544,17 +382,20 @@ func (p *Procstat) cgroupPIDs() []PidsTags {
items, err := filepath.Glob(procsPath)
if err != nil {
return []PidsTags{{nil, nil, fmt.Errorf("glob failed: %w", err)}}
return nil, fmt.Errorf("glob failed: %w", err)
}
pidTags := make([]PidsTags, 0, len(items))
for _, item := range items {
pids, err := p.singleCgroupPIDs(item)
if err != nil {
return nil, err
}
tags := map[string]string{"cgroup": p.CGroup, "cgroup_full": item}
pidTags = append(pidTags, PidsTags{pids, tags, err})
pidTags = append(pidTags, PidsTags{pids, tags})
}
return pidTags
return pidTags, nil
}
func (p *Procstat) singleCgroupPIDs(path string) ([]PID, error) {
@ -610,6 +451,6 @@ func (p *Procstat) winServicePIDs() ([]PID, error) {
func init() {
inputs.Add("procstat", func() telegraf.Input {
return &Procstat{createProcess: NewProc}
return &Procstat{createProcess: newProc}
})
}

462
plugins/inputs/procstat/procstat_test.go
View File

@ -10,10 +10,11 @@ import (
"testing"
"time"
"github.com/shirou/gopsutil/v3/cpu"
"github.com/shirou/gopsutil/v3/process"
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/testutil"
)
@ -78,12 +79,10 @@ type testPgrep struct {
err error
}
func pidFinder(pids []PID) func() (PIDFinder, error) {
return func() (PIDFinder, error) {
return &testPgrep{
pids: pids,
err: nil,
}, nil
func newTestFinder(pids []PID) PIDFinder {
return &testPgrep{
pids: pids,
err: nil,
}
}
@ -107,7 +106,7 @@ func (pg *testPgrep) FullPattern(_ string) ([]PID, error) {
return pg.pids, pg.err
}
func (pg *testPgrep) ChildPattern(_ string) ([]PID, error) {
func (pg *testPgrep) Children(_ PID) ([]PID, error) {
pids := []PID{7311, 8111, 8112}
return pids, pg.err
}
@ -128,77 +127,75 @@ func (p *testProc) PID() PID {
return p.pid
}
func (p *testProc) Username() (string, error) {
return "testuser", nil
func (p *testProc) Name() (string, error) {
return "test_proc", nil
}
func (p *testProc) Tags() map[string]string {
return p.tags
}
func (p *testProc) PageFaults() (*process.PageFaultsStat, error) {
return &process.PageFaultsStat{}, nil
}
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) SetTag(k, v string) {
p.tags[k] = v
}
func (p *testProc) MemoryMaps(bool) (*[]process.MemoryMapsStat, error) {
return &[]process.MemoryMapsStat{}, nil
}
func (p *testProc) Name() (string, error) {
return "test_proc", nil
}
func (p *testProc) Metric(prefix string, cmdLineTag, _ bool) telegraf.Metric {
if prefix != "" {
prefix += "_"
}
func (p *testProc) NumCtxSwitches() (*process.NumCtxSwitchesStat, error) {
return &process.NumCtxSwitchesStat{}, nil
}
fields := map[string]interface{}{
"pid": int32(p.pid),
"ppid": int32(0),
prefix + "num_fds": int32(0),
prefix + "num_threads": int32(0),
prefix + "voluntary_context_switches": int64(0),
prefix + "involuntary_context_switches": int64(0),
prefix + "minor_faults": uint64(0),
prefix + "major_faults": uint64(0),
prefix + "child_major_faults": uint64(0),
prefix + "child_minor_faults": uint64(0),
prefix + "read_bytes": uint64(0),
prefix + "read_count": uint64(0),
prefix + "write_bytes": uint64(0),
prefix + "write_count": uint64(0),
prefix + "created_at": int64(0),
prefix + "cpu_time_user": float64(0),
prefix + "cpu_time_system": float64(0),
prefix + "cpu_time_iowait": float64(0),
prefix + "cpu_usage": float64(0),
prefix + "memory_rss": uint64(0),
prefix + "memory_vms": uint64(0),
prefix + "memory_usage": float32(0),
prefix + "status": "running",
}
func (p *testProc) NumFDs() (int32, error) {
return 0, nil
}
tags := map[string]string{
"process_name": "test_proc",
"user": "testuser",
}
for k, v := range p.tags {
tags[k] = v
}
func (p *testProc) NumThreads() (int32, error) {
return 0, nil
}
func (p *testProc) Percent(_ time.Duration) (float64, error) {
return 0, nil
}
func (p *testProc) MemoryPercent() (float32, error) {
return 0, nil
}
func (p *testProc) CreateTime() (int64, error) {
return 0, nil
}
func (p *testProc) Times() (*cpu.TimesStat, error) {
return &cpu.TimesStat{}, nil
}
func (p *testProc) RlimitUsage(_ bool) ([]process.RlimitStat, error) {
return []process.RlimitStat{}, nil
}
func (p *testProc) Ppid() (int32, error) {
return 0, nil
}
func (p *testProc) Status() ([]string, error) {
return []string{"running"}, nil
if cmdLineTag {
tags["cmdline"] = "test_proc"
}
return metric.New("procstat", tags, fields, time.Time{})
}
var pid = PID(42)
var exe = "foo"
func TestInitInvalidFinder(t *testing.T) {
plugin := Procstat{
PidFinder: "foo",
Log: testutil.Logger{},
createProcess: newTestProc,
}
require.Error(t, plugin.Init())
}
func TestInitRequiresChildDarwin(t *testing.T) {
if runtime.GOOS != "darwin" {
t.Skip("Skipping test on non-darwin platform")
@ -208,196 +205,228 @@ func TestInitRequiresChildDarwin(t *testing.T) {
Pattern: "somepattern",
SupervisorUnit: []string{"a_unit"},
PidFinder: "native",
Log: testutil.Logger{},
}
require.ErrorContains(t, p.Init(), "requires the 'pgrep' finder")
}
func TestGather_CreateProcessErrorOk(t *testing.T) {
var acc testutil.Accumulator
func TestInitMissingPidMethod(t *testing.T) {
p := Procstat{
Exe: exe,
createPIDFinder: pidFinder([]PID{pid}),
Log: testutil.Logger{},
createProcess: newTestProc,
}
require.ErrorContains(t, p.Init(), "require filter option but none set")
}
func TestGather_CreateProcessErrorOk(t *testing.T) {
p := Procstat{
Exe: exe,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: func(PID) (Process, error) {
return nil, fmt.Errorf("createProcess error")
},
}
require.NoError(t, acc.GatherError(p.Gather))
}
require.NoError(t, p.Init())
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, acc.GatherError(p.Gather))
require.NoError(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}),
createProcess: newTestProc,
Exe: exe,
ProcessName: "custom_name",
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.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}),
createProcess: newTestProc,
Exe: exe,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.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}),
createProcess: newTestProc,
Exe: exe,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.True(t, acc.HasInt32Field("procstat", "pid"))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.True(t, acc.HasInt64Field("procstat", "pid"))
require.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}),
createProcess: newTestProc,
Exe: exe,
PidTag: true,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.Equal(t, "42", acc.TagValue("procstat", "pid"))
require.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}),
createProcess: newTestProc,
Exe: exe,
Prefix: "custom_prefix",
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.True(t, acc.HasInt32Field("procstat", "custom_prefix_num_fds"))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.True(t, acc.HasInt64Field("procstat", "custom_prefix_num_fds"))
}
func TestGather_Exe(t *testing.T) {
var acc testutil.Accumulator
p := Procstat{
Exe: exe,
createPIDFinder: pidFinder([]PID{pid}),
createProcess: newTestProc,
Exe: exe,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.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}),
createProcess: newTestProc,
User: user,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.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}),
createProcess: newTestProc,
Pattern: pattern,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.Equal(t, pattern, acc.TagValue("procstat", "pattern"))
}
func TestGather_MissingPidMethod(t *testing.T) {
var acc testutil.Accumulator
p := Procstat{
createPIDFinder: pidFinder([]PID{pid}),
createProcess: newTestProc,
}
require.Error(t, acc.GatherError(p.Gather))
}
func TestGather_PidFile(t *testing.T) {
var acc testutil.Accumulator
pidfile := "/path/to/pidfile"
p := Procstat{
PidFile: pidfile,
createPIDFinder: pidFinder([]PID{pid}),
createProcess: newTestProc,
PidFile: pidfile,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.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}),
createProcess: NewProc,
Pattern: "foo",
PidTag: true,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.True(t, acc.HasFloatField("procstat", "cpu_time_user"))
require.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}),
createProcess: NewProc,
Pattern: "foo",
PidTag: true,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
require.NoError(t, p.Gather(&acc))
require.True(t, acc.HasFloatField("procstat", "cpu_time_user"))
require.True(t, acc.HasFloatField("procstat", "cpu_usage"))
@ -405,17 +434,19 @@ func TestGather_PercentSecondPass(t *testing.T) {
func TestGather_systemdUnitPIDs(t *testing.T) {
p := Procstat{
createPIDFinder: pidFinder([]PID{}),
SystemdUnits: "TestGather_systemdUnitPIDs",
SystemdUnits: "TestGather_systemdUnitPIDs",
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
}
pidsTags := p.findPids()
require.NoError(t, p.Init())
pidsTags, err := p.findPids()
require.NoError(t, err)
for _, pidsTag := range pidsTags {
pids := pidsTag.PIDS
tags := pidsTag.Tags
err := pidsTag.Err
require.NoError(t, err)
require.Equal(t, []PID{11408}, pids)
require.Equal(t, "TestGather_systemdUnitPIDs", tags["systemd_unit"])
require.Equal(t, []PID{11408}, pidsTag.PIDs)
require.Equal(t, "TestGather_systemdUnitPIDs", pidsTag.Tags["systemd_unit"])
}
}
@ -429,41 +460,50 @@ func TestGather_cgroupPIDs(t *testing.T) {
require.NoError(t, err)
p := Procstat{
createPIDFinder: pidFinder([]PID{}),
CGroup: td,
CGroup: td,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
}
pidsTags := p.findPids()
require.NoError(t, p.Init())
pidsTags, err := p.findPids()
require.NoError(t, err)
for _, pidsTag := range pidsTags {
pids := pidsTag.PIDS
tags := pidsTag.Tags
err := pidsTag.Err
require.NoError(t, err)
require.Equal(t, []PID{1234, 5678}, pids)
require.Equal(t, td, tags["cgroup"])
require.Equal(t, []PID{1234, 5678}, pidsTag.PIDs)
require.Equal(t, td, pidsTag.Tags["cgroup"])
}
}
func TestProcstatLookupMetric(t *testing.T) {
p := Procstat{
createPIDFinder: pidFinder([]PID{543}),
createProcess: NewProc,
Exe: "-Gsys",
Exe: "-Gsys",
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{543}),
createProcess: newProc,
}
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, acc.GatherError(p.Gather))
require.Len(t, acc.Metrics, len(p.procs)+1)
require.NoError(t, p.Gather(&acc))
require.Len(t, acc.GetTelegrafMetrics(), 1)
}
func TestGather_SameTimestamps(t *testing.T) {
var acc testutil.Accumulator
pidfile := "/path/to/pidfile"
p := Procstat{
PidFile: pidfile,
createPIDFinder: pidFinder([]PID{pid}),
createProcess: newTestProc,
PidFile: pidfile,
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
createProcess: newTestProc,
}
require.NoError(t, acc.GatherError(p.Gather))
require.NoError(t, p.Init())
var acc testutil.Accumulator
require.NoError(t, p.Gather(&acc))
procstat, _ := acc.Get("procstat")
procstatLookup, _ := acc.Get("procstat_lookup")
@ -473,40 +513,42 @@ func TestGather_SameTimestamps(t *testing.T) {
func TestGather_supervisorUnitPIDs(t *testing.T) {
p := Procstat{
createPIDFinder: pidFinder([]PID{}),
SupervisorUnit: []string{"TestGather_supervisorUnitPIDs"},
SupervisorUnit: []string{"TestGather_supervisorUnitPIDs"},
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
}
pidsTags := p.findPids()
require.NoError(t, p.Init())
pidsTags, err := p.findPids()
require.NoError(t, err)
for _, pidsTag := range pidsTags {
pids := pidsTag.PIDS
tags := pidsTag.Tags
err := pidsTag.Err
require.NoError(t, err)
require.Equal(t, []PID{7311, 8111, 8112}, pids)
require.Equal(t, "TestGather_supervisorUnitPIDs", tags["supervisor_unit"])
require.Equal(t, []PID{7311, 8111, 8112}, pidsTag.PIDs)
require.Equal(t, "TestGather_supervisorUnitPIDs", pidsTag.Tags["supervisor_unit"])
}
}
func TestGather_MoresupervisorUnitPIDs(t *testing.T) {
p := Procstat{
createPIDFinder: pidFinder([]PID{}),
Pattern: "7311",
SupervisorUnit: []string{"TestGather_STARTINGsupervisorUnitPIDs", "TestGather_FATALsupervisorUnitPIDs"},
SupervisorUnit: []string{"TestGather_STARTINGsupervisorUnitPIDs", "TestGather_FATALsupervisorUnitPIDs"},
PidFinder: "test",
Log: testutil.Logger{},
finder: newTestFinder([]PID{pid}),
}
pidsTags := p.findPids()
require.NoError(t, p.Init())
pidsTags, err := p.findPids()
require.NoError(t, err)
for _, pidsTag := range pidsTags {
pids := pidsTag.PIDS
tags := pidsTag.Tags
err := pidsTag.Err
require.Empty(t, pids)
require.Contains(t, []string{"TestGather_STARTINGsupervisorUnitPIDs", "TestGather_FATALsupervisorUnitPIDs"}, tags["supervisor_unit"])
if tags["supervisor_unit"] == "TestGather_STARTINGsupervisorUnitPIDs" {
require.Equal(t, "STARTING", tags["status"])
require.NoError(t, err)
} else if tags["supervisor_unit"] == "TestGather_FATALsupervisorUnitPIDs" {
require.Equal(t, "FATAL", tags["status"])
require.NoError(t, err)
require.Equal(t, "Exited too quickly (process log may have details)", tags["error"])
require.Empty(t, pidsTag.PIDs)
switch pidsTag.Tags["supervisor_unit"] {
case "TestGather_STARTINGsupervisorUnitPIDs":
require.Equal(t, "STARTING", pidsTag.Tags["status"])
case "TestGather_FATALsupervisorUnitPIDs":
require.Equal(t, "FATAL", pidsTag.Tags["status"])
require.Equal(t, "Exited too quickly (process log may have details)", pidsTag.Tags["error"])
default:
t.Fatalf("unexpected value for tag 'supervisor_unit': %q", pidsTag.Tags["supervisor_unit"])
}
}
}

11
plugins/inputs/procstat/win_service_notwindows.go
View File

@ -1,11 +0,0 @@
//go:build !windows
package procstat
import (
"fmt"
)
func queryPidWithWinServiceName(_ string) (uint32, error) {
return 0, fmt.Errorf("os not support win_service option")
}