Implement distribution metrics in the Sentry, with arbitrary number of fields.

Distribution metrics are well-suited for recording "events" and the time these
events take, for performance measurements. They bucket durations in buckets,
and keep track of the number of samples in each bucket. As this structure also
inherently keeps track of the *total* number of samples, it can be used as a
simple event counter as well, obviating the need for a counter metric next to
it counting the same thing.

In order to be compatible with the needs of the KVM platform to track events
that happen where new memory allocations would not be possible, the code for
adding a sample to a distribution is optimized to be fast and allocation-free.
The tradeoff there mostly comes in the form of memory, such as requiring a
weird new `fieldMapper` recursive struct that acts as a lookup table for the
concatenated key containing the values of all the fields for which the sample
is being recorded. Since we do not expect to deal with large number of field
combinations, this should not be a problem.

Another tradeoff this imposes is the lack of support for a generic `Bucketer`
interface allowing users to define their own bucketing scheme, as we would not
be able to enforce the lack of allocations in custom `Bucketer`
implementations, nor enforce `+checkescape` on them. However, since in
practice all bucketing implementations will probably reside in `metric.go`,
this is worked around by just having the distribution metric code refer to
`Bucketer` implementations as references and call them directly (without the
interface indirection). Since there is only one implementation currently
(`ExponentialBucketer`), this is faster than using the interface.

PiperOrigin-RevId: 436614053

pkg/metric/BUILD

