telegraf/plugins/inputs/exec/exec.go

233 lines
5.0 KiB
Go
Raw Permalink Normal View History

2015-08-06 00:29:27 +00:00
package exec
import (
"bytes"
"fmt"
"os/exec"
"path/filepath"
"runtime"
"strings"
"sync"
"syscall"
"time"
"github.com/kballard/go-shellquote"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/internal/errchan"
2016-01-20 18:57:35 +00:00
"github.com/influxdata/telegraf/plugins/inputs"
"github.com/influxdata/telegraf/plugins/parsers"
"github.com/influxdata/telegraf/plugins/parsers/nagios"
2015-08-06 00:29:27 +00:00
)
const sampleConfig = `
## Commands array
commands = [
"/tmp/test.sh",
"/usr/bin/mycollector --foo=bar",
"/tmp/collect_*.sh"
]
## Timeout for each command to complete.
timeout = "5s"
## measurement name suffix (for separating different commands)
2016-01-07 20:39:43 +00:00
name_suffix = "_mycollector"
## Data format to consume.
## Each data format has it's own unique set of configuration options, read
## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
data_format = "influx"
2015-09-23 18:21:42 +00:00
`
2015-08-06 00:29:27 +00:00
type Exec struct {
Commands []string
Command string
Timeout internal.Duration
parser parsers.Parser
runner Runner
errChan chan error
}
func NewExec() *Exec {
return &Exec{
runner: CommandRunner{},
Timeout: internal.Duration{Duration: time.Second * 5},
}
2015-08-06 00:29:27 +00:00
}
2015-09-23 18:21:42 +00:00
type Runner interface {
Run(*Exec, string, telegraf.Accumulator) ([]byte, error)
2015-08-06 00:29:27 +00:00
}
2015-09-23 18:21:42 +00:00
type CommandRunner struct{}
func AddNagiosState(exitCode error, acc telegraf.Accumulator) error {
nagiosState := 0
if exitCode != nil {
exiterr, ok := exitCode.(*exec.ExitError)
if ok {
status, ok := exiterr.Sys().(syscall.WaitStatus)
if ok {
nagiosState = status.ExitStatus()
} else {
return fmt.Errorf("exec: unable to get nagios plugin exit code")
}
} else {
return fmt.Errorf("exec: unable to get nagios plugin exit code")
}
}
fields := map[string]interface{}{"state": nagiosState}
acc.AddFields("nagios_state", fields, nil)
return nil
}
func (c CommandRunner) Run(
e *Exec,
command string,
acc telegraf.Accumulator,
) ([]byte, error) {
split_cmd, err := shellquote.Split(command)
2015-09-23 18:21:42 +00:00
if err != nil || len(split_cmd) == 0 {
return nil, fmt.Errorf("exec: unable to parse command, %s", err)
}
cmd := exec.Command(split_cmd[0], split_cmd[1:]...)
2015-08-06 00:29:27 +00:00
var out bytes.Buffer
cmd.Stdout = &out
if err := internal.RunTimeout(cmd, e.Timeout.Duration); err != nil {
switch e.parser.(type) {
case *nagios.NagiosParser:
AddNagiosState(err, acc)
default:
return nil, fmt.Errorf("exec: %s for command '%s'", err, command)
}
} else {
switch e.parser.(type) {
case *nagios.NagiosParser:
AddNagiosState(nil, acc)
}
2015-08-06 00:29:27 +00:00
}
out = removeCarriageReturns(out)
2015-08-06 00:29:27 +00:00
return out.Bytes(), nil
}
// removeCarriageReturns removes all carriage returns from the input if the
// OS is Windows. It does not return any errors.
func removeCarriageReturns(b bytes.Buffer) bytes.Buffer {
if runtime.GOOS == "windows" {
var buf bytes.Buffer
for {
byt, er := b.ReadBytes(0x0D)
end := len(byt)
if nil == er {
end -= 1
}
if nil != byt {
buf.Write(byt[:end])
} else {
break
}
if nil != er {
break
}
}
b = buf
}
return b
}
func (e *Exec) ProcessCommand(command string, acc telegraf.Accumulator, wg *sync.WaitGroup) {
defer wg.Done()
out, err := e.runner.Run(e, command, acc)
if err != nil {
e.errChan <- err
return
}
metrics, err := e.parser.Parse(out)
if err != nil {
e.errChan <- err
} else {
for _, metric := range metrics {
acc.AddFields(metric.Name(), metric.Fields(), metric.Tags(), metric.Time())
}
}
}
2015-08-06 00:29:27 +00:00
func (e *Exec) SampleConfig() string {
return sampleConfig
}
func (e *Exec) Description() string {
return "Read metrics from one or more commands that can output to stdout"
2015-08-06 00:29:27 +00:00
}
func (e *Exec) SetParser(parser parsers.Parser) {
e.parser = parser
}
func (e *Exec) Gather(acc telegraf.Accumulator) error {
var wg sync.WaitGroup
// Legacy single command support
if e.Command != "" {
e.Commands = append(e.Commands, e.Command)
e.Command = ""
}
commands := make([]string, 0, len(e.Commands))
for _, pattern := range e.Commands {
cmdAndArgs := strings.SplitN(pattern, " ", 2)
if len(cmdAndArgs) == 0 {
continue
}
matches, err := filepath.Glob(cmdAndArgs[0])
if err != nil {
return err
}
if len(matches) == 0 {
// There were no matches with the glob pattern, so let's assume
// that the command is in PATH and just run it as it is
commands = append(commands, pattern)
} else {
// There were matches, so we'll append each match together with
// the arguments to the commands slice
for _, match := range matches {
if len(cmdAndArgs) == 1 {
commands = append(commands, match)
} else {
commands = append(commands,
strings.Join([]string{match, cmdAndArgs[1]}, " "))
}
}
}
}
errChan := errchan.New(len(commands))
e.errChan = errChan.C
wg.Add(len(commands))
for _, command := range commands {
go e.ProcessCommand(command, acc, &wg)
}
wg.Wait()
return errChan.Error()
2015-08-06 00:29:27 +00:00
}
func init() {
inputs.Add("exec", func() telegraf.Input {
2015-08-06 00:29:27 +00:00
return NewExec()
})
}