Implement a per-output fixed size metric buffer

Also moved some objects out of config.go and put them in their own
package, internal/models

fixes influxdata#568
closes influxdata#285

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ### Release Notes:
 
+- Telegraf now keeps a fixed-length buffer of metrics per-output. This buffer
+defaults to 10,000 metrics, and is adjustable. The buffer is cleared when a
+successful write to that output occurs.
 - The docker plugin has been significantly overhauled to add more metrics
 and allow for docker-machine (incl OSX) support.
 [See the readme](https://github.com/influxdata/telegraf/blob/master/plugins/inputs/docker/README.md)
@@ -26,6 +29,7 @@ specifying a docker endpoint to get metrics from.
 - [#553](https://github.com/influxdata/telegraf/pull/553): Amazon CloudWatch output. thanks @skwong2!
 - [#503](https://github.com/influxdata/telegraf/pull/503): Support docker endpoint configuration.
 - [#563](https://github.com/influxdata/telegraf/pull/563): Docker plugin overhaul.
+- [#285](https://github.com/influxdata/telegraf/issues/285): Fixed-size buffer of points.
 
 ### Bugfixes
 - [#506](https://github.com/influxdata/telegraf/pull/506): Ping input doesn't return response time metric when timeout. Thanks @titilambert!
@@ -34,6 +38,7 @@ specifying a docker endpoint to get metrics from.
 - [#543](https://github.com/influxdata/telegraf/issues/543): Statsd Packet size sometimes truncated.
 - [#440](https://github.com/influxdata/telegraf/issues/440): Don't query filtered devices for disk stats.
 - [#463](https://github.com/influxdata/telegraf/issues/463): Docker plugin not working on AWS Linux
+- [#568](https://github.com/influxdata/telegraf/issues/568): Multiple output race condition.
 
 ## v0.10.0 [2016-01-12]
 

accumulator.go

@@ -7,7 +7,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/influxdata/telegraf/internal/config"
+	"github.com/influxdata/telegraf/internal/models"
 
 	"github.com/influxdata/influxdb/client/v2"
 )
@@ -29,7 +29,7 @@ type Accumulator interface {
 }
 
 func NewAccumulator(
-	inputConfig *config.InputConfig,
+	inputConfig *models.InputConfig,
 	points chan *client.Point,
 ) Accumulator {
 	acc := accumulator{}
@@ -47,7 +47,7 @@ type accumulator struct {
 
 	debug bool
 
-	inputConfig *config.InputConfig
+	inputConfig *models.InputConfig
 
 	prefix string
 }

agent.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf/internal/config"
+	"github.com/influxdata/telegraf/internal/models"
 	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/outputs"
 
@@ -101,7 +102,7 @@ func (a *Agent) gatherParallel(pointChan chan *client.Point) error {
 
 		wg.Add(1)
 		counter++
-		go func(input *config.RunningInput) {
+		go func(input *models.RunningInput) {
 			defer wg.Done()
 
 			acc := NewAccumulator(input.Config, pointChan)
@@ -144,7 +145,7 @@ func (a *Agent) gatherParallel(pointChan chan *client.Point) error {
 // reporting interval.
 func (a *Agent) gatherSeparate(
 	shutdown chan struct{},
-	input *config.RunningInput,
+	input *models.RunningInput,
 	pointChan chan *client.Point,
 ) error {
 	ticker := time.NewTicker(input.Config.Interval)
@@ -202,7 +203,6 @@ func (a *Agent) Test() error {
 	for _, input := range a.Config.Inputs {
 		acc := NewAccumulator(input.Config, pointChan)
 		acc.SetDebug(true)
-		// acc.SetPrefix(input.Name + "_")
 
 		fmt.Printf("* Plugin: %s, Collection 1\n", input.Name)
 		if input.Config.Interval != 0 {
@@ -228,93 +228,45 @@ func (a *Agent) Test() error {
 	return nil
 }
 
-// writeOutput writes a list of points to a single output, with retries.
-// Optionally takes a `done` channel to indicate that it is done writing.
-func (a *Agent) writeOutput(
-	points []*client.Point,
-	ro *config.RunningOutput,
-	shutdown chan struct{},
-	wg *sync.WaitGroup,
-) {
-	defer wg.Done()
-	if len(points) == 0 {
-		return
-	}
-	retry := 0
-	retries := a.Config.Agent.FlushRetries
-	start := time.Now()
-
-	for {
-		filtered := ro.FilterPoints(points)
-		err := ro.Output.Write(filtered)
-		if err == nil {
-			// Write successful
-			elapsed := time.Since(start)
-			if !a.Config.Agent.Quiet {
-				log.Printf("Flushed %d metrics to output %s in %s\n",
-					len(filtered), ro.Name, elapsed)
-			}
-			return
-		}
-
-		select {
-		case <-shutdown:
-			return
-		default:
-			if retry >= retries {
-				// No more retries
-				msg := "FATAL: Write to output [%s] failed %d times, dropping" +
-					" %d metrics\n"
-				log.Printf(msg, ro.Name, retries+1, len(points))
-				return
-			} else if err != nil {
-				// Sleep for a retry
-				log.Printf("Error in output [%s]: %s, retrying in %s",
-					ro.Name, err.Error(), a.Config.Agent.FlushInterval.Duration)
-				time.Sleep(a.Config.Agent.FlushInterval.Duration)
-			}
-		}
-
-		retry++
-	}
-}
-
 // flush writes a list of points to all configured outputs
-func (a *Agent) flush(
-	points []*client.Point,
-	shutdown chan struct{},
-	wait bool,
-) {
+func (a *Agent) flush() {
 	var wg sync.WaitGroup
+
+	wg.Add(len(a.Config.Outputs))
 	for _, o := range a.Config.Outputs {
-		wg.Add(1)
-		go a.writeOutput(points, o, shutdown, &wg)
-	}
-	if wait {
-		wg.Wait()
+		go func(output *models.RunningOutput) {
+			defer wg.Done()
+			err := output.Write()
+			if err != nil {
+				log.Printf("Error writing to output [%s]: %s\n",
+					output.Name, err.Error())
+			}
+		}(o)
 	}
+
+	wg.Wait()
 }
 
 // flusher monitors the points input channel and flushes on the minimum interval
 func (a *Agent) flusher(shutdown chan struct{}, pointChan chan *client.Point) error {
 	// Inelegant, but this sleep is to allow the Gather threads to run, so that
 	// the flusher will flush after metrics are collected.
-	time.Sleep(time.Millisecond * 100)
+	time.Sleep(time.Millisecond * 200)
 
 	ticker := time.NewTicker(a.Config.Agent.FlushInterval.Duration)
-	points := make([]*client.Point, 0)
 
 	for {
 		select {
 		case <-shutdown:
 			log.Println("Hang on, flushing any cached points before shutdown")
-			a.flush(points, shutdown, true)
+			a.flush()
 			return nil
 		case <-ticker.C:
-			a.flush(points, shutdown, false)
-			points = make([]*client.Point, 0)
+			a.flush()
 		case pt := <-pointChan:
-			points = append(points, pt)
+			for _, o := range a.Config.Outputs {
+				o.AddPoint(pt)
+			}
 		}
 	}
 }
@@ -389,7 +341,7 @@ func (a *Agent) Run(shutdown chan struct{}) error {
 		// configured. Default intervals are handled below with gatherParallel
 		if input.Config.Interval != 0 {
 			wg.Add(1)
-			go func(input *config.RunningInput) {
+			go func(input *models.RunningInput) {
 				defer wg.Done()
 				if err := a.gatherSeparate(shutdown, input, pointChan); err != nil {
 					log.Printf(err.Error())