func (a *Agent) Run(shutdown chan struct{}) error {
	var wg sync.WaitGroup

	log.Printf("I! Agent Config: Interval:%s, Quiet:%#v, Hostname:%#v, "+
		"Flush Interval:%s \n",
		a.Config.Agent.Interval.Duration, a.Config.Agent.Quiet,
		a.Config.Agent.Hostname, a.Config.Agent.FlushInterval.Duration)

	// channel shared between all input threads for accumulating metrics
	metricC := make(chan telegraf.Metric, 100)
	aggC := make(chan telegraf.Metric, 100)

	// Start all ServicePlugins
	for _, input := range a.Config.Inputs {
		input.SetDefaultTags(a.Config.Tags)
		switch p := input.Input.(type) {
		case telegraf.ServiceInput:
			acc := NewAccumulator(input, metricC)
			// Service input plugins should set their own precision of their
			// metrics.
			acc.SetPrecision(time.Nanosecond, 0)
			if err := p.Start(acc); err != nil {
				log.Printf("E! Service for input %s failed to start, exiting\n%s\n",
					input.Name(), err.Error())
				return err
			}
			defer p.Stop()
		}
	}

	// Round collection to nearest interval by sleeping
	if a.Config.Agent.RoundInterval {
		i := int64(a.Config.Agent.Interval.Duration)
		time.Sleep(time.Duration(i - (time.Now().UnixNano() % i)))
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := a.flusher(shutdown, metricC, aggC); err != nil {
			log.Printf("E! Flusher routine failed, exiting: %s\n", err.Error())
			close(shutdown)
		}
	}()

	wg.Add(len(a.Config.Aggregators))
	for _, aggregator := range a.Config.Aggregators {
		go func(agg *models.RunningAggregator) {
			defer wg.Done()
			acc := NewAccumulator(agg, aggC)
			acc.SetPrecision(a.Config.Agent.Precision.Duration,
				a.Config.Agent.Interval.Duration)
			agg.Run(acc, shutdown)
		}(aggregator)
	}

	wg.Add(len(a.Config.Inputs))
	for _, input := range a.Config.Inputs {
		interval := a.Config.Agent.Interval.Duration
		// overwrite global interval if this plugin has it's own.
		if input.Config.Interval != 0 {
			interval = input.Config.Interval
		}
		go func(in *models.RunningInput, interv time.Duration) {
			defer wg.Done()
			a.gatherer(shutdown, in, interv, metricC)
		}(input, interval)
	}

	wg.Wait()
	a.Close()
	return nil
}
// Run runs the agent daemon, gathering every Interval
Run runs the agent daemon, gathering every Interval
// Run runs the agent daemon gathering every Interval
func (a *Agent) Run(shutdown chan struct{}) error {
	var wg sync.WaitGroup
	log.Printf("I! Agent Config: Interval:%s, Quiet:%#v, Hostname:%#v, "+
		"Flush Interval:%s \n",
		a.Config.Agent.Interval.Duration, a.Config.Agent.Quiet,
		a.Config.Agent.Hostname, a.Config.Agent.FlushInterval.Duration)
	metricC := make(chan telegraf.Metric, 100)
	aggC := make(chan telegraf.Metric, 100)
	for _, input := range a.Config.Inputs {
		input.SetDefaultTags(a.Config.Tags)
		switch p := input.Input.(type) {
		case telegraf.ServiceInput:
			acc := NewAccumulator(input, metricC)
			acc.SetPrecision(time.Nanosecond, 0)
			if err := p.Start(acc); err != nil {
				log.Printf("E! Service for input %s failed to start, exiting\n%s\n",
					input.Name(), err.Error())
				return err
			}
			defer p.Stop()
		}
	}
	if a.Config.Agent.RoundInterval {
		i := int64(a.Config.Agent.Interval.Duration)
		time.Sleep(time.Duration(i - (time.Now().UnixNano() % i)))
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := a.flusher(shutdown, metricC, aggC); err != nil {
			log.Printf("E! Flusher routine failed, exiting: %s\n", err.Error())
			close(shutdown)
		}
	}()
	wg.Add(len(a.Config.Aggregators))
	for _, aggregator := range a.Config.Aggregators {
		go func(agg *models.RunningAggregator) {
			defer wg.Done()
			acc := NewAccumulator(agg, aggC)
			acc.SetPrecision(a.Config.Agent.Precision.Duration,
				a.Config.Agent.Interval.Duration)
			agg.Run(acc, shutdown)
		}(aggregator)
	}
	wg.Add(len(a.Config.Inputs))
	for _, input := range a.Config.Inputs {
		interval := a.Config.Agent.Interval.Duration
		if input.Config.Interval != 0 {
			interval = input.Config.Interval
		}
		go func(in *models.RunningInput, interv time.Duration) {
			defer wg.Done()
			a.gatherer(shutdown, in, interv, metricC)
		}(input, interval)
	}
	wg.Wait()
	a.Close()
	return nil
}
func ( a * Agent ) Run ( shutdown chan struct { } ) error { var wg sync . WaitGroup log . Printf ( " " + " \n " , a . Config . Agent . Interval . Duration , a . Config . Agent . Quiet , a . Config . Agent . Hostname , a . Config . Agent . FlushInterval . Duration ) metricC := make ( chan telegraf . Metric , 100 ) aggC := make ( chan telegraf . Metric , 100 ) for _ , input := range a . Config . Inputs { input . SetDefaultTags ( a . Config . Tags ) switch p := input . Input . ( type ) { case telegraf . ServiceInput : acc := NewAccumulator ( input , metricC ) acc . SetPrecision ( time . Nanosecond , 0 ) if err := p . Start ( acc ) ; err != nil { log . Printf ( " \n \n " , input . Name ( ) , err . Error ( ) ) return err } defer p . Stop ( ) } } if a . Config . Agent . RoundInterval { i := int64 ( a . Config . Agent . Interval . Duration ) time . Sleep ( time . Duration ( i - ( time . Now ( ) . UnixNano ( ) % i ) ) ) } wg . Add ( 1 ) go func ( ) { defer wg . Done ( ) if err := a . flusher ( shutdown , metricC , aggC ) ; err != nil { log . Printf ( " \n " , err . Error ( ) ) close ( shutdown ) } } ( ) wg . Add ( len ( a . Config . Aggregators ) ) for _ , aggregator := range a . Config . Aggregators { go func ( agg * models . RunningAggregator ) { defer wg . Done ( ) acc := NewAccumulator ( agg , aggC ) acc . SetPrecision ( a . Config . Agent . Precision . Duration , a . Config . Agent . Interval . Duration ) agg . Run ( acc , shutdown ) } ( aggregator ) } wg . Add ( len ( a . Config . Inputs ) ) for _ , input := range a . Config . Inputs { interval := a . Config . Agent . Interval . Duration if input . Config . Interval != 0 { interval = input . Config . Interval } go func ( in * models . RunningInput , interv time . Duration ) { defer wg . Done ( ) a . gatherer ( shutdown , in , interv , metricC ) } ( input , interval ) } wg . Wait ( ) a . Close ( ) return nil }
Run runs the agent daemon, gathering every Interval
Run runs the agent daemon gathering every Interval
// channel shared between all input threads for accumulating metrics // Start all ServicePlugins // Service input plugins should set their own precision of their // metrics. // Round collection to nearest interval by sleeping // overwrite global interval if this plugin has it's own.