feat(inputs): Add option to choose the metric time source

config/config.go

@@ -1488,6 +1488,8 @@ func (c *Config) buildInput(name string, tbl *ast.Table) (*models.InputConfig, e
 	cp.CollectionJitter, _ = c.getFieldDuration(tbl, "collection_jitter")
 	cp.CollectionOffset, _ = c.getFieldDuration(tbl, "collection_offset")
 	cp.StartupErrorBehavior = c.getFieldString(tbl, "startup_error_behavior")
+	cp.TimeSource = models.TimeSourceType(c.getFieldString(tbl, "time_source"))
+
 	cp.MeasurementPrefix = c.getFieldString(tbl, "name_prefix")
 	cp.MeasurementSuffix = c.getFieldString(tbl, "name_suffix")
 	cp.NameOverride = c.getFieldString(tbl, "name_override")

docs/CONFIGURATION.md

@@ -396,6 +396,14 @@ Parameters that can be used with any input plugin:
 
   When this value is set on a service input, multiple events occurring at the
   same timestamp may be merged by the output database.
+- **time_source**:
+  Specifies the source of the timestamp on metrics. Possible values are:
+  - `metric` will not alter the metric (default)
+  - `collection_start` sets the timestamp to when collection started
+  - `collection_end` set the timestamp to when collection finished
+
+  `time_source` will NOT be used for service inputs. It is up to each individual
+  service input to set the timestamp.
 - **collection_jitter**:
   Overrides the `collection_jitter` setting of the [agent][Agent] for the
   plugin.  Collection jitter is used to jitter the collection by a random

models/running_input.go

@@ -17,16 +17,26 @@ var (
 	GlobalGatherTimeouts  = selfstat.Register("agent", "gather_timeouts", map[string]string{})
 )
 
+type TimeSourceType string
+
+const (
+	TimeSourceMetric          TimeSourceType = "metric"
+	TimeSourceCollectionStart TimeSourceType = "collection_start"
+	TimeSourceCollectionEnd   TimeSourceType = "collection_end"
+)
+
 type RunningInput struct {
 	Input  telegraf.Input
 	Config *InputConfig
 
 	log         telegraf.Logger
 	defaultTags map[string]string
 
-	startAcc telegraf.Accumulator
-	started  bool
-	retries  uint64
+	startAcc    telegraf.Accumulator
+	started     bool
+	retries     uint64
+	gatherStart time.Time
+	gatherEnd   time.Time
 
 	MetricsGathered selfstat.Stat
 	GatherTime      selfstat.Stat
@@ -87,6 +97,7 @@ type InputConfig struct {
 	CollectionJitter     time.Duration
 	CollectionOffset     time.Duration
 	Precision            time.Duration
+	TimeSource           TimeSourceType
 	StartupErrorBehavior string
 	LogLevel             string
 
@@ -114,6 +125,16 @@ func (r *RunningInput) Init() error {
 		return fmt.Errorf("invalid 'startup_error_behavior' setting %q", r.Config.StartupErrorBehavior)
 	}
 
+	if _, ok := r.Input.(telegraf.ServiceInput); !ok {
+		switch r.Config.TimeSource {
+		case "":
+			r.Config.TimeSource = TimeSourceMetric
+		case TimeSourceMetric, TimeSourceCollectionStart, TimeSourceCollectionEnd:
+		default:
+			return fmt.Errorf("invalid 'time_source' setting %q", r.Config.TimeSource)
+		}
+	}
+
 	if p, ok := r.Input.(telegraf.Initializer); ok {
 		return p.Init()
 	}
@@ -206,6 +227,16 @@ func (r *RunningInput) MakeMetric(metric telegraf.Metric) telegraf.Metric {
 		makemetric(metric, "", "", "", local, global)
 	}
 
+	if _, ok := r.Input.(telegraf.ServiceInput); !ok {
+		switch r.Config.TimeSource {
+		case TimeSourceCollectionStart:
+			metric.SetTime(r.gatherStart)
+		case TimeSourceCollectionEnd:
+			metric.SetTime(r.gatherEnd)
+		default:
+		}
+	}
+
 	r.MetricsGathered.Incr(1)
 	GlobalMetricsGathered.Incr(1)
 	return metric
@@ -228,10 +259,11 @@ func (r *RunningInput) Gather(acc telegraf.Accumulator) error {
 		}
 	}
 
-	start := time.Now()
+	r.gatherStart = time.Now()
 	err := r.Input.Gather(acc)
-	elapsed := time.Since(start)
-	r.GatherTime.Incr(elapsed.Nanoseconds())
+	r.gatherEnd = time.Now()
+
+	r.GatherTime.Incr(r.gatherEnd.Sub(r.gatherStart).Nanoseconds())
 	return err
 }
 

models/running_input_test.go

@@ -428,6 +428,65 @@ func TestMakeMetricWithAlwaysKeepingPluginTagsEnabled(t *testing.T) {
 	require.Equal(t, expected, actual)
 }
 
+func TestMakeMetricWithGatherMetricTimeSource(t *testing.T) {
+	ri := NewRunningInput(&testInput{}, &InputConfig{
+		Name:                    "TestRunningInput",
+		Tags:                    make(map[string]string),
+		Filter:                  Filter{},
+		AlwaysIncludeLocalTags:  false,
+		AlwaysIncludeGlobalTags: false,
+		TimeSource:              TimeSourceMetric,
+	})
+	start := time.Now()
+	ri.gatherStart = start
+	ri.gatherEnd = start.Add(time.Second)
+
+	expected := testutil.MockMetrics()[0]
+
+	m := testutil.MockMetrics()[0]
+	actual := ri.MakeMetric(m)
+
+	require.Equal(t, expected, actual)
+}
+
+func TestMakeMetricWithGatherStartTimeSource(t *testing.T) {
+	start := time.Now()
+	ri := NewRunningInput(&testInput{}, &InputConfig{
+		Name:                    "TestRunningInput",
+		Tags:                    make(map[string]string),
+		Filter:                  Filter{},
+		AlwaysIncludeLocalTags:  false,
+		AlwaysIncludeGlobalTags: false,
+		TimeSource:              TimeSourceCollectionStart,
+	})
+	ri.gatherStart = start
+
+	expected := testutil.MockMetrics()[0]
+	expected.SetTime(start)
+
+	m := testutil.MockMetrics()[0]
+	actual := ri.MakeMetric(m)
+
+	require.Equal(t, expected, actual)
+}
+
+func TestMakeMetricWithGatherEndTimeSource(t *testing.T) {
+	end := time.Now()
+	ri := NewRunningInput(&testInput{}, &InputConfig{
+		Name:       "TestRunningInput",
+		TimeSource: TimeSourceCollectionEnd,
+	})
+	ri.gatherEnd = end
+
+	expected := testutil.MockMetrics()[0]
+	expected.SetTime(end)
+
+	m := testutil.MockMetrics()[0]
+	actual := ri.MakeMetric(m)
+
+	require.Equal(t, expected, actual)
+}
+
 type testInput struct{}
 
 func (t *testInput) Description() string                 { return "" }