feat(outputs.datadog): Add support for submitting alongside dd-agent

plugins/outputs/datadog/README.md

@@ -36,6 +36,14 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## Override the default (none) compression used to send data.
   ## Supports: "zlib", "none"
   # compression = "none"
+
+  ## Convert counts to rates
+  ## Use this to be able to submit metrics from telegraf alongside Datadog agent
+  # should_rate_counts = false
+
+  ## Overrides the default rate interval used to divide count metrics by
+  ## when should_rate_counts is enabled
+  # rate_interval = 10
 ```
 
 ## Metrics
@@ -46,11 +54,12 @@ field key with a `.` character.
 Field values are converted to floating point numbers.  Strings and floats that
 cannot be sent over JSON, namely NaN and Inf, are ignored.
 
-We do not send `Rate` types. Counts are sent as `count`, with an
-interval hard-coded to 1. Note that this behavior does *not* play
-super-well if running simultaneously with current Datadog agents; they
-will attempt to change to `Rate` with `interval=10`. We prefer this
-method, however, as it reflects the raw data more accurately.
+Enabling the `should_rate_counts` will convert `count` metrics to `rate`
+and divide it by the `rate_interval` before submitting to Datadog.
+This allows telegraf to submit metrics alongside Datadog agents.
+Note that this only supports metrics ingested via `inputs.statsd` given
+the dependency on the `metric_type` tag it creates. There is only support for
+`counter` metrics, and `count` values from `timing` and `histogram` metrics.
 
 [metrics]: https://docs.datadoghq.com/api/v1/metrics/#submit-metrics
 [apikey]: https://app.datadoghq.com/account/settings#api

plugins/outputs/datadog/datadog.go

@@ -25,11 +25,12 @@ import (
 var sampleConfig string
 
 type Datadog struct {
-	Apikey      string          `toml:"apikey"`
-	Timeout     config.Duration `toml:"timeout"`
-	URL         string          `toml:"url"`
-	Compression string          `toml:"compression"`
-	Log         telegraf.Logger `toml:"-"`
+	Apikey           string          `toml:"apikey"`
+	Timeout          config.Duration `toml:"timeout"`
+	URL              string          `toml:"url"`
+	Compression      string          `toml:"compression"`
+	RateInterval     int64           `toml:"rate_interval"`
+	Log              telegraf.Logger `toml:"-"`
 
 	client *http.Client
 	proxy.HTTPProxy
@@ -75,15 +76,16 @@ func (d *Datadog) Connect() error {
 	return nil
 }
 
-func (d *Datadog) Write(metrics []telegraf.Metric) error {
-	ts := TimeSeries{}
+func (d *Datadog) convertToDatadogMetric(metrics []telegraf.Metric) ([]*Metric, int) {
 	tempSeries := []*Metric{}
 	metricCounter := 0
 
 	for _, m := range metrics {
 		if dogMs, err := buildMetrics(m); err == nil {
 			metricTags := buildTags(m.TagList())
 			host, _ := m.GetTag("host")
+			// Retrieve the metric_type tag created by inputs.statsd
+			statsDMetricType, _ := m.GetTag("metric_type")
 
 			if len(dogMs) == 0 {
 				continue
@@ -99,9 +101,19 @@ func (d *Datadog) Write(metrics []telegraf.Metric) error {
 					dname = m.Name() + "." + fieldName
 				}
 				var tname string
+				var interval int64
+				interval = 1
 				switch m.Type() {
-				case telegraf.Counter:
-					tname = "count"
+				case telegraf.Counter, telegraf.Untyped:
+					if d.RateInterval > 0 && isRateable(statsDMetricType, fieldName) {
+						interval = d.RateInterval
+						dogM[1] = dogM[1] / float64(interval)
+						tname = "rate"
+					} else if m.Type() == telegraf.Counter {
+						tname = "count"
+					} else {
+						tname = ""
+					}
 				case telegraf.Gauge:
 					tname = "gauge"
 				default:
@@ -112,7 +124,7 @@ func (d *Datadog) Write(metrics []telegraf.Metric) error {
 					Tags:     metricTags,
 					Host:     host,
 					Type:     tname,
-					Interval: 1,
+					Interval: interval,
 				}
 				metric.Points[0] = dogM
 				tempSeries = append(tempSeries, metric)
@@ -122,6 +134,12 @@ func (d *Datadog) Write(metrics []telegraf.Metric) error {
 			d.Log.Infof("Unable to build Metric for %s due to error '%v', skipping", m.Name(), err)
 		}
 	}
+	return tempSeries, metricCounter
+}
+
+func (d *Datadog) Write(metrics []telegraf.Metric) error {
+	ts := TimeSeries{}
+	tempSeries, metricCounter := d.convertToDatadogMetric(metrics)
 
 	if len(tempSeries) == 0 {
 		return nil
@@ -220,6 +238,20 @@ func verifyValue(v interface{}) bool {
 	return true
 }
 
+func isRateable(statsDMetricType string, fieldName string) bool {
+	switch statsDMetricType {
+	case
+		"counter":
+		return true
+	case
+		"timing",
+		"histogram":
+		return fieldName == "count"
+	default:
+		return false
+	}
+}
+
 func (p *Point) setValue(v interface{}) error {
 	switch d := v.(type) {
 	case int64:
@@ -246,8 +278,8 @@ func (d *Datadog) Close() error {
 func init() {
 	outputs.Add("datadog", func() telegraf.Output {
 		return &Datadog{
-			URL:         datadogAPI,
-			Compression: "none",
+			URL:          datadogAPI,
+			Compression:  "none",
 		}
 	})
 }