Run ServiceInputs during test mode; add --test-wait option (#5911)

This commit is contained in:
Daniel Nelson 2019-06-14 12:06:25 -07:00 committed by GitHub
parent fec1b3ec19
commit de096428be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 56 additions and 31 deletions

View File

@ -136,7 +136,7 @@ func (a *Agent) Run(ctx context.Context) error {
} }
// Test runs the inputs once and prints the output to stdout in line protocol. // Test runs the inputs once and prints the output to stdout in line protocol.
func (a *Agent) Test(ctx context.Context) error { func (a *Agent) Test(ctx context.Context, waitDuration time.Duration) error {
var wg sync.WaitGroup var wg sync.WaitGroup
metricC := make(chan telegraf.Metric) metricC := make(chan telegraf.Metric)
nulC := make(chan telegraf.Metric) nulC := make(chan telegraf.Metric)
@ -156,8 +156,8 @@ func (a *Agent) Test(ctx context.Context) error {
octets, err := s.Serialize(metric) octets, err := s.Serialize(metric)
if err == nil { if err == nil {
fmt.Print("> ", string(octets)) fmt.Print("> ", string(octets))
} }
metric.Reject()
} }
}() }()
@ -168,20 +168,32 @@ func (a *Agent) Test(ctx context.Context) error {
} }
}() }()
hasServiceInputs := false
for _, input := range a.Config.Inputs {
if _, ok := input.Input.(telegraf.ServiceInput); ok {
hasServiceInputs = true
break
}
}
if hasServiceInputs {
log.Printf("D! [agent] Starting service inputs")
err := a.startServiceInputs(ctx, metricC)
if err != nil {
return err
}
}
for _, input := range a.Config.Inputs { for _, input := range a.Config.Inputs {
select { select {
case <-ctx.Done(): case <-ctx.Done():
return nil return nil
default: default:
if _, ok := input.Input.(telegraf.ServiceInput); ok { break
log.Printf("W!: [agent] skipping plugin [[%s]]: service inputs not supported in --test mode",
input.Name())
continue
} }
acc := NewAccumulator(input, metricC) acc := NewAccumulator(input, metricC)
acc.SetPrecision(a.Precision()) acc.SetPrecision(a.Precision())
input.SetDefaultTags(a.Config.Tags)
// Special instructions for some inputs. cpu, for example, needs to be // Special instructions for some inputs. cpu, for example, needs to be
// run twice in order to return cpu usage percentages. // run twice in order to return cpu usage percentages.
@ -190,19 +202,25 @@ func (a *Agent) Test(ctx context.Context) error {
nulAcc := NewAccumulator(input, nulC) nulAcc := NewAccumulator(input, nulC)
nulAcc.SetPrecision(a.Precision()) nulAcc.SetPrecision(a.Precision())
if err := input.Input.Gather(nulAcc); err != nil { if err := input.Input.Gather(nulAcc); err != nil {
return err acc.AddError(err)
} }
time.Sleep(500 * time.Millisecond) time.Sleep(500 * time.Millisecond)
if err := input.Input.Gather(acc); err != nil { if err := input.Input.Gather(acc); err != nil {
return err acc.AddError(err)
} }
default: default:
if err := input.Input.Gather(acc); err != nil { if err := input.Input.Gather(acc); err != nil {
return err acc.AddError(err)
} }
} }
} }
if hasServiceInputs {
log.Printf("D! [agent] Waiting for service inputs")
internal.SleepContext(ctx, waitDuration)
log.Printf("D! [agent] Stopping service inputs")
a.stopServiceInputs()
} }
return nil return nil

View File

@ -13,6 +13,7 @@ import (
"runtime" "runtime"
"strings" "strings"
"syscall" "syscall"
"time"
"github.com/influxdata/telegraf/agent" "github.com/influxdata/telegraf/agent"
"github.com/influxdata/telegraf/internal" "github.com/influxdata/telegraf/internal"
@ -33,7 +34,8 @@ var pprofAddr = flag.String("pprof-addr", "",
"pprof address to listen on, not activate pprof if empty") "pprof address to listen on, not activate pprof if empty")
var fQuiet = flag.Bool("quiet", false, var fQuiet = flag.Bool("quiet", false,
"run in quiet mode") "run in quiet mode")
var fTest = flag.Bool("test", false, "gather metrics, print them out, and exit") var fTest = flag.Bool("test", false, "enable test mode: gather metrics, print them out, and exit")
var fTestWait = flag.Int("test-wait", 0, "wait up to this many seconds for service inputs to complete in test mode")
var fConfig = flag.String("config", "", "configuration file to load") var fConfig = flag.String("config", "", "configuration file to load")
var fConfigDirectory = flag.String("config-directory", "", var fConfigDirectory = flag.String("config-directory", "",
"directory containing additional *.conf files") "directory containing additional *.conf files")
@ -167,8 +169,9 @@ func runAgent(ctx context.Context,
logger.SetupLogging(logConfig) logger.SetupLogging(logConfig)
if *fTest { if *fTest || *fTestWait != 0 {
return ag.Test(ctx) testWaitDuration := time.Duration(*fTestWait) * time.Second
return ag.Test(ctx, testWaitDuration)
} }
log.Printf("I! Loaded inputs: %s", strings.Join(c.InputNames(), " ")) log.Printf("I! Loaded inputs: %s", strings.Join(c.InputNames(), " "))

View File

@ -31,6 +31,8 @@ The commands & flags are:
--sample-config print out full sample configuration --sample-config print out full sample configuration
--test gather metrics, print them out, and exit; --test gather metrics, print them out, and exit;
processors, aggregators, and outputs are not run processors, aggregators, and outputs are not run
--test-wait wait up to this many seconds for service
inputs to complete in test mode
--usage <plugin> print usage for a plugin, ie, 'telegraf --usage mysql' --usage <plugin> print usage for a plugin, ie, 'telegraf --usage mysql'
--version display the version and exit --version display the version and exit

View File

@ -31,6 +31,8 @@ The commands & flags are:
'processors', 'aggregators' and 'inputs' 'processors', 'aggregators' and 'inputs'
--test gather metrics, print them out, and exit; --test gather metrics, print them out, and exit;
processors, aggregators, and outputs are not run processors, aggregators, and outputs are not run
--test-wait wait up to this many seconds for service
inputs to complete in test mode
--usage <plugin> print usage for a plugin, ie, 'telegraf --usage mysql' --usage <plugin> print usage for a plugin, ie, 'telegraf --usage mysql'
--version display the version and exit --version display the version and exit