func (a *Agent) once(ctx context.Context, wait time.Duration) error {
	log.Printf("D! [agent] Initializing plugins")
	err := a.initPlugins()
	if err != nil {
		return err
	}
	startTime := time.Now()
	log.Printf("D! [agent] Connecting outputs")
	next, ou, err := a.startOutputs(ctx, a.Config.Outputs)
	if err != nil {
		return err
	}
	var apu []*processorUnit
	var au *aggregatorUnit
	if len(a.Config.Aggregators) != 0 {
		procC := next
		if len(a.Config.AggProcessors) != 0 {
			procC, apu, err = a.startProcessors(next, a.Config.AggProcessors)
			if err != nil {
				return err
			}
		}
		next, au = a.startAggregators(procC, next, a.Config.Aggregators)
	}
	var pu []*processorUnit
	if len(a.Config.Processors) != 0 {
		next, pu, err = a.startProcessors(next, a.Config.Processors)
		if err != nil {
			return err
		}
	}
	iu := a.testStartInputs(next, a.Config.Inputs)
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		a.runOutputs(ou)
	}()
	if au != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runProcessors(apu)
		}()
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runAggregators(startTime, au)
		}()
	}
	if pu != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runProcessors(pu)
		}()
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		a.testRunInputs(ctx, wait, iu)
	}()
	wg.Wait()
	log.Printf("D! [agent] Stopped Successfully")
	return nil
}
On runs the agent and performs a single gather sending output to the
outputF.  After gathering pauses for the wait duration to allow service
inputs to run.
package agent
import (
	"context"
	"fmt"
	"log"
	"os"
	"runtime"
	"sort"
	"sync"
	"time"
	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/config"
	"github.com/influxdata/telegraf/internal"
	"github.com/influxdata/telegraf/models"
	"github.com/influxdata/telegraf/plugins/serializers/influx"
)
type Agent struct {
	Config *config.Config
}
func NewAgent(config *config.Config) (*Agent, error) {
	a := &Agent{
		Config: config,
	}
	return a, nil
}
type inputUnit struct {
	dst    chan<- telegraf.Metric
	inputs []*models.RunningInput
}
type processorUnit struct {
	src       <-chan telegraf.Metric
	dst       chan<- telegraf.Metric
	processor *models.RunningProcessor
}
type aggregatorUnit struct {
	src         <-chan telegraf.Metric
	aggC        chan<- telegraf.Metric
	outputC     chan<- telegraf.Metric
	aggregators []*models.RunningAggregator
}
type outputUnit struct {
	src     <-chan telegraf.Metric
	outputs []*models.RunningOutput
}
func (a *Agent) Run(ctx context.Context) error {
	log.Printf("I! [agent] Config: Interval:%s, Quiet:%#v, Hostname:%#v, "+
		"Flush Interval:%s",
		time.Duration(a.Config.Agent.Interval), a.Config.Agent.Quiet,
		a.Config.Agent.Hostname, time.Duration(a.Config.Agent.FlushInterval))
	log.Printf("D! [agent] Initializing plugins")
	err := a.initPlugins()
	if err != nil {
		return err
	}
	startTime := time.Now()
	log.Printf("D! [agent] Connecting outputs")
	next, ou, err := a.startOutputs(ctx, a.Config.Outputs)
	if err != nil {
		return err
	}
	var apu []*processorUnit
	var au *aggregatorUnit
	if len(a.Config.Aggregators) != 0 {
		aggC := next
		if len(a.Config.AggProcessors) != 0 {
			aggC, apu, err = a.startProcessors(next, a.Config.AggProcessors)
			if err != nil {
				return err
			}
		}
		next, au = a.startAggregators(aggC, next, a.Config.Aggregators)
	}
	var pu []*processorUnit
	if len(a.Config.Processors) != 0 {
		next, pu, err = a.startProcessors(next, a.Config.Processors)
		if err != nil {
			return err
		}
	}
	iu, err := a.startInputs(next, a.Config.Inputs)
	if err != nil {
		return err
	}
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		a.runOutputs(ou)
	}()
	if au != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runProcessors(apu)
		}()
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runAggregators(startTime, au)
		}()
	}
	if pu != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runProcessors(pu)
		}()
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		a.runInputs(ctx, startTime, iu)
	}()
	wg.Wait()
	log.Printf("D! [agent] Stopped Successfully")
	return err
}
func (a *Agent) initPlugins() error {
	for _, input := range a.Config.Inputs {
		err := input.Init()
		if err != nil {
			return fmt.Errorf("could not initialize input %s: %v",
				input.LogName(), err)
		}
	}
	for _, processor := range a.Config.Processors {
		err := processor.Init()
		if err != nil {
			return fmt.Errorf("could not initialize processor %s: %v",
				processor.Config.Name, err)
		}
	}
	for _, aggregator := range a.Config.Aggregators {
		err := aggregator.Init()
		if err != nil {
			return fmt.Errorf("could not initialize aggregator %s: %v",
				aggregator.Config.Name, err)
		}
	}
	for _, processor := range a.Config.AggProcessors {
		err := processor.Init()
		if err != nil {
			return fmt.Errorf("could not initialize processor %s: %v",
				processor.Config.Name, err)
		}
	}
	for _, output := range a.Config.Outputs {
		err := output.Init()
		if err != nil {
			return fmt.Errorf("could not initialize output %s: %v",
				output.Config.Name, err)
		}
	}
	return nil
}
func (a *Agent) startInputs(
	dst chan<- telegraf.Metric,
	inputs []*models.RunningInput,
) (*inputUnit, error) {
	log.Printf("D! [agent] Starting service inputs")
	unit := &inputUnit{
		dst: dst,
	}
	for _, input := range inputs {
		if si, ok := input.Input.(telegraf.ServiceInput); ok {
			var interval time.Duration
			var precision time.Duration
			if input.Config.Precision != 0 {
				precision = input.Config.Precision
			}
			acc := NewAccumulator(input, dst)
			acc.SetPrecision(getPrecision(precision, interval))
			err := si.Start(acc)
			if err != nil {
				stopServiceInputs(unit.inputs)
				return nil, fmt.Errorf("starting input %s: %w", input.LogName(), err)
			}
		}
		unit.inputs = append(unit.inputs, input)
	}
	return unit, nil
}
func (a *Agent) runInputs(
	ctx context.Context,
	startTime time.Time,
	unit *inputUnit,
) {
	var wg sync.WaitGroup
	for _, input := range unit.inputs {
		interval := time.Duration(a.Config.Agent.Interval)
		if input.Config.Interval != 0 {
			interval = input.Config.Interval
		}
		precision := time.Duration(a.Config.Agent.Precision)
		if input.Config.Precision != 0 {
			precision = input.Config.Precision
		}
		jitter := time.Duration(a.Config.Agent.CollectionJitter)
		if input.Config.CollectionJitter != 0 {
			jitter = input.Config.CollectionJitter
		}
		var ticker Ticker
		if a.Config.Agent.RoundInterval {
			ticker = NewAlignedTicker(startTime, interval, jitter)
		} else {
			ticker = NewUnalignedTicker(interval, jitter)
		}
		defer ticker.Stop()
		acc := NewAccumulator(input, unit.dst)
		acc.SetPrecision(getPrecision(precision, interval))
		wg.Add(1)
		go func(input *models.RunningInput) {
			defer wg.Done()
			a.gatherLoop(ctx, acc, input, ticker, interval)
		}(input)
	}
	wg.Wait()
	log.Printf("D! [agent] Stopping service inputs")
	stopServiceInputs(unit.inputs)
	close(unit.dst)
	log.Printf("D! [agent] Input channel closed")
}
func (a *Agent) testStartInputs(
	dst chan<- telegraf.Metric,
	inputs []*models.RunningInput,
) *inputUnit {
	log.Printf("D! [agent] Starting service inputs")
	unit := &inputUnit{
		dst: dst,
	}
	for _, input := range inputs {
		if si, ok := input.Input.(telegraf.ServiceInput); ok {
			acc := NewAccumulator(input, dst)
			acc.SetPrecision(time.Nanosecond)
			err := si.Start(acc)
			if err != nil {
				log.Printf("E! [agent] Starting input %s: %v", input.LogName(), err)
			}
		}
		unit.inputs = append(unit.inputs, input)
	}
	return unit
}
func (a *Agent) testRunInputs(
	ctx context.Context,
	wait time.Duration,
	unit *inputUnit,
) {
	var wg sync.WaitGroup
	nul := make(chan telegraf.Metric)
	go func() {
		for range nul {
		}
	}()
	for _, input := range unit.inputs {
		wg.Add(1)
		go func(input *models.RunningInput) {
			defer wg.Done()
			interval := time.Duration(a.Config.Agent.Interval)
			if input.Config.Interval != 0 {
				interval = input.Config.Interval
			}
			precision := time.Duration(a.Config.Agent.Precision)
			if input.Config.Precision != 0 {
				precision = input.Config.Precision
			}
			switch input.Config.Name {
			case "cpu", "mongodb", "procstat":
				nulAcc := NewAccumulator(input, nul)
				nulAcc.SetPrecision(getPrecision(precision, interval))
				if err := input.Input.Gather(nulAcc); err != nil {
					nulAcc.AddError(err)
				}
				time.Sleep(500 * time.Millisecond)
			}
			acc := NewAccumulator(input, unit.dst)
			acc.SetPrecision(getPrecision(precision, interval))
			if err := input.Input.Gather(acc); err != nil {
				acc.AddError(err)
			}
		}(input)
	}
	wg.Wait()
	internal.SleepContext(ctx, wait)
	log.Printf("D! [agent] Stopping service inputs")
	stopServiceInputs(unit.inputs)
	close(unit.dst)
	log.Printf("D! [agent] Input channel closed")
}
func stopServiceInputs(inputs []*models.RunningInput) {
	for _, input := range inputs {
		if si, ok := input.Input.(telegraf.ServiceInput); ok {
			si.Stop()
		}
	}
}
func stopRunningOutputs(outputs []*models.RunningOutput) {
	for _, output := range outputs {
		output.Close()
	}
}
func (a *Agent) gatherLoop(
	ctx context.Context,
	acc telegraf.Accumulator,
	input *models.RunningInput,
	ticker Ticker,
	interval time.Duration,
) {
	defer panicRecover(input)
	for {
		select {
		case <-ticker.Elapsed():
			err := a.gatherOnce(acc, input, ticker, interval)
			if err != nil {
				acc.AddError(err)
			}
		case <-ctx.Done():
			return
		}
	}
}
func (a *Agent) gatherOnce(
	acc telegraf.Accumulator,
	input *models.RunningInput,
	ticker Ticker,
	interval time.Duration,
) error {
	done := make(chan error)
	go func() {
		done <- input.Gather(acc)
	}()
	slowWarning := time.NewTicker(interval)
	defer slowWarning.Stop()
	for {
		select {
		case err := <-done:
			return err
		case <-slowWarning.C:
			log.Printf("W! [%s] Collection took longer than expected; not complete after interval of %s",
				input.LogName(), interval)
		case <-ticker.Elapsed():
			log.Printf("D! [%s] Previous collection has not completed; scheduled collection skipped",
				input.LogName())
		}
	}
}
func (a *Agent) startProcessors(
	dst chan<- telegraf.Metric,
	processors models.RunningProcessors,
) (chan<- telegraf.Metric, []*processorUnit, error) {
	var units []*processorUnit
	sort.SliceStable(processors, func(i, j int) bool {
		return processors[i].Config.Order > processors[j].Config.Order
	})
	var src chan telegraf.Metric
	for _, processor := range processors {
		src = make(chan telegraf.Metric, 100)
		acc := NewAccumulator(processor, dst)
		err := processor.Start(acc)
		if err != nil {
			for _, u := range units {
				u.processor.Stop()
				close(u.dst)
			}
			return nil, nil, fmt.Errorf("starting processor %s: %w", processor.LogName(), err)
		}
		units = append(units, &processorUnit{
			src:       src,
			dst:       dst,
			processor: processor,
		})
		dst = src
	}
	return src, units, nil
}
func (a *Agent) runProcessors(
	units []*processorUnit,
) {
	var wg sync.WaitGroup
	for _, unit := range units {
		wg.Add(1)
		go func(unit *processorUnit) {
			defer wg.Done()
			acc := NewAccumulator(unit.processor, unit.dst)
			for m := range unit.src {
				if err := unit.processor.Add(m, acc); err != nil {
					acc.AddError(err)
					m.Drop()
				}
			}
			unit.processor.Stop()
			close(unit.dst)
			log.Printf("D! [agent] Processor channel closed")
		}(unit)
	}
	wg.Wait()
}
func (a *Agent) startAggregators(
	aggC chan<- telegraf.Metric,
	outputC chan<- telegraf.Metric,
	aggregators []*models.RunningAggregator,
) (chan<- telegraf.Metric, *aggregatorUnit) {
	src := make(chan telegraf.Metric, 100)
	unit := &aggregatorUnit{
		src:         src,
		aggC:        aggC,
		outputC:     outputC,
		aggregators: aggregators,
	}
	return src, unit
}
func (a *Agent) runAggregators(
	startTime time.Time,
	unit *aggregatorUnit,
) {
	ctx, cancel := context.WithCancel(context.Background())
	for _, agg := range a.Config.Aggregators {
		since, until := updateWindow(startTime, a.Config.Agent.RoundInterval, agg.Period())
		agg.UpdateWindow(since, until)
	}
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for metric := range unit.src {
			var dropOriginal bool
			for _, agg := range a.Config.Aggregators {
				if ok := agg.Add(metric); ok {
					dropOriginal = true
				}
			}
			if !dropOriginal {
				unit.outputC <- metric  
			} else {
				metric.Drop()
			}
		}
		cancel()
	}()
	for _, agg := range a.Config.Aggregators {
		wg.Add(1)
		go func(agg *models.RunningAggregator) {
			defer wg.Done()
			interval := time.Duration(a.Config.Agent.Interval)
			precision := time.Duration(a.Config.Agent.Precision)
			acc := NewAccumulator(agg, unit.aggC)
			acc.SetPrecision(getPrecision(precision, interval))
			a.push(ctx, agg, acc)
		}(agg)
	}
	wg.Wait()
	close(unit.aggC)
	log.Printf("D! [agent] Aggregator channel closed")
}
func updateWindow(start time.Time, roundInterval bool, period time.Duration) (time.Time, time.Time) {
	var until time.Time
	if roundInterval {
		until = internal.AlignTime(start, period)
		if until == start {
			until = internal.AlignTime(start.Add(time.Nanosecond), period)
		}
	} else {
		until = start.Add(period)
	}
	since := until.Add(-period)
	return since, until
}
func (a *Agent) push(
	ctx context.Context,
	aggregator *models.RunningAggregator,
	acc telegraf.Accumulator,
) {
	for {
		until := time.Until(aggregator.EndPeriod())
		select {
		case <-time.After(until):
			aggregator.Push(acc)
			break
		case <-ctx.Done():
			aggregator.Push(acc)
			return
		}
	}
}
func (a *Agent) startOutputs(
	ctx context.Context,
	outputs []*models.RunningOutput,
) (chan<- telegraf.Metric, *outputUnit, error) {
	src := make(chan telegraf.Metric, 100)
	unit := &outputUnit{src: src}
	for _, output := range outputs {
		err := a.connectOutput(ctx, output)
		if err != nil {
			for _, output := range unit.outputs {
				output.Close()
			}
			return nil, nil, fmt.Errorf("connecting output %s: %w", output.LogName(), err)
		}
		unit.outputs = append(unit.outputs, output)
	}
	return src, unit, nil
}
func (a *Agent) connectOutput(ctx context.Context, output *models.RunningOutput) error {
	log.Printf("D! [agent] Attempting connection to [%s]", output.LogName())
	err := output.Output.Connect()
	if err != nil {
		log.Printf("E! [agent] Failed to connect to [%s], retrying in 15s, "+
			"error was '%s'", output.LogName(), err)
		err := internal.SleepContext(ctx, 15*time.Second)
		if err != nil {
			return err
		}
		err = output.Output.Connect()
		if err != nil {
			return fmt.Errorf("Error connecting to output %q: %w", output.LogName(), err)
		}
	}
	log.Printf("D! [agent] Successfully connected to %s", output.LogName())
	return nil
}
func (a *Agent) runOutputs(
	unit *outputUnit,
) {
	var wg sync.WaitGroup
	interval := time.Duration(a.Config.Agent.FlushInterval)
	jitter := time.Duration(a.Config.Agent.FlushJitter)
	ctx, cancel := context.WithCancel(context.Background())
	for _, output := range unit.outputs {
		interval := interval
		if output.Config.FlushInterval != 0 {
			interval = output.Config.FlushInterval
		}
		jitter := jitter
		if output.Config.FlushJitter != 0 {
			jitter = output.Config.FlushJitter
		}
		wg.Add(1)
		go func(output *models.RunningOutput) {
			defer wg.Done()
			ticker := NewRollingTicker(interval, jitter)
			defer ticker.Stop()
			a.flushLoop(ctx, output, ticker)
		}(output)
	}
	for metric := range unit.src {
		for i, output := range unit.outputs {
			if i == len(a.Config.Outputs)-1 {
				output.AddMetric(metric)
			} else {
				output.AddMetric(metric.Copy())
			}
		}
	}
	log.Println("I! [agent] Hang on, flushing any cached metrics before shutdown")
	cancel()
	wg.Wait()
	log.Println("I! [agent] Stopping running outputs")
	stopRunningOutputs(unit.outputs)
}
func (a *Agent) flushLoop(
	ctx context.Context,
	output *models.RunningOutput,
	ticker Ticker,
) {
	logError := func(err error) {
		if err != nil {
			log.Printf("E! [agent] Error writing to %s: %v", output.LogName(), err)
		}
	}
	flushRequested := make(chan os.Signal, 1)
	watchForFlushSignal(flushRequested)
	defer stopListeningForFlushSignal(flushRequested)
	for {
		select {
		case <-ctx.Done():
			logError(a.flushOnce(output, ticker, output.Write))
			return
		default:
		}
		select {
		case <-ctx.Done():
			logError(a.flushOnce(output, ticker, output.Write))
			return
		case <-ticker.Elapsed():
			logError(a.flushOnce(output, ticker, output.Write))
		case <-flushRequested:
			logError(a.flushOnce(output, ticker, output.Write))
		case <-output.BatchReady:
			select {
			case <-ticker.Elapsed():
				logError(a.flushOnce(output, ticker, output.Write))
			default:
				logError(a.flushOnce(output, ticker, output.WriteBatch))
			}
		}
	}
}
func (a *Agent) flushOnce(
	output *models.RunningOutput,
	ticker Ticker,
	writeFunc func() error,
) error {
	done := make(chan error)
	go func() {
		done <- writeFunc()
	}()
	for {
		select {
		case err := <-done:
			output.LogBufferStatus()
			return err
		case <-ticker.Elapsed():
			log.Printf("W! [agent] [%q] did not complete within its flush interval",
				output.LogName())
			output.LogBufferStatus()
		}
	}
}
func (a *Agent) Test(ctx context.Context, wait time.Duration) error {
	src := make(chan telegraf.Metric, 100)
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		s := influx.NewSerializer()
		s.SetFieldSortOrder(influx.SortFields)
		for metric := range src {
			octets, err := s.Serialize(metric)
			if err == nil {
				fmt.Print("> ", string(octets))
			}
			metric.Reject()
		}
	}()
	err := a.test(ctx, wait, src)
	if err != nil {
		return err
	}
	wg.Wait()
	if models.GlobalGatherErrors.Get() != 0 {
		return fmt.Errorf("input plugins recorded %d errors", models.GlobalGatherErrors.Get())
	}
	return nil
}
func (a *Agent) test(ctx context.Context, wait time.Duration, outputC chan<- telegraf.Metric) error {
	log.Printf("D! [agent] Initializing plugins")
	err := a.initPlugins()
	if err != nil {
		return err
	}
	startTime := time.Now()
	next := outputC
	var apu []*processorUnit
	var au *aggregatorUnit
	if len(a.Config.Aggregators) != 0 {
		procC := next
		if len(a.Config.AggProcessors) != 0 {
			procC, apu, err = a.startProcessors(next, a.Config.AggProcessors)
			if err != nil {
				return err
			}
		}
		next, au = a.startAggregators(procC, next, a.Config.Aggregators)
	}
	var pu []*processorUnit
	if len(a.Config.Processors) != 0 {
		next, pu, err = a.startProcessors(next, a.Config.Processors)
		if err != nil {
			return err
		}
	}
	iu := a.testStartInputs(next, a.Config.Inputs)
	var wg sync.WaitGroup
	if au != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runProcessors(apu)
		}()
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runAggregators(startTime, au)
		}()
	}
	if pu != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			a.runProcessors(pu)
		}()
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		a.testRunInputs(ctx, wait, iu)
	}()
	wg.Wait()
	log.Printf("D! [agent] Stopped Successfully")
	return nil
}
func (a *Agent) Once(ctx context.Context, wait time.Duration) error {
	err := a.once(ctx, wait)
	if err != nil {
		return err
	}
	if models.GlobalGatherErrors.Get() != 0 {
		return fmt.Errorf("input plugins recorded %d errors", models.GlobalGatherErrors.Get())
	}
	unsent := 0
	for _, output := range a.Config.Outputs {
		unsent += output.BufferLength()
	}
	if unsent != 0 {
		return fmt.Errorf("output plugins unable to send %d metrics", unsent)
	}
	return nil
}