Repository: atlassian/gostatsd
Pull request #245 (2019-06-19 13:04:48)

Latency Histograms
Implementing latency histograms, based on previous work in #222.

Latency histograms can be enabled with "configuration tag" `gsd_histogram` where its value says what thresholds should be used to output counter metrics.

E.g. timer `t1` with tag `gsd_histogram:2.5_20_100_2000` will output additional 5 cumulative counters
* `t1.histogram` with tag `le:2.5`
* `t1.histogram` with tag `le:20`
* `t1.histogram` with tag `le:100`
* `t1.histogram` with tag `le:2000`
* `t1.histogram` with tag `le:+Inf`

where `le` means `less or equal` and follows [Prometheus convention](https://prometheus.io/docs/practices/histograms/)

counter values will be equal to the number of data points with values `<=` configured histogram thresholds.
`le:+Inf` will naturally contain total number of data points as all values are `<= +Inf`

tmszdmsk: @tiedotguy @adampetrovic @MovieStoreGuy 

 I am thinking about configuration options & restrictions around histograms and I am not sure what do we need, here are my thoughts:
1. configuration to enable/disable support for `gsd_histogram:`, enabled by default
1. being able to specify the maximum number of histogram buckets (what should be the default? 10?)
1. I am not really convinced that we should do aggregations XOR histograms. It can slow down the adoption because you will have to sacrifice something upfront. Maybe it should be 2 buckets "for free" + every next bucket eliminates one aggregation, in some specific order. But that's probably too intricate. What do you think? 
1. What other configuration options?
tiedotguy: For the configuration, rather than two options for enable/disable and count, I would have it as a single configuration setting, where 0 is disabled (with special logic to check for 0), and a default of `math.MaxInt32` - effectively unlimited, but practically bound by the UDP datagram size.

They should absolutely be distinct.  You don't sacrifice anything - you make a choice to use it or not.  If you want both, you send twice.  If they're not distinct then there's no way to opt out of the wasteful additional metrics.  If I was running the deployment, I would look carefully at the cardinality impact, then set the maximum buckets to 2, one of which is +Inf.

To put it bluntly, there's a lot of bad tagging that goes on, and this is a chance to clean house.  I hope for two things to come out of this work:
1. because it will give better data, people will want to adopt it (aka the carrot)
2. because it's new, strict controls can be put in place from day 0 to ensure only good data is accepted (aka the stick)

On a more technical level, this is something I may re-factor at some point to change from a timer to a new type (histogram) when the tag is present, because there's some efficiency gains to be made by pushing the work to a local process.  This would be internal only - the statsd interface would be unchanged.

I don't think there's any other configuration options needed.  From memory, in the original branch it was using "named sets", ie `gsd_histogram:pow2`, and the administrator had to define the sets ahead of time, but this approach bypasses it.
tmszdmsk: I thought more about operational backward-compatibility, not binary, but if it's being used as a library then it makes sense. I'll bump that.

Should I add support for `nil`able values in `Timer` or should the `if` be replicated in all the backends? (The  one seems cleaner ;)

Also, I am gonna pause work on this for now as obs needs to decide whether to proceed with this change or not. Thanks, Harold!
tiedotguy: No problem, thanks for working on it - it's definitely very close to ready if they decide to proceed.  There was something I remembered today that needs changing, but now I can't recall it.  I'll give the whole thing a fresh review if it proceeds.

Not certain what you mean by `nil`lable values in `Timer` though?
tmszdmsk: @tiedotguy 
> Not certain what you mean by nillable values in Timer though?

There is no way to set default aggregations (`timer.Count`, `timer.Sum`, ...) to "no value" (`nil`) so that they are not outputted in the backends. I don't want to add knowledge about `gsd_histogram` meta tag to the backends.
My idea is to change the definition of `Timer` to
```go
type Timer struct {
	Count        *int         // The number of timers in the series
	SampledCount *float64     // Number of timings received, divided by sampling rate
        ...
```
so that we can assign `nil`s when histogram is enabled and check that in the backends before outputting.
tiedotguy: Ahh, understood.  I believe each backend should just check for `Timer.Histogram == nil`.  It might be worth pulling each path (regular timer + histogram) out to a function, just to clean things up though, but I won't hold you to it.

I'm also considering adding Datadog distributions through the timer statsd type (with a tag), so at some point in the future I may add a `Timer.Type` (regular/histogram/distribution), or possibly make the tags route to more concrete types (ie, timer type + `gsd_histogram` becomes a `gostatsd.Histogram` instead of a `gostatsd.Timer`).

For now, I think it's ok not to do any big refactor to support it.
tiedotguy: Also if you like, next time I'm in the office I can whiteboard why I'm paranoid about the counter vs gauge vs rate thing.
MovieStoreGuy: This branch is lagging behind master a lot, is this something we are going to keep going with?
tmszdmsk: > This branch is lagging behind master a lot, is this something we are going to keep going with?

Yes, I have some changes on my localhost, but wanted to address all the comments first
tmszdmsk: master branch is merged & I think all comments are resolved, please @aelse, @tiedotguy review once more.
tmszdmsk: Also, how can I test it with one of my projects in e.g. staging?
tiedotguy: Thanks! I'll try and give it a thorough review over the weekend. For testing, you'll need to discuss with the team, because it's the aggregation tier that's changing, not the on-host tier.
adampetrovic: FYI i'll be testing this branch out in an innovation week this week. Will report back any issues I find.
tiedotguy: Sounds good to me. Sync up the README.md and cut a release. It's clearly listed as experimental, so if breaking changes are needed, that's also groovy. Looking at the requested changes I had historically they appear to be done.
tiedotguy: Also, let's make sure this has a squash merge, it's a little noisy :)

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e48a487..4f875c5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,8 @@
+
+20.3.0
+------
+- Added experimental support for generating latency histograms for timers, more details in [README.md]
+
 20.2.0
 ------
 - Support for `response-header-timeout` for `transport` section
diff --git a/README.md b/README.md
index cf09ec8..7a29cba 100644
--- a/README.md
+++ b/README.md
@@ -184,7 +184,40 @@ upper-pct=false
 
 By default (for compatibility), they are all false and the metrics will be emitted.
 
+Timer histograms (experimental feature)
+----------------
 
+Timer histograms inspired by [Prometheus implementation](https://prometheus.io/docs/concepts/metric_types/#histogram) can be
+enabled on a per time series basis using `gsd_histogram` meta tag with value containing histogram bucketing definition (joined with `_`)
+e.g. `gsd_histogram:-10_0_2.5_5_10_25_50`.
+
+It will:
+* output additional counter time series with name `<base>.histogram` and `le` tags specifying histogram buckets.
+* disable default sub-aggregations for timers e.g. `<base>.Count`, `<base>.Mean`, `<base>.Upper`, `<base>.Upper_XX`, etc.
+
+For timer with `gsd_histogram:-10_0_2.5_5_10_25_50` meta tag, following time series will be generated
+* `<base>.histogram` with tag `le:-10`
+* `<base>.histogram` with tag `le:0`
+* `<base>.histogram` with tag `le:2.5`
+* `<base>.histogram` with tag `le:5`
+* `<base>.histogram` with tag `le:10`
+* `<base>.histogram` with tag `le:25`
+* `<base>.histogram` with tag `le:50`
+* `<base>.histogram` with tag `le:+Inf`
+
+Each time series will contain a total number of timer data points that had a value less or equal `le` value, e.g. counter `<base>.histogram` with the tag `le:5` will contain the number of all observations that had a value not bigger than `5`.
+Counter `<base>.histogram` with tag `le:+Inf` is equivalent to `<base>.count` and contains the total number.
+
+All original timer tags are preserved and added to all the time series.
+
+To limit cardinality, `timer-histogram-limit` option can be specified to limit the number of buckets that will be created (default is `math.MaxUint32`).
+Value of `0` won't disable the feature, `0` buckets will be emitted which effectively drops metrics with `gsd_hostogram` tags.
+
+Incorrect meta tag values will be handled in best effort manner, i.e.
+* `gsd_histogram:10__20_50` & `gsd_histogram:10_incorrect_20_50` will generate `le:10`, `le:20`, `le:50` and `le:+Inf` buckets
+* `gsd_histogram:incorrect` will result in only `le:+Inf` bucket
+
+This is an experimental feature and it may be removed or changed in future versions.
 
 Sending metrics
 ---------------
diff --git a/cmd/gostatsd/main.go b/cmd/gostatsd/main.go
index 629709b..06cd4ba 100644
--- a/cmd/gostatsd/main.go
+++ b/cmd/gostatsd/main.go
@@ -145,6 +145,7 @@ func constructServer(v *viper.Viper) (*statsd.Server, error) {
 		},
 		DisabledSubTypes:          gostatsd.DisabledSubMetrics(v),
 		BadLineRateLimitPerSecond: rate.Limit(v.GetFloat64(statsd.ParamBadLinesPerMinute) / 60.0),
+		HistogramLimit:            v.GetUint32(statsd.ParamTimerHistogramLimit),
 		Viper:                     v,
 		TransportPool:             pool,
 	}, nil
