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 }