Switch all metrics to distributions via v1 distribution_points API

Use distributions instead of gauges so we get p50/p95/p99 across all
cache keys. Add speed_mbps and size_bytes to every command. Remove
high-cardinality branch tag from delta metrics.

Author: joshfriend

cmd/gradle-cache/main.go

@@ -301,8 +301,13 @@ func (c *RestoreCmd) Run(ctx context.Context, metrics metricsClient) error {
 	// total ≈ download time + a small flush of buffered pipeline stages.
 	slog.Info("restore pipeline complete",
 		"total_duration", totalElapsed.Round(time.Millisecond))
-	emitTiming(metrics, "gradle_cache.restore.duration_ms", totalElapsed.Milliseconds(), "cache_key:"+c.CacheKey)
-	emitGauge(metrics, "gradle_cache.restore.size_bytes", cb.n, "cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.restore.duration_ms", float64(totalElapsed.Milliseconds()), "cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.restore.size_bytes", float64(cb.n), "cache_key:"+c.CacheKey)
+	if !cb.eofAt.IsZero() {
+		dlElapsed := cb.eofAt.Sub(dlStart)
+		mbps := float64(cb.n) / dlElapsed.Seconds() / 1e6
+		metrics.distribution("gradle_cache.restore.speed_mbps", mbps, "cache_key:"+c.CacheKey)
+	}
 
 	// Write a marker recording when the base restore finished.
 	// save-delta compares file mtimes against this to identify files created
@@ -458,8 +463,16 @@ func (c *RestoreDeltaCmd) Run(ctx context.Context, metrics metricsClient) error
 	deltaElapsed := time.Since(dlStart)
 	slog.Info("applied delta bundle", "branch", c.Branch, "cache-key", c.CacheKey,
 		"total_duration", deltaElapsed.Round(time.Millisecond))
-	emitTiming(metrics, "gradle_cache.restore_delta.duration_ms", deltaElapsed.Milliseconds(),
-		"cache_key:"+c.CacheKey, "branch:"+c.Branch)
+	metrics.distribution("gradle_cache.restore_delta.duration_ms", float64(deltaElapsed.Milliseconds()),
+		"cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.restore_delta.size_bytes", float64(cb.n),
+		"cache_key:"+c.CacheKey)
+	if !cb.eofAt.IsZero() {
+		dlElapsed := cb.eofAt.Sub(dlStart)
+		mbps := float64(cb.n) / dlElapsed.Seconds() / 1e6
+		metrics.distribution("gradle_cache.restore_delta.speed_mbps", mbps,
+			"cache_key:"+c.CacheKey)
+	}
 	return nil
 }
 
@@ -580,8 +593,9 @@ func (c *SaveCmd) Run(ctx context.Context, metrics metricsClient) error {
 		"size_mb", fmt.Sprintf("%.1f", float64(size)/1e6),
 		"speed_mbps", fmt.Sprintf("%.1f", mbps))
 	slog.Info("saved bundle", "commit", c.Commit[:min(8, len(c.Commit))], "cache-key", c.CacheKey)
-	emitTiming(metrics, "gradle_cache.save.duration_ms", elapsed.Milliseconds(), "cache_key:"+c.CacheKey)
-	emitGauge(metrics, "gradle_cache.save.size_bytes", size, "cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.save.duration_ms", float64(elapsed.Milliseconds()), "cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.save.size_bytes", float64(size), "cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.save.speed_mbps", mbps, "cache_key:"+c.CacheKey)
 	return nil
 }
 
@@ -692,10 +706,12 @@ func (c *SaveDeltaCmd) Run(ctx context.Context, metrics metricsClient) error {
 		"duration", elapsed.Round(time.Millisecond),
 		"size_mb", fmt.Sprintf("%.1f", float64(size)/1e6),
 		"speed_mbps", fmt.Sprintf("%.1f", mbps))
-	emitTiming(metrics, "gradle_cache.save_delta.duration_ms", elapsed.Milliseconds(),
-		"cache_key:"+c.CacheKey, "branch:"+c.Branch)
-	emitGauge(metrics, "gradle_cache.save_delta.size_bytes", size,
-		"cache_key:"+c.CacheKey, "branch:"+c.Branch)
+	metrics.distribution("gradle_cache.save_delta.duration_ms", float64(elapsed.Milliseconds()),
+		"cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.save_delta.size_bytes", float64(size),
+		"cache_key:"+c.CacheKey)
+	metrics.distribution("gradle_cache.save_delta.speed_mbps", mbps,
+		"cache_key:"+c.CacheKey)
 	return nil
 }
 

cmd/gradle-cache/metrics.go

@@ -13,12 +13,11 @@ import (
 	"time"
 )
 
-// metricsClient emits timing and gauge metrics to a backend.
+// metricsClient emits distribution metrics to a backend.
+// All metrics are submitted as distributions to support percentile aggregation.
 type metricsClient interface {
-	// timing records a duration metric in milliseconds.
-	timing(name string, ms int64, tags ...string)
-	// gauge records a point-in-time value.
-	gauge(name string, value int64, tags ...string)
+	// distribution records a single sample for a distribution metric.
+	distribution(name string, value float64, tags ...string)
 	// close flushes any pending data.
 	close()
 }
@@ -27,9 +26,8 @@ type metricsClient interface {
 // It exists because kong cannot bind a nil interface value.
 type noopMetrics struct{}
 
-func (noopMetrics) timing(string, int64, ...string) {}
-func (noopMetrics) gauge(string, int64, ...string)  {}
-func (noopMetrics) close()                          {}
+func (noopMetrics) distribution(string, float64, ...string) {}
+func (noopMetrics) close()                                  {}
 
 // metricsFlags are CLI flags for configuring metrics emission.
 type metricsFlags struct {
@@ -94,12 +92,8 @@ func newStatsdClient(addr string, baseTags []string) *statsdClient {
 	return &statsdClient{conn: conn, tags: baseTags}
 }
 
-func (s *statsdClient) timing(name string, ms int64, tags ...string) {
-	s.send(fmt.Sprintf("%s:%d|d", name, ms), tags) // |d = distribution for DD percentile support
-}
-
-func (s *statsdClient) gauge(name string, value int64, tags ...string) {
-	s.send(fmt.Sprintf("%s:%d|g", name, value), tags)
+func (s *statsdClient) distribution(name string, value float64, tags ...string) {
+	s.send(fmt.Sprintf("%s:%g|d", name, value), tags)
 }
 
 func (s *statsdClient) send(stat string, extraTags []string) {
@@ -114,9 +108,9 @@ func (s *statsdClient) close() {
 	s.conn.Close() //nolint:errcheck,gosec
 }
 
-// ── DataDog HTTP API ────────────────────────────────────────────────────────
+// ── DataDog HTTP API (v1 distribution_points) ───────────────────────────────
 
-const datadogSeriesURL = "https://api.datadoghq.com/api/v2/series"
+const datadogDistURL = "https://api.datadoghq.com/api/v1/distribution_points"
 
 type datadogAPIClient struct {
 	apiKey string
@@ -132,34 +126,17 @@ func newDatadogAPIClient(apiKey string, baseTags []string) *datadogAPIClient {
 	}
 }
 
-// Datadog v2 metric type enum values.
-const (
-	ddMetricTypeGauge = 3
-	ddMetricTypeCount = 1
-)
-
-func (d *datadogAPIClient) timing(name string, ms int64, tags ...string) {
-	d.submit(name, float64(ms), ddMetricTypeGauge, tags)
-}
-
-func (d *datadogAPIClient) gauge(name string, value int64, tags ...string) {
-	d.submit(name, float64(value), ddMetricTypeGauge, tags)
-}
-
-func (d *datadogAPIClient) submit(name string, value float64, metricType int, extraTags []string) {
-	allTags := append(d.tags, extraTags...)
+func (d *datadogAPIClient) distribution(name string, value float64, tags ...string) {
+	allTags := append(d.tags, tags...)
 	now := time.Now().Unix()
 
+	// v1 distribution_points format: points is [[timestamp, [value, ...]]]
 	payload := map[string]interface{}{
 		"series": []map[string]interface{}{
 			{
 				"metric": name,
-				"type":   metricType,
-				"points": []map[string]interface{}{
-					{
-						"timestamp": now,
-						"value":     value,
-					},
+				"points": []interface{}{
+					[]interface{}{now, []float64{value}},
 				},
 				"tags": allTags,
 			},
@@ -172,7 +149,7 @@ func (d *datadogAPIClient) submit(name string, value float64, metricType int, ex
 		return
 	}
 
-	req, err := http.NewRequest("POST", datadogSeriesURL, bytes.NewReader(body))
+	req, err := http.NewRequest("POST", datadogDistURL, bytes.NewReader(body))
 	if err != nil {
 		slog.Warn("metrics: failed to create request", "error", err)
 		return
@@ -193,11 +170,3 @@ func (d *datadogAPIClient) submit(name string, value float64, metricType int, ex
 }
 
 func (d *datadogAPIClient) close() {}
-
-func emitTiming(m metricsClient, name string, ms int64, tags ...string) {
-	m.timing(name, ms, tags...)
-}
-
-func emitGauge(m metricsClient, name string, value int64, tags ...string) {
-	m.gauge(name, value, tags...)
-}

cmd/gradle-cache/metrics_test.go

@@ -6,7 +6,7 @@ import (
 	"time"
 )
 
-func TestStatsdClientTiming(t *testing.T) {
+func TestStatsdClientDistribution(t *testing.T) {
 	// Start a UDP listener to capture the statsd packet.
 	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
 	if err != nil {
@@ -21,7 +21,7 @@ func TestStatsdClientTiming(t *testing.T) {
 	}
 	defer client.close()
 
-	client.timing("gradle_cache.restore.duration_ms", 1234, "cache_key:foo")
+	client.distribution("gradle_cache.restore.duration_ms", 1234, "cache_key:foo")
 
 	buf := make([]byte, 1024)
 	_ = pc.SetReadDeadline(time.Now().Add(2 * time.Second))
@@ -37,7 +37,7 @@ func TestStatsdClientTiming(t *testing.T) {
 	}
 }
 
-func TestStatsdClientGauge(t *testing.T) {
+func TestStatsdClientDistributionFloat(t *testing.T) {
 	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatal(err)
@@ -51,7 +51,7 @@ func TestStatsdClientGauge(t *testing.T) {
 	}
 	defer client.close()
 
-	client.gauge("gradle_cache.save.size_bytes", 5678)
+	client.distribution("gradle_cache.restore.speed_mbps", 155.6)
 
 	buf := make([]byte, 1024)
 	_ = pc.SetReadDeadline(time.Now().Add(2 * time.Second))
@@ -61,7 +61,7 @@ func TestStatsdClientGauge(t *testing.T) {
 	}
 
 	got := string(buf[:n])
-	want := "gradle_cache.save.size_bytes:5678|g"
+	want := "gradle_cache.restore.speed_mbps:155.6|d"
 	if got != want {
 		t.Errorf("got %q, want %q", got, want)
 	}
@@ -70,8 +70,7 @@ func TestStatsdClientGauge(t *testing.T) {
 func TestNoopMetrics(t *testing.T) {
 	// noopMetrics should not panic.
 	var m metricsClient = noopMetrics{}
-	emitTiming(m, "test.metric", 100)
-	emitGauge(m, "test.metric", 200)
+	m.distribution("test.metric", 100)
 	m.close()
 }