integrity verifier: support sampling subset like 1/4 of all files on a single scan

closes #266

Author: joonas-fi

pkg/stoserver/commandhandlers.go

@@ -412,11 +412,26 @@ func (c *cHandlers) VolumeMigrateData(cmd *stoservertypes.VolumeMigrateData, ctx
 func (c *cHandlers) VolumeVerifyIntegrity(cmd *stoservertypes.VolumeVerifyIntegrity, ctx *command.Ctx) error {
 	jobID := stoutils.NewIntegrityVerificationJobID()
 
+	sampleSpecification, err := func() (*string, error) {
+		if samplingSpec := cmd.Sampling; samplingSpec != "" {
+			// validate the sampling spec
+			_, err := stointegrityverifier.CreateSampler(&samplingSpec)
+
+			return &samplingSpec, err
+		} else {
+			return nil, nil
+		}
+	}()
+	if err != nil {
+		return err
+	}
+
 	if err := c.db.Update(func(tx *bbolt.Tx) error {
 		job := &stotypes.IntegrityVerificationJob{
-			ID:       jobID,
-			Started:  ctx.Meta.Timestamp,
-			VolumeID: cmd.Id,
+			ID:                  jobID,
+			Started:             ctx.Meta.Timestamp,
+			VolumeID:            cmd.Id,
+			SampleSpecification: sampleSpecification,
 		}
 
 		return stodb.IntegrityVerificationJobRepository.Update(job, tx)

pkg/stoserver/stointegrityverifier/batchsampler.go

@@ -0,0 +1,60 @@
+package stointegrityverifier
+
+// With sampling we can select only a subset of the blobs to visit.
+// https://en.wikipedia.org/wiki/Sampling_(statistics)
+
+import (
+	"encoding/binary"
+	"fmt"
+	"strconv"
+
+	"github.com/function61/varasto/pkg/stotypes"
+)
+
+// answers whether we should visit a blob
+type batchSampler func(stotypes.BlobRef) bool
+
+func CreateSampler(sampleSpecificationMaybe *string) (batchSampler, error) {
+	if sampleSpecification := sampleSpecificationMaybe; sampleSpecification != nil {
+		// bit string like `1111` to number (`15`)
+		num, err := strconv.ParseUint(*sampleSpecification, 2, 32)
+		if err != nil {
+			return nil, fmt.Errorf("invalid sampling spec. expected binary string like 01; got '%s'", *sampleSpecification)
+		}
+		bitCount := len(*sampleSpecification)
+
+		return prefixSampler(uint32(num), uint8(bitCount)), nil
+	} else {
+		return func(_ stotypes.BlobRef) bool { return true }, nil
+	}
+}
+
+// only accepts blob refs that start with a specific bit pattern. that means that we can accept blob refs starting with:
+// - `0b0` => accepts 1/2 of refs
+// - `0b00` => accepts 1/4 of refs
+// - `0b000` => accepts 1/8 of refs
+// - and so on...
+//
+// the interesting property of this, as opposed to something like using random sampling for acceptance is that
+// this is deterministic based on blob ref and thus we can integrity verify first batch today and next back next week and
+// we are guaranteed that the next batch won't re-visit blobs from first batch.
+//
+// let's take the 1/4 acceptance as example. we have four batches: 1) `0b00` 2) `0b01` 3) `0b10` 4) `0b11`
+//
+// we could visit those four batches in four different weeks to guaranteed visit all blobs (except those added later to earlier batches' "partitions")
+func prefixSampler(value uint32, bitCount uint8) batchSampler {
+	return func(blobRef stotypes.BlobRef) bool {
+		// counter-intuitively: use big-endian encoding to *not* filter on the very first bits of the blob ref, because
+		// the verifier is iterating the blobs ordered on the blob ref. we don't want us to consecutively ignore a large
+		// portion of the scan but instead we want to accept blobs uniformly over time
+		blobRefUint32 := binary.BigEndian.Uint32(blobRef)
+		return blobRefUint32&bitmask(bitCount) == value
+	}
+}
+
+// `1` => `0b1`
+// `3` => `0b111`
+// `8` => `0b11111111`
+func bitmask(n uint8) uint32 {
+	return (1 << n) - 1
+}

pkg/stoserver/stointegrityverifier/batchsampler_test.go

@@ -0,0 +1,34 @@
+package stointegrityverifier
+
+import (
+	"testing"
+
+	"github.com/function61/gokit/assert"
+	"github.com/function61/varasto/pkg/stotypes"
+)
+
+func TestNoSampler(t *testing.T) {
+	sample, err := CreateSampler(nil)
+	assert.Ok(t, err)
+
+	assert.Assert(t, sample(stotypes.BlobRef{0x00, 0x00, 0x00, 0x00}) == true)
+	assert.Assert(t, sample(stotypes.BlobRef{0xFF, 0xFF, 0xFF, 0xFF}) == true)
+}
+
+func TestSampler(t *testing.T) {
+	sampleSpec := "11"
+	sample, err := CreateSampler(&sampleSpec)
+	assert.Ok(t, err)
+
+	assert.Assert(t, sample(stotypes.BlobRef{0x00, 0x00, 0x00, 0x00}) == false)
+	assert.Assert(t, sample(stotypes.BlobRef{0x00, 0x00, 0x00, 0b01}) == false)
+	assert.Assert(t, sample(stotypes.BlobRef{0x00, 0x00, 0x00, 0b10}) == false)
+	assert.Assert(t, sample(stotypes.BlobRef{0x00, 0x00, 0x00, 0b11}) == true)
+	assert.Assert(t, sample(stotypes.BlobRef{0xFF, 0xFF, 0xFF, 0xFF}) == true)
+}
+
+func TestInvalidSpec(t *testing.T) {
+	sampleSpec := "wrong"
+	_, err := CreateSampler(&sampleSpec)
+	assert.EqualString(t, err.Error(), "invalid sampling spec. expected binary string like 01; got 'wrong'")
+}

pkg/stoserver/stointegrityverifier/controller.go

@@ -197,6 +197,12 @@ func (c *Controller) resumeJobWorker(
 		return nil
 	}
 
+	// depending on the sampling function we may during this run visit 100 % of blobs, 25 % of blobs or so on
+	sample, err := CreateSampler(job.SampleSpecification)
+	if err != nil {
+		return err
+	}
+
 	batchLimit := 1000
 
 	for {
@@ -232,11 +238,18 @@ func (c *Controller) resumeJobWorker(
 				}
 			}
 
+			sizeOnDisk := uint64(blob.SizeOnDisk)
+
 			blobExistsOnVolumeToVerify := slices.Contains(blob.Volumes, job.VolumeID)
 			if !blobExistsOnVolumeToVerify {
 				continue
 			}
 
+			if !sample(blob.Ref) {
+				job.BytesSkipped += sizeOnDisk
+				continue
+			}
+
 			bytesScanned, err := c.diskAccess.Scrub(blob.Ref, job.VolumeID)
 			if err != nil {
 				descr := fmt.Sprintf("blob %s: %v\n", blob.Ref.AsHex(), err)
@@ -251,7 +264,7 @@ func (c *Controller) resumeJobWorker(
 				}
 			}
 
-			job.BytesScanned += uint64(blob.SizeOnDisk)
+			job.BytesScanned += sizeOnDisk
 		}
 
 		if len(blobBatch) < batchLimit { // fewer blobs than requested, so there will be no more

pkg/stoserver/stoservertypes/commands.json

@@ -23,7 +23,8 @@
 		"title": "Verify data integrity",
 		"fields": [
 			{ "key": "Id", "type": "integer", "hideIfDefaultValue": true },
-			{ "key": "Start", "title": "Start immediately", "type": "checkbox" }
+			{ "key": "Start", "title": "Start immediately", "type": "checkbox" },
+			{ "key": "Sampling", "title": "Instead of verifying all content, verify a subset by sampling", "placeholder": "00", "optional": true }
 		]
 	},
 	{

pkg/stotypes/dbtypes.go

@@ -145,9 +145,11 @@ type IntegrityVerificationJob struct {
 	ID                   string
 	Started              time.Time
 	Completed            time.Time
-	VolumeID             int `msgpack:"VolumeId"`
+	SampleSpecification  *string // sample only a subset of items. half: `0`. if you later want to process the second half: `1`. if you want to process quarter: `00`. then the remaining quarters would be (`01`, `10` and `11`). 1 % (can be approximated as 1/128) would be `0000000`.
+	VolumeID             int     `msgpack:"VolumeId"`
 	LastCompletedBlobRef BlobRef
 	BytesScanned         uint64
+	BytesSkipped         uint64 // if sampling is in place, we'll be skipping some blobs
 	ErrorsFound          int
 	Report               string
 }