diff --git a/pkg/backends/cloudwatch/cloudwatch.go b/pkg/backends/cloudwatch/cloudwatch.go
index 6353ee3..62c7073 100644
--- a/pkg/backends/cloudwatch/cloudwatch.go
+++ b/pkg/backends/cloudwatch/cloudwatch.go
@@ -2,6 +2,8 @@ package cloudwatch
 
 import (
 	"context"
+	"math"
+	"strconv"
 	"strings"
 	"time"
 
@@ -125,35 +127,46 @@ func (client Client) buildMetricData(metrics *gostatsd.MetricMap) (metricData []
 
 	prefix = "stats.timers."
 	metrics.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
-		if !disabled.Lower {
-			addMetricData(key+".lower", "Milliseconds", timer.Min, timer.Tags)
-		}
-		if !disabled.Upper {
-			addMetricData(key+".upper", "Milliseconds", timer.Max, timer.Tags)
-		}
-		if !disabled.Count {
-			addMetricData(key+".count", "Count", float64(timer.Count), timer.Tags)
-		}
-		if !disabled.CountPerSecond {
-			addMetricData(key+".count_ps", "Count/Second", timer.PerSecond, timer.Tags)
-		}
-		if !disabled.Mean {
-			addMetricData(key+".mean", "Milliseconds", timer.Mean, timer.Tags)
-		}
-		if !disabled.Median {
-			addMetricData(key+".median", "Milliseconds", timer.Median, timer.Tags)
-		}
-		if !disabled.StdDev {
-			addMetricData(key+".std", "Milliseconds", timer.StdDev, timer.Tags)
-		}
-		if !disabled.Sum {
-			addMetricData(key+".sum", "Milliseconds", timer.Sum, timer.Tags)
-		}
-		if !disabled.SumSquares {
-			addMetricData(key+".sum_squares", "Milliseconds", timer.SumSquares, timer.Tags)
-		}
-		for _, pct := range timer.Percentiles {
-			addMetricData(key+"."+pct.Str, "Milliseconds", pct.Float, timer.Tags)
+		if timer.Histogram != nil {
+			for histogramThreshold, count := range timer.Histogram {
+				bucketTag := "le:+Inf"
+				if !math.IsInf(float64(histogramThreshold), 1) {
+					bucketTag = "le:" + strconv.FormatFloat(float64(histogramThreshold), 'f', -1, 64)
+				}
+				newTags := timer.Tags.Concat(gostatsd.Tags{bucketTag})
+				addMetricData(key+".histogram", "Count", float64(count), newTags)
+			}
+		} else {
+			if !disabled.Lower {
+				addMetricData(key+".lower", "Milliseconds", timer.Min, timer.Tags)
+			}
+			if !disabled.Upper {
+				addMetricData(key+".upper", "Milliseconds", timer.Max, timer.Tags)
+			}
+			if !disabled.Count {
+				addMetricData(key+".count", "Count", float64(timer.Count), timer.Tags)
+			}
+			if !disabled.CountPerSecond {
+				addMetricData(key+".count_ps", "Count/Second", timer.PerSecond, timer.Tags)
+			}
+			if !disabled.Mean {
+				addMetricData(key+".mean", "Milliseconds", timer.Mean, timer.Tags)
+			}
+			if !disabled.Median {
+				addMetricData(key+".median", "Milliseconds", timer.Median, timer.Tags)
+			}
+			if !disabled.StdDev {
+				addMetricData(key+".std", "Milliseconds", timer.StdDev, timer.Tags)
+			}
+			if !disabled.Sum {
+				addMetricData(key+".sum", "Milliseconds", timer.Sum, timer.Tags)
+			}
+			if !disabled.SumSquares {
+				addMetricData(key+".sum_squares", "Milliseconds", timer.SumSquares, timer.Tags)
+			}
+			for _, pct := range timer.Percentiles {
+				addMetricData(key+"."+pct.Str, "Milliseconds", pct.Float, timer.Tags)
+			}
 		}
 	})
 
diff --git a/pkg/backends/datadog/datadog.go b/pkg/backends/datadog/datadog.go
index f80db81..c4873e7 100644
--- a/pkg/backends/datadog/datadog.go
+++ b/pkg/backends/datadog/datadog.go
@@ -7,9 +7,11 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"net/url"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -20,7 +22,7 @@ import (
 	"github.com/atlassian/gostatsd/pkg/util"
 
 	"github.com/cenkalti/backoff"
-	jsoniter "github.com/json-iterator/go"
+	"github.com/json-iterator/go"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )
@@ -167,35 +169,47 @@ func (d *Client) processMetrics(metrics *gostatsd.MetricMap, cb func(*timeSeries
 	})
 
 	metrics.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
-		if !d.disabledSubtypes.Lower {
-			fl.addMetricf(gauge, timer.Min, timer.Hostname, timer.Tags, "%s.lower", key)
-		}
-		if !d.disabledSubtypes.Upper {
-			fl.addMetricf(gauge, timer.Max, timer.Hostname, timer.Tags, "%s.upper", key)
-		}
-		if !d.disabledSubtypes.Count {
-			fl.addMetricf(gauge, float64(timer.Count), timer.Hostname, timer.Tags, "%s.count", key)
-		}
-		if !d.disabledSubtypes.CountPerSecond {
-			fl.addMetricf(rate, timer.PerSecond, timer.Hostname, timer.Tags, "%s.count_ps", key)
-		}
-		if !d.disabledSubtypes.Mean {
-			fl.addMetricf(gauge, timer.Mean, timer.Hostname, timer.Tags, "%s.mean", key)
-		}
-		if !d.disabledSubtypes.Median {
-			fl.addMetricf(gauge, timer.Median, timer.Hostname, timer.Tags, "%s.median", key)
-		}
-		if !d.disabledSubtypes.StdDev {
-			fl.addMetricf(gauge, timer.StdDev, timer.Hostname, timer.Tags, "%s.std", key)
-		}
-		if !d.disabledSubtypes.Sum {
-			fl.addMetricf(gauge, timer.Sum, timer.Hostname, timer.Tags, "%s.sum", key)
-		}
-		if !d.disabledSubtypes.SumSquares {
-			fl.addMetricf(gauge, timer.SumSquares, timer.Hostname, timer.Tags, "%s.sum_squares", key)
-		}
-		for _, pct := range timer.Percentiles {
-			fl.addMetricf(gauge, pct.Float, timer.Hostname, timer.Tags, "%s.%s", key, pct.Str)
+		if timer.Histogram != nil {
+			for histogramThreshold, count := range timer.Histogram {
+				bucketTag := "le:+Inf"
+				if !math.IsInf(float64(histogramThreshold), 1) {
+					bucketTag = "le:" + strconv.FormatFloat(float64(histogramThreshold), 'f', -1, 64)
+				}
+				newTags := timer.Tags.Concat(gostatsd.Tags{bucketTag})
+				fl.addMetricf(counter, float64(count), timer.Hostname, newTags, "%s.histogram", key)
+			}
+		} else {
+
+			if !d.disabledSubtypes.Lower {
+				fl.addMetricf(gauge, timer.Min, timer.Hostname, timer.Tags, "%s.lower", key)
+			}
+			if !d.disabledSubtypes.Upper {
+				fl.addMetricf(gauge, timer.Max, timer.Hostname, timer.Tags, "%s.upper", key)
+			}
+			if !d.disabledSubtypes.Count {
+				fl.addMetricf(gauge, float64(timer.Count), timer.Hostname, timer.Tags, "%s.count", key)
+			}
+			if !d.disabledSubtypes.CountPerSecond {
+				fl.addMetricf(rate, timer.PerSecond, timer.Hostname, timer.Tags, "%s.count_ps", key)
+			}
+			if !d.disabledSubtypes.Mean {
+				fl.addMetricf(gauge, timer.Mean, timer.Hostname, timer.Tags, "%s.mean", key)
+			}
+			if !d.disabledSubtypes.Median {
+				fl.addMetricf(gauge, timer.Median, timer.Hostname, timer.Tags, "%s.median", key)
+			}
+			if !d.disabledSubtypes.StdDev {
+				fl.addMetricf(gauge, timer.StdDev, timer.Hostname, timer.Tags, "%s.std", key)
+			}
+			if !d.disabledSubtypes.Sum {
+				fl.addMetricf(gauge, timer.Sum, timer.Hostname, timer.Tags, "%s.sum", key)
+			}
+			if !d.disabledSubtypes.SumSquares {
+				fl.addMetricf(gauge, timer.SumSquares, timer.Hostname, timer.Tags, "%s.sum_squares", key)
+			}
+			for _, pct := range timer.Percentiles {
+				fl.addMetricf(gauge, pct.Float, timer.Hostname, timer.Tags, "%s.%s", key, pct.Str)
+			}
 		}
 		fl.maybeFlush()
 	})
