Add timestamps to points in Kafka/AMQP outputs

outputs/amqp/amqp.go

@@ -2,6 +2,7 @@ package amqp
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/influxdb/influxdb/client"
 	"github.com/influxdb/telegraf/outputs"
@@ -72,6 +73,7 @@ func (q *AMQP) Write(bp client.BatchPoints) error {
 		return nil
 	}
 
+	var zero_time time.Time
 	for _, p := range bp.Points {
 		// Combine tags from Point and BatchPoints and grab the resulting
 		// line-protocol output string to write to AMQP
@@ -85,6 +87,13 @@ func (q *AMQP) Write(bp client.BatchPoints) error {
 				}
 				p.Tags[k] = v
 			}
+			if p.Time == zero_time {
+				if bp.Time == zero_time {
+					p.Time = time.Now()
+				} else {
+					p.Time = bp.Time
+				}
+			}
 			value = p.MarshalString()
 		}
 

outputs/kafka/kafka.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"github.com/influxdb/influxdb/client"
@@ -56,6 +57,7 @@ func (k *Kafka) Write(bp client.BatchPoints) error {
 		return nil
 	}
 
+	var zero_time time.Time
 	for _, p := range bp.Points {
 		// Combine tags from Point and BatchPoints and grab the resulting
 		// line-protocol output string to write to Kafka
@@ -69,6 +71,13 @@ func (k *Kafka) Write(bp client.BatchPoints) error {
 				}
 				p.Tags[k] = v
 			}
+			if p.Time == zero_time {
+				if bp.Time == zero_time {
+					p.Time = time.Now()
+				} else {
+					p.Time = bp.Time
+				}
+			}
 			value = p.MarshalString()
 		}