Add blocks storage consistency check in the querier (#2593)

* Added blocks storage consistency check in the querier

Signed-off-by: Marco Pracucci <[email protected]>

* Log each time a consistency check fails

Signed-off-by: Marco Pracucci <[email protected]>

* Log the trace ID too when logging consistency check failure

Signed-off-by: Marco Pracucci <[email protected]>

* Update pkg/querier/blocks_consistency_checker.go

Signed-off-by: Marco Pracucci <[email protected]>

Co-authored-by: Peter Štibraný <[email protected]>

* Update pkg/querier/blocks_consistency_checker.go

Signed-off-by: Marco Pracucci <[email protected]>

Co-authored-by: Peter Štibraný <[email protected]>

* Update pkg/querier/blocks_consistency_checker.go

Signed-off-by: Marco Pracucci <[email protected]>

Co-authored-by: Peter Štibraný <[email protected]>

* Update pkg/querier/blocks_consistency_checker.go

Signed-off-by: Marco Pracucci <[email protected]>

Co-authored-by: Peter Štibraný <[email protected]>

* Merged TimeMillisecond() and TimeToMillis() utility functions together

Signed-off-by: Marco Pracucci <[email protected]>

* Disabled consistency check by default

Signed-off-by: Marco Pracucci <[email protected]>

Co-authored-by: Peter Štibraný <[email protected]>

Author: pracucci

development/tsdb-blocks-storage-s3/config/prometheus.yaml

@@ -39,16 +39,16 @@ scrape_configs:
       - targets: ['query-frontend:8007']
         labels:
           container: 'query-frontend'
-- job_name: cortex-store-gateway-1
-  static_configs:
-    - targets: ['store-gateway-1:8008']
-      labels:
-        container: 'store-gateway-1'
-- job_name: cortex-store-gateway-2
-  static_configs:
-    - targets: ['store-gateway-2:8009']
-      labels:
-        container: 'store-gateway-2'
+  - job_name: cortex-store-gateway-1
+    static_configs:
+      - targets: ['store-gateway-1:8008']
+        labels:
+          container: 'store-gateway-1'
+  - job_name: cortex-store-gateway-2
+    static_configs:
+      - targets: ['store-gateway-2:8009']
+        labels:
+          container: 'store-gateway-2'
 
 remote_write:
   - url: http://distributor:8001/api/prom/push

docs/configuration/config-file-reference.md

@@ -664,6 +664,19 @@ store_gateway_client:
   # TLS CA path for the client
   # CLI flag: -experimental.querier.store-gateway-client.tls-ca-path
   [tls_ca_path: <string> | default = ""]
+
+# Configures the consistency check done by the querier on queried blocks when
+# running the experimental blocks storage.
+blocks_consistency_check:
+  # Whether the querier should run a consistency check to ensure all expected
+  # blocks have been queried.
+  # CLI flag: -experimental.querier.blocks-consistency-check.enabled
+  [enabled: <boolean> | default = false]
+
+  # The grace period allowed before a new block is included in the consistency
+  # check.
+  # CLI flag: -experimental.querier.blocks-consistency-check.upload-grace-period
+  [upload_grace_period: <duration> | default = 1h]
 ```
 
 ### `query_frontend_config`

pkg/querier/blocks_consistency_checker.go

@@ -0,0 +1,90 @@
+package querier
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/oklog/ulid"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+	"github.com/thanos-io/thanos/pkg/store/hintspb"
+)
+
+type BlocksConsistencyChecker struct {
+	uploadGracePeriod   time.Duration
+	deletionGracePeriod time.Duration
+
+	checksTotal  prometheus.Counter
+	checksFailed prometheus.Counter
+}
+
+func NewBlocksConsistencyChecker(uploadGracePeriod, deletionGracePeriod time.Duration, reg prometheus.Registerer) *BlocksConsistencyChecker {
+	return &BlocksConsistencyChecker{
+		uploadGracePeriod:   uploadGracePeriod,
+		deletionGracePeriod: deletionGracePeriod,
+		checksTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "cortex_querier_blocks_consistency_checks_total",
+			Help: "Total number of consistency checks run on queried blocks.",
+		}),
+		checksFailed: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "cortex_querier_blocks_consistency_checks_failed_total",
+			Help: "Total number of consistency checks failed on queried blocks.",
+		}),
+	}
+}
+
+func (c *BlocksConsistencyChecker) Check(expectedBlocks []ulid.ULID, knownDeletionMarks map[ulid.ULID]*metadata.DeletionMark, queriedBlocks map[string][]hintspb.Block) error {
+	c.checksTotal.Inc()
+
+	// Reverse the map of queried blocks, so that we can easily look for missing ones
+	// while keeping the information about which store-gateways have already been queried
+	// for that block.
+	actualBlocks := map[string][]string{}
+	for gatewayAddr, blocks := range queriedBlocks {
+		for _, b := range blocks {
+			actualBlocks[b.Id] = append(actualBlocks[b.Id], gatewayAddr)
+		}
+	}
+
+	// Look for any missing block.
+	missingBlocks := map[string][]string{}
+	var missingBlockIDs []string
+
+	for _, blockID := range expectedBlocks {
+		// Some recently uploaded blocks, already discovered by the querier, may not have been discovered
+		// and loaded by the store-gateway yet. In order to avoid false positives, we grant some time
+		// to the store-gateway to discover them. It's safe to exclude recently uploaded blocks because:
+		// - Blocks uploaded by ingesters: we will continue querying them from ingesters for a while (depends
+		//   on the configured retention period).
+		// - Blocks uploaded by compactor: the source blocks are marked for deletion but will continue to be
+		//   queried by store-gateways for a while (depends on the configured deletion marks delay).
+		if ulid.Now()-blockID.Time() < uint64(c.uploadGracePeriod.Milliseconds()) {
+			continue
+		}
+
+		// The store-gateway may offload blocks before the querier. If that happens, the querier will run a consistency check
+		// on blocks that can't be queried because they were offloaded. For this reason, we don't run the consistency check on any block
+		// which has been marked for deletion more then "grace period" time ago. Basically, the grace period is the time
+		// we still expect a block marked for deletion to be still queried.
+		if mark := knownDeletionMarks[blockID]; mark != nil {
+			if time.Since(time.Unix(mark.DeletionTime, 0)) > c.deletionGracePeriod {
+				continue
+			}
+		}
+
+		id := blockID.String()
+		if gatewayAddrs, ok := actualBlocks[id]; !ok {
+			missingBlocks[id] = gatewayAddrs
+			missingBlockIDs = append(missingBlockIDs, id)
+		}
+	}
+
+	if len(missingBlocks) == 0 {
+		return nil
+	}
+
+	c.checksFailed.Inc()
+	return fmt.Errorf("consistency check failed because some blocks were not queried: %s", strings.Join(missingBlockIDs, " "))
+}

pkg/querier/blocks_consistency_checker_test.go

@@ -0,0 +1,118 @@
+package querier
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+	"github.com/thanos-io/thanos/pkg/store/hintspb"
+
+	"github.com/cortexproject/cortex/pkg/util"
+)
+
+func TestBlocksConsistencyChecker_Check(t *testing.T) {
+	now := time.Now()
+	uploadGracePeriod := 10 * time.Minute
+	deletionGracePeriod := 5 * time.Minute
+
+	block1 := ulid.MustNew(uint64(util.TimeToMillis(now.Add(-uploadGracePeriod*2))), nil)
+	block2 := ulid.MustNew(uint64(util.TimeToMillis(now.Add(-uploadGracePeriod*3))), nil)
+	block3 := ulid.MustNew(uint64(util.TimeToMillis(now.Add(-uploadGracePeriod*4))), nil)
+	blockRecentlyUploaded := ulid.MustNew(uint64(util.TimeToMillis(now)), nil)
+
+	tests := map[string]struct {
+		expectedBlocks     []ulid.ULID
+		knownDeletionMarks map[ulid.ULID]*metadata.DeletionMark
+		queriedBlocks      map[string][]hintspb.Block
+		expectedErr        error
+	}{
+		"no expected blocks": {
+			expectedBlocks:     []ulid.ULID{},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{},
+			queriedBlocks:      map[string][]hintspb.Block{},
+		},
+		"all expected blocks have been queried from a single store-gateway": {
+			expectedBlocks:     []ulid.ULID{block1, block2},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}, {Id: block2.String()}},
+			},
+		},
+		"all expected blocks have been queried from multiple store-gateway": {
+			expectedBlocks:     []ulid.ULID{block1, block2},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}},
+				"2.2.2.2": {{Id: block2.String()}},
+			},
+		},
+		"store-gateway has queried more blocks than expected": {
+			expectedBlocks:     []ulid.ULID{block1, block2},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}},
+				"2.2.2.2": {{Id: block2.String()}, {Id: block3.String()}},
+			},
+		},
+		"store-gateway has queried less blocks than expected": {
+			expectedBlocks:     []ulid.ULID{block1, block2, block3},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}},
+				"2.2.2.2": {{Id: block3.String()}},
+			},
+			expectedErr: fmt.Errorf("consistency check failed because some blocks were not queried: %s", block2.String()),
+		},
+		"store-gateway has queried less blocks than expected, but the missing block has been recently uploaded": {
+			expectedBlocks:     []ulid.ULID{block1, block2, blockRecentlyUploaded},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}},
+				"2.2.2.2": {{Id: block2.String()}},
+			},
+		},
+		"store-gateway has queried less blocks than expected and the missing block has been recently marked for deletion": {
+			expectedBlocks: []ulid.ULID{block1, block2, block3},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{
+				block3: {DeletionTime: now.Add(-deletionGracePeriod / 2).Unix()},
+			},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}},
+				"2.2.2.2": {{Id: block2.String()}},
+			},
+			expectedErr: fmt.Errorf("consistency check failed because some blocks were not queried: %s", block3.String()),
+		},
+		"store-gateway has queried less blocks than expected and the missing block has been marked for deletion long time ago": {
+			expectedBlocks: []ulid.ULID{block1, block2, block3},
+			knownDeletionMarks: map[ulid.ULID]*metadata.DeletionMark{
+				block3: {DeletionTime: now.Add(-deletionGracePeriod * 2).Unix()},
+			},
+			queriedBlocks: map[string][]hintspb.Block{
+				"1.1.1.1": {{Id: block1.String()}},
+				"2.2.2.2": {{Id: block2.String()}},
+			},
+		},
+	}
+
+	for testName, testData := range tests {
+		t.Run(testName, func(t *testing.T) {
+			reg := prometheus.NewPedanticRegistry()
+			c := NewBlocksConsistencyChecker(uploadGracePeriod, deletionGracePeriod, reg)
+
+			err := c.Check(testData.expectedBlocks, testData.knownDeletionMarks, testData.queriedBlocks)
+			assert.Equal(t, testData.expectedErr, err)
+			assert.Equal(t, float64(1), testutil.ToFloat64(c.checksTotal))
+
+			if testData.expectedErr != nil {
+				assert.Equal(t, float64(1), testutil.ToFloat64(c.checksFailed))
+			} else {
+				assert.Equal(t, float64(0), testutil.ToFloat64(c.checksFailed))
+			}
+		})
+	}
+}