From 1c9ce540b8b946a112492d5803e864fcd20fa3c2 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Tue, 22 Mar 2022 17:43:24 -0700 Subject: [PATCH] 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 | 2 + pkg/metric/metric.go | 495 ++++++++++++++++++++++++++++++++++-- pkg/metric/metric.proto | 35 ++- pkg/metric/metric_test.go | 278 +++++++++++++++++--- pkg/metric/metric_unsafe.go | 47 ++++ 5 files changed, 802 insertions(+), 55 deletions(-) create mode 100644 pkg/metric/metric_unsafe.go diff --git a/pkg/metric/BUILD b/pkg/metric/BUILD index c08792751..570bf9aa6 100644 --- a/pkg/metric/BUILD +++ b/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", diff --git a/pkg/metric/metric.go b/pkg/metric/metric.go index ac38ec894..68c962c92 100644 --- a/pkg/metric/metric.go +++ b/pkg/metric/metric.go @@ -18,7 +18,9 @@ package metric import ( "errors" "fmt" + "math" "sort" + "strings" "sync/atomic" "time" @@ -38,6 +40,10 @@ var ( // new metric after initialization. ErrInitializationDone = errors.New("metric cannot be created after initialization is complete") + // ErrFieldValueContainsIllegalChar indicates that the value of a metric + // field had an invalid character in it. + ErrFieldValueContainsIllegalChar = errors.New("metric field value contains illegal character") + // WeirdnessMetric is a metric with fields created to track the number // of weird occurrences such as time fallback, partial_result, vsyscall // count, watchdog startup timeouts and stuck tasks. @@ -118,7 +124,10 @@ func Initialize() error { } m := pb.MetricRegistration{} - for _, v := range allMetrics.m { + for _, v := range allMetrics.uint64Metrics { + m.Metrics = append(m.Metrics, v.metadata) + } + for _, v := range allMetrics.distributionMetrics { m.Metrics = append(m.Metrics, v.metadata) } m.Stages = make([]string, 0, len(allStages)) @@ -172,6 +181,132 @@ type Field struct { allowedValues []string } +// NewField defines a new Field that can be used to break down a metric. +func NewField(name string, allowedValues []string) Field { + return Field{ + name: name, + allowedValues: allowedValues, + } +} + +// toProto returns the proto definition of this field, for use in metric +// metadata. +func (f Field) toProto() *pb.MetricMetadata_Field { + return &pb.MetricMetadata_Field{ + FieldName: f.name, + AllowedValues: f.allowedValues, + } +} + +// multiFieldToKey returns a concatenated version of the given fields. +// It can be used as a unique key within multi-dimensional metrics. +// Does not allow commas as valid character within field values. +func multiFieldToKey(fields ...string) (string, error) { + if len(fields) == 0 { + return "", nil + } + for _, f := range fields { + if strings.ContainsRune(f, ',') { + return "", ErrFieldValueContainsIllegalChar + } + } + return strings.Join(fields, ","), nil +} + +// keyToMultiField is the reverse of multiFieldToKey. +func keyToMultiField(key string) []string { + if key == "" { + return nil + } + return strings.Split(key, ",") +} + +// fieldMapper provides multi-dimensional fields to a single concatenated key +// that can be used as string key for multi-dimensional metrics. +// fieldMapper is a recursive struct, but its lookup function is not. +// It pays for its allocation-free, low-stack lookup by preallocating a map of +// all possible field values, so it is memory-hungry. +type fieldMapper struct { + // depth is 0 at the lowest level of fieldMapper. + depth int + // key is set only at the lowest level of fieldMapper, i.e. depth == 0. + // It contains the full concatenated key of all the parent field values. + key string + // children is set only at depth > 0. + // For depth=d, children[fields[d]] is the fieldMapper that can be used to + // look up keys for fields[d+1:]. + children map[string]fieldMapper +} + +// newFieldMapper returns a new fieldMapper for the given set of fields. +func newFieldMapper(fields ...Field) (fieldMapper, error) { + var initFieldMapper func(values []string, remaining ...Field) (fieldMapper, error) + initFieldMapper = func(values []string, remaining ...Field) (fieldMapper, error) { + depth := len(remaining) + if depth == 0 { + key, err := multiFieldToKey(values...) + if err != nil { + return fieldMapper{}, err + } + return fieldMapper{key: key}, nil + } + current := remaining[0] + children := make(map[string]fieldMapper, len(current.allowedValues)) + for _, value := range current.allowedValues { + newValues := make([]string, len(values)+1) + copy(newValues, values) + newValues[len(values)] = value + child, err := initFieldMapper(newValues, remaining[1:]...) + if err != nil { + return fieldMapper{}, err + } + children[value] = child + } + return fieldMapper{ + depth: depth, + children: children, + }, nil + } + return initFieldMapper(nil, fields...) +} + +// lookup looks up a key within the fieldMapper. +// It needs to allocate no memory and be nosplit-compatible, so it cannot be +// recursive. +// This *must* be called with the correct number of fields, or it will panic. +// +checkescape:all +//go:nosplit +func (m fieldMapper) lookup(fields ...string) string { + depth := len(fields) + if depth != m.depth { + panic("invalid field lookup depth") + } + var found bool + for i := 0; i < depth; i++ { + if m, found = m.children[fields[i]]; !found { + panic("disallowed field value") + } + } + return m.key +} + +// all iterates over all keys within the fieldMapper. +func (m fieldMapper) all() []string { + var all []string + var visit func(fm fieldMapper) + visit = func(fm fieldMapper) { + if fm.depth == 0 { + all = append(all, fm.key) + } else { + for _, child := range fm.children { + visit(child) + } + } + } + visit(m) + return all +} + // RegisterCustomUint64Metric registers a metric with the given name. // // Register must only be called at init and will return and error if called @@ -186,11 +321,14 @@ func RegisterCustomUint64Metric(name string, cumulative, sync bool, units pb.Met return ErrInitializationDone } - if _, ok := allMetrics.m[name]; ok { + if _, ok := allMetrics.uint64Metrics[name]; ok { + return ErrNameInUse + } + if _, ok := allMetrics.distributionMetrics[name]; ok { return ErrNameInUse } - allMetrics.m[name] = customUint64Metric{ + allMetrics.uint64Metrics[name] = customUint64Metric{ metadata: &pb.MetricMetadata{ Name: name, Description: description, @@ -208,10 +346,7 @@ func RegisterCustomUint64Metric(name string, cumulative, sync bool, units pb.Met } for _, field := range fields { - allMetrics.m[name].metadata.Fields = append(allMetrics.m[name].metadata.Fields, &pb.MetricMetadata_Field{ - FieldName: field.name, - AllowedValues: field.allowedValues, - }) + allMetrics.uint64Metrics[name].metadata.Fields = append(allMetrics.uint64Metrics[name].metadata.Fields, field.toProto()) } return nil } @@ -314,6 +449,241 @@ func (m *Uint64Metric) IncrementBy(v uint64, fieldValues ...string) { } } +// Bucketer is an interface to bucket values into finite, distinct buckets. +type Bucketer interface { + // NumFiniteBuckets is the number of finite buckets in the distribution. + // This is only called once and never expected to return a different value. + NumFiniteBuckets() int + + // LowerBound takes the index of a bucket (within [0, NumBuckets()]) and + // returns the inclusive lower bound of that bucket. + // In other words, the lowest value of `x` for which `BucketIndex(x) == i` + // should be `x = LowerBound(i)`. + // The upper bound of a bucket is the lower bound of the next bucket. + // The last bucket (with `bucketIndex == NumFiniteBuckets()`) is infinite, + // i.e. it has no upper bound (but it still has a lower bound). + LowerBound(bucketIndex int) int64 + + // BucketIndex takes a sample and returns the index of the bucket that the + // sample should fall into. + // Must return either: + // - A value within [0, NumBuckets() -1] if the sample falls within a + // finite bucket + // - NumBuckets() if the sample falls within the last (infinite) bucket + // - '-1' if the sample is lower than what any bucket can represent, i.e. + // the sample should be in the implicit "underflow" bucket. + // This function must be go:nosplit-compatible and have no escapes. + // +checkescape:all + BucketIndex(sample int64) int +} + +// ExponentialBucketer implements Bucketer, with the first bucket starting +// with 0 as lowest bound with `Width` width, and each subsequent bucket being +// wider by a scaled exponentially-growing series, until `NumFiniteBuckets` +// buckets exist. +type ExponentialBucketer struct { + // numFinitebuckets is the total number of finite buckets in the scheme. + numFiniteBuckets int + + // width is the size of the first (0-th) finite bucket. + width float64 + + // scale is a factor applied uniformly to the exponential growth portion + // of the bucket size. + scale float64 + + // growth is the exponential growth factor for finite buckets. + // The n-th bucket is `growth` times wider than the (n-1)-th bucket. + // Bucket sizes are floored, so `width` and `growth` must be large enough + // such that the second bucket is actually wider than the first after + // flooring (unless, of course, fixed-width buckets are what's desired). + growth float64 + + // growthLog is math.Log(growth). + growthLog float64 + + // maxSample is the max sample value which can be represented in a finite + // bucket. + maxSample int64 + + // lowerbounds is a precomputed set of lower bounds of the buckets. + // The "underflow" bucket has no lower bound, so it is not included here. + // lowerBounds[0] is the lower bound of the first finite bucket, which is + // also the upper bound of the underflow bucket. + // lowerBounds[numFiniteBuckets] is the lower bound of the overflow bucket. + lowerBounds []int64 +} + +// NewExponentialBucketer returns a new Bucketer with exponential buckets. +func NewExponentialBucketer(numFiniteBuckets int, width uint64, scale, growth float64) *ExponentialBucketer { + b := &ExponentialBucketer{ + numFiniteBuckets: numFiniteBuckets, + width: float64(width), + scale: scale, + growth: growth, + growthLog: math.Log(growth), + lowerBounds: make([]int64, numFiniteBuckets+1), + } + b.lowerBounds[0] = 0 + for i := 1; i <= numFiniteBuckets; i++ { + b.lowerBounds[i] = int64(b.width*float64(i) + b.scale*math.Pow(b.growth, float64(i-1))) + } + b.maxSample = b.lowerBounds[numFiniteBuckets] - 1 + return b +} + +// NumFiniteBuckets implements Bucketer.NumFiniteBuckets. +func (b *ExponentialBucketer) NumFiniteBuckets() int { + return int(b.numFiniteBuckets) +} + +// LowerBound implements Bucketer.LowerBound. +func (b *ExponentialBucketer) LowerBound(bucketIndex int) int64 { + return b.lowerBounds[bucketIndex] +} + +// BucketIndex implements Bucketer.BucketIndex. +// +checkescape:all +//go:nosplit +func (b *ExponentialBucketer) BucketIndex(sample int64) int { + if sample < 0 { + return -1 + } + if sample == 0 { + return 0 + } + if sample > b.maxSample { + return b.numFiniteBuckets + } + // Do a binary search. For the number of buckets we expect to deal with in + // this code (a few dozen at most), this may be faster than computing a + // logarithm. We can't use recursion because this would violate go:nosplit. + lowIndex := 0 + highIndex := b.numFiniteBuckets + for { + pivotIndex := (highIndex + lowIndex) >> 1 + lowerBound := b.lowerBounds[pivotIndex] + if sample < lowerBound { + highIndex = pivotIndex + continue + } + upperBound := b.lowerBounds[pivotIndex+1] + if sample >= upperBound { + lowIndex = pivotIndex + continue + } + return pivotIndex + } +} + +// Verify that ExponentialBucketer implements Bucketer. +var _ = (Bucketer)((*ExponentialBucketer)(nil)) + +// DistributionMetric represents a distribution of values in finite buckets. +// It also separately keeps track of min/max in order to ascertain whether the +// buckets can faithfully represent the range of values encountered in the +// distribution. +type DistributionMetric struct { + // exponentialBucketer is the bucketing scheme used for this metric. + // Because we need DistributionMetric.AddSample to be go:nosplit-compatible, + // we cannot use an interface reference here, as we would not be able to call + // it in AddSample. Instead, we need one field per Bucketer implementation, + // and we call whichever one is in use in AddSample. + exponentialBucketer *ExponentialBucketer + + // metadata is the metadata about this metric. + metadata *pb.MetricMetadata + + // fieldsToKey converts a multi-dimensional fields to a single string to use + // as key for `samples`. + fieldsToKey fieldMapper + + // samples is the number of samples that fell within each bucket. + // It is mapped by the concatenation of the fields, using fieldsToKey. + // The value is a list of bucket sample counts, with the 0-th being the + // "underflow bucket", i.e. the bucket of samples which cannot fall into + // any bucket that the bucketer supports. + // The i-th value is the number of samples that fell into the bucketer's + // (i-1)-th finite bucket. + // The last value is the number of samples that fell into the bucketer's + // last (i.e. infinite) bucket. + samples map[string][]uint64 +} + +// NewDistributionMetric creates and registers a new distribution metric. +func NewDistributionMetric(name string, sync bool, bucketer Bucketer, unit pb.MetricMetadata_Units, description string, fields ...Field) (*DistributionMetric, error) { + if initialized { + return nil, ErrInitializationDone + } + if _, ok := allMetrics.uint64Metrics[name]; ok { + return nil, ErrNameInUse + } + if _, ok := allMetrics.distributionMetrics[name]; ok { + return nil, ErrNameInUse + } + + var exponentialBucketer *ExponentialBucketer + if expBucketer, ok := bucketer.(*ExponentialBucketer); ok { + exponentialBucketer = expBucketer + } else { + return nil, fmt.Errorf("unsupported bucketer implementation: %T", bucketer) + } + fieldsToKey, err := newFieldMapper(fields...) + if err != nil { + return nil, err + } + allKeys := fieldsToKey.all() + samples := make(map[string][]uint64, len(allKeys)) + numFiniteBuckets := bucketer.NumFiniteBuckets() + for _, key := range allKeys { + samples[key] = make([]uint64, numFiniteBuckets+2) + } + protoFields := make([]*pb.MetricMetadata_Field, len(fields)) + for i, f := range fields { + protoFields[i] = f.toProto() + } + lowerBounds := make([]int64, numFiniteBuckets+1) + for i := 0; i <= numFiniteBuckets; i++ { + lowerBounds[i] = bucketer.LowerBound(i) + } + allMetrics.distributionMetrics[name] = &DistributionMetric{ + exponentialBucketer: exponentialBucketer, + fieldsToKey: fieldsToKey, + samples: samples, + metadata: &pb.MetricMetadata{ + Name: name, + Description: description, + Cumulative: false, + Sync: sync, + Type: pb.MetricMetadata_TYPE_DISTRIBUTION, + Units: unit, + Fields: protoFields, + DistributionBucketLowerBounds: lowerBounds, + }, + } + return allMetrics.distributionMetrics[name], nil +} + +// MustRegisterDistributionMetric creates and registers a distribution metric. +// If an error occurs, it panics. +func MustRegisterDistributionMetric(name string, sync bool, bucketer Bucketer, unit pb.MetricMetadata_Units, description string, fields ...Field) *DistributionMetric { + distrib, err := NewDistributionMetric(name, sync, bucketer, unit, description, fields...) + if err != nil { + panic(err) + } + return distrib +} + +// AddSample adds a sample to the distribution. +// This *must* be called with the correct number of fields, or it will panic. +// +checkescape:all +//go:nosplit +func (d *DistributionMetric) AddSample(sample int64, fields ...string) { + key := d.fieldsToKey.lookup(fields...) + bucket := d.exponentialBucketer.BucketIndex(sample) + atomic.AddUint64(&d.samples[key][bucket+1], 1) +} + // stageTiming contains timing data for an initialization stage. type stageTiming struct { stage InitStage @@ -329,8 +699,11 @@ func (s stageTiming) inProgress() bool { // metricSet holds metric data. type metricSet struct { - // Map of metrics. - m map[string]customUint64Metric + // Map of uint64 metrics. + uint64Metrics map[string]customUint64Metric + + // Map of distribution metrics. + distributionMetrics map[string]*DistributionMetric // mu protects the fields below. mu sync.RWMutex @@ -346,8 +719,9 @@ type metricSet struct { // makeMetricSet returns a new metricSet. func makeMetricSet() metricSet { return metricSet{ - m: make(map[string]customUint64Metric), - finished: make([]stageTiming, 0, len(allStages)), + uint64Metrics: make(map[string]customUint64Metric), + distributionMetrics: make(map[string]*DistributionMetric), + finished: make([]stageTiming, 0, len(allStages)), } } @@ -358,34 +732,74 @@ func (m *metricSet) Values() metricValues { m.mu.Unlock() vals := metricValues{ - m: make(map[string]interface{}, len(m.m)), - stages: stages, + uint64Metrics: make(map[string]interface{}, len(m.uint64Metrics)), + distributionMetrics: make(map[string]map[string][]uint64, len(m.distributionMetrics)), + distributionTotalSamples: make(map[string]map[string]uint64, len(m.distributionMetrics)), + stages: stages, } - - for k, v := range m.m { + for k, v := range m.uint64Metrics { fields := v.metadata.GetFields() switch len(fields) { case 0: - vals.m[k] = v.value() + vals.uint64Metrics[k] = v.value() case 1: values := fields[0].GetAllowedValues() fieldsMap := make(map[string]uint64) for _, fieldValue := range values { fieldsMap[fieldValue] = v.value(fieldValue) } - vals.m[k] = fieldsMap + vals.uint64Metrics[k] = fieldsMap default: panic(fmt.Sprintf("Unsupported number of metric fields: %d", len(fields))) } } + for name, metric := range m.distributionMetrics { + fieldKeysToValues := make(map[string][]uint64, len(metric.samples)) + fieldKeysToTotalSamples := make(map[string]uint64, len(metric.samples)) + for fieldKey, samples := range metric.samples { + samplesSnapshot := snapshotDistribution(samples) + totalSamples := uint64(0) + for _, bucket := range samplesSnapshot { + totalSamples += bucket + } + if totalSamples == 0 { + // No samples recorded for this combination of field, so leave + // the maps for this fieldKey as nil. This lessens the memory cost + // of distributions with unused field combinations. + fieldKeysToTotalSamples[fieldKey] = 0 + fieldKeysToValues[fieldKey] = nil + } else { + fieldKeysToTotalSamples[fieldKey] = totalSamples + fieldKeysToValues[fieldKey] = samplesSnapshot + } + } + vals.distributionMetrics[name] = fieldKeysToValues + vals.distributionTotalSamples[name] = fieldKeysToTotalSamples + } return vals } // metricValues contains a copy of the values of all metrics. type metricValues struct { - // m is a map with key as metric name and value can be either uint64 or - // map[string]uint64 to support metrics with one field. - m map[string]interface{} + // uint64Metrics is a map of uint64 metrics, + // with key as metric name. Value can be either uint64, or map[string]uint64 + // to support metrics with one field. + uint64Metrics map[string]interface{} + + // distributionMetrics is a map of distribution metrics. + // The first key level is the metric name. + // The second key level is the concatenated view of the fields. + // The value is the number of samples in each bucket of the distribution, + // with the first (0-th) element being the underflow bucket and the last + // element being the "infinite" (overflow) bucket. + distributionMetrics map[string]map[string][]uint64 + + // distributionTotalSamples is the total number of samples for each + // distribution metric and field values. + // It allows performing a quick diff between snapshots without having to + // iterate over all the buckets individually, so that distributions with + // no new samples are not retransmitted. + distributionTotalSamples map[string]map[string]uint64 // Information on when initialization stages were reached. Does not include // the currently-ongoing stage, if any. @@ -419,8 +833,8 @@ func EmitMetricUpdate() { m := pb.MetricUpdate{} // On the first call metricsAtLastEmit will be empty. Include all // metrics then. - for k, v := range snapshot.m { - prev, ok := metricsAtLastEmit.m[k] + for k, v := range snapshot.uint64Metrics { + prev, ok := metricsAtLastEmit.uint64Metrics[k] switch t := v.(type) { case uint64: // Metric exists and value did not change. @@ -450,6 +864,43 @@ func EmitMetricUpdate() { } } } + for name, dist := range snapshot.distributionTotalSamples { + prev, ok := metricsAtLastEmit.distributionTotalSamples[name] + for fieldKey, currentTotal := range dist { + if currentTotal == 0 { + continue + } + if ok { + if prevTotal, ok2 := prev[fieldKey]; ok2 && prevTotal == currentTotal { + continue + } + } + oldSamples := metricsAtLastEmit.distributionMetrics[name][fieldKey] + var newSamples []uint64 + if oldSamples != nil { + currentSamples := snapshot.distributionMetrics[name][fieldKey] + numBuckets := len(currentSamples) + newSamples = make([]uint64, numBuckets) + for i := 0; i < numBuckets; i++ { + newSamples[i] = currentSamples[i] - oldSamples[i] + } + } else { + // oldSamples == nil means that the previous snapshot has no samples. + // This means the delta is the current number of samples, no need for + // a copy. + newSamples = snapshot.distributionMetrics[name][fieldKey] + } + m.Metrics = append(m.Metrics, &pb.MetricValue{ + Name: name, + FieldValues: keyToMultiField(fieldKey), + Value: &pb.MetricValue_DistributionValue{ + DistributionValue: &pb.Samples{ + NewSamples: newSamples, + }, + }, + }) + } + } for s := len(metricsAtLastEmit.stages); s < len(snapshot.stages); s++ { newStage := snapshot.stages[s] diff --git a/pkg/metric/metric.proto b/pkg/metric/metric.proto index d466b6904..59dfdf6fe 100644 --- a/pkg/metric/metric.proto +++ b/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; diff --git a/pkg/metric/metric_test.go b/pkg/metric/metric_test.go index 0654bdf07..1d188c8b1 100644 --- a/pkg/metric/metric_test.go +++ b/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) + } + } + }) + } +} diff --git a/pkg/metric/metric_unsafe.go b/pkg/metric/metric_unsafe.go new file mode 100644 index 000000000..a092844bd --- /dev/null +++ b/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 +}