From 8e6f57da4a88e4f543d3551db14959f9d27aa849 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Mon, 5 Jun 2023 15:50:54 -0700 Subject: [PATCH] Metric verification lib: Port over data potentially missing from new snapshot. The Prometheus metric verification library uses a `numberPacker` to pack the numbers it must retain from snapshot to snapshot into a tiny amount of space. This involves putting those that fit in a few bits as-is, but for the larger ones, they are stored in the `numberPacker` struct itself, and the packed number represents an offset within that struct instead. When the library verifies a new snapshot, it instantiates a new `numberPacker` so that numbers that are no longer referenced are not kept around forever. This works well in most cases, but in the case where a new snapshot does *not* contain a particular metric for whatever reason, the library will still report it as existing, but the indirectly-referenced numbers will no longer exist. This CL reworks how `numberPacker` is used such that this usage of indirectly-stored numbers is tracked more precisely across snapshots, and ensures that all such numbers are ported over from a snapshot to the next even if the next snapshot only contains a partial result. It also changes the `numberPacker` semantics to have its storage be explicitly allocated, and will `panic` if asked to store more than that. This means the `numberPacker` never needs to allocate memory, so (as a bonus) it can use `go:nosplit`. Additionally, the Verifier checks that it uses exactly all the storage slots it thinks it will need, which ensures that it has correctly tracked the expected usage. The tests are minimal but will be further reinforced in a future CL which adds additional distribution statistics into distribution metrics. One of these (the sum-of-squared-deviations statistic) is a floating-point number which almost always requires indirect storage, and thus provides more consistent coverage. Previous unit tests almost never actually required indirect storage, hence this bug not having been found until now. PiperOrigin-RevId: 538002697 --- pkg/prometheus/prometheus_test.go | 190 +++++++++++++++------ pkg/prometheus/prometheus_verify.go | 252 ++++++++++++++++++++++------ 2 files changed, 346 insertions(+), 96 deletions(-) diff --git a/pkg/prometheus/prometheus_test.go b/pkg/prometheus/prometheus_test.go index 6244c9090..8bc7b1547 100644 --- a/pkg/prometheus/prometheus_test.go +++ b/pkg/prometheus/prometheus_test.go @@ -851,6 +851,15 @@ func TestVerifier(t *testing.T) { }, ), }, + { + Name: "partial incremental snapshot needing indirection", + Registration: newMetricRegistration(fooCounter), + WantSuccess: []*Snapshot{ + newSnapshotAt(epsilon(-2)).Add(fooCounter.int(int64(maxDirectUint + 2))), + newSnapshotAt(epsilon(-1)).Add(), + newSnapshotAt(epsilon(0)).Add(fooCounter.int(int64(maxDirectUint + 3))), + }, + }, { Name: "worked example", Registration: newMetricRegistration( @@ -944,16 +953,34 @@ func TestVerifier(t *testing.T) { } } else { for i, snapshot := range test.WantSuccess { - if err = verifier.Verify(snapshot); err != nil { - t.Fatalf("snapshot WantSuccess[%d] failed verification: %v", i, err) - } + func() { + defer func() { + panicErr := recover() + t.Helper() + if panicErr != nil { + t.Fatalf("panic during verification of WantSuccess[%d] snapshot: %v", i, panicErr) + } + }() + if err = verifier.Verify(snapshot); err != nil { + t.Fatalf("snapshot WantSuccess[%d] failed verification: %v", i, err) + } + }() } if test.WantFail != nil { - if err = verifier.Verify(test.WantFail); err == nil { - t.Error("WantFail snapshot unexpectedly succeeded verification") - } else { - t.Logf("WantFail snapshot failed verification (as expected by this test): %v", err) - } + func() { + defer func() { + panicErr := recover() + t.Helper() + if panicErr != nil { + t.Fatalf("panic during verification of WantFail snapshot: %v", panicErr) + } + }() + if err = verifier.Verify(test.WantFail); err == nil { + t.Error("WantFail snapshot unexpectedly succeeded verification") + } else { + t.Logf("WantFail snapshot failed verification (as expected by this test): %v", err) + } + }() } } }) @@ -1550,6 +1577,8 @@ func TestNumberPacker(t *testing.T) { } } for _, i := range []int64{ + 0, + -1, math.MinInt, math.MaxInt, math.MinInt8, @@ -1563,6 +1592,7 @@ func TestNumberPacker(t *testing.T) { math.MaxUint32, math.MinInt64, math.MaxInt64, + int64(maxDirectUint), } { for d := int64(-3); d <= int64(3); d++ { interestingIntegers[uint64(i+d)] = struct{}{} @@ -1577,22 +1607,61 @@ func TestNumberPacker(t *testing.T) { interestingIntegers[math.MaxUint64-1] = struct{}{} interestingIntegers[math.MaxUint64] = struct{}{} - p := &numberPacker{} + interestingFloats := make(map[float64]struct{}, len(interestingIntegers)+21*21+17) + for divExp := -10; divExp < 10; divExp++ { + div := math.Pow(10, float64(divExp)) + for i := -10; i < 10; i++ { + interestingFloats[float64(i)*div] = struct{}{} + } + } + interestingFloats[0.0] = struct{}{} + interestingFloats[math.NaN()] = struct{}{} + interestingFloats[math.Inf(1)] = struct{}{} + interestingFloats[math.Inf(-1)] = struct{}{} + interestingFloats[math.Pi] = struct{}{} + interestingFloats[math.Sqrt2] = struct{}{} + interestingFloats[math.E] = struct{}{} + interestingFloats[math.SqrtE] = struct{}{} + interestingFloats[math.Ln2] = struct{}{} + interestingFloats[math.MaxFloat32] = struct{}{} + interestingFloats[-math.MaxFloat32] = struct{}{} + interestingFloats[math.MaxFloat64] = struct{}{} + interestingFloats[-math.MaxFloat64] = struct{}{} + interestingFloats[math.SmallestNonzeroFloat32] = struct{}{} + interestingFloats[-math.SmallestNonzeroFloat32] = struct{}{} + interestingFloats[math.SmallestNonzeroFloat64] = struct{}{} + interestingFloats[-math.SmallestNonzeroFloat64] = struct{}{} + for interestingInt := range interestingIntegers { + interestingFloats[math.Float64frombits(interestingInt)] = struct{}{} + } + + p := &numberPacker{ + data: make([]uint64, 0, len(interestingIntegers)+len(interestingFloats)), + } + t.Run("integers", func(t *testing.T) { seenDirectInteger := false seenIndirectInteger := false for interestingInt := range interestingIntegers { orig := NewInt(int64(interestingInt)) - packed, err := p.pack(orig) - if err != nil { - t.Fatalf("integer %v (bits=%x): cannot pack: %v", orig, interestingInt, err) - } + packed := p.pack(orig) unpacked := p.unpack(packed) if !orig.SameType(unpacked) || orig.Int != unpacked.Int { t.Errorf("integer %v (bits=%x): got packed=%v => unpacked version %v (int: %d)", orig, interestingInt, uint32(packed), unpacked, unpacked.Int) } - seenDirectInteger = seenDirectInteger || (uint32(packed)&storageField) == storageFieldDirect - seenIndirectInteger = seenIndirectInteger || (uint32(packed)&storageField) == storageFieldIndirect + needsIndirection := needsPackerStorage(orig) + switch uint32(packed) & storageField { + case storageFieldDirect: + seenDirectInteger = true + if needsIndirection != 0 { + t.Errorf("integer %v (bits=%x): got needsIndirection=%v want %v", orig, interestingInt, needsIndirection, 0) + } + case storageFieldIndirect: + seenIndirectInteger = true + if needsIndirection != 1 { + t.Errorf("integer %v (bits=%x): got needsIndirection=%v want %v", orig, interestingInt, needsIndirection, 1) + } + } } if !seenDirectInteger { t.Error("did not encounter any integer that could be packed directly") @@ -1608,41 +1677,11 @@ func TestNumberPacker(t *testing.T) { } }) t.Run("floats", func(t *testing.T) { - interestingFloats := make(map[float64]struct{}, len(interestingIntegers)+21*21+17) - for divExp := -10; divExp < 10; divExp++ { - div := math.Pow(10, float64(divExp)) - for i := -10; i < 10; i++ { - interestingFloats[float64(i)*div] = struct{}{} - } - } - interestingFloats[0.0] = struct{}{} - interestingFloats[math.NaN()] = struct{}{} - interestingFloats[math.Inf(1)] = struct{}{} - interestingFloats[math.Inf(-1)] = struct{}{} - interestingFloats[math.Pi] = struct{}{} - interestingFloats[math.Sqrt2] = struct{}{} - interestingFloats[math.E] = struct{}{} - interestingFloats[math.SqrtE] = struct{}{} - interestingFloats[math.Ln2] = struct{}{} - interestingFloats[math.MaxFloat32] = struct{}{} - interestingFloats[-math.MaxFloat32] = struct{}{} - interestingFloats[math.MaxFloat64] = struct{}{} - interestingFloats[-math.MaxFloat64] = struct{}{} - interestingFloats[math.SmallestNonzeroFloat32] = struct{}{} - interestingFloats[-math.SmallestNonzeroFloat32] = struct{}{} - interestingFloats[math.SmallestNonzeroFloat64] = struct{}{} - interestingFloats[-math.SmallestNonzeroFloat64] = struct{}{} - for interestingInt := range interestingIntegers { - interestingFloats[math.Float64frombits(interestingInt)] = struct{}{} - } seenDirectFloat := false seenIndirectFloat := false for interestingFloat := range interestingFloats { orig := NewFloat(interestingFloat) - packed, err := p.pack(orig) - if err != nil { - t.Fatalf("float %v (64bits=%x, 32bits=%x. float32-encodable=%v): cannot pack: %v", orig, math.Float64bits(interestingFloat), math.Float32bits(float32(interestingFloat)), float64(float32(interestingFloat)) == interestingFloat, err) - } + packed := p.pack(orig) unpacked := p.unpack(packed) switch { case interestingFloat == 0: // Zero-valued float becomes an integer. @@ -1660,8 +1699,19 @@ func TestNumberPacker(t *testing.T) { t.Errorf("float %v (64bits=%x, 32bits=%x, float32-encodable=%v): got packed=%x => unpacked version %v (float: %f)", orig, math.Float64bits(interestingFloat), math.Float32bits(float32(interestingFloat)), float64(float32(interestingFloat)) == interestingFloat, uint32(packed), unpacked, unpacked.Float) } } - seenDirectFloat = seenDirectFloat || (uint32(packed)&storageField) == storageFieldDirect - seenIndirectFloat = seenIndirectFloat || (uint32(packed)&storageField) == storageFieldIndirect + needsIndirection := needsPackerStorage(orig) + switch uint32(packed) & storageField { + case storageFieldDirect: + seenDirectFloat = true + if needsIndirection != 0 { + t.Errorf("float %v (64bits=%x): got needsIndirection=%v want %v", orig, math.Float64bits(interestingFloat), needsIndirection, 0) + } + case storageFieldIndirect: + seenIndirectFloat = true + if needsIndirection != 1 { + t.Errorf("float %v (bits=%x): got needsIndirection=%v want %v", orig, math.Float64bits(interestingFloat), needsIndirection, 1) + } + } } if !seenDirectFloat { t.Error("did not encounter any float that could be packed directly") @@ -1671,3 +1721,47 @@ func TestNumberPacker(t *testing.T) { } }) } + +func TestNumberPackerCapacity(t *testing.T) { + packer := &numberPacker{ + data: make([]uint64, 0, 2), + } + checkPanic := func(want bool, fn func()) { + t.Helper() + defer func() { + panicErr := recover() + t.Helper() + if want && panicErr == nil { + t.Error("function did not panic but wanted it to") + } else if !want && panicErr != nil { + t.Errorf("function unexpectedly panic'd: %v", panicErr) + } + }() + fn() + } + t.Run("number that does not need indirection", func(t *testing.T) { + checkPanic(false, func() { + packer.pack(&Number{Int: 1}) + }) + }) + t.Run("first number that needs indirection fits", func(t *testing.T) { + checkPanic(false, func() { + packer.pack(&Number{Int: int64(maxDirectUint + 3)}) + }) + }) + t.Run("second number that needs indirection also fits", func(t *testing.T) { + checkPanic(false, func() { + packer.pack(&Number{Int: int64(maxDirectUint + 2)}) + }) + }) + t.Run("third number that needs indirection does not", func(t *testing.T) { + checkPanic(true, func() { + packer.pack(&Number{Int: int64(maxDirectUint + 1)}) + }) + }) + t.Run("second number that does not need indirection still fits", func(t *testing.T) { + checkPanic(false, func() { + packer.pack(&Number{Int: int64(maxDirectUint)}) + }) + }) +} diff --git a/pkg/prometheus/prometheus_verify.go b/pkg/prometheus/prometheus_verify.go index 881c8a00d..fccbd86c4 100644 --- a/pkg/prometheus/prometheus_verify.go +++ b/pkg/prometheus/prometheus_verify.go @@ -152,6 +152,11 @@ func globalInternVerifierReleased() { // numberPacker holds packedNumber data. It is useful to store large amounts of Number structs in a // small memory footprint. type numberPacker struct { + // `data` *must* be pre-allocated if there is any number to be stored in it. + // Attempts to pack a number that cannot fit into the existing space + // allocated for this slice will cause a panic. + // Callers may use `needsIndirection` to determine whether a number needs + // space in this slice or not ahead of packing it. data []uint64 } @@ -187,6 +192,7 @@ const ( storageFieldDirect = uint32(0) storageFieldIndirect = uint32(storageField) valueField = uint32(1<<30 - 1) + maxDirectUint = uint64(valueField) float32ExponentField = uint32(0x7f800000) float32ExponentShift = uint32(23) float32ExponentBias = uint32(127) @@ -198,35 +204,75 @@ const ( packedFloatInf = packedNumber(typeFieldFloat | storageFieldDirect | 0x30000000) ) -// errOutOfPackerMemory is returned when the number cannot be packed into a numberPacker. -var errOutOfPackerMemory = errors.New("out of numberPacker memory") - -// pack packs a Number into a packedNumber. -func (p *numberPacker) pack(n *Number) (packedNumber, error) { +// needsPackerStorage returns 0 for numbers that can be +// stored directly into the 32 bits of a packedNumber, or 1 for numbers that +// need more bits and would need to be stored into a numberPacker's `data` +// field. +// +//go:nosplit +func needsPackerStorage(n *Number) uint64 { if n.Float == 0.0 { v := n.Int if v >= 0 && v <= int64(valueField) { - // We can store the integer value directly. - return packedNumber(typeFieldInteger | storageFieldDirect | uint32(v)), nil + return 0 } - // Need to allocate a new entry in packedNumber. - newIndex := uint32(len(p.data)) - if newIndex > valueField { - return 0, errOutOfPackerMemory + return 1 + } + // n is a float. + v := n.Float + if math.IsNaN(v) || v == math.Inf(-1) || v == math.Inf(1) { + return 0 + } + if v >= 0.0 && float64(float32(v)) == v { + float32Bits := math.Float32bits(float32(v)) + exponent := (float32Bits&float32ExponentField)>>float32ExponentShift - float32ExponentBias + packedExponent := (exponent + packedFloatExponentBias) << float32ExponentShift + if packedExponent&packedFloatExponentField == packedExponent { + return 0 + } + } + return 1 +} + +// isIndirect returns 1 iff this packedNumber needs storage in a numberPacker. +// +//go:nosplit +func (n packedNumber) isIndirect() uint64 { + if uint32(n)&storageField == storageFieldIndirect { + return 1 + } + return 0 +} + +// errOutOfPackerMemory is emitted when the number cannot be packed into a numberPacker. +var errOutOfPackerMemory = errors.New("out of numberPacker memory") + +// pack packs a Number into a packedNumber. +// +//go:nosplit +func (p *numberPacker) pack(n *Number) packedNumber { + if n.Float == 0.0 { + v := n.Int + if v >= 0 && v <= int64(maxDirectUint) { + // We can store the integer value directly. + return packedNumber(typeFieldInteger | storageFieldDirect | uint32(v)) + } + if len(p.data) == cap(p.data) { + panic(errOutOfPackerMemory) } p.data = append(p.data, uint64(v)) - return packedNumber(typeFieldInteger | storageFieldIndirect | newIndex), nil + return packedNumber(typeFieldInteger | storageFieldIndirect | uint32(len(p.data)-1)) } // n is a float. v := n.Float if math.IsNaN(v) { - return packedFloatNaN, nil + return packedFloatNaN } if v == math.Inf(-1) { - return packedFloatNegInf, nil + return packedFloatNegInf } if v == math.Inf(1) { - return packedFloatInf, nil + return packedFloatInf } if v >= 0.0 && float64(float32(v)) == v { float32Bits := math.Float32bits(float32(v)) @@ -234,34 +280,30 @@ func (p *numberPacker) pack(n *Number) (packedNumber, error) { packedExponent := (exponent + packedFloatExponentBias) << float32ExponentShift if packedExponent&packedFloatExponentField == packedExponent { float32Fraction := float32Bits & float32FractionField - return packedNumber(typeFieldFloat | storageFieldDirect | packedExponent | float32Fraction), nil + return packedNumber(typeFieldFloat | storageFieldDirect | packedExponent | float32Fraction) } } - // Need to allocate a new entry in packedNumber. - newIndex := uint32(len(p.data)) - if newIndex > valueField { - return 0, errOutOfPackerMemory + if len(p.data) == cap(p.data) { + panic(errOutOfPackerMemory) } p.data = append(p.data, math.Float64bits(v)) - return packedNumber(typeFieldFloat | storageFieldIndirect | newIndex), nil + return packedNumber(typeFieldFloat | storageFieldIndirect | uint32(len(p.data)-1)) } -func (p *numberPacker) mustPack(n *Number) packedNumber { - packed, err := p.pack(n) - if err != nil { - panic(err) - } - return packed -} - -func (p *numberPacker) mustPackInt(val int64) packedNumber { +// packInt packs an integer. +// +//go:nosplit +func (p *numberPacker) packInt(val int64) packedNumber { n := Number{Int: val} - return p.mustPack(&n) + return p.pack(&n) } -func (p *numberPacker) mustPackFloat(val float64) packedNumber { +// packFloat packs a floating-point number. +// +//go:nosplit +func (p *numberPacker) packFloat(val float64) packedNumber { n := Number{Float: val} - return p.mustPack(&n) + return p.pack(&n) } // unpack unpacks a packedNumber back into a Number. @@ -296,6 +338,8 @@ func (p *numberPacker) unpack(n packedNumber) *Number { panic("unreachable") } +// mustUnpackInt unpacks an integer. +// It panics if the packedNumber is not an integer. func (p *numberPacker) mustUnpackInt(n packedNumber) int64 { num := p.unpack(n) if !num.IsInteger() { @@ -304,6 +348,21 @@ func (p *numberPacker) mustUnpackInt(n packedNumber) int64 { return num.Int } +// portTo ports over a packedNumber from this numberPacker to a new one. +// It is equivalent to `p.pack(other.unpack(n))` but avoids +// allocations in the overwhelmingly-common case where the number is direct. +func (p *numberPacker) portTo(other *numberPacker, n packedNumber) packedNumber { + if uint32(n)&storageField == storageFieldDirect { + // `n` is self-contained, just return as-is. + return n + } + if len(other.data) == cap(other.data) { + panic(errOutOfPackerMemory) + } + other.data = append(other.data, p.data[uint32(n)&valueField]) + return packedNumber(uint32(n)&(typeField|storageField) | uint32(len(other.data)-1)) +} + // verifiableMetric verifies a single metric within a Verifier. type verifiableMetric struct { metadata *pb.MetricMetadata @@ -311,7 +370,7 @@ type verifiableMetric struct { numFields uint32 verifier *Verifier allowedFieldValues map[string]map[string]struct{} - wantBucketUpperBounds []packedNumber + wantBucketUpperBounds []Number // The following fields are used to verify that values are actually increasing monotonically. // They are only read and modified when the parent Verifier.mu is held. @@ -395,11 +454,11 @@ func newVerifiableMetric(metadata *pb.MetricMetadata, verifier *Verifier) (*veri if numBuckets <= 1 || numBuckets > 256 { return nil, fmt.Errorf("unsupported number of buckets: %d", numBuckets) } - v.wantBucketUpperBounds = make([]packedNumber, numBuckets) + v.wantBucketUpperBounds = make([]Number, numBuckets) for i, boundary := range metadata.GetDistributionBucketLowerBounds() { - v.wantBucketUpperBounds[i] = verifier.permanentPacker.mustPackInt(boundary) + v.wantBucketUpperBounds[i] = Number{Int: boundary} } - v.wantBucketUpperBounds[numBuckets-1] = verifier.permanentPacker.mustPackFloat(math.Inf(1)) + v.wantBucketUpperBounds[numBuckets-1] = Number{Float: math.Inf(1)} v.lastBucketSamples = make(map[string][]packedNumber, numFieldCombinations) default: return nil, fmt.Errorf("invalid type: %v", metadata.GetType()) @@ -416,6 +475,7 @@ func (v *verifiableMetric) numFieldCombinations() int { // field values that have already been seen. `verify` should populate this. // `dataToFieldsSeen` is passed across calls to `verify` and other methods of `verifiableMetric`. // It is used to store the canonical representation of the field values seen for each *Data. +// // Precondition: `Verifier.mu` is held. func (v *verifiableMetric) verify(data *Data, metricFieldsSeen map[string]struct{}, dataToFieldsSeen map[*Data]string) error { if *data.Metric != v.wantMetric { @@ -477,7 +537,7 @@ func (v *verifiableMetric) verify(data *Data, metricFieldsSeen map[string]struct return fmt.Errorf("invalid number of buckets: got %d want %d", len(data.HistogramValue.Buckets), len(v.wantBucketUpperBounds)) } for i, b := range data.HistogramValue.Buckets { - if want := v.wantBucketUpperBounds[i]; b.UpperBound != *v.verifier.permanentPacker.unpack(want) { + if want := v.wantBucketUpperBounds[i]; b.UpperBound != want { return fmt.Errorf("invalid upper bound for bucket %d (0-based): got %v want %v", i, b.UpperBound, want) } } @@ -493,6 +553,7 @@ func (v *verifiableMetric) verify(data *Data, metricFieldsSeen map[string]struct } // verifyIncrement verifies that incremental metrics are monotonically increasing. +// // Preconditions: `verify` has succeeded on the given `data`, and `Verifier.mu` is held. func (v *verifiableMetric) verifyIncrement(data *Data, fieldValues string, packer *numberPacker) error { switch v.wantMetric.Type { @@ -519,16 +580,90 @@ func (v *verifiableMetric) verifyIncrement(data *Data, fieldValues string, packe return nil } +// packerCapacityNeeded returns the `numberPacker` capacity to store `Data`. +func (v *verifiableMetric) packerCapacityNeededForData(data *Data, fieldValues string) uint64 { + switch v.wantMetric.Type { + case TypeCounter: + return needsPackerStorage(data.Number) + case TypeHistogram: + var toPack uint64 + var buf Number + for _, b := range data.HistogramValue.Buckets { + buf = Number{Int: int64(b.Samples)} + toPack += needsPackerStorage(&buf) + } + return toPack + default: + return 0 + } +} + +// packerCapacityNeededForLast returns the `numberPacker` capacity needed to +// store the last snapshot's data that was not seen in the current snapshot +// (aka not in metricFieldsSeen). +func (v *verifiableMetric) packerCapacityNeededForLast(metricFieldsSeen map[string]struct{}) uint64 { + var capacity uint64 + switch v.wantMetric.Type { + case TypeCounter: + for fieldValues, lastCounterValue := range v.lastCounterValue { + if _, found := metricFieldsSeen[fieldValues]; found { + continue + } + capacity += lastCounterValue.isIndirect() + } + case TypeHistogram: + for fieldValues, bucketSamples := range v.lastBucketSamples { + if _, found := metricFieldsSeen[fieldValues]; found { + continue + } + for _, b := range bucketSamples { + capacity += b.isIndirect() + } + } + } + return capacity +} + // update updates incremental metrics' "last seen" data. -// Preconditions: `verifyIncrement` has succeeded on the given `data`, and `Verifier.mu` is held. +// +// Preconditions: `verifyIncrement` has succeeded on the given `data`, `Verifier.mu` is held, +// and `packer` is guaranteed to have enough room to store all numbers. func (v *verifiableMetric) update(data *Data, fieldValues string, packer *numberPacker) { switch v.wantMetric.Type { case TypeCounter: - v.lastCounterValue[v.verifier.internMap.Intern(fieldValues)] = packer.mustPack(data.Number) + v.lastCounterValue[v.verifier.internMap.Intern(fieldValues)] = packer.pack(data.Number) case TypeHistogram: lastBucketSamples := v.lastBucketSamples[v.verifier.internMap.Intern(fieldValues)] for i, b := range data.HistogramValue.Buckets { - lastBucketSamples[i] = packer.mustPackInt(int64(b.Samples)) + lastBucketSamples[i] = packer.packInt(int64(b.Samples)) + } + } +} + +// repackUnseen packs all numbers that must be carried over from snapshot to snapshot and which were +// not seen in the latest snapshot's data. +// This function should carry over all numbers typically packed in `v.update` but for all metric +// field combinations that are not in `metricFieldsSeen`. +// +// Preconditions: `verifyIncrement` has succeeded on the given `data`, +// and `newPacker` is guaranteed to have enough room to store all numbers. +func (v *verifiableMetric) repackUnseen(metricFieldsSeen map[string]struct{}, oldPacker, newPacker *numberPacker) { + switch v.wantMetric.Type { + case TypeCounter: + for fieldValues, lastCounterValue := range v.lastCounterValue { + if _, found := metricFieldsSeen[fieldValues]; found { + continue + } + v.lastCounterValue[fieldValues] = oldPacker.portTo(newPacker, lastCounterValue) + } + case TypeHistogram: + for fieldValues, bucketSamples := range v.lastBucketSamples { + if _, found := metricFieldsSeen[fieldValues]; found { + continue + } + for i, b := range bucketSamples { + bucketSamples[i] = oldPacker.portTo(newPacker, b) + } } } } @@ -540,9 +675,6 @@ func (v *verifiableMetric) update(data *Data, fieldValues string, packer *number type Verifier struct { knownMetrics map[string]*verifiableMetric - // permanentPacker is used to pack numbers seen at Verifier initialization time only. - permanentPacker *numberPacker - // mu protects the fields below. mu sync.Mutex @@ -564,9 +696,8 @@ type Verifier struct { func NewVerifier(registration *pb.MetricRegistration) (*Verifier, func(), error) { globalInternVerifierCreated() verifier := &Verifier{ - knownMetrics: make(map[string]*verifiableMetric), - permanentPacker: &numberPacker{}, - internMap: make(internedStringMap), + knownMetrics: make(map[string]*verifiableMetric), + internMap: make(internedStringMap), } for _, metric := range registration.GetMetrics() { metricName := metric.GetPrometheusName() @@ -625,18 +756,43 @@ func (v *Verifier) Verify(snapshot *Snapshot) error { if v.lastTimestamp.After(snapshot.When) { return fmt.Errorf("consecutive snapshots are not chronologically ordered: last verified snapshot was exported at %v, this one is from %v", v.lastTimestamp, snapshot.When) } + for _, data := range snapshot.Data { - if err = v.knownMetrics[data.Metric.Name].verifyIncrement(data, dataToFieldsSeen[data], v.lastPacker); err != nil { + if err := v.knownMetrics[data.Metric.Name].verifyIncrement(data, dataToFieldsSeen[data], v.lastPacker); err != nil { return fmt.Errorf("metric %q: %v", data.Metric.Name, err) } } + var neededPackerCapacity uint64 + for _, data := range snapshot.Data { + neededPackerCapacity += v.knownMetrics[data.Metric.Name].packerCapacityNeededForData(data, dataToFieldsSeen[data]) + } + for name, metric := range v.knownMetrics { + neededPackerCapacity += metric.packerCapacityNeededForLast(fieldsSeen[name]) + } + if neededPackerCapacity > uint64(valueField) { + return fmt.Errorf("snapshot contains too many large numbers to fit into packer memory (%d numbers needing indirection)", neededPackerCapacity) + } // All checks succeeded, update last-seen data. + // We need to be guaranteed to not fail past this point in the function. newPacker := &numberPacker{} + if neededPackerCapacity != 0 { + newPacker.data = make([]uint64, 0, neededPackerCapacity) + } v.lastTimestamp = snapshot.When for _, data := range snapshot.Data { v.knownMetrics[globalIntern(data.Metric.Name)].update(data, v.internMap.Intern(dataToFieldsSeen[data]), newPacker) } + if uint64(len(newPacker.data)) != neededPackerCapacity { + for name, metric := range v.knownMetrics { + metric.repackUnseen(fieldsSeen[name], v.lastPacker, newPacker) + } + } + if uint64(len(newPacker.data)) != neededPackerCapacity { + // We panic here because this represents an internal logic error, + // not something the user did wrong. + panic(fmt.Sprintf("did not pack the expected number of numbers in numberPacker: packed %d, expected %d; this indicates a logic error in verifyIncrement", len(newPacker.data), neededPackerCapacity)) + } v.lastPacker = newPacker return nil }