diff --git a/pkg/backends/datadog/flush.go b/pkg/backends/datadog/flush.go
index 5431370..53d00c1 100644
--- a/pkg/backends/datadog/flush.go
+++ b/pkg/backends/datadog/flush.go
@@ -13,6 +13,8 @@ const (
 	gauge metricType = "gauge"
 	// rate is datadog rate type.
 	rate metricType = "rate"
+	// counter is a datadog counter type.
+	counter metricType = "count"
 )
 
 // flush represents a send operation.
diff --git a/pkg/backends/graphite/graphite.go b/pkg/backends/graphite/graphite.go
index 0fd6690..0b18b02 100644
--- a/pkg/backends/graphite/graphite.go
+++ b/pkg/backends/graphite/graphite.go
@@ -4,8 +4,10 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"math"
 	"net"
 	"regexp"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -159,35 +161,46 @@ func (client *Client) preparePayload(metrics *gostatsd.MetricMap, ts time.Time)
 		})
 	}
 	metrics.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
-		if !client.disabledSubtypes.Lower {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "lower", timer.Hostname, timer.Tags), timer.Min, now)
-		}
-		if !client.disabledSubtypes.Upper {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "upper", timer.Hostname, timer.Tags), timer.Max, now)
-		}
-		if !client.disabledSubtypes.Count {
-			_, _ = fmt.Fprintf(buf, "%s %d %d\n", client.prepareName(client.timerNamespace, key, "count", timer.Hostname, timer.Tags), timer.Count, now)
-		}
-		if !client.disabledSubtypes.CountPerSecond {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "count_ps", timer.Hostname, timer.Tags), timer.PerSecond, now)
-		}
-		if !client.disabledSubtypes.Mean {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "mean", timer.Hostname, timer.Tags), timer.Mean, now)
-		}
-		if !client.disabledSubtypes.Median {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "median", timer.Hostname, timer.Tags), timer.Median, now)
-		}
-		if !client.disabledSubtypes.StdDev {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "std", timer.Hostname, timer.Tags), timer.StdDev, now)
-		}
-		if !client.disabledSubtypes.Sum {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "sum", timer.Hostname, timer.Tags), timer.Sum, now)
-		}
-		if !client.disabledSubtypes.SumSquares {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "sum_squares", timer.Hostname, timer.Tags), timer.SumSquares, now)
-		}
-		for _, pct := range timer.Percentiles {
-			_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, pct.Str, timer.Hostname, timer.Tags), pct.Float, now)
+		if timer.Histogram != nil {
+			for histogramThreshold, count := range timer.Histogram {
+				bucketTag := "le:+Inf"
+				if !math.IsInf(float64(histogramThreshold), 1) {
+					bucketTag = "le:" + strconv.FormatFloat(float64(histogramThreshold), 'f', -1, 64)
+				}
+				newTags := timer.Tags.Concat(gostatsd.Tags{bucketTag})
+				_, _ = fmt.Fprintf(buf, "%s %d %d\n", client.prepareName(client.counterNamespace, key, "histogram", timer.Hostname, newTags), count, now)
+			}
+		} else {
+			if !client.disabledSubtypes.Lower {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "lower", timer.Hostname, timer.Tags), timer.Min, now)
+			}
+			if !client.disabledSubtypes.Upper {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "upper", timer.Hostname, timer.Tags), timer.Max, now)
+			}
+			if !client.disabledSubtypes.Count {
+				_, _ = fmt.Fprintf(buf, "%s %d %d\n", client.prepareName(client.timerNamespace, key, "count", timer.Hostname, timer.Tags), timer.Count, now)
+			}
+			if !client.disabledSubtypes.CountPerSecond {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "count_ps", timer.Hostname, timer.Tags), timer.PerSecond, now)
+			}
+			if !client.disabledSubtypes.Mean {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "mean", timer.Hostname, timer.Tags), timer.Mean, now)
+			}
+			if !client.disabledSubtypes.Median {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "median", timer.Hostname, timer.Tags), timer.Median, now)
+			}
+			if !client.disabledSubtypes.StdDev {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "std", timer.Hostname, timer.Tags), timer.StdDev, now)
+			}
+			if !client.disabledSubtypes.Sum {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "sum", timer.Hostname, timer.Tags), timer.Sum, now)
+			}
+			if !client.disabledSubtypes.SumSquares {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, "sum_squares", timer.Hostname, timer.Tags), timer.SumSquares, now)
+			}
+			for _, pct := range timer.Percentiles {
+				_, _ = fmt.Fprintf(buf, "%s %f %d\n", client.prepareName(client.timerNamespace, key, pct.Str, timer.Hostname, timer.Tags), pct.Float, now)
+			}
 		}
 	})
 	metrics.Gauges.Each(func(key, tagsKey string, gauge gostatsd.Gauge) {
diff --git a/pkg/backends/newrelic/newrelic.go b/pkg/backends/newrelic/newrelic.go
index 468dd49..aa8b22c 100644
--- a/pkg/backends/newrelic/newrelic.go
+++ b/pkg/backends/newrelic/newrelic.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"runtime"
 	"strconv"
@@ -214,7 +215,18 @@ func (n *Client) processMetrics(metrics *gostatsd.MetricMap, cb func(*timeSeries
 	})
 
 	metrics.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
-		fl.addTimerMetric(n, "timer", timer, tagsKey, key)
+		if timer.Histogram != nil {
+			for histogramThreshold, count := range timer.Histogram {
+				bucketTag := "le:infinity"
+				if !math.IsInf(float64(histogramThreshold), 1) {
+					bucketTag = "le:" + strconv.FormatFloat(float64(histogramThreshold), 'f', -1, 64)
+				}
+				newTags := timer.Tags.Concat(gostatsd.Tags{bucketTag})
+				fl.addMetric(n, "counter", float64(count), 0, timer.Hostname, newTags, key+".histogram", timer.Timestamp)
+			}
+		} else {
+			fl.addTimerMetric(n, "timer", timer, tagsKey, key)
+		}
 		fl.maybeFlush()
 	})
 
diff --git a/pkg/backends/stdout/stdout.go b/pkg/backends/stdout/stdout.go
index 1dc5eaf..e687f67 100644
--- a/pkg/backends/stdout/stdout.go
+++ b/pkg/backends/stdout/stdout.go
@@ -4,12 +4,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"math"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/transport"
-
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )
@@ -81,35 +82,46 @@ func preparePayload(metrics *gostatsd.MetricMap, disabled *gostatsd.TimerSubtype
 	})
 	metrics.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
 		nk := composeMetricName(key, tagsKey)
