Don't skip metrics during startup in aggregate phase (#4230)

agent/agent.go

@@ -362,24 +362,6 @@ func (a *Agent) Run(shutdown chan struct{}) error {
 	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)
@@ -419,6 +401,25 @@ func (a *Agent) Run(shutdown chan struct{}) error {
 		}(input, interval)
 	}
 
+	// Start all ServicePlugins inputs after all other
+	// plugins are loaded so that no metrics get dropped
+	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()
+		}
+	}
+
 	wg.Wait()
 	a.Close()
 	return nil

internal/models/running_aggregator.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"log"
 	"time"
 
 	"github.com/influxdata/telegraf"
@@ -153,6 +154,7 @@ func (r *RunningAggregator) Run(
 				m.Time().After(r.periodEnd.Add(truncation).Add(r.Config.Delay)) {
 				// the metric is outside the current aggregation period, so
 				// skip it.
+				log.Printf("D! aggregator: metric \"%s\" is not in the current timewindow, skipping", m.Name())
 				continue
 			}
 			r.add(m)