diff --git a/outputs/amqp/amqp.go b/outputs/amqp/amqp.go index 070793fdb..9ca2783f1 100644 --- a/outputs/amqp/amqp.go +++ b/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() } diff --git a/outputs/kafka/kafka.go b/outputs/kafka/kafka.go index 370f1ba95..ce9946ba3 100644 --- a/outputs/kafka/kafka.go +++ b/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() }