-		if !disabled.Lower {
-			fmt.Fprintf(buf, "stats.timers.%s.lower %f %d\n", nk, timer.Min, now) // #nosec
-		}
-		if !disabled.Upper {
-			fmt.Fprintf(buf, "stats.timers.%s.upper %f %d\n", nk, timer.Max, now) // #nosec
-		}
-		if !disabled.Count {
-			fmt.Fprintf(buf, "stats.timers.%s.count %d %d\n", nk, timer.Count, now) // #nosec
-		}
-		if !disabled.CountPerSecond {
-			fmt.Fprintf(buf, "stats.timers.%s.count_ps %f %d\n", nk, timer.PerSecond, now) // #nosec
-		}
-		if !disabled.Mean {
-			fmt.Fprintf(buf, "stats.timers.%s.mean %f %d\n", nk, timer.Mean, now) // #nosec
-		}
-		if !disabled.Median {
-			fmt.Fprintf(buf, "stats.timers.%s.median %f %d\n", nk, timer.Median, now) // #nosec
-		}
-		if !disabled.StdDev {
-			fmt.Fprintf(buf, "stats.timers.%s.std %f %d\n", nk, timer.StdDev, now) // #nosec
-		}
-		if !disabled.Sum {
-			fmt.Fprintf(buf, "stats.timers.%s.sum %f %d\n", nk, timer.Sum, now) // #nosec
-		}
-		if !disabled.SumSquares {
-			fmt.Fprintf(buf, "stats.timers.%s.sum_squares %f %d\n", nk, timer.SumSquares, now) // #nosec
-		}
-		for _, pct := range timer.Percentiles {
-			fmt.Fprintf(buf, "stats.timers.%s.%s %f %d\n", nk, pct.Str, pct.Float, now) // #nosec
+		if timer.Histogram != nil {
+			nk := composeMetricName(key, tagsKey)
+			for histogramThreshold, count := range timer.Histogram {
+				bucketTag := "le:+Inf"
+				if !math.IsInf(float64(histogramThreshold), 1) {
+					bucketTag = "le:" + strconv.FormatFloat(float64(histogramThreshold), 'f', -1, 64)
+				}
+				fmt.Fprintf(buf, "stats.timers.%s.histogram.%s %d %d\n", nk, bucketTag, count, now) // #nosec
+			}
+		} else {
+			if !disabled.Lower {
+				fmt.Fprintf(buf, "stats.timers.%s.lower %f %d\n", nk, timer.Min, now) // #nosec
+			}
+			if !disabled.Upper {
+				fmt.Fprintf(buf, "stats.timers.%s.upper %f %d\n", nk, timer.Max, now) // #nosec
+			}
+			if !disabled.Count {
+				fmt.Fprintf(buf, "stats.timers.%s.count %d %d\n", nk, timer.Count, now) // #nosec
+			}
+			if !disabled.CountPerSecond {
+				fmt.Fprintf(buf, "stats.timers.%s.count_ps %f %d\n", nk, timer.PerSecond, now) // #nosec
+			}
+			if !disabled.Mean {
+				fmt.Fprintf(buf, "stats.timers.%s.mean %f %d\n", nk, timer.Mean, now) // #nosec
+			}
+			if !disabled.Median {
+				fmt.Fprintf(buf, "stats.timers.%s.median %f %d\n", nk, timer.Median, now) // #nosec
+			}
+			if !disabled.StdDev {
+				fmt.Fprintf(buf, "stats.timers.%s.std %f %d\n", nk, timer.StdDev, now) // #nosec
+			}
+			if !disabled.Sum {
+				fmt.Fprintf(buf, "stats.timers.%s.sum %f %d\n", nk, timer.Sum, now) // #nosec
+			}
+			if !disabled.SumSquares {
+				fmt.Fprintf(buf, "stats.timers.%s.sum_squares %f %d\n", nk, timer.SumSquares, now) // #nosec
+			}
+			for _, pct := range timer.Percentiles {
+				fmt.Fprintf(buf, "stats.timers.%s.%s %f %d\n", nk, pct.Str, pct.Float, now) // #nosec
+			}
 		}
 	})
 	metrics.Gauges.Each(func(key, tagsKey string, gauge gostatsd.Gauge) {
diff --git a/pkg/statsd/aggregator.go b/pkg/statsd/aggregator.go
index f50031b..0d6d917 100644
--- a/pkg/statsd/aggregator.go
+++ b/pkg/statsd/aggregator.go
@@ -30,11 +30,12 @@ type MetricAggregator struct {
 	now                func() time.Time // Returns current time. Useful for testing.
 	statser            stats.Statser
 	disabledSubtypes   gostatsd.TimerSubtypes
+	histogramLimit     uint32
 	metricMap          *gostatsd.MetricMap
 }
 
 // NewMetricAggregator creates a new MetricAggregator object.
-func NewMetricAggregator(percentThresholds []float64, expiryInterval time.Duration, disabled gostatsd.TimerSubtypes) *MetricAggregator {
+func NewMetricAggregator(percentThresholds []float64, expiryInterval time.Duration, disabled gostatsd.TimerSubtypes, histogramLimit uint32) *MetricAggregator {
 	a := MetricAggregator{
 		expiryInterval:    expiryInterval,
 		percentThresholds: make(map[float64]percentStruct, len(percentThresholds)),
@@ -42,6 +43,7 @@ func NewMetricAggregator(percentThresholds []float64, expiryInterval time.Durati
 		statser:           stats.NewNullStatser(), // Will probably be replaced via RunMetrics
 		metricMap:         gostatsd.NewMetricMap(),
 		disabledSubtypes:  disabled,
+		histogramLimit:    histogramLimit,
 	}
 	for _, pct := range percentThresholds {
 		sPct := strconv.Itoa(int(pct))
@@ -76,6 +78,12 @@ func (a *MetricAggregator) Flush(flushInterval time.Duration) {
 	})
 
 	a.metricMap.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
+		if hasHistogramTag(timer) {
+			timer.Histogram = latencyHistogram(timer, a.histogramLimit)
+			a.metricMap.Timers[key][tagsKey] = timer
+			return
+		}
+
 		if count := len(timer.Values); count > 0 {
 			sort.Float64s(timer.Values)
 			timer.Min = timer.Values[0]
@@ -162,13 +170,12 @@ func (a *MetricAggregator) Flush(flushInterval time.Duration) {
 
 			timer.Count = int(round(timer.SampledCount))
 			timer.PerSecond = timer.SampledCount / flushInSeconds
-
-			a.metricMap.Timers[key][tagsKey] = timer
 		} else {
 			timer.Count = 0
 			timer.SampledCount = 0
 			timer.PerSecond = 0
 		}
+		a.metricMap.Timers[key][tagsKey] = timer
 	})
 }
 
@@ -213,11 +220,21 @@ func (a *MetricAggregator) Reset() {
 		if a.isExpired(nowNano, timer.Timestamp) {
 			deleteMetric(key, tagsKey, a.metricMap.Timers)
 		} else {
-			a.metricMap.Timers[key][tagsKey] = gostatsd.Timer{
-				Timestamp: timer.Timestamp,
-				Hostname:  timer.Hostname,
-				Tags:      timer.Tags,
-				Values:    timer.Values[:0],
+			if hasHistogramTag(timer) {
+				a.metricMap.Timers[key][tagsKey] = gostatsd.Timer{
+					Timestamp: timer.Timestamp,
+					Hostname:  timer.Hostname,
+					Tags:      timer.Tags,
+					Values:    timer.Values[:0],
+					Histogram: emptyHistogram(timer, a.histogramLimit),
+				}
+			} else {
+				a.metricMap.Timers[key][tagsKey] = gostatsd.Timer{
+					Timestamp: timer.Timestamp,
+					Hostname:  timer.Hostname,
+					Tags:      timer.Tags,
+					Values:    timer.Values[:0],
+				}
 			}
 		}
 	})
diff --git a/pkg/statsd/latency_histogram.go b/pkg/statsd/latency_histogram.go
new file mode 100644
index 0000000..717f2ad
--- /dev/null
+++ b/pkg/statsd/latency_histogram.go
@@ -0,0 +1,105 @@
+package statsd
+
+import (
+	"math"
+	"strconv"
+	"strings"
+
+	"github.com/atlassian/gostatsd"
+)
+
+const (
+	histogramThresholdsTagPrefix = "gsd_histogram:"
+	histogramThresholdsSeparator = "_"
+)
+
+func latencyHistogram(timer gostatsd.Timer, bucketLimit uint32) map[gostatsd.HistogramThreshold]int {
+	result := emptyHistogram(timer, bucketLimit)
+
+	if len(result) == 0 {
+		return result
+	}
+
+	infiniteThreshold := gostatsd.HistogramThreshold(math.Inf(1))
+
+	for _, value := range timer.Values {
+		for latencyBucket := range result {
+			if value <= float64(latencyBucket) {
+				result[latencyBucket] += 1
+			}
+		}
+	}
+	result[infiniteThreshold] = len(timer.Values)
+
+	return result
+}
+
+func emptyHistogram(timer gostatsd.Timer, bucketLimit uint32) map[gostatsd.HistogramThreshold]int {
+	result := make(map[gostatsd.HistogramThreshold]int)
+
+	if bucketLimit == 0 {
+		return result
+	}
+
+	thresholds := retrieveThresholds(timer, bucketLimit)
+
+	if thresholds == nil {
+		return nil
+	}
+	infiniteThreshold := gostatsd.HistogramThreshold(math.Inf(1))
+
+	for _, histogramThreshold := range thresholds {
+		result[histogramThreshold] = 0
+	}
+
+	result[infiniteThreshold] = 0
+
+	return result
+}
+
+func retrieveThresholds(timer gostatsd.Timer, bucketlimit uint32) []gostatsd.HistogramThreshold {
+	tag, found := findTag(timer.Tags, histogramThresholdsTagPrefix)
+	if found {
+		bucketsTagValue := tag[len(histogramThresholdsTagPrefix):]
+		stringThresholds := strings.Split(bucketsTagValue, histogramThresholdsSeparator)
+		floatThresholds := mapToThresholds(stringThresholds)
+		floatThresholds = floatThresholds[:(min(uint32(len(floatThresholds)), bucketlimit))]
+		if floatThresholds == nil {
+			return []gostatsd.HistogramThreshold{}
+		}
+		return floatThresholds
+	}
+	return nil
+}
+
+func mapToThresholds(vs []string) []gostatsd.HistogramThreshold {
+	var lb []gostatsd.HistogramThreshold
+	for _, v := range vs {
+		floatBucket, err := strconv.ParseFloat(v, 64)
+		if err == nil {
+			lb = append(lb, gostatsd.HistogramThreshold(floatBucket))
+		}
+	}
+	return lb
+}
+
+func hasHistogramTag(timer gostatsd.Timer) bool {
+	_, found := findTag(timer.Tags, histogramThresholdsTagPrefix)
+	return found
+}
+
+func findTag(a []string, prefix string) (string, bool) {
+	for _, n := range a {
+		if strings.HasPrefix(n, prefix) {
+			return n, true
+		}
+	}
+	return "", false
+}
+
+func min(a, b uint32) uint32 {
+	if a < b {
+		return a
+	}
+	return b
+}
diff --git a/pkg/statsd/statsd.go b/pkg/statsd/statsd.go
index f0c9a35..68e7f58 100644
--- a/pkg/statsd/statsd.go
+++ b/pkg/statsd/statsd.go
@@ -48,6 +48,7 @@ type Server struct {
 	HeartbeatTags             gostatsd.Tags
 	ReceiveBatchSize          int
 	DisabledSubTypes          gostatsd.TimerSubtypes
+	HistogramLimit            uint32
 	BadLineRateLimitPerSecond rate.Limit
 	ServerMode                string
 	Hostname                  string
@@ -98,6 +99,7 @@ func (s *Server) createStandaloneSink() (gostatsd.PipelineHandler, []gostatsd.Ru
 		percentThresholds: s.PercentThreshold,
 		expiryInterval:    s.ExpiryInterval,
 		disabledSubtypes:  s.DisabledSubTypes,
+		histogramLimit:    s.HistogramLimit,
 	}
 
 	backendHandler := NewBackendHandler(s.Backends, uint(s.MaxConcurrentEvents), s.MaxWorkers, s.MaxQueueSize, &factory)
@@ -272,10 +274,11 @@ type agrFactory struct {
 	percentThresholds []float64
 	expiryInterval    time.Duration
 	disabledSubtypes  gostatsd.TimerSubtypes
+	histogramLimit    uint32
 }
 
 func (af *agrFactory) Create() Aggregator {
-	return NewMetricAggregator(af.percentThresholds, af.expiryInterval, af.disabledSubtypes)
+	return NewMetricAggregator(af.percentThresholds, af.expiryInterval, af.disabledSubtypes, af.histogramLimit)
 }
 
 func toStringSlice(fs []float64) []string {
diff --git a/pkg/statsd/statsd_defaults_and_params.go b/pkg/statsd/statsd_defaults_and_params.go
index cb61f3a..a1f07b8 100644
--- a/pkg/statsd/statsd_defaults_and_params.go
+++ b/pkg/statsd/statsd_defaults_and_params.go
@@ -1,6 +1,7 @@
 package statsd
 
 import (
+	"math"
 	"runtime"
 	"strings"
 	"time"
@@ -122,6 +123,8 @@ const (
 	DefaultBadLinesPerMinute = 0
 	// DefaultServerMode is the default mode to run as, standalone|forwarder
 	DefaultServerMode = "standalone"
+	// DefaultTimerHistogramLimit default upper limit for timer histograms (effectively unlimited)
+	DefaultTimerHistogramLimit = math.MaxUint32
 	// DefaultLogRawMetric is the default value for whether to log the metrics received from network
 	DefaultLogRawMetric = false
 )
@@ -187,6 +190,8 @@ const (
 	ParamServerMode = "server-mode"
 	// ParamHostname allows hostname overrides
 	ParamHostname = "hostname"
+	// ParamTimerHistogramLimit upper limit of timer histogram buckets that can be specified
+	ParamTimerHistogramLimit = "timer-histogram-limit"
 	// ParamLogRawMetric enables custom metrics to be printed to stdout
 	ParamLogRawMetric = "log-raw-metric"
 )
@@ -222,6 +227,7 @@ func AddFlags(fs *pflag.FlagSet) {
 	fs.Bool(ParamConnPerReader, DefaultConnPerReader, "Create a separate connection per reader (requires system support for reusing addresses)")
 	fs.String(ParamServerMode, DefaultServerMode, "The server mode to run in")
 	fs.String(ParamHostname, getHost(), "overrides the hostname of the server")
+	fs.Uint32(ParamTimerHistogramLimit, DefaultTimerHistogramLimit, "upper limit of timer histogram buckets (MaxUint32 by default)")
 	fs.Bool(ParamLogRawMetric, DefaultLogRawMetric, "Print metrics received from network to stdout in JSON format")
 }
 
diff --git a/timers.go b/timers.go
index 8b1530f..b164a92 100644
--- a/timers.go
+++ b/timers.go
@@ -19,8 +19,14 @@ type Timer struct {
 	Timestamp    Nanotime    // Last time value was updated
 	Hostname     string      // Hostname of the source of the metric
 	Tags         Tags        // The tags for the timer
+
+	// Map bounds to count of measures seen in that bucket.
+	// This map only non-empty if the metric specifies histogram aggregation in its tags.
+	Histogram map[HistogramThreshold]int
 }
 
+type HistogramThreshold float64
+
 // NewTimer initialises a new timer.
 func NewTimer(timestamp Nanotime, values []float64, hostname string, tags Tags) Timer {
 	return Timer{Values: values, Timestamp: timestamp, Hostname: hostname, Tags: tags.Copy(), SampledCount: float64(len(values))}


Test patch:
diff --git a/pkg/backends/cloudwatch/cloudwatch_test.go b/pkg/backends/cloudwatch/cloudwatch_test.go
index 2e0fca9..ee45691 100644
--- a/pkg/backends/cloudwatch/cloudwatch_test.go
+++ b/pkg/backends/cloudwatch/cloudwatch_test.go
@@ -2,6 +2,7 @@ package cloudwatch
 
 import (
 	"context"
+	"math"
 	"testing"
 
 	"github.com/sirupsen/logrus"
@@ -189,3 +190,69 @@ func metricsOneOfEach() *gostatsd.MetricMap {
 		},
 	}
 }
+
+func TestSendHistogram(t *testing.T) {
+	t.Parallel()
+
+	p := transport.NewTransportPool(logrus.New(), viper.New())
+	cli, err := NewClient("ns", "default", gostatsd.TimerSubtypes{}, p)
+	require.NoError(t, err)
+
+	metricMap := &gostatsd.MetricMap{
+		Timers: gostatsd.Timers{
+			"t1": map[string]gostatsd.Timer{
+				"tag2": {
+					Values:    []float64{0, 1},
+					Timestamp: gostatsd.Nanotime(200),
+					Hostname:  "h2",
+					Tags:      gostatsd.Tags{"tag2", "gsd_histogram:20_30_40_50_60"},
+					Histogram: map[gostatsd.HistogramThreshold]int{
+						20:                                       5,
+						30:                                       10,
+						40:                                       10,
+						50:                                       10,
+						60:                                       19,
+						gostatsd.HistogramThreshold(math.Inf(1)): 19,
+					},
+				},
+			},
+		},
+	}
+
+	cli.cloudwatch = &mockedCloudwatch{
+		PutMetricDataHandler: func(input *cloudwatch.PutMetricDataInput) (*cloudwatch.PutMetricDataOutput, error) {
+			assert.Equal(t, *findMetricDatum(input, "stats.timers.t1.histogram", "le", "20").Value, float64(5))
+			assert.Equal(t, *findMetricDatum(input, "stats.timers.t1.histogram", "le", "30").Value, float64(10))
+			assert.Equal(t, *findMetricDatum(input, "stats.timers.t1.histogram", "le", "40").Value, float64(10))
+			assert.Equal(t, *findMetricDatum(input, "stats.timers.t1.histogram", "le", "50").Value, float64(10))
+			assert.Equal(t, *findMetricDatum(input, "stats.timers.t1.histogram", "le", "60").Value, float64(19))
+			assert.Equal(t, *findMetricDatum(input, "stats.timers.t1.histogram", "le", "+Inf").Value, float64(19))
+			assert.Len(t, input.MetricData, 6)
+			return nil, nil
+		},
+	}
+
+	res := make(chan []error, 1)
+	cli.SendMetricsAsync(context.Background(), metricMap, func(errs []error) {
+		res <- errs
+	})
+	errs := <-res
+	for _, err := range errs {
+		assert.NoError(t, err)
+	}
+}
+
+func findMetricDatum(input *cloudwatch.PutMetricDataInput, name string, dimenensionName string, dimensionValue string) *cloudwatch.MetricDatum {
+	for _, data := range input.MetricData {
+		MetricName := *data.MetricName
+		if MetricName != name {
+			continue
+		}
+		for _, dimension := range data.Dimensions {
+			if *dimension.Name == dimenensionName && *dimension.Value == dimensionValue {
+				return data
+			}
+		}
+	}
+	return nil
+}
diff --git a/pkg/backends/datadog/datadog_test.go b/pkg/backends/datadog/datadog_test.go
index b03cdad..fa7e468 100644
--- a/pkg/backends/datadog/datadog_test.go
+++ b/pkg/backends/datadog/datadog_test.go
@@ -4,7 +4,9 @@ import (
 	"bytes"
 	"compress/zlib"
 	"context"
+	"index/suffixarray"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"net/http/httptest"
 	"sync/atomic"
@@ -16,7 +18,6 @@ import (
 
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/transport"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -209,3 +210,83 @@ func metricsOneOfEach() *gostatsd.MetricMap {
 		},
 	}
 }
+
+func TestSendHistogram(t *testing.T) {
+	t.Parallel()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/series", func(w http.ResponseWriter, r *http.Request) {
+		data, err := ioutil.ReadAll(r.Body)
+		if !assert.NoError(t, err) {
+			return
+		}
+		enc := r.Header.Get("Content-Encoding")
+		if enc == "deflate" {
+			decompressor, err := zlib.NewReader(bytes.NewReader(data))
+			if !assert.NoError(t, err) {
+				return
+			}
+			data, err = ioutil.ReadAll(decompressor)
+			assert.NoError(t, err)
+		}
+		expected := []string{
+			`{"host":"h2","interval":1.1,"metric":"t1.histogram","points":[[100,5]],"tags":["tag2","gsd_histogram:20_30_40_50_60","le:20"],"type":"count"}`,
+			`{"host":"h2","interval":1.1,"metric":"t1.histogram","points":[[100,10]],"tags":["tag2","gsd_histogram:20_30_40_50_60","le:30"],"type":"count"}`,
+			`{"host":"h2","interval":1.1,"metric":"t1.histogram","points":[[100,10]],"tags":["tag2","gsd_histogram:20_30_40_50_60","le:40"],"type":"count"}`,
+			`{"host":"h2","interval":1.1,"metric":"t1.histogram","points":[[100,10]],"tags":["tag2","gsd_histogram:20_30_40_50_60","le:50"],"type":"count"}`,
+			`{"host":"h2","interval":1.1,"metric":"t1.histogram","points":[[100,19]],"tags":["tag2","gsd_histogram:20_30_40_50_60","le:60"],"type":"count"}`,
+			`{"host":"h2","interval":1.1,"metric":"t1.histogram","points":[[100,19]],"tags":["tag2","gsd_histogram:20_30_40_50_60","le:+Inf"],"type":"count"}`,
+		}
+
+		for _, e := range expected {
+			assert.Contains(t, string(data), e)
+		}
+		assert.Equal(t, 6, countMatches(string(data), "metric"))
+	})
+	ts := httptest.NewServer(mux)
+	defer ts.Close()
+
+	v := viper.New()
+	v.Set("transport.default.client-timeout", 1*time.Second)
+	p := transport.NewTransportPool(logrus.New(), v)
+	client, err := NewClient(ts.URL, "apiKey123", "agent", "default", 1000, defaultMaxRequests, true, 2*time.Second, 1100*time.Millisecond, gostatsd.TimerSubtypes{}, p)
+	require.NoError(t, err)
+	client.now = func() time.Time {
+		return time.Unix(100, 0)
+	}
+	res := make(chan []error, 1)
+	client.SendMetricsAsync(context.Background(), metricsWithHistogram(), func(errs []error) {
+		res <- errs
+	})
+	errs := <-res
+	for _, err := range errs {
+		assert.NoError(t, err)
+	}
+}
+
+func metricsWithHistogram() *gostatsd.MetricMap {
+	return &gostatsd.MetricMap{
+		Timers: gostatsd.Timers{
+			"t1": map[string]gostatsd.Timer{
+				"tag2": {
+					Values:    []float64{0, 1},
+					Timestamp: gostatsd.Nanotime(200),
+					Hostname:  "h2",
+					Tags:      gostatsd.Tags{"tag2", "gsd_histogram:20_30_40_50_60"},
+					Histogram: map[gostatsd.HistogramThreshold]int{
+						20:                                       5,
+						30:                                       10,
+						40:                                       10,
+						50:                                       10,
+						60:                                       19,
+						gostatsd.HistogramThreshold(math.Inf(1)): 19,
+					},
+				},
+			},
+		},
+	}
+}
+
+func countMatches(s string, m string) int {
+	index := suffixarray.New([]byte(s))
+	return len(index.Lookup([]byte(m), -1))
+}
diff --git a/pkg/backends/graphite/graphite_test.go b/pkg/backends/graphite/graphite_test.go
index 4832751..e250b3b 100644
--- a/pkg/backends/graphite/graphite_test.go
+++ b/pkg/backends/graphite/graphite_test.go
@@ -3,6 +3,7 @@ package graphite
 import (
 	"context"
 	"io"
+	"math"
 	"net"
 	"sort"
 	"strings"
@@ -110,6 +111,25 @@ func TestPreparePayloadTags(t *testing.T) {
 	require.Equal(t, expected, actual)
 }
 
+func TestPreparePayloadHistogram(t *testing.T) {
+	t.Parallel()
+	metrics := metricsWithHistogram()
+	expected :=
+		"gp.pc.t1.histogram.gs;le=20 5 1234\n" +
+			"gp.pc.t1.histogram.gs;le=30 10 1234\n" +
+			"gp.pc.t1.histogram.gs;le=40 10 1234\n" +
+			"gp.pc.t1.histogram.gs;le=50 10 1234\n" +
+			"gp.pc.t1.histogram.gs;le=60 19 1234\n" +
+			"gp.pc.t1.histogram.gs;le=+Inf 19 1234\n"
+
+	cl, err := NewClient("127.0.0.1:9", 1*time.Second, 1*time.Second, "gp", "pc", "pt", "pg", "ps", "gs", "tags", gostatsd.TimerSubtypes{})
+	require.NoError(t, err)
+	b := cl.preparePayload(metrics, time.Unix(1234, 0))
+	expected = sortLines(expected)
+	actual := sortLines(b.String())
+	require.Equal(t, expected, actual)
+}
+
 func sortLines(s string) string {
 	lines := strings.Split(s, "\n")
 	sort.Strings(lines)
@@ -177,6 +197,22 @@ func metrics() *gostatsd.MetricMap {
 	return mm
 }
 
+func metricsWithHistogram() *gostatsd.MetricMap {
+	timestamp := gostatsd.Nanotime(time.Unix(123456, 0).UnixNano())
+
+	mm := gostatsd.NewMetricMap()
+	mm.Timers["t1"] = map[string]gostatsd.Timer{}
+	mm.Timers["t1"]["gsd_histogram:20_30_40_50_60"] = gostatsd.Timer{Values: []float64{10}, Timestamp: timestamp, Histogram: map[gostatsd.HistogramThreshold]int{
+		20:                                       5,
+		30:                                       10,
+		40:                                       10,
+		50:                                       10,
+		60:                                       19,
+		gostatsd.HistogramThreshold(math.Inf(1)): 19,
+	}}
+	return mm
+}
+
 func metricsWithTags() *gostatsd.MetricMap {
 	timestamp := gostatsd.Nanotime(time.Unix(123456, 0).UnixNano())
 
diff --git a/pkg/backends/newrelic/newrelic_test.go b/pkg/backends/newrelic/newrelic_test.go
index e2d7faf..6b58feb 100644
--- a/pkg/backends/newrelic/newrelic_test.go
+++ b/pkg/backends/newrelic/newrelic_test.go
@@ -4,7 +4,9 @@ import (
 	"compress/gzip"
 	"context"
 	"encoding/json"
+	"index/suffixarray"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"net/http/httptest"
 	"sync/atomic"
@@ -215,6 +217,54 @@ func decodeBody(enc string, r *http.Request, t *testing.T) (string, bool) {
 	return body, false
 }
 
+func TestSendMetricsWithHistogram(t *testing.T) {
+	t.Parallel()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/v1/data", func(w http.ResponseWriter, r *http.Request) {
+		data, err := ioutil.ReadAll(r.Body)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		expected := []string{
+			`{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"le":20,"metric_name":"t1.histogram","metric_per_second":0,"metric_type":"counter","metric_value":5,"timestamp":0}`,
+			`{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"le":30,"metric_name":"t1.histogram","metric_per_second":0,"metric_type":"counter","metric_value":10,"timestamp":0}`,
+			`{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"le":40,"metric_name":"t1.histogram","metric_per_second":0,"metric_type":"counter","metric_value":10,"timestamp":0}`,
+			`{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"le":50,"metric_name":"t1.histogram","metric_per_second":0,"metric_type":"counter","metric_value":10,"timestamp":0}`,
+			`{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"le":60,"metric_name":"t1.histogram","metric_per_second":0,"metric_type":"counter","metric_value":19,"timestamp":0}`,
+			`{"event_type":"GoStatsD","integration_version":"2.3.0","interval":1,"le":"infinity","metric_name":"t1.histogram","metric_per_second":0,"metric_type":"counter","metric_value":19,"timestamp":0}`,
+		}
+
+		for _, e := range expected {
+			assert.Contains(t, string(data), e)
+		}
+		assert.Equal(t, 6, countMatches(string(data), "metric_name"))
+	})
+	ts := httptest.NewServer(mux)
+	defer ts.Close()
+
+	v := viper.New()
+	v.SetDefault("transport.default.client-timeout", 1*time.Second)
+	p := transport.NewTransportPool(logrus.New(), v)
+	client, err := NewClient("default", ts.URL+"/v1/data", "", "GoStatsD", "", "", "", "metric_name", "metric_type",
+		"metric_per_second", "metric_value", "samples_min", "samples_max", "samples_count",
+		"samples_mean", "samples_median", "samples_std_dev", "samples_sum", "samples_sum_squares", "agent",
+		defaultMetricsPerBatch, defaultMaxRequests, 2*time.Second, 1*time.Second, gostatsd.TimerSubtypes{}, p)
+
+	require.NoError(t, err)
+	client.now = func() time.Time {
+		return time.Unix(100, 0)
+	}
+	res := make(chan []error, 1)
+	client.SendMetricsAsync(context.Background(), metricsWithHistogram(), func(errs []error) {
+		res <- errs
+	})
+	errs := <-res
+	for _, err := range errs {
+		assert.NoError(t, err)
+	}
+}
+
 // twoCounters returns two counters.
 func twoCounters() *gostatsd.MetricMap {
 	return &gostatsd.MetricMap{
@@ -280,6 +330,23 @@ func metricsOneOfEach() *gostatsd.MetricMap {
 	}
 }
 
+func metricsWithHistogram() *gostatsd.MetricMap {
+	mm := gostatsd.NewMetricMap()
+	mm.Timers["t1"] = map[string]gostatsd.Timer{}
+	mm.Timers["t1"]["gsd_histogram:20_30_40_50_60"] = gostatsd.Timer{
+		Values:    []float64{10},
+		Timestamp: 0,
+		Histogram: map[gostatsd.HistogramThreshold]int{
+			20:                                       5,
+			30:                                       10,
+			40:                                       10,
+			50:                                       10,
+			60:                                       19,
+			gostatsd.HistogramThreshold(math.Inf(1)): 19,
+		}}
+	return mm
+}
+
 func TestEventFormatter(t *testing.T) {
 	t.Parallel()
 
@@ -322,3 +389,8 @@ func TestEventFormatter(t *testing.T) {
 		})
 	}
 }
+
+func countMatches(s string, m string) int {
+	index := suffixarray.New([]byte(s))
+	return len(index.Lookup([]byte(m), -1))
+}
diff --git a/pkg/statsd/aggregator_test.go b/pkg/statsd/aggregator_test.go
index b216d89..23784e0 100644
--- a/pkg/statsd/aggregator_test.go
+++ b/pkg/statsd/aggregator_test.go
@@ -2,6 +2,7 @@ package statsd
 
 import (
 	"context"
+	"math"
 	"runtime"
 	"testing"
 	"time"
@@ -17,6 +18,7 @@ func newFakeAggregator() *MetricAggregator {
 		[]float64{90},
 		5*time.Minute,
 		gostatsd.TimerSubtypes{},
+		math.MaxUint32,
 	)
 }
 
@@ -49,6 +51,71 @@ func TestNewAggregator(t *testing.T) {
 	}
 }
 
+func TestLatencyHistograms(t *testing.T) {
+	t.Parallel()
+	assrt := assert.New(t)
+	ma := newFakeAggregator()
+	ma.metricMap.Timers["testTimer"] = make(map[string]gostatsd.Timer)
+	values := gostatsd.NewTimerValues([]float64{10.0, 20.0, 29.9, 2000.0, -38.0, -5.0})
+	values.Tags = gostatsd.Tags{histogramThresholdsTagPrefix + "-10_0_2.5_20_50_5000"}
+	ma.metricMap.Timers["testTimer"]["simple"] = values
+
+	ma.Flush(10)
+
+	result := ma.metricMap.Timers["testTimer"]["simple"]
+	assrt.Len(result.Histogram, 7)
+	assrt.Equal(1, result.Histogram[gostatsd.HistogramThreshold(-10)])
+	assrt.Equal(2, result.Histogram[gostatsd.HistogramThreshold(0)])
+	assrt.Equal(2, result.Histogram[gostatsd.HistogramThreshold(2.5)])
+	assrt.Equal(5, result.Histogram[gostatsd.HistogramThreshold(50)])
+	assrt.Equal(6, result.Histogram[gostatsd.HistogramThreshold(5000)])
+	assrt.Equal(6, result.Histogram[gostatsd.HistogramThreshold(math.Inf(1))])
+}
+
+func TestLatencyHistogramWithNoValuesOutputHistogramWithZeros(t *testing.T) {
+	t.Parallel()
+	assrt := assert.New(t)
+	ma := newFakeAggregator()
+	ma.metricMap.Timers["testTimer"] = make(map[string]gostatsd.Timer)
+	values := gostatsd.NewTimerValues([]float64{})
+	values.Tags = gostatsd.Tags{histogramThresholdsTagPrefix + "10_20_50_5000"}
+	ma.metricMap.Timers["testTimer"]["simple"] = values
+
+	ma.Flush(10)
+
+	result := ma.metricMap.Timers["testTimer"]["simple"]
+	assrt.Len(result.Histogram, 5)
+	assrt.Equal(0, result.Histogram[gostatsd.HistogramThreshold(10)])
+	assrt.Equal(0, result.Histogram[gostatsd.HistogramThreshold(20)])
+	assrt.Equal(0, result.Histogram[gostatsd.HistogramThreshold(50)])
+	assrt.Equal(0, result.Histogram[gostatsd.HistogramThreshold(5000)])
+	assrt.Equal(0, result.Histogram[gostatsd.HistogramThreshold(math.Inf(1))])
+}
+
+func TestLatencyHistogramDisablesAggregations(t *testing.T) {
+	t.Parallel()
+	assrt := assert.New(t)
+	ma := newFakeAggregator()
+	ma.metricMap.Timers["testTimer"] = make(map[string]gostatsd.Timer)
+	values := gostatsd.NewTimerValues([]float64{10.0, 20.0, 29.9, 2000.0})
+	values.Tags = gostatsd.Tags{histogramThresholdsTagPrefix + "20_50_5000"}
+	ma.metricMap.Timers["testTimer"]["simple"] = values
+
+	ma.Flush(10)
+
+	result := ma.metricMap.Timers["testTimer"]["simple"]
+
+	assrt.Equal(0, result.Count)
+	assrt.Equal(0.0, result.PerSecond)
+	assrt.Equal(0.0, result.Mean)
+	assrt.Equal(0.0, result.Min)
+	assrt.Equal(0.0, result.Max)
+	assrt.Equal(0.0, result.StdDev)
+	assrt.Equal(0.0, result.Sum)
+	assrt.Equal(0.0, result.SumSquares)
+	assrt.Nil(result.Percentiles)
+}
+
 func TestFlush(t *testing.T) {
 	t.Parallel()
 	assrt := assert.New(t)
@@ -191,6 +258,27 @@ func TestReset(t *testing.T) {
 
 	assrt.Equal(expected.metricMap.Timers, actual.metricMap.Timers)
 
+	actual = newFakeAggregator()
+	actual.metricMap.Timers["histogram"] = map[string]gostatsd.Timer{
+		histogramThresholdsTagPrefix + "10_20_30": gostatsd.NewTimer(nowNano, []float64{}, host, gostatsd.Tags{histogramThresholdsTagPrefix + "10_20_30"}),
+	}
+
+	expected = newFakeAggregator()
+	expectedTimer := gostatsd.NewTimer(nowNano, []float64{}, host, gostatsd.Tags{histogramThresholdsTagPrefix + "10_20_30"})
+	expectedTimer.Histogram = map[gostatsd.HistogramThreshold]int{
+		gostatsd.HistogramThreshold(10):          0,
+		gostatsd.HistogramThreshold(20):          0,
+		gostatsd.HistogramThreshold(30):          0,
+		gostatsd.HistogramThreshold(math.Inf(1)): 0,
+	}
+	expected.metricMap.Timers["histogram"] = map[string]gostatsd.Timer{
+		histogramThresholdsTagPrefix + "10_20_30": expectedTimer,
+	}
+	expected.now = nowFn
+
+	actual.Reset()
+	assrt.Equal(expected.metricMap.Timers, actual.metricMap.Timers)
+
 	actual = newFakeAggregator()
 	actual.metricMap.Gauges["some"] = map[string]gostatsd.Gauge{
 		"thing":       gostatsd.NewGauge(nowNano, 50, host, nil),
@@ -370,6 +458,7 @@ func TestDisabledLower(t *testing.T) {
 		[]float64{-90},
 		5*time.Minute,
 		gostatsd.TimerSubtypes{},
+		math.MaxUint32,
 	)
 	ma.disabledSubtypes.LowerPct = true
 	ma.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
diff --git a/pkg/statsd/latency_histogram_test.go b/pkg/statsd/latency_histogram_test.go
new file mode 100644
index 0000000..bed6ed8
--- /dev/null
+++ b/pkg/statsd/latency_histogram_test.go
@@ -0,0 +1,143 @@
+package statsd
+
+import (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/atlassian/gostatsd"
+)
+
+func timer(histogramThresholds string, values ...float64) gostatsd.Timer {
+	return gostatsd.Timer{
+		Values:    values,
+		Tags:      []string{histogramThresholdsTagPrefix + histogramThresholds},
+		Histogram: map[gostatsd.HistogramThreshold]int{},
+	}
+}
+
+func TestLatencyHistogram(t *testing.T) {
+	tests := []struct {
+		name  string
+		timer gostatsd.Timer
+		want  map[gostatsd.HistogramThreshold]int
+	}{
+		{
+			name:  "happy path",
+			timer: timer("10_30_45", 1, 10, 11, 12, 29, 30, 31, 45, 100, 100000),
+			want: map[gostatsd.HistogramThreshold]int{
+				10:                                       2,
+				30:                                       6,
+				45:                                       8,
+				gostatsd.HistogramThreshold(math.Inf(1)): 10,
+			},
+		},
+		{
+			name:  "empty threshold",
+			timer: timer("10__45", 1, 10, 11, 12, 29, 30, 31, 45, 100, 100000),
+			want: map[gostatsd.HistogramThreshold]int{
+				10:                                       2,
+				45:                                       8,
+				gostatsd.HistogramThreshold(math.Inf(1)): 10,
+			},
+		},
+		{
+			name:  "float thresholds",
+			timer: timer("1.5_4_7.0", 1.4999, 1.5, 1.51, 4.0, 7.01),
+			want: map[gostatsd.HistogramThreshold]int{
+				1.5:                                      2,
+				4:                                        4,
+				7.0:                                      4,
+				gostatsd.HistogramThreshold(math.Inf(1)): 5,
+			},
+		},
+		{
+			name:  "no timer values",
+			timer: timer("1_5_10"),
+			want: map[gostatsd.HistogramThreshold]int{
+				1:                                        0,
+				5:                                        0,
+				10:                                       0,
+				gostatsd.HistogramThreshold(math.Inf(1)): 0,
+			},
+		},
+		{
+			name:  "one non parsable thresholds",
+			timer: timer("1_incorrect_10", 0, 10, 20),
+			want: map[gostatsd.HistogramThreshold]int{
+				1:                                        1,
+				10:                                       2,
+				gostatsd.HistogramThreshold(math.Inf(1)): 3,
+			},
+		},
+		{
+			name:  "totally unparsable tag",
+			timer: timer("nothresoldsatall", 0, 10, 20),
+			want: map[gostatsd.HistogramThreshold]int{
+				gostatsd.HistogramThreshold(math.Inf(1)): 3,
+			},
+		},
+		{
+			name: "timer without the tag at all",
+			timer: gostatsd.Timer{
+				Values:    []float64{1, 2, 3},
+				Tags:      []string{"some_different_tag:yep"},
+				Histogram: map[gostatsd.HistogramThreshold]int{},
+			},
+			want: nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			buckets := latencyHistogram(tt.timer, math.MaxUint32)
+			assert.Equal(t, tt.want, buckets)
+		})
+	}
+}
+
+func TestBucketLimit(t *testing.T) {
+	timer := timer("10_20_30_50", 1, 10, 20, 30, 40, 50, 60)
+	infinity := gostatsd.HistogramThreshold(math.Inf(1))
+	tests := []struct {
+		name  string
+		limit uint32
+		want  []gostatsd.HistogramThreshold
+	}{
+		{
+			name:  "no limit",
+			limit: math.MaxUint32,
+			want:  []gostatsd.HistogramThreshold{10, 20, 30, 50, infinity},
+		},
+		{
+			name:  "histogram disabled",
+			limit: 0,
+			want:  []gostatsd.HistogramThreshold{},
+		},
+		{
+			name:  "histogram buckets limited",
+			limit: 2,
+			want:  []gostatsd.HistogramThreshold{10, 20, infinity},
+		},
+		{
+			name:  "limit equal to thresholds",
+			limit: 4,
+			want:  []gostatsd.HistogramThreshold{10, 20, 30, 50, infinity},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			buckets := thresholds(latencyHistogram(timer, tt.limit))
+			assert.ElementsMatch(t, buckets, tt.want)
+		})
+	}
+}
+
+func thresholds(buckets map[gostatsd.HistogramThreshold]int) []gostatsd.HistogramThreshold {
+	keys := make([]gostatsd.HistogramThreshold, 0, len(buckets))
+	for key := range buckets {
+		keys = append(keys, key)
+	}
+	return keys
+}