Add native histogram reporting to the reporter interfaces (#278)

Tally's timers and histograms are pre-bucketed, so percentiles cannot be
merged across sources. Native histograms derive their buckets from the
data and merge cleanly, but every value-carrying method in tally today is
a fixed-width number; a native histogram travels as a marshalled blob.
Nothing existing can carry one, so the reporter interfaces have to break.

This adds:

  StatsReporter.ReportNativeHistogram(name, tags, payload, samples)
  CachedStatsReporter.AllocateNativeHistogram(name, tags, maxBuckets)
  CachedNativeHistogram.ReportNativeHistogram(payload, samples)

There is deliberately no per-bucket equivalent of CachedHistogramBucket.
A native histogram rescales its buckets as it observes values, so there
is no stable set of bounds to pre-allocate handles against; the whole
distribution is reported as one payload.

Every in-repo implementer is updated so the tree compiles. Most of them
are no-ops, because their wire formats cannot carry an opaque payload:

  statsd      carries scalar values only.
  prometheus  has native histograms of its own, but tally hands the
              reporter an already-serialized payload in whatever encoding
              the application chose, and a blob cannot become a
              prometheus.Collector.
  m3          speaks thrift over UDP with no payload field. This one is
              not a plain no-op: it counts drops into
              tally.internal.num-native-histograms-dropped, reported only
              when non-zero so that fleets which never use native
              histograms do not gain a permanently-zero series.

This commit lands before the NativeHistogram type it carries, so that the
type and the Scope surface can each be reviewed against an interface that
already exists.

Author: rluong003

example/main.go

@@ -69,6 +69,16 @@ func (r *printStatsReporter) ReportHistogramDurationSamples(
 		name, bucketLowerBound, bucketUpperBound, samples)
 }
 
+func (r *printStatsReporter) ReportNativeHistogram(
+	name string,
+	_ map[string]string,
+	payload []byte,
+	samples uint64,
+) {
+	fmt.Printf("native histogram %s payload %d bytes samples %d\n",
+		name, len(payload), samples)
+}
+
 func (r *printStatsReporter) Capabilities() tally.Capabilities {
 	return r
 }

m3/reporter.go

@@ -134,6 +134,9 @@ type reporter struct {
 	numWriteErrors        atomic.Int64
 	numWriteErrorsCounter tally.CachedCount
 	numTagCacheCounter    tally.CachedCount
+
+	numNativeHistogramsDropped        atomic.Int64
+	numNativeHistogramsDroppedCounter tally.CachedCount
 }
 
 // Options is a set of options for the M3 reporter.
@@ -301,6 +304,9 @@ func NewReporter(opts Options) (Reporter, error) {
 	r.numMetricsCounter = r.AllocateCounter("tally.internal.num-metrics", internalTags)
 	r.numWriteErrorsCounter = r.AllocateCounter("tally.internal.num-write-errors", internalTags)
 	r.numTagCacheCounter = r.AllocateCounter("tally.internal.num-tag-cache", internalTags)
+	r.numNativeHistogramsDroppedCounter = r.AllocateCounter(
+		"tally.internal.num-native-histograms-dropped", internalTags,
+	)
 	r.wg.Add(1)
 	go func() {
 		defer r.wg.Done()
@@ -441,6 +447,20 @@ func (r *reporter) AllocateHistogram(
 	}
 }
 
+// AllocateNativeHistogram implements tally.CachedStatsReporter.
+//
+// The M3 thrift wire format carries only count, gauge and timer values, so
+// there is no field a serialized native histogram payload can travel in.
+// Rather than silently discard them, the returned handle counts every dropped
+// report into tally.internal.num-native-histograms-dropped.
+func (r *reporter) AllocateNativeHistogram(
+	name string,
+	tags map[string]string,
+	maxBuckets int,
+) tally.CachedNativeHistogram {
+	return cachedNativeHistogram{reporter: r}
+}
+
 func (r *reporter) valueBucketString(v float64) string {
 	if v == math.MaxFloat64 {
 		return "infinity"
@@ -699,6 +719,13 @@ func (r *reporter) reportInternalMetrics() {
 	r.numMetricsCounter.ReportCount(metrics)
 	r.numWriteErrorsCounter.ReportCount(writeErrors)
 	r.numTagCacheCounter.ReportCount(int64(r.tagCache.Len()))
+
+	// Reported only when non-zero. Emitting it unconditionally would add a
+	// permanently-zero series to every m3 reporter in a fleet, since almost
+	// no caller uses native histograms with this reporter.
+	if dropped := r.numNativeHistogramsDropped.Swap(0); dropped > 0 {
+		r.numNativeHistogramsDroppedCounter.ReportCount(dropped)
+	}
 }
 
 func (r *reporter) timeLoop() {
@@ -735,6 +762,16 @@ func (c cachedMetric) ReportTimer(interval time.Duration) {
 	c.reporter.reportCopyMetric(c.metric, c.size, "", "")
 }
 
+// cachedNativeHistogram discards native histogram payloads, which the M3
+// thrift wire format cannot represent, and counts the drops.
+type cachedNativeHistogram struct {
+	reporter *reporter
+}
+
+func (c cachedNativeHistogram) ReportNativeHistogram(payload []byte, samples uint64) {
+	c.reporter.numNativeHistogramsDropped.Inc()
+}
+
 type noopMetric struct{}
 
 func (c noopMetric) ReportCount(value int64)            {}

multi/reporter.go

@@ -103,6 +103,17 @@ func (r *multi) ReportHistogramDurationSamples(
 	}
 }
 
+func (r *multi) ReportNativeHistogram(
+	name string,
+	tags map[string]string,
+	payload []byte,
+	samples uint64,
+) {
+	for _, r := range r.reporters {
+		r.ReportNativeHistogram(name, tags, payload, samples)
+	}
+}
+
 func (r *multi) Capabilities() tally.Capabilities {
 	return r.multiBaseReporters.Capabilities()
 }
@@ -175,6 +186,18 @@ func (r *multiCached) AllocateHistogram(
 	return multiMetric{histograms: metrics}
 }
 
+func (r *multiCached) AllocateNativeHistogram(
+	name string,
+	tags map[string]string,
+	maxBuckets int,
+) tally.CachedNativeHistogram {
+	metrics := make([]tally.CachedNativeHistogram, 0, len(r.reporters))
+	for _, r := range r.reporters {
+		metrics = append(metrics, r.AllocateNativeHistogram(name, tags, maxBuckets))
+	}
+	return multiMetric{nativeHistograms: metrics}
+}
+
 func (r *multiCached) Capabilities() tally.Capabilities {
 	return r.multiBaseReporters.Capabilities()
 }
@@ -184,10 +207,11 @@ func (r *multiCached) Flush() {
 }
 
 type multiMetric struct {
-	counters   []tally.CachedCount
-	gauges     []tally.CachedGauge
-	timers     []tally.CachedTimer
-	histograms []tally.CachedHistogram
+	counters         []tally.CachedCount
+	gauges           []tally.CachedGauge
+	timers           []tally.CachedTimer
+	histograms       []tally.CachedHistogram
+	nativeHistograms []tally.CachedNativeHistogram
 }
 
 func (m multiMetric) ReportCount(value int64) {
@@ -230,6 +254,12 @@ func (m multiMetric) DurationBucket(
 	return multiHistogramBucket{multi}
 }
 
+func (m multiMetric) ReportNativeHistogram(payload []byte, samples uint64) {
+	for _, m := range m.nativeHistograms {
+		m.ReportNativeHistogram(payload, samples)
+	}
+}
+
 type multiHistogramBucket struct {
 	multi []tally.CachedHistogramBucket
 }

multi/reporter_test.go

@@ -44,6 +44,7 @@ func TestMultiReporter(t *testing.T) {
 		{"foo": "qux"},
 		{"foo": "bzz"},
 		{"foo": "buz"},
+		{"foo": "nhg"},
 	}
 
 	valueBuckets := tally.MustMakeLinearValueBuckets(0, 2, 5)
@@ -57,6 +58,7 @@ func TestMultiReporter(t *testing.T) {
 		2.0, 4.0, 3)
 	r.ReportHistogramDurationSamples("buz", tags[4], durationBuckets,
 		2*time.Second, 4*time.Second, 3)
+	r.ReportNativeHistogram("nhg", tags[5], []byte("payload"), 7)
 	for _, r := range all {
 		require.Equal(t, 2, len(r.counts))
 
@@ -87,6 +89,12 @@ func TestMultiReporter(t *testing.T) {
 		assert.Equal(t, 2*time.Second, r.histogramDurationSamples[0].bucketLowerBound)
 		assert.Equal(t, 4*time.Second, r.histogramDurationSamples[0].bucketUpperBound)
 		assert.Equal(t, int64(3), r.histogramDurationSamples[0].samples)
+
+		require.Equal(t, 1, len(r.nativeHistograms))
+		assert.Equal(t, "nhg", r.nativeHistograms[0].name)
+		assert.Equal(t, tags[5], r.nativeHistograms[0].tags)
+		assert.Equal(t, []byte("payload"), r.nativeHistograms[0].payload)
+		assert.Equal(t, uint64(7), r.nativeHistograms[0].samples)
 	}
 
 	assert.NotNil(t, r.Capabilities())
@@ -112,6 +120,7 @@ func TestMultiCachedReporter(t *testing.T) {
 		{"foo": "qux"},
 		{"foo": "bzz"},
 		{"foo": "buz"},
+		{"foo": "nhg"},
 	}
 
 	valueBuckets := tally.MustMakeLinearValueBuckets(0, 2, 5)
@@ -133,6 +142,9 @@ func TestMultiCachedReporter(t *testing.T) {
 	dhist := r.AllocateHistogram("buz", tags[4], durationBuckets)
 	dhist.DurationBucket(2*time.Second, 4*time.Second).ReportSamples(3)
 
+	nhist := r.AllocateNativeHistogram("nhg", tags[5], 160)
+	nhist.ReportNativeHistogram([]byte("payload"), 7)
+
 	for _, r := range all {
 		require.Equal(t, 2, len(r.counts))
 
@@ -163,6 +175,12 @@ func TestMultiCachedReporter(t *testing.T) {
 		assert.Equal(t, 2*time.Second, r.histogramDurationSamples[0].bucketLowerBound)
 		assert.Equal(t, 4*time.Second, r.histogramDurationSamples[0].bucketUpperBound)
 		assert.Equal(t, int64(3), r.histogramDurationSamples[0].samples)
+
+		require.Equal(t, 1, len(r.nativeHistograms))
+		assert.Equal(t, "nhg", r.nativeHistograms[0].name)
+		assert.Equal(t, tags[5], r.nativeHistograms[0].tags)
+		assert.Equal(t, []byte("payload"), r.nativeHistograms[0].payload)
+		assert.Equal(t, uint64(7), r.nativeHistograms[0].samples)
 	}
 
 	assert.NotNil(t, r.Capabilities())
@@ -179,6 +197,7 @@ type capturingStatsReporter struct {
 	timers                   []capturedTimer
 	histogramValueSamples    []capturedHistogramValueSamples
 	histogramDurationSamples []capturedHistogramDurationSamples
+	nativeHistograms         []capturedNativeHistogram
 	capabilities             int
 	flush                    int
 }
@@ -217,6 +236,13 @@ type capturedHistogramDurationSamples struct {
 	samples          int64
 }
 
+type capturedNativeHistogram struct {
+	name    string
+	tags    map[string]string
+	payload []byte
+	samples uint64
+}
+
 func newCapturingStatsReporter() *capturingStatsReporter {
 	return &capturingStatsReporter{}
 }
@@ -275,6 +301,16 @@ func (r *capturingStatsReporter) ReportHistogramDurationSamples(
 	r.histogramDurationSamples = append(r.histogramDurationSamples, elem)
 }
 
+func (r *capturingStatsReporter) ReportNativeHistogram(
+	name string,
+	tags map[string]string,
+	payload []byte,
+	samples uint64,
+) {
+	elem := capturedNativeHistogram{name, tags, payload, samples}
+	r.nativeHistograms = append(r.nativeHistograms, elem)
+}
+
 func (r *capturingStatsReporter) AllocateCounter(
 	name string,
 	tags map[string]string,
@@ -323,6 +359,17 @@ func (r *capturingStatsReporter) AllocateHistogram(
 	}
 }
 
+func (r *capturingStatsReporter) AllocateNativeHistogram(
+	name string,
+	tags map[string]string,
+	maxBuckets int,
+) tally.CachedNativeHistogram {
+	return cachedNativeHistogram{fn: func(payload []byte, samples uint64) {
+		elem := capturedNativeHistogram{name, tags, payload, samples}
+		r.nativeHistograms = append(r.nativeHistograms, elem)
+	}}
+}
+
 func (r *capturingStatsReporter) Capabilities() tally.Capabilities {
 	r.capabilities++
 	return r
@@ -364,6 +411,14 @@ func (c cachedTimer) ReportTimer(value time.Duration) {
 	c.fn(value)
 }
 
+type cachedNativeHistogram struct {
+	fn func(payload []byte, samples uint64)
+}
+
+func (h cachedNativeHistogram) ReportNativeHistogram(payload []byte, samples uint64) {
+	h.fn(payload, samples)
+}
+
 type cachedHistogram struct {
 	valueFn    func(bucketLowerBound, bucketUpperBound float64, samples int64)
 	durationFn func(bucketLowerBound, bucketUpperBound time.Duration, samples int64)

prometheus/reporter.go

@@ -216,6 +216,9 @@ func (m noopMetric) ReportCount(value int64)            {}
 func (m noopMetric) ReportGauge(value float64)          {}
 func (m noopMetric) ReportTimer(interval time.Duration) {}
 func (m noopMetric) ReportSamples(value int64)          {}
+
+func (m noopMetric) ReportNativeHistogram(payload []byte, samples uint64) {}
+
 func (m noopMetric) ValueBucket(lower, upper float64) tally.CachedHistogramBucket {
 	return m
 }
@@ -571,6 +574,21 @@ func (r *reporter) AllocateHistogram(
 	return &cachedMetric{histogram: histogramVec.With(tags)}
 }
 
+// AllocateNativeHistogram implements tally.CachedStatsReporter.
+//
+// Prometheus has native histograms of its own, but tally hands this method an
+// already-serialized payload in whatever encoding the application chose via
+// ScopeOptions.NativeHistogramFactory. An opaque blob cannot be turned into a
+// prometheus.Collector, so there is nothing to register and the metric is
+// dropped.
+func (r *reporter) AllocateNativeHistogram(
+	name string,
+	tags map[string]string,
+	maxBuckets int,
+) tally.CachedNativeHistogram {
+	return noopMetric{}
+}
+
 func (r *reporter) Capabilities() tally.Capabilities {
 	return r
 }

reporter.go

@@ -75,6 +75,16 @@ type StatsReporter interface {
 		bucketUpperBound time.Duration,
 		samples int64,
 	)
+
+	// ReportNativeHistogram reports a serialized native histogram covering
+	// the samples observed since the last report. The payload encoding is
+	// determined by the NativeHistogramData the scope was configured with.
+	ReportNativeHistogram(
+		name string,
+		tags map[string]string,
+		payload []byte,
+		samples uint64,
+	)
 }
 
 // CachedStatsReporter is a backend for Scopes that pre allocates all
@@ -107,6 +117,14 @@ type CachedStatsReporter interface {
 		tags map[string]string,
 		buckets Buckets,
 	) CachedHistogram
+
+	// AllocateNativeHistogram pre allocates a native histogram data structure
+	// with name, tags and a bucket budget.
+	AllocateNativeHistogram(
+		name string,
+		tags map[string]string,
+		maxBuckets int,
+	) CachedNativeHistogram
 }
 
 // CachedCount interface for reporting an individual counter
@@ -138,3 +156,13 @@ type CachedHistogram interface {
 type CachedHistogramBucket interface {
 	ReportSamples(value int64)
 }
+
+// CachedNativeHistogram interface for reporting an individual native histogram.
+//
+// There is no per-bucket equivalent of CachedHistogramBucket: a native
+// histogram rescales its buckets as it observes values, so there is no stable
+// set of bounds to pre-allocate handles against. The whole distribution is
+// reported as one payload instead.
+type CachedNativeHistogram interface {
+	ReportNativeHistogram(payload []byte, samples uint64)
+}

scope_benchmark_test.go

@@ -290,6 +290,8 @@ func (s noopStat) DurationBucket(bucketLowerBound, bucketUpperBound time.Duratio
 }
 func (s noopStat) ReportSamples(value int64) {}
 
+func (s noopStat) ReportNativeHistogram(payload []byte, samples uint64) {}
+
 type noopCachedReporter struct{}
 
 func (n noopCachedReporter) Capabilities() Capabilities {
@@ -311,6 +313,9 @@ func (n noopCachedReporter) ReportHistogramValueSamples(name string, tags map[st
 func (n noopCachedReporter) ReportHistogramDurationSamples(name string, tags map[string]string, buckets Buckets, bucketLowerBound time.Duration, bucketUpperBound time.Duration, samples int64) {
 }
 
+func (n noopCachedReporter) ReportNativeHistogram(name string, tags map[string]string, payload []byte, samples uint64) {
+}
+
 func (n noopCachedReporter) AllocateCounter(name string, tags map[string]string) CachedCount {
 	return noopStat{}
 }
@@ -325,3 +330,7 @@ func (n noopCachedReporter) AllocateTimer(name string, tags map[string]string) C
 func (n noopCachedReporter) AllocateHistogram(name string, tags map[string]string, buckets Buckets) CachedHistogram {
 	return noopStat{}
 }
+
+func (n noopCachedReporter) AllocateNativeHistogram(name string, tags map[string]string, maxBuckets int) CachedNativeHistogram {
+	return noopStat{}
+}