stats: add option for percentiles to be emitted by client for distribution metrics

datadog.go

@@ -6,6 +6,7 @@
 package datadog
 
 import (
+ "fmt"
  "log"
  "regexp"
  "strings"
@@ -73,6 +74,43 @@ type Options struct {
  // GlobalTags holds a set of tags that will automatically be applied to all
  // exported spans.
  GlobalTags map[string]interface{}
+
+ // DisableCountPerBuckets specifies whether to emit count_per_bucket metrics
+ DisableCountPerBuckets bool
+
+ // EmitPercentiles given a list of percentiles [0.5, 0.95, 0.99], for each one will estimate the percentile
+ // from the Distribution metric and emit a unique metric for each
+ //
+ // Example: []Percentile{{0.5, "50p"},{0.95, "95p"}, {0.99, "99p"}}
+ EmitPercentiles []Percentile
+}
+
+// Percentile indicates the percentile to calculate for a distribution metric and its corresponding name
+type Percentile struct {
+ // Percentile must be in range [0, 100].
+ Percentile float64
+
+ // MetricSuffix (optional) controls the suffix of the metric name that is emitted. If not provided,
+ // will be formatted to 2 decimal places.
+ //
+ // Example: "99p" would emit `server_latency.99p if the distribution metric name was server_latency
+ MetricSuffix string
+}
+
+// Commonly used percentiles
+var (
+ Percentile50th = Percentile{Percentile: 0.5, MetricSuffix: "median"}
+ Percentile75th = Percentile{Percentile: 0.75, MetricSuffix: "75p"}
+ Percentile95th = Percentile{Percentile: 0.95, MetricSuffix: "95p"}
+ Percentile99th = Percentile{Percentile: 0.99, MetricSuffix: "99p"}
+ Percentile999th = Percentile{Percentile: 0.999, MetricSuffix: "999p"}
+)
+
+func (p Percentile) buildMetricSuffix() string {
+ if p.MetricSuffix != "" {
+ return p.MetricSuffix
+ }
+ return fmt.Sprintf("%.2fp", p.Percentile*100)
 }
 
 func (o *Options) onError(err error) {

datadog_test.go

@@ -272,3 +272,46 @@ func TestHistogram(t *testing.T) {
  t.Errorf("Expected: %v, Got: %v\n", vd, actual)
  }
 }
+
+func TestPercentile_buildMetricSuffix(t *testing.T) {
+ tsts := []struct {
+ Percentile
+ Expected string
+ }{
+ // Common
+ {
+ Percentile50th,
+ "median",
+ },
+ {
+ Percentile75th,
+ "75p",
+ },
+ {
+ Percentile95th,
+ "95p",
+ },
+ {
+ Percentile99th,
+ "99p",
+ },
+ {
+ Percentile999th,
+ "999p",
+ },
+ // Formatted
+ {
+ Percentile{Percentile: 0.92},
+ "92.00p",
+ },
+ }
+
+ for _, tst := range tsts {
+ t.Run(fmt.Sprintf("%f", tst.Percentile.Percentile), func(t *testing.T) {
+ got := tst.buildMetricSuffix()
+ if got != tst.Expected {
+ t.Errorf("Expected: %v, Got %v\n", tst.Expected, got)
+ }
+ })
+ }
+}

stats.go

@@ -7,6 +7,7 @@ package datadog
 
 import (
  "fmt"
+ "math"
  "sync"
 
  "github.com/DataDog/datadog-go/statsd"
@@ -92,17 +93,40 @@ func (s *statsExporter) submitMetric(v *view.View, row *view.Row, metricName str
  "avg": data.Mean,
  "squared_dev_sum": data.SumOfSquaredDev,
  }
+ for _, percentile := range s.opts.EmitPercentiles {
+ metrics[percentile.buildMetricSuffix()] = calculatePercentile(percentile.Percentile, v.Aggregation.Buckets, data.CountPerBucket)
+ }
 
  for name, value := range metrics {
  err = client.Gauge(metricName+"."+name, value, opt.tagMetrics(row.Tags, tags), rate)
  }
-
- for x := range data.CountPerBucket {
- addlTags := []string{"bucket_idx:" + fmt.Sprint(x)}
- err = client.Gauge(metricName+".count_per_bucket", float64(data.CountPerBucket[x]), opt.tagMetrics(row.Tags, addlTags), rate)
+ if !s.opts.DisableCountPerBuckets {
+ for x := range data.CountPerBucket {
+ addlTags := []string{"bucket_idx:" + fmt.Sprint(x)}
+ err = client.Gauge(metricName+".count_per_bucket", float64(data.CountPerBucket[x]), opt.tagMetrics(row.Tags, addlTags), rate)
+ }
  }
  return err
  default:
  return fmt.Errorf("aggregation %T is not supported", v.Aggregation)
  }
 }
+
+func calculatePercentile(percentile float64, buckets []float64, countPerBucket []int64) float64 {
+ cumulativePerBucket := make([]int64, len(countPerBucket))
+ var sum int64
+ for n, count := range countPerBucket {
+ sum += count
+ cumulativePerBucket[n] = sum
+ }
+ atBin := int64(math.Floor(percentile * float64(sum)))
+
+ var previousCount int64
+ for n, count := range cumulativePerBucket {
+ if atBin >= previousCount && atBin <= count {
+ return buckets[n]
+ }
+ previousCount = count
+ }
+ return buckets[len(buckets)-1]
+}

stats_test.go

@@ -7,6 +7,7 @@ package datadog
 
 import (
  "fmt"
+ "math/rand"
  "testing"
  "time"
 
@@ -103,3 +104,78 @@ func TestNilAggregation(t *testing.T) {
  t.Errorf("Expected: %v, Got: %v", fmt.Errorf("aggregation *view.Aggregation is not supported"), actual)
  }
 }
+
+func Test_calculatePercentile(t *testing.T) {
+ var buckets []float64
+ for i := float64(-100); i < 100; i += 0.1 {
+ buckets = append(buckets, i)
+ }
+
+ normalDistribution := calculateNormalDistribution(buckets, 0, 100)
+ tsts := []struct {
+ expected int64
+ percentile float64
+ buckets []float64
+ countsPerBucket []int64
+ }{
+ {
+ 0,
+ 0.5,
+ buckets,
+ normalDistribution,
+ },
+ {
+ 44,
+ 0.75,
+ buckets,
+ normalDistribution,
+ },
+ {
+ 86,
+ 0.95,
+ buckets,
+ normalDistribution,
+ },
+ {
+ 97,
+ 0.99,
+ buckets,
+ normalDistribution,
+ },
+ {
+ 99,
+ 0.999,
+ buckets,
+ normalDistribution,
+ },
+ }
+
+ for _, tst := range tsts {
+ t.Run(fmt.Sprintf("%v", tst.percentile), func(t *testing.T) {
+ got := calculatePercentile(tst.percentile, tst.buckets, tst.countsPerBucket)
+
+ if tst.expected != int64(got) {
+ t.Errorf("Expected: %v, Got: %v", tst.expected, got)
+ }
+ })
+
+ }
+}
+
+func calculateNormalDistribution(buckets []float64, seed int64, standardDeviation float64) []int64 {
+ r := rand.New(rand.NewSource(seed))
+
+ normalDistribution := make([]int64, len(buckets))
+ for n := 0; n < 1e6; n++ {
+ rnd := r.NormFloat64() * standardDeviation
+ var previousBucket float64
+ for bidx, bucket := range buckets {
+ if rnd > previousBucket && rnd <= bucket {
+ normalDistribution[bidx]++
+ break
+ }
+ previousBucket = bucket
+ }
+ }
+ return normalDistribution
+}