Allow procstat plugin to handle multiple PIDs from pgrep

Closes #248

CHANGELOG.md

@@ -13,6 +13,7 @@
 - Memory plugin: cached and buffered measurements re-added
 - Logging: additional logging for each collection interval, track the number
 of metrics collected and from how many plugins.
+- [#240](https://github.com/influxdb/telegraf/pull/240): procstat plugin, thanks @ranjib!
 
 ### Bugfixes
 - [#228](https://github.com/influxdb/telegraf/pull/228): New version of package will replace old one. Thanks @ekini!

README.md

@@ -179,6 +179,7 @@ Telegraf currently has support for collecting metrics from
 * nginx
 * ping
 * postgresql
+* procstat
 * prometheus
 * rabbitmq
 * redis
@@ -207,6 +208,7 @@ found by running `telegraf -sample-config`
 * datadog
 * opentsdb
 * amqp (rabbitmq)
+* mqtt
 
 ## Contributing
 

plugins/procstat/README.md

@@ -5,9 +5,11 @@
 The procstat plugin can be used to monitor system resource usage by an
 individual process using their /proc data.
 
+The plugin will tag processes by their PID and their process name.
+
 Processes can be specified either by pid file or by executable name. Procstat
 plugin will use `pgrep` when executable name is provided to obtain the pid.
-Proctsta plugin will transmit IO, memory, cpu, file descriptor related
+Proctstas plugin will transmit IO, memory, cpu, file descriptor related
 measurements for every process specified. A prefix can be set to isolate
 individual process specific measurements.
 
@@ -22,39 +24,49 @@ Example:
 
     [[procstat.specifications]]
     pid_file = "/var/run/lxc/dnsmasq.pid"
-  prefix = "dnsmasq"
+```
+
+The above configuration would result in output like:
+
+```
+[...]
+> [name="dnsmasq" pid="44979"] procstat_cpu_user value=0.14
+> [name="dnsmasq" pid="44979"] procstat_cpu_system value=0.07
+[...]
+> [name="influxd" pid="34337"] procstat_influxd_cpu_user value=25.43
+> [name="influxd" pid="34337"] procstat_influxd_cpu_system value=21.82
 ```
 
 # Measurements
-Note: prefix will set by the user, per process.
+Note: prefix can be set by the user, per process.
 
 File descriptor related measurement names:
-- procstat_prefix_num_fds value=4
+- procstat_[prefix_]num_fds value=4
 
 Context switch related measurement names:
-- procstat_prefix_voluntary_context_switches value=250
+- procstat_[prefix_]voluntary_context_switches value=250
-- procstat_prefix_involuntary_context_switches value=0
+- procstat_[prefix_]involuntary_context_switches value=0
 
 I/O related measurement names:
-- procstat_prefix_read_count value=396
+- procstat_[prefix_]read_count value=396
-- procstat_prefix_write_count value=1
+- procstat_[prefix_]write_count value=1
-- procstat_prefix_read_bytes value=1019904
+- procstat_[prefix_]read_bytes value=1019904
-- procstat_prefix_write_bytes value=1
+- procstat_[prefix_]write_bytes value=1
 
 CPU related measurement names:
-- procstat_prefix_cpu_user value=0
+- procstat_[prefix_]cpu_user value=0
-- procstat_prefix_cpu_system value=0.01
+- procstat_[prefix_]cpu_system value=0.01
-- procstat_prefix_cpu_idle value=0
+- procstat_[prefix_]cpu_idle value=0
-- procstat_prefix_cpu_nice value=0
+- procstat_[prefix_]cpu_nice value=0
-- procstat_prefix_cpu_iowait value=0
+- procstat_[prefix_]cpu_iowait value=0
-- procstat_prefix_cpu_irq value=0
+- procstat_[prefix_]cpu_irq value=0
-- procstat_prefix_cpu_soft_irq value=0
+- procstat_[prefix_]cpu_soft_irq value=0
-- procstat_prefix_cpu_soft_steal value=0
+- procstat_[prefix_]cpu_soft_steal value=0
-- procstat_prefix_cpu_soft_stolen value=0
+- procstat_[prefix_]cpu_soft_stolen value=0
-- procstat_prefix_cpu_soft_guest value=0
+- procstat_[prefix_]cpu_soft_guest value=0
-- procstat_prefix_cpu_soft_guest_nice value=0
+- procstat_[prefix_]cpu_soft_guest_nice value=0
 
 Memory related measurement names:
-- procstat_prefix_memory_rss value=1777664
+- procstat_[prefix_]memory_rss value=1777664
-- procstat_prefix_memory_vms value=24227840
+- procstat_[prefix_]memory_vms value=24227840
-- procstat_prefix_memory_swap value=282624
+- procstat_[prefix_]memory_swap value=282624

plugins/procstat/procstat.go

@@ -3,6 +3,7 @@ package procstat
 import (
 	"fmt"
 	"io/ioutil"
+	"log"
 	"os/exec"
 	"strconv"
 	"strings"
@@ -29,7 +30,7 @@ func NewProcstat() *Procstat {
 
 var sampleConfig = `
 	[[procstat.specifications]]
-	prefix = "nginx" # required
+	prefix = "" # optional string to prefix measurements
 	# Use one of pid_file or exe to find process
 	pid_file = "/var/run/nginx.pid"
 	# executable name (used by pgrep)
@@ -46,39 +47,65 @@ func (_ *Procstat) Description() string {
 
 func (p *Procstat) Gather(acc plugins.Accumulator) error {
 	var wg sync.WaitGroup
-	var outerr error
+
 	for _, specification := range p.Specifications {
 		wg.Add(1)
 		go func(spec *Specification, acc plugins.Accumulator) {
 			defer wg.Done()
-			proc, err := spec.createProcess()
+			procs, err := spec.createProcesses()
 			if err != nil {
-				outerr = err
+				log.Printf("Error: procstat getting process, exe: %s, pidfile: %s, %s",
+					spec.Exe, spec.PidFile, err.Error())
 			} else {
-				outerr = NewSpecProcessor(spec.Prefix, acc, proc).pushMetrics()
+				for _, proc := range procs {
+					p := NewSpecProcessor(spec.Prefix, acc, proc)
+					p.pushMetrics()
+				}
 			}
 		}(specification, acc)
 	}
 	wg.Wait()
-	return outerr
+
+	return nil
 }
 
-func (spec *Specification) createProcess() (*process.Process, error) {
+func (spec *Specification) createProcesses() ([]*process.Process, error) {
+	var out []*process.Process
+	var errstring string
+	var outerr error
+	var pids []int32
+
 	if spec.PidFile != "" {
 		pid, err := pidFromFile(spec.PidFile)
 		if err != nil {
-			return nil, err
+			errstring += err.Error() + " "
-		}
-		return process.NewProcess(int32(pid))
-	} else if spec.Exe != "" {
-		pid, err := pidFromExe(spec.Exe)
-		if err != nil {
-			return nil, err
-		}
-		return process.NewProcess(int32(pid))
 		} else {
-		return nil, fmt.Errorf("Either exe or pid_file has to be specified")
+			pids = append(pids, int32(pid))
 		}
+	} else if spec.Exe != "" {
+		exepids, err := pidsFromExe(spec.Exe)
+		if err != nil {
+			errstring += err.Error() + " "
+		}
+		pids = append(pids, exepids...)
+	} else {
+		errstring += fmt.Sprintf("Either exe or pid_file has to be specified")
+	}
+
+	for _, pid := range pids {
+		p, err := process.NewProcess(int32(pid))
+		if err == nil {
+			out = append(out, p)
+		} else {
+			errstring += err.Error() + " "
+		}
+	}
+
+	if errstring != "" {
+		outerr = fmt.Errorf("%s", errstring)
+	}
+
+	return out, outerr
 }
 
 func pidFromFile(file string) (int, error) {
@@ -90,14 +117,25 @@ func pidFromFile(file string) (int, error) {
 	}
 }
 
-func pidFromExe(exe string) (int, error) {
+func pidsFromExe(exe string) ([]int32, error) {
-	pidString, err := exec.Command("pgrep", exe).Output()
+	var out []int32
+	var outerr error
+	pgrep, err := exec.Command("pgrep", exe).Output()
 	if err != nil {
-		return -1, fmt.Errorf("Failed to execute pgrep. Error: '%s'", err)
+		return out, fmt.Errorf("Failed to execute pgrep. Error: '%s'", err)
 	} else {
-		return strconv.Atoi(strings.TrimSpace(string(pidString)))
+		pids := strings.Fields(string(pgrep))
+		for _, pid := range pids {
+			ipid, err := strconv.Atoi(pid)
+			if err == nil {
+				out = append(out, int32(ipid))
+			} else {
+				outerr = err
 			}
 		}
+	}
+	return out, outerr
+}
 
 func init() {
 	plugins.Add("procstat", func() plugins.Plugin {

plugins/procstat/spec_processor.go

@@ -17,7 +17,13 @@ type SpecProcessor struct {
 }
 
 func (p *SpecProcessor) add(metric string, value interface{}) {
-	p.acc.Add(p.Prefix+"_"+metric, value, p.tags)
+	var mname string
+	if p.Prefix == "" {
+		mname = metric
+	} else {
+		mname = p.Prefix + "_" + metric
+	}
+	p.acc.Add(mname, value, p.tags)
 }
 
 func NewSpecProcessor(
@@ -27,6 +33,9 @@ func NewSpecProcessor(
 ) *SpecProcessor {
 	tags := make(map[string]string)
 	tags["pid"] = fmt.Sprintf("%v", p.Pid)
+	if name, err := p.Name(); err == nil {
+		tags["name"] = name
+	}
 	return &SpecProcessor{
 		Prefix: prefix,
 		tags:   tags,
@@ -35,23 +44,22 @@ func NewSpecProcessor(
 	}
 }
 
-func (p *SpecProcessor) pushMetrics() error {
+func (p *SpecProcessor) pushMetrics() {
 	if err := p.pushFDStats(); err != nil {
-		log.Printf(err.Error())
+		log.Printf("procstat, fd stats not available: %s", err.Error())
 	}
 	if err := p.pushCtxStats(); err != nil {
-		log.Printf(err.Error())
+		log.Printf("procstat, ctx stats not available: %s", err.Error())
 	}
 	if err := p.pushIOStats(); err != nil {
-		log.Printf(err.Error())
+		log.Printf("procstat, io stats not available: %s", err.Error())
 	}
 	if err := p.pushCPUStats(); err != nil {
-		log.Printf(err.Error())
+		log.Printf("procstat, cpu stats not available: %s", err.Error())
 	}
 	if err := p.pushMemoryStats(); err != nil {
-		log.Printf(err.Error())
+		log.Printf("procstat, mem stats not available: %s", err.Error())
 	}
-	return nil
 }
 
 func (p *SpecProcessor) pushFDStats() error {