From 235e7e0fff76df5a673450bde39c8170f6111505 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Tue, 22 Mar 2022 20:57:29 -0700 Subject: [PATCH] Sentry: Implement timer metrics. Timer metrics are metrics that measure nanosecond-precision operations within the sentry, and present the results in an aggregated manner using distribution bucketing. They have a more convenient API than using distribution metrics directly, but otherwise are just a convenient wrapper on top of them. This requires exposing `runtime.nanotime()` from the Go runtime in order to get the current time without causing system calls. Intended usage: ```go m := NewTimerMetric(...) ... op := m.Start() // Starts measuring time from this point ... do something interesting... op.Finish() // End the stopwatch. ``` Operation structs are meant to be self-contained to be able to be passed around in code. Additionally, the timer metrics support multiple fields, specifiable both at operation `Start` and `Finish` time. This can be useful for measuring operations that can go down multiple distinct branches, and creating aggregates that can selectively differentiate between them. For example: ```go m := NewTimerMetric("/packet_processing_time", ..., "protocol", "path") ... func HandleTCPPacket(pkt TCPPacket) { op := m.Start("tcp") if fast path { ... do fast TCP handling... op.Finish("fast") return } doSlowTCPHandling(pkt, op) // `op` can be passed around } func doSlowTCPHandling(pkt TCPPacket, op TimerOperation) { ... do slow TCP handling... op.Finish("slow") } func HandleUDPPacket(pkt UDPPacket) { op := m.Start("udp") ... do UDP packet handling... op.Finish("") } ``` PiperOrigin-RevId: 436641265 --- pkg/gohacks/gohacks_test.go | 11 +++ pkg/gohacks/gohacks_unsafe.go | 11 +++ pkg/metric/BUILD | 1 + pkg/metric/metric.go | 138 ++++++++++++++++++++++++++++++++- pkg/metric/metric_test.go | 141 ++++++++++++++++++++++++++++++++-- pkg/metric/metric_unsafe.go | 6 ++ tools/checklinkname/known.go | 3 + 7 files changed, 304 insertions(+), 7 deletions(-) diff --git a/pkg/gohacks/gohacks_test.go b/pkg/gohacks/gohacks_test.go index e18c8abc7..08277b9d1 100644 --- a/pkg/gohacks/gohacks_test.go +++ b/pkg/gohacks/gohacks_test.go @@ -20,6 +20,7 @@ import ( "os" "runtime/debug" "testing" + "time" "golang.org/x/sys/unix" ) @@ -95,3 +96,13 @@ func TestSigbusOnMemmove(t *testing.T) { t.Fatalf("testCopy didn't panic when it should have") } } + +func TestNanotime(t *testing.T) { + // Verify that nanotime increases over time. + nano1 := Nanotime() + time.Sleep(10 * time.Millisecond) + nano2 := Nanotime() + if nano2 <= nano1 { + t.Errorf("runtime.nanotime() did not increase after 10ms: %d vs %d", nano1, nano2) + } +} diff --git a/pkg/gohacks/gohacks_unsafe.go b/pkg/gohacks/gohacks_unsafe.go index c4bf360a5..c183a6ac9 100644 --- a/pkg/gohacks/gohacks_unsafe.go +++ b/pkg/gohacks/gohacks_unsafe.go @@ -95,3 +95,14 @@ func Memmove(to, from unsafe.Pointer, n uintptr) { //go:linkname memmove runtime.memmove //go:noescape func memmove(to, from unsafe.Pointer, n uintptr) + +// Nanotime is runtime.nanotime. +// +//go:nosplit +func Nanotime() int64 { + return nanotime() +} + +//go:linkname nanotime runtime.nanotime +//go:noescape +func nanotime() int64 diff --git a/pkg/metric/BUILD b/pkg/metric/BUILD index 570bf9aa6..ca43e6fc4 100644 --- a/pkg/metric/BUILD +++ b/pkg/metric/BUILD @@ -35,6 +35,7 @@ go_test( deps = [ ":metric_go_proto", "//pkg/eventchannel", + "//pkg/sync", "@org_golang_google_protobuf//proto:go_default_library", ], ) diff --git a/pkg/metric/metric.go b/pkg/metric/metric.go index 68c962c92..f43cc0370 100644 --- a/pkg/metric/metric.go +++ b/pkg/metric/metric.go @@ -290,6 +290,33 @@ func (m fieldMapper) lookup(fields ...string) string { return m.key } +// lookupConcat looks up a key within the fieldMapper where the fields are +// the concatenation of two list of fields. +// It needs to allocate no memory and be nosplit-compatible, so it cannot be +// recursive, and cannot allocate a concatenated []string. +// This *must* be called with the correct number of fields, or it will panic. +// +checkescape:all +//go:nosplit +func (m fieldMapper) lookupConcat(fields1, fields2 []string) string { + depth1 := len(fields1) + depth2 := len(fields2) + if depth1+depth2 != m.depth { + panic("invalid field lookup depth") + } + var found bool + for i := 0; i < depth1; i++ { + if m, found = m.children[fields1[i]]; !found { + panic("disallowed field value") + } + } + for i := 0; i < depth2; i++ { + if m, found = m.children[fields2[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 @@ -514,8 +541,17 @@ type ExponentialBucketer struct { lowerBounds []int64 } +// Minimum/maximum finite buckets for exponential bucketers. +const ( + exponentialMinBuckets = 1 + exponentialMaxBuckets = 100 +) + // NewExponentialBucketer returns a new Bucketer with exponential buckets. func NewExponentialBucketer(numFiniteBuckets int, width uint64, scale, growth float64) *ExponentialBucketer { + if numFiniteBuckets < exponentialMinBuckets || numFiniteBuckets > exponentialMaxBuckets { + panic(fmt.Sprintf("number of finite buckets must be in [%d, %d]", exponentialMinBuckets, exponentialMaxBuckets)) + } b := &ExponentialBucketer{ numFiniteBuckets: numFiniteBuckets, width: float64(width), @@ -679,11 +715,111 @@ func MustRegisterDistributionMetric(name string, sync bool, bucketer Bucketer, u // +checkescape:all //go:nosplit func (d *DistributionMetric) AddSample(sample int64, fields ...string) { - key := d.fieldsToKey.lookup(fields...) + d.addSampleByKey(sample, d.fieldsToKey.lookup(fields...)) +} + +// addSampleByKey works like AddSample, with the field key already known. +// +checkescape:all +//go:nosplit +func (d *DistributionMetric) addSampleByKey(sample int64, key string) { bucket := d.exponentialBucketer.BucketIndex(sample) atomic.AddUint64(&d.samples[key][bucket+1], 1) } +// Minimum number of buckets for NewDurationBucket. +const durationMinBuckets = 3 + +// NewDurationBucketer returns a Bucketer well-suited for measuring durations in +// nanoseconds. Useful for NewTimerMetric. +// minDuration and maxDuration are conservative estimates of the minimum and +// maximum durations expected to be accurately measured by the Bucketer. +func NewDurationBucketer(numFiniteBuckets int, minDuration, maxDuration time.Duration) Bucketer { + if numFiniteBuckets < durationMinBuckets { + panic(fmt.Sprintf("duration bucketer must have at least %d buckets, got %d", durationMinBuckets, numFiniteBuckets)) + } + minNs := minDuration.Nanoseconds() + exponentCoversNs := float64(maxDuration.Nanoseconds()-int64(numFiniteBuckets-durationMinBuckets)*minNs) / float64(minNs) + exponent := math.Log(exponentCoversNs) / math.Log(float64(numFiniteBuckets-durationMinBuckets)) + minNs = int64(float64(minNs) / exponent) + return NewExponentialBucketer(numFiniteBuckets, uint64(minNs), float64(minNs), exponent) +} + +// TimerMetric wraps a distribution metric with convenience functions for +// latency measurements, which is a popular specialization of distribution +// metrics. +type TimerMetric struct { + DistributionMetric +} + +// NewTimerMetric provides a convenient way to measure latencies. +// The arguments are the same as `NewDistributionMetric`, except: +// - `nanoBucketer`: Same as `NewDistribution`'s `bucketer`, expected to hold +// durations in nanoseconds. Adjust parameters accordingly. +// NewDurationBucketer may be helpful here. +func NewTimerMetric(name string, nanoBucketer Bucketer, description string, fields ...Field) (*TimerMetric, error) { + distrib, err := NewDistributionMetric(name, false, nanoBucketer, pb.MetricMetadata_UNITS_NANOSECONDS, description, fields...) + if err != nil { + return nil, err + } + return &TimerMetric{ + DistributionMetric: *distrib, + }, nil +} + +// MustRegisterTimerMetric creates and registers a timer metric. +// If an error occurs, it panics. +func MustRegisterTimerMetric(name string, nanoBucketer Bucketer, description string, fields ...Field) *TimerMetric { + timer, err := NewTimerMetric(name, nanoBucketer, description, fields...) + if err != nil { + panic(err) + } + return timer +} + +// TimedOperation is used by TimerMetric to keep track of the time elapsed +// between an operation starting and stopping. +type TimedOperation struct { + // metric is a reference to the timer metric for the operation. + metric *TimerMetric + + // partialFields is a prefix of the fields used in this operation. + // The rest of the fields is provided in TimedOperation.Finish. + partialFields []string + + // startedNs is the number of nanoseconds measured in TimerMetric.Start(). + startedNs int64 +} + +// Start starts a timer measurement for the given combination of fields. +// It returns a TimedOperation which can be passed around as necessary to +// measure the duration of the operation. +// Once the operation is finished, call Finish on the TimedOperation. +// The fields passed to Start may be partially specified; if so, the remaining +// fields must be passed to TimedOperation.Finish. This is useful for cases +// where which path an operation took is only known after it happens. This +// path can be part of the fields passed to Finish. +//+checkescape:all +//go:nosplit +func (t *TimerMetric) Start(fields ...string) TimedOperation { + return TimedOperation{ + metric: t, + partialFields: fields, + startedNs: CheapNowNano(), + } +} + +// Finish marks an operation as finished and records its duration. +// `extraFields` is the rest of the fields appended to the fields passed to +// `TimerMetric.Start`. The concatenation of these two must be the exact +// number of fields that the underlying metric has. +//+checkescape:all +//go:nosplit +func (o TimedOperation) Finish(extraFields ...string) { + ended := CheapNowNano() + fieldKey := o.metric.fieldsToKey.lookupConcat(o.partialFields, extraFields) + o.metric.addSampleByKey(ended-o.startedNs, fieldKey) +} + // stageTiming contains timing data for an initialization stage. type stageTiming struct { stage InitStage diff --git a/pkg/metric/metric_test.go b/pkg/metric/metric_test.go index 1d188c8b1..200615c17 100644 --- a/pkg/metric/metric_test.go +++ b/pkg/metric/metric_test.go @@ -16,12 +16,14 @@ package metric import ( "math" + "reflect" "testing" "time" "google.golang.org/protobuf/proto" "gvisor.dev/gvisor/pkg/eventchannel" pb "gvisor.dev/gvisor/pkg/metric/metric_go_proto" + "gvisor.dev/gvisor/pkg/sync" ) // sliceEmitter implements eventchannel.Emitter by appending all messages to a @@ -638,13 +640,77 @@ func TestMetricUpdateStageTiming(t *testing.T) { } } +func TestTimerMetric(t *testing.T) { + defer reset() + // This bucketer just has 2 finite buckets: [0, 500ms) and [500ms, 1s). + bucketer := NewExponentialBucketer(2, uint64((500 * time.Millisecond).Nanoseconds()), 0, 1) + field1 := NewField("field1", []string{"foo", "bar"}) + field2 := NewField("field2", []string{"baz", "quux"}) + timer, err := NewTimerMetric("/timer", bucketer, "a timer metric", field1, field2) + if err != nil { + t.Fatalf("NewTimerMetric: %v", err) + } + if err := Initialize(); err != nil { + t.Fatalf("Initialize(): %s", err) + } + // Don't care about the registration metrics. + emitter.Reset() + + // Create timer data. + var wg sync.WaitGroup + for i := 0; i < 5; i++ { + wg.Add(1) + go func() { + defer wg.Done() + op := timer.Start("foo") + defer op.Finish("quux") + time.Sleep(250 * time.Millisecond) + }() + } + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { + defer wg.Done() + op := timer.Start() + defer op.Finish("foo", "quux") + time.Sleep(750 * time.Millisecond) + }() + } + wg.Wait() + EmitMetricUpdate() + if len(emitter) != 1 { + t.Fatalf("EmitMetricUpdate emitted %d events want %d", len(emitter), 1) + } + m := emitter[0].(*pb.MetricUpdate).Metrics[0] + wantFields := []string{"foo", "quux"} + if !reflect.DeepEqual(m.GetFieldValues(), wantFields) { + t.Errorf("%+v: got fields %v want %v", m, m.GetFieldValues(), wantFields) + } + dv, ok := m.Value.(*pb.MetricValue_DistributionValue) + if !ok { + t.Fatalf("%+v: want pb.MetricValue_DistributionValue", m) + } + samples := dv.DistributionValue.GetNewSamples() + if len(samples) != 4 { + t.Fatalf("%+v: got %d buckets, want %d", dv.DistributionValue, len(samples), 4) + } + wantSamples := []uint64{0, 5, 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]) + } + } +} + func TestBucketer(t *testing.T) { for _, test := range []struct { - name string - bucketer Bucketer - minSample int64 - maxSample int64 - firstFewLowerBounds []int64 + name string + bucketer Bucketer + minSample int64 + maxSample int64 + step int64 + firstFewLowerBounds []int64 + successiveBucketSamples []int64 }{ { name: "static-sized buckets", @@ -667,9 +733,33 @@ func TestBucketer(t *testing.T) { 50 + int64(math.Floor(2*1.5*1.5*1.5*1.5)), }, }, + { + name: "timer buckets", + bucketer: NewDurationBucketer(8, time.Second, time.Minute), + minSample: 0, + maxSample: (5 * time.Minute).Nanoseconds(), + step: (500 * time.Millisecond).Nanoseconds(), + successiveBucketSamples: []int64{ + // Roughly exponential successive durations: + (500 * time.Millisecond).Nanoseconds(), + (1200 * time.Millisecond).Nanoseconds(), + (2500 * time.Millisecond).Nanoseconds(), + (5 * time.Second).Nanoseconds(), + (15 * time.Second).Nanoseconds(), + (35 * time.Second).Nanoseconds(), + (75 * time.Second).Nanoseconds(), + (3 * time.Minute).Nanoseconds(), + (7 * time.Minute).Nanoseconds(), + }, + }, } { t.Run(test.name, func(t *testing.T) { numFiniteBuckets := test.bucketer.NumFiniteBuckets() + t.Logf("Underflow bucket has bounds (-inf, %d)", test.bucketer.LowerBound(0)) + for b := 0; b < numFiniteBuckets; b++ { + t.Logf("Bucket %d has bounds [%d, %d)", b, test.bucketer.LowerBound(b), test.bucketer.LowerBound(b+1)) + } + t.Logf("Overflow bucket has bounds [%d, +inf)", test.bucketer.LowerBound(numFiniteBuckets)) testAround := func(bound int64, bucketIndex int) { for sample := bound - 2; sample <= bound+2; sample++ { gotIndex := test.bucketer.BucketIndex(sample) @@ -678,7 +768,11 @@ func TestBucketer(t *testing.T) { } } } - for sample := test.minSample; sample <= test.maxSample; sample++ { + step := test.step + if step == 0 { + step = 1 + } + for sample := test.minSample; sample <= test.maxSample; sample += step { bucket := test.bucketer.BucketIndex(sample) if bucket == -1 { lowestBound := test.bucketer.LowerBound(0) @@ -712,6 +806,41 @@ func TestBucketer(t *testing.T) { t.Errorf("bucket %d has lower bound %d, want %d", bi, got, want) } } + previousBucket := -1 + for i, sample := range test.successiveBucketSamples { + gotBucket := test.bucketer.BucketIndex(sample) + if gotBucket != previousBucket+1 { + t.Errorf("successive-bucket sample #%d (%d) fell in bucket %d whereas previous sample fell in bucket %d", i, sample, gotBucket, previousBucket) + } + previousBucket = gotBucket + } + }) + } +} + +func TestBucketerPanics(t *testing.T) { + for name, fn := range map[string]func(){ + "NewExponentialBucketer @ 0": func() { + NewExponentialBucketer(0, 2, 0, 1) + }, + "NewExponentialBucketer @ 120": func() { + NewExponentialBucketer(120, 2, 0, 1) + }, + "NewDurationBucketer @ 2": func() { + NewDurationBucketer(2, time.Second, time.Minute) + }, + } { + t.Run(name, func(t *testing.T) { + var recovered interface{} + func() { + defer func() { + recovered = recover() + }() + fn() + }() + if recovered == nil { + t.Error("did not panic") + } }) } } diff --git a/pkg/metric/metric_unsafe.go b/pkg/metric/metric_unsafe.go index a092844bd..c5a8e579a 100644 --- a/pkg/metric/metric_unsafe.go +++ b/pkg/metric/metric_unsafe.go @@ -45,3 +45,9 @@ func snapshotDistribution(samples []uint64) []uint64 { } return snapshot } + +// CheapNowNano returns the current unix timestamp in nanoseconds. +//go:nosplit +func CheapNowNano() int64 { + return gohacks.Nanotime() +} diff --git a/tools/checklinkname/known.go b/tools/checklinkname/known.go index 455d80855..aa4d61745 100644 --- a/tools/checklinkname/known.go +++ b/tools/checklinkname/known.go @@ -76,6 +76,9 @@ var knownLinknames = map[string]map[string]linknameSignatures{ "wakep": linknameSignatures{ local: "func()", }, + "nanotime": linknameSignatures{ + local: "func() int64", + }, }, "sync": map[string]linknameSignatures{ "runtime_canSpin": linknameSignatures{