@@ -6,11 +6,13 @@ go_library(
     name = "metric",
     srcs = [
         "metric.go",
+        "metric_unsafe.go",
     ],
     visibility = ["//:sandbox"],
     deps = [
         ":metric_go_proto",
         "//pkg/eventchannel",
+        "//pkg/gohacks",
         "//pkg/log",
         "//pkg/sync",
         "@org_golang_google_protobuf//types/known/timestamppb",

pkg/metric/metric.proto

@@ -28,6 +28,7 @@ message MetricMetadata {
   string description = 2;
 
   // cumulative indicates that this metric is never decremented.
+  // Only applies for uint64-type metrics.
   bool cumulative = 3;
 
   // sync indicates that values from the final metric event should be
@@ -38,7 +39,10 @@ message MetricMetadata {
   // the monitoring system.
   bool sync = 4;
 
-  enum Type { TYPE_UINT64 = 0; }
+  enum Type {
+    TYPE_UINT64 = 0;
+    TYPE_DISTRIBUTION = 1;
+  }
 
   // type is the type of the metric value.
   Type type = 5;
@@ -56,9 +60,15 @@ message MetricMetadata {
     repeated string allowed_values = 2;
   }
 
-  // fields contains the metric fields. Currently a metric can have at most
-  // one field.
+  // fields contains the metric fields for this metric.
   repeated Field fields = 7;
+
+  // For distribution-typed metrics, this list contains the lower bound of all
+  // buckets (other than the underflow bucket, which has no lower bound).
+  // A distribution with n finite buckets should have n+1 values here.
+  // The (n+1)-th value is the upper bound of the n-th bucket, and the lower
+  // bound of the "overflow" bucket (which has no upper bound).
+  repeated int64 distribution_bucket_lower_bounds = 8;
 }
 
 // MetricRegistration contains the metadata for all metrics that will be in
@@ -68,6 +78,24 @@ message MetricRegistration {
   repeated string stages = 2;
 }
 
+// Samples contains the number of samples in each bucket of a distribution.
+message Samples {
+  // new_samples contains the number of *new* samples in each bucket of a
+  // distribution metric. "New" means the number of new samples added since the
+  // last MetricValue update for this metric and combination of fields.
+  // Given a distribution metrics with `num_finite_buckets` finite buckets,
+  // this means:
+  //   - num_samples[0] is the number of new samples in the "underflow" bucket,
+  //     i.e. samples which are smaller than the lower bound of the
+  //     distribution's first (0-th) finite bucket.
+  //   - num_samples[i] is the number of new samples in the distribution's
+  //     0-based (i-1)-th finite bucket.
+  //   - num_samples[num_finite_buckets+1] is the number of new samples in the
+  //     distribution's last bucket, which is infinite (i.e. it has a lower
+  //     bound but no upper bound).
+  repeated uint64 new_samples = 1;
+}
+
 // MetricValue the value of a metric at a single point in time.
 message MetricValue {
   // name is the unique name of the metric, as in MetricMetadata.
@@ -77,6 +105,7 @@ message MetricValue {
   // depends on the type of the metric.
   oneof value {
     uint64 uint64_value = 2;
+    Samples distribution_value = 3;
   }
 
   repeated string field_values = 4;

pkg/metric/metric_test.go

@@ -15,6 +15,7 @@
 package metric
 
 import (
+	"math"
 	"testing"
 	"time"
 
@@ -65,6 +66,7 @@ const (
 	fooDescription     = "Foo!"
 	barDescription     = "Bar Baz"
 	counterDescription = "Counter"
+	distribDescription = "A distribution metric for testing"
 )
 
 func TestInitialize(t *testing.T) {
@@ -80,6 +82,14 @@ func TestInitialize(t *testing.T) {
 		t.Fatalf("NewUint64Metric got err %v want nil", err)
 	}
 
+	bucketer := NewExponentialBucketer(3, 2, 0, 1)
+	field1 := NewField("field1", []string{"foo", "bar"})
+	field2 := NewField("field2", []string{"baz", "quux"})
+	_, err = NewDistributionMetric("/distrib", true, bucketer, pb.MetricMetadata_UNITS_NANOSECONDS, distribDescription, field1, field2)
+	if err != nil {
+		t.Fatalf("NewDistributionMetric got err %v want nil", err)
+	}
+
 	if err := Initialize(); err != nil {
 		t.Fatalf("Initialize(): %s", err)
 	}
@@ -93,23 +103,23 @@ func TestInitialize(t *testing.T) {
 		t.Fatalf("emitter %v got %T want pb.MetricRegistration", emitter[0], emitter[0])
 	}
 
-	if len(mr.Metrics) != 2 {
-		t.Errorf("MetricRegistration got %d metrics want 2", len(mr.Metrics))
+	if len(mr.Metrics) != 3 {
+		t.Errorf("MetricRegistration got %d metrics want %d", len(mr.Metrics), 3)
 	}
 
 	foundFoo := false
 	foundBar := false
+	foundDistrib := false
 	for _, m := range mr.Metrics {
-		if m.Type != pb.MetricMetadata_TYPE_UINT64 {
-			t.Errorf("Metadata %+v Type got %v want pb.MetricMetadata_TYPE_UINT64", m, m.Type)
-		}
-		if !m.Cumulative {
-			t.Errorf("Metadata %+v Cumulative got false want true", m)
-		}
-
 		switch m.Name {
 		case "/foo":
 			foundFoo = true
+			if m.Type != pb.MetricMetadata_TYPE_UINT64 {
+				t.Errorf("Metadata %+v Type got %v want pb.MetricMetadata_TYPE_UINT64", m, m.Type)
+			}
+			if !m.Cumulative {
+				t.Errorf("Metadata %+v Cumulative got false want true", m)
+			}
 			if m.Description != fooDescription {
 				t.Errorf("/foo %+v Description got %q want %q", m, m.Description, fooDescription)
 			}
@@ -121,6 +131,12 @@ func TestInitialize(t *testing.T) {
 			}
 		case "/bar":
 			foundBar = true
+			if m.Type != pb.MetricMetadata_TYPE_UINT64 {
+				t.Errorf("Metadata %+v Type got %v want pb.MetricMetadata_TYPE_UINT64", m, m.Type)
+			}
+			if !m.Cumulative {
+				t.Errorf("Metadata %+v Cumulative got false want true", m)
+			}
 			if m.Description != barDescription {
 				t.Errorf("/bar %+v Description got %q want %q", m, m.Description, barDescription)
 			}
@@ -130,6 +146,23 @@ func TestInitialize(t *testing.T) {
 			if m.Units != pb.MetricMetadata_UNITS_NANOSECONDS {
 				t.Errorf("/bar %+v Units got %v want %v", m, m.Units, pb.MetricMetadata_UNITS_NANOSECONDS)
 			}
+		case "/distrib":
+			foundDistrib = true
+			want := &pb.MetricMetadata{
+				Name:        "/distrib",
+				Type:        pb.MetricMetadata_TYPE_DISTRIBUTION,
+				Units:       pb.MetricMetadata_UNITS_NANOSECONDS,
+				Description: distribDescription,
+				Sync:        true,
+				Fields: []*pb.MetricMetadata_Field{
+					{FieldName: "field1", AllowedValues: []string{"foo", "bar"}},
+					{FieldName: "field2", AllowedValues: []string{"baz", "quux"}},
+				},
+				DistributionBucketLowerBounds: []int64{0, 2, 4, 6},
+			}
+			if !proto.Equal(m, want) {
+				t.Fatalf("got /distrib metadata:\n%v\nwant:\n%v", m, want)
+			}
 		}
 	}
 
@@ -139,6 +172,9 @@ func TestInitialize(t *testing.T) {
 	if !foundBar {
 		t.Errorf("/bar not found: %+v", emitter)
 	}
+	if !foundDistrib {
+		t.Errorf("/distrib not found: %+v", emitter)
+	}
 }
 
 func TestDisable(t *testing.T) {
@@ -154,6 +190,11 @@ func TestDisable(t *testing.T) {
 		t.Fatalf("NewUint64Metric got err %v want nil", err)
 	}
 
+	_, err = NewDistributionMetric("/distrib", false, NewExponentialBucketer(2, 2, 0, 1), pb.MetricMetadata_UNITS_NONE, distribDescription)
+	if err != nil {
+		t.Fatalf("NewDistributionMetric got err %v want nil", err)
+	}
+
 	if err := Disable(); err != nil {
 		t.Fatalf("Disable(): %s", err)
 	}
@@ -185,6 +226,14 @@ func TestEmitMetricUpdate(t *testing.T) {
 		t.Fatalf("NewUint64Metric got err %v want nil", err)
 	}
 
+	bucketer := NewExponentialBucketer(2, 2, 0, 1)
+	field1 := NewField("field1", []string{"foo", "bar"})
+	field2 := NewField("field2", []string{"baz", "quux"})
+	distrib, err := NewDistributionMetric("/distrib", false, bucketer, pb.MetricMetadata_UNITS_NONE, distribDescription, field1, field2)
+	if err != nil {
+		t.Fatalf("NewDistributionMetric: %v", err)
+	}
+
 	if err := Initialize(); err != nil {
 		t.Fatalf("Initialize(): %s", err)
 	}
@@ -194,7 +243,7 @@ func TestEmitMetricUpdate(t *testing.T) {
 	EmitMetricUpdate()
 
 	if len(emitter) != 1 {
-		t.Fatalf("EmitMetricUpdate emitted %d events want 1", len(emitter))
+		t.Fatalf("EmitMetricUpdate emitted %d events want %d", len(emitter), 1)
 	}
 
 	update, ok := emitter[0].(*pb.MetricUpdate)
@@ -203,26 +252,31 @@ func TestEmitMetricUpdate(t *testing.T) {
 	}
 
 	if len(update.Metrics) != 2 {
-		t.Errorf("MetricUpdate got %d metrics want 2", len(update.Metrics))
+		t.Errorf("MetricUpdate got %d metrics want %d", len(update.Metrics), 2)
 	}
 
 	// Both are included for their initial values.
 	foundFoo := false
 	foundBar := false
+	foundDistrib := false
 	for _, m := range update.Metrics {
 		switch m.Name {
 		case "/foo":
 			foundFoo = true
 		case "/bar":
 			foundBar = true
+		case "/distrib":
+			foundDistrib = true
 		}
-		uv, ok := m.Value.(*pb.MetricValue_Uint64Value)
-		if !ok {
-			t.Errorf("%+v: value %v got %T want pb.MetricValue_Uint64Value", m, m.Value, m.Value)
-			continue
-		}
-		if uv.Uint64Value != 0 {
-			t.Errorf("%v: Value got %v want 0", m, uv.Uint64Value)
+		if m.Name != "/distrib" {
+			uv, ok := m.Value.(*pb.MetricValue_Uint64Value)
+			if !ok {
+				t.Errorf("%+v: value %v got %T want pb.MetricValue_Uint64Value", m, m.Value, m.Value)
+				continue
+			}
+			if uv.Uint64Value != 0 {
+				t.Errorf("%v: Value got %v want %d", m, uv.Uint64Value, 0)
+			}
 		}
 	}
 
@@ -232,38 +286,124 @@ func TestEmitMetricUpdate(t *testing.T) {
 	if !foundBar {
 		t.Errorf("/bar not found: %+v", emitter)
 	}
+	if foundDistrib {
+		t.Errorf("/distrib unexpectedly found: %+v", emitter)
+	}
+	if t.Failed() {
+		t.Fatal("Aborting test so far due to earlier errors.")
+	}
 
 	// Increment foo. Only it is included in the next update.
 	foo.Increment()
-
+	foo.Increment()
+	foo.Increment()
 	emitter.Reset()
 	EmitMetricUpdate()
-
 	if len(emitter) != 1 {
 		t.Fatalf("EmitMetricUpdate emitted %d events want 1", len(emitter))
 	}
-
 	update, ok = emitter[0].(*pb.MetricUpdate)
 	if !ok {
 		t.Fatalf("emitter %v got %T want pb.MetricUpdate", emitter[0], emitter[0])
 	}
-
 	if len(update.Metrics) != 1 {
-		t.Errorf("MetricUpdate got %d metrics want 1", len(update.Metrics))
+		t.Fatalf("MetricUpdate got %d metrics want %d", len(update.Metrics), 1)
 	}
-
 	m := update.Metrics[0]
-
 	if m.Name != "/foo" {
-		t.Errorf("Metric %+v name got %q want '/foo'", m, m.Name)
+		t.Fatalf("Metric %+v name got %q want '/foo'", m, m.Name)
 	}
-
 	uv, ok := m.Value.(*pb.MetricValue_Uint64Value)
 	if !ok {
-		t.Errorf("%+v: value %v got %T want pb.MetricValue_Uint64Value", m, m.Value, m.Value)
+		t.Fatalf("%+v: value %v got %T want pb.MetricValue_Uint64Value", m, m.Value, m.Value)
 	}
-	if uv.Uint64Value != 1 {
-		t.Errorf("%v: Value got %v want 1", m, uv.Uint64Value)
+	if uv.Uint64Value != 3 {
+		t.Errorf("%v: Value got %v want %d", m, uv.Uint64Value, 3)
+	}
+
+	// Add a few samples to the distribution metric.
+	distrib.AddSample(1, "foo", "baz")
+	distrib.AddSample(1, "foo", "baz")
+	distrib.AddSample(3, "foo", "baz")
+	distrib.AddSample(-1, "foo", "quux")
+	distrib.AddSample(1, "foo", "quux")
+	distrib.AddSample(100, "foo", "quux")
+	emitter.Reset()
+	EmitMetricUpdate()
+	if len(emitter) != 1 {
+		t.Fatalf("EmitMetricUpdate emitted %d events want %d", len(emitter), 1)
+	}
+	update, ok = emitter[0].(*pb.MetricUpdate)
+	if !ok {
+		t.Fatalf("emitter %v got %T want pb.MetricUpdate", emitter[0], emitter[0])
+	}
+	if len(update.Metrics) != 2 {
+		t.Fatalf("MetricUpdate got %d metrics want %d", len(update.Metrics), 1)
+	}
+	for _, m := range update.Metrics {
+		if m.Name != "/distrib" {
+			t.Fatalf("Metric %+v name got %q want '/distrib'", m, m.Name)
+		}
+		if len(m.FieldValues) != 2 {
+			t.Fatalf("Metric %+v fields: got %v want %d fields", m, m.FieldValues, 2)
+		}
+		if m.FieldValues[0] != "foo" {
+			t.Fatalf("Metric %+v field 0: got %v want %v", m, m.FieldValues[0], "foo")
+		}
+		dv, ok := m.Value.(*pb.MetricValue_DistributionValue)
+		if !ok {
+			t.Fatalf("%+v: value %v got %T want pb.MetricValue_DistributionValue", m, m.Value, m.Value)
+		}
+		samples := dv.DistributionValue.GetNewSamples()
+		if len(samples) != 4 {
+			t.Fatalf("%+v: got %d buckets, want %d", dv.DistributionValue, len(samples), 4)
+		}
+		var wantSamples []uint64
+		switch m.FieldValues[1] {
+		case "baz":
+			wantSamples = []uint64{0, 2, 1, 0}
+		case "quux":
+			wantSamples = []uint64{1, 1, 0, 1}
+		default:
+			t.Fatalf("%+v: got unexpected field[1]: %q", m, m.FieldValues[1])
+		}
+		for i, s := range samples {
+			if s != wantSamples[i] {
+				t.Errorf("%+v [fields %v]: sample %d: got %d want %d", dv.DistributionValue, m.FieldValues, i, s, wantSamples[i])
+			}
+		}
+	}
+
+	// Add more samples to the distribution metric, check that we get the delta.
+	distrib.AddSample(3, "foo", "baz")
+	distrib.AddSample(2, "foo", "baz")
+	distrib.AddSample(1, "foo", "baz")
+	distrib.AddSample(3, "foo", "baz")
+	emitter.Reset()
+	EmitMetricUpdate()
+	if len(emitter) != 1 {
+		t.Fatalf("EmitMetricUpdate emitted %d events want %d", len(emitter), 1)
+	}
+	dv, ok := emitter[0].(*pb.MetricUpdate).Metrics[0].Value.(*pb.MetricValue_DistributionValue)
+	if !ok {
+		t.Fatalf("%+v: want pb.MetricValue_DistributionValue", emitter)
+	}
+	samples := dv.DistributionValue.GetNewSamples()
+	if len(samples) != 4 {
+		t.Fatalf("%+v: got %d buckets, want %d", dv.DistributionValue, len(samples), 4)
+	}
+	wantSamples := []uint64{0, 1, 3, 0}
+	for i, s := range samples {
+		if s != wantSamples[i] {
+			t.Errorf("%+v: sample %d: got %d want %d", dv.DistributionValue, i, s, wantSamples[i])
+		}
+	}
+
+	// Change nothing but still call EmitMetricUpdate. Verify that nothing gets sent.
+	emitter.Reset()
+	EmitMetricUpdate()
+	if len(emitter) != 0 {
+		t.Fatalf("EmitMetricUpdate emitted %d events want %d", len(emitter), 0)
 	}
 }
 
@@ -497,3 +637,81 @@ func TestMetricUpdateStageTiming(t *testing.T) {
 		checkStage(update.StageTiming[1], "last_stage_2")
 	}
 }
+
+func TestBucketer(t *testing.T) {
+	for _, test := range []struct {
+		name                string
+		bucketer            Bucketer
+		minSample           int64
+		maxSample           int64
+		firstFewLowerBounds []int64
+	}{
+		{
+			name:                "static-sized buckets",
+			bucketer:            NewExponentialBucketer(10, 10, 0, 1),
+			minSample:           -5,
+			maxSample:           105,
+			firstFewLowerBounds: []int64{0, 10, 20, 30, 40, 50},
+		},
+		{
+			name:      "exponential buckets",
+			bucketer:  NewExponentialBucketer(10, 10, 2, 1.5),
+			minSample: -5,
+			maxSample: int64(20 * math.Pow(1.5, 12)),
+			firstFewLowerBounds: []int64{
+				0,
+				10 + 2,
+				20 + int64(2*1.5),
+				30 + int64(math.Floor(2*1.5*1.5)),
+				40 + int64(math.Floor(2*1.5*1.5*1.5)),
+				50 + int64(math.Floor(2*1.5*1.5*1.5*1.5)),
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			numFiniteBuckets := test.bucketer.NumFiniteBuckets()
+			testAround := func(bound int64, bucketIndex int) {
+				for sample := bound - 2; sample <= bound+2; sample++ {
+					gotIndex := test.bucketer.BucketIndex(sample)
+					if sample < bound && gotIndex != bucketIndex-1 || sample >= bound && gotIndex != bucketIndex {
+						t.Errorf("LowerBound(%d) = %d, yet BucketIndex(%d) = %d", bucketIndex, bound, sample, gotIndex)
+					}
+				}
+			}
+			for sample := test.minSample; sample <= test.maxSample; sample++ {
+				bucket := test.bucketer.BucketIndex(sample)
+				if bucket == -1 {
+					lowestBound := test.bucketer.LowerBound(0)
+					if sample >= lowestBound {
+						t.Errorf("sample %d: got bucket %d but lowest bound %d", sample, bucket, lowestBound)
+					}
+					testAround(lowestBound, 0)
+				} else if bucket > numFiniteBuckets {
+					t.Errorf("sample %d: got bucket with 0-based index %d but bucketer supposedly only has %d buckets", sample, bucket, numFiniteBuckets)
+				} else if bucket == numFiniteBuckets {
+					lastBucketBound := test.bucketer.LowerBound(bucket)
+					if sample < lastBucketBound {
+						t.Errorf("sample %d: got bucket %d but it has lower bound %d", sample, bucket, lastBucketBound)
+					}
+					testAround(lastBucketBound, bucket)
+				} else {
+					lowerBound := test.bucketer.LowerBound(bucket)
+					upperBound := test.bucketer.LowerBound(bucket + 1)
+					if upperBound <= lowerBound {
+						t.Errorf("sample %d: got bucket %d, upperbound %d <= lowerbound %d", sample, bucket, upperBound, lowerBound)
+					}
+					if sample < lowerBound || sample >= upperBound {
+						t.Errorf("sample %d: got bucket %d which has range [%d, %d)", sample, bucket, lowerBound, upperBound)
+					}
+					testAround(lowerBound, bucket)
+					testAround(upperBound, bucket+1)
+				}
+			}
+			for bi, want := range test.firstFewLowerBounds {
+				if got := test.bucketer.LowerBound(bi); got != want {
+					t.Errorf("bucket %d has lower bound %d, want %d", bi, got, want)
+				}
+			}
+		})
+	}
+}

pkg/metric/metric_unsafe.go

@@ -0,0 +1,47 @@
+// Copyright 2022 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metric
+
+import (
+	"unsafe"
+
+	"gvisor.dev/gvisor/pkg/gohacks"
+	"gvisor.dev/gvisor/pkg/sync"
+)
+
+// snapshotDistribution snapshots the sample data of distribution metrics in
+// a non-consistent manner.
+// Distribution metrics don't need to be read consistently, because any
+// inconsistency (i.e. increments that race with the snapshot) will simply be
+// detected during the next snapshot instead. Reading them consistently would
+// require more synchronization during increments, which we need to be cheap.
+func snapshotDistribution(samples []uint64) []uint64 {
+	// The number of buckets within a distribution never changes, so there is
+	// no race condition from getting the number of buckets upfront.
+	numBuckets := len(samples)
+	snapshot := make([]uint64, numBuckets)
+	samplesHeader := (*gohacks.SliceHeader)(unsafe.Pointer(&samples))
+	snapshotHeader := (*gohacks.SliceHeader)(unsafe.Pointer(&snapshot))
+	if sync.RaceEnabled {
+		// runtime.RaceDisable() doesn't actually stop the race detector, so it
+		// can't help us here. Instead, call runtime.memmove directly, which is
+		// not instrumented by the race detector.
+		gohacks.Memmove(snapshotHeader.Data, samplesHeader.Data, unsafe.Sizeof(uint64(0))*uintptr(numBuckets))
+	} else {
+		// Just use copy.
+		copy(snapshot, samples)
+	}
+	return snapshot
+}