feat(serializers.influx): Add option to omit timestamp (#15220)

plugins/outputs/influxdb/README.md

@@ -105,6 +105,12 @@ to use them.
   ## integer values.  Enabling this option will result in field type errors if
   ## existing data has been written.
   # influx_uint_support = false
+
+  ## When true, Telegraf will omit the timestamp on data to allow InfluxDB
+  ## to set the timestamp of the data during ingestion. This is generally NOT
+  ## what you want as it can lead to data points captured at different times
+  ## getting omitted due to similar data.
+  # influx_omit_timestamp = false
 ```
 
 To send every metrics into multiple influxdb,

plugins/outputs/influxdb/influxdb.go

@@ -58,6 +58,7 @@ type InfluxDB struct {
 	ContentEncoding           string            `toml:"content_encoding"`
 	SkipDatabaseCreation      bool              `toml:"skip_database_creation"`
 	InfluxUintSupport         bool              `toml:"influx_uint_support"`
+	OmitTimestamp             bool              `toml:"influx_omit_timestamp"`
 	Precision                 string            `toml:"precision" deprecated:"1.0.0;option is ignored"`
 	Log                       telegraf.Logger   `toml:"-"`
 	tls.ClientConfig
@@ -202,7 +203,10 @@ func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
 }
 
 func (i *InfluxDB) udpClient(address *url.URL, localAddr *net.UDPAddr) (Client, error) {
-	serializer := &influx.Serializer{UintSupport: i.InfluxUintSupport}
+	serializer := &influx.Serializer{
+		UintSupport:   i.InfluxUintSupport,
+		OmitTimestamp: i.OmitTimestamp,
+	}
 	if err := serializer.Init(); err != nil {
 		return nil, err
 	}
@@ -229,7 +233,10 @@ func (i *InfluxDB) httpClient(ctx context.Context, address *url.URL, localAddr *
 		return nil, err
 	}
 
-	serializer := &influx.Serializer{UintSupport: i.InfluxUintSupport}
+	serializer := &influx.Serializer{
+		UintSupport:   i.InfluxUintSupport,
+		OmitTimestamp: i.OmitTimestamp,
+	}
 	if err := serializer.Init(); err != nil {
 		return nil, err
 	}

plugins/outputs/influxdb/sample.conf

@@ -79,3 +79,9 @@
   ## integer values.  Enabling this option will result in field type errors if
   ## existing data has been written.
   # influx_uint_support = false
+
+  ## When true, Telegraf will omit the timestamp on data to allow InfluxDB
+  ## to set the timestamp of the data during ingestion. This is generally NOT
+  ## what you want as it can lead to data points captured at different times
+  ## getting omitted due to similar data.
+  # influx_omit_timestamp = false

plugins/outputs/influxdb_v2/README.md

@@ -71,6 +71,12 @@ to use them.
   ## Enable or disable uint support for writing uints influxdb 2.0.
   # influx_uint_support = false
 
+  ## When true, Telegraf will omit the timestamp on data to allow InfluxDB
+  ## to set the timestamp of the data during ingestion. This is generally NOT
+  ## what you want as it can lead to data points captured at different times
+  ## getting omitted due to similar data.
+  # influx_omit_timestamp = false
+
   ## HTTP/2 Timeouts
   ## The following values control the HTTP/2 client's timeouts. These settings
   ## are generally not required unless a user is seeing issues with client

plugins/outputs/influxdb_v2/influxdb_v2.go

@@ -50,6 +50,7 @@ type InfluxDB struct {
 	UserAgent        string            `toml:"user_agent"`
 	ContentEncoding  string            `toml:"content_encoding"`
 	UintSupport      bool              `toml:"influx_uint_support"`
+	OmitTimestamp    bool              `toml:"influx_omit_timestamp"`
 	PingTimeout      config.Duration   `toml:"ping_timeout"`
 	ReadIdleTimeout  config.Duration   `toml:"read_idle_timeout"`
 	tls.ClientConfig
@@ -158,7 +159,10 @@ func (i *InfluxDB) getHTTPClient(address *url.URL, localAddr *net.TCPAddr, proxy
 		return nil, err
 	}
 
-	serializer := &influx.Serializer{UintSupport: i.UintSupport}
+	serializer := &influx.Serializer{
+		UintSupport:   i.UintSupport,
+		OmitTimestamp: i.OmitTimestamp,
+	}
 	if err := serializer.Init(); err != nil {
 		return nil, err
 	}

plugins/outputs/influxdb_v2/sample.conf

@@ -47,6 +47,12 @@
   ## Enable or disable uint support for writing uints influxdb 2.0.
   # influx_uint_support = false
 
+  ## When true, Telegraf will omit the timestamp on data to allow InfluxDB
+  ## to set the timestamp of the data during ingestion. This is generally NOT
+  ## what you want as it can lead to data points captured at different times
+  ## getting omitted due to similar data.
+  # influx_omit_timestamp = false
+
   ## HTTP/2 Timeouts
   ## The following values control the HTTP/2 client's timeouts. These settings
   ## are generally not required unless a user is seeing issues with client

plugins/serializers/influx/README.md

@@ -30,6 +30,12 @@ for interoperability.
   ## integer values.  Enabling this option will result in field type errors if
   ## existing data has been written.
   influx_uint_support = false
+
+  ## When true, Telegraf will omit the timestamp on data to allow InfluxDB
+  ## to set the timestamp of the data during ingestion. This is generally NOT
+  ## what you want as it can lead to data points captured at different times
+  ## getting omitted due to similar data.
+  # influx_omit_timestamp = false
 ```
 
 ## Metrics

plugins/serializers/influx/influx.go

@@ -46,9 +46,10 @@ func (e FieldError) Error() string {
 
 // Serializer is a serializer for line protocol.
 type Serializer struct {
-	MaxLineBytes int  `toml:"influx_max_line_bytes"`
+	MaxLineBytes  int  `toml:"influx_max_line_bytes"`
-	SortFields   bool `toml:"influx_sort_fields"`
+	SortFields    bool `toml:"influx_sort_fields"`
-	UintSupport  bool `toml:"influx_uint_support"`
+	UintSupport   bool `toml:"influx_uint_support"`
+	OmitTimestamp bool `toml:"influx_omit_timestamp"`
 
 	bytesWritten int
 
@@ -153,8 +154,10 @@ func (s *Serializer) buildHeader(m telegraf.Metric) error {
 
 func (s *Serializer) buildFooter(m telegraf.Metric) {
 	s.footer = s.footer[:0]
-	s.footer = append(s.footer, ' ')
+	if !s.OmitTimestamp {
-	s.footer = strconv.AppendInt(s.footer, m.Time().UnixNano(), 10)
+		s.footer = append(s.footer, ' ')
+		s.footer = strconv.AppendInt(s.footer, m.Time().UnixNano(), 10)
+	}
 	s.footer = append(s.footer, '\n')
 }
 

plugins/serializers/influx/influx_test.go

@@ -497,6 +497,24 @@ func TestSerializer(t *testing.T) {
 	}
 }
 
+func TestOmitTimestamp(t *testing.T) {
+	m := metric.New(
+		"cpu",
+		map[string]string{},
+		map[string]interface{}{
+			"value": 42.0,
+		},
+		time.Unix(1519194109, 42),
+	)
+
+	serializer := &Serializer{
+		OmitTimestamp: true,
+	}
+	output, err := serializer.Serialize(m)
+	require.NoError(t, err)
+	require.Equal(t, []byte("cpu value=42\n"), output)
+}
+
 func BenchmarkSerializer(b *testing.B) {
 	for _, tt := range tests {
 		b.Run(tt.name, func(b *testing.B) {

plugins/serializers/registry.go

@@ -94,6 +94,9 @@ type Config struct {
 	// Support unsigned integer output; influx format only
 	InfluxUintSupport bool `toml:"influx_uint_support"`
 
+	// Omit timestamp from output; influx format only
+	InfluxOmitTimestamp bool `toml:"influx_omit_timestamp"`
+
 	// Prefix to add to all measurements, only supports Graphite
 	Prefix string `toml:"prefix"`