Repository: cortexproject/cortex (go)

Issue:
Race condition in the compactor can leave blocks with just the deletion mark file
There's a race condition in the Cortex compactor due to the fact that the blocks cleaner and the blocks compactor have been fully decoupled in Cortex (contrary to Thanos). This change has been introduced in the PR #2623.

The race condition occurs when the blocks cleaner and the compactor's garbage collection start nearly at the same time. We've observed happening it few times, always with the same pattern (see below).

Blocks cleaner:
```
ts=2020-06-18T02:19:30.566213195Z caller=blocks_cleaner.go:99 component=cleaner msg="started hard deletion of blocks marked for deletion"
ts=2020-06-18T02:19:37.877100828Z caller=blocks_cleaner.go:53 component=cleaner org_id=10428 msg="deleted block marked for deletion" block=01EB15VDNGAZEYSG9JESYX8MH9
```

Compactor:
```
ts=2020-06-18T02:19:36.718176479Z caller=compact.go:937 component=compactor org_id=10428 msg="start sync of metas"
ts=2020-06-18T02:19:41.993683461Z caller=compact.go:942 component=compactor org_id=10428 msg="start of GC"
ts=2020-06-18T02:19:41.993527445Z caller=fetcher.go:452 component=compactor org_id=10428 component=block.BaseFetcher msg="successfully synchronized block metadata" duration=5.275308446s cached=301 returned=138 partial=113
ts=2020-06-18T02:19:42.129782115Z caller=block.go:149 component=compactor org_id=10428 msg="block has been marked for deletion" block=01EB15VDNGAZEYSG9JESYX8MH9
ts=2020-06-18T02:19:42.097451326Z caller=compact.go:182 component=compactor org_id=10428 msg="marking outdated block for deletion" block=01EB15VDNGAZEYSG9JESYX8MH9
```

What happened:
- Blocks cleaner and compactor started garbage collection nearly at the same time
- Blocks cleaner and compactor scanned the bucket nearly at the same time and found out the block `01EB15VDNGAZEYSG9JESYX8MH9`
- Blocks cleaner deleted block `01EB15VDNGAZEYSG9JESYX8MH9`
- Compactor syncer's metadata fetcher called the ignore deletion mark filter for block `01EB15VDNGAZEYSG9JESYX8MH9` and it returned "no deletion mark" because the block has been deleted in the meanwhile
- Compactor syncer's GarbageCollect() created the deletion mark but the block was already deleted so we ended up with a block just containing the deletion mark, which is considered partial and never cleaned up


Pull request description:
Cleanup partial blocks with deletion mark expired
**What this PR does**:
I've investigated #2748 and fixing it may be non trivial without a coordination between compactor and blocks cleaner. Another option to fix it would be disabling the compactor garbage collection at all, but would require a discussion in Thanos first and deeply understand the impact of it.

Since Thanos and the Cortex blocks storage can deal with partial blocks (it's an expected state because a block is partial also each time it's upload is in progress from the ingester) I propose to keep it simple and just add a cleanup of partial blocks marked for deletion in the blocks cleaner.

**Which issue(s) this PR fixes**:
Fixes #2748

**Checklist**
- [x] Tests updated
- [ ] Documentation added
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6f159896e9..c108e395ea 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -118,6 +118,7 @@
   - `cortex_purger_load_pending_requests_attempts_total`: Number of attempts that were made to load pending requests with status.
   - `cortex_purger_oldest_pending_delete_request_age_seconds`: Age of oldest pending delete request in seconds.
   - `cortex_purger_pending_delete_requests_count`: Count of requests which are in process or are ready to be processed.
+* [ENHANCEMENT] Experimental TSDB: Improved compactor to hard-delete also partial blocks with an deletion mark (even if the deletion mark threshold has not been reached). #2751
 * [BUGFIX] Ruler: Ensure temporary rule files with special characters are properly mapped and cleaned up. #2506
 * [BUGFIX] Fixes #2411, Ensure requests are properly routed to the prometheus api embedded in the query if `-server.path-prefix` is set. #2372
 * [BUGFIX] Experimental TSDB: fixed chunk data corruption when querying back series using the experimental blocks storage. #2400
diff --git a/development/tsdb-blocks-storage-s3-single-binary/config/grafana-agent.yaml b/development/tsdb-blocks-storage-s3-single-binary/config/grafana-agent.yaml
index 046c77b0b4..1061dfb44f 100644
--- a/development/tsdb-blocks-storage-s3-single-binary/config/grafana-agent.yaml
+++ b/development/tsdb-blocks-storage-s3-single-binary/config/grafana-agent.yaml
@@ -6,7 +6,7 @@ prometheus:
   global:
     scrape_interval: 5s
     external_labels:
-      origin: grafana-agent
+      scraped_by: grafana-agent
   configs:
     - name: local
       host_filter: false
diff --git a/development/tsdb-blocks-storage-s3-single-binary/config/prometheus.yaml b/development/tsdb-blocks-storage-s3-single-binary/config/prometheus.yaml
index b28541feba..c430d766bf 100644
--- a/development/tsdb-blocks-storage-s3-single-binary/config/prometheus.yaml
+++ b/development/tsdb-blocks-storage-s3-single-binary/config/prometheus.yaml
@@ -1,7 +1,7 @@
 global:
   scrape_interval: 5s
   external_labels:
-    origin: prometheus
+    scraped_by: prometheus
 
 scrape_configs:
   - job_name: cortex-1
diff --git a/development/tsdb-blocks-storage-s3/config/grafana-agent.yaml b/development/tsdb-blocks-storage-s3/config/grafana-agent.yaml
index b24a1fadb8..9b62fdf055 100644
--- a/development/tsdb-blocks-storage-s3/config/grafana-agent.yaml
+++ b/development/tsdb-blocks-storage-s3/config/grafana-agent.yaml
@@ -6,7 +6,7 @@ prometheus:
   global:
     scrape_interval: 5s
     external_labels:
-      origin: grafana-agent
+      scraped_by: grafana-agent
   configs:
     - name: local
       host_filter: false
diff --git a/development/tsdb-blocks-storage-s3/config/prometheus.yaml b/development/tsdb-blocks-storage-s3/config/prometheus.yaml
index 336106b249..ffd1c68b1f 100644
--- a/development/tsdb-blocks-storage-s3/config/prometheus.yaml
+++ b/development/tsdb-blocks-storage-s3/config/prometheus.yaml
@@ -1,7 +1,7 @@
 global:
   scrape_interval: 5s
   external_labels:
-    origin: prometheus
+    scraped_by: prometheus
 
 scrape_configs:
   - job_name: tsdb-blocks-storage-s3/distributor
diff --git a/docs/configuration/arguments.md b/docs/configuration/arguments.md
index b2ec00c7fb..2f66ce51d5 100644
--- a/docs/configuration/arguments.md
+++ b/docs/configuration/arguments.md
@@ -132,7 +132,7 @@ The ingester query API was improved over time, but defaults to the old behaviour
 
    Set this flag to `true` for the new behaviour.
 
-   Important to note is that when setting this flag to `true`, it has to be set on both the distributor and the querier. If the flag is only set on the distributor and not on the querier, you will get incomplete query results because not all ingesters are queried.
+   Important to note is that when setting this flag to `true`, it has to be set on both the distributor and the querier (called `-distributor.shard-by-all-labels` on Querier as well). If the flag is only set on the distributor and not on the querier, you will get incomplete query results because not all ingesters are queried.
 
    **Upgrade notes**: As this flag also makes all queries always read from all ingesters, the upgrade path is pretty trivial; just enable the flag. When you do enable it, you'll see a spike in the number of active series as the writes are "reshuffled" amongst the ingesters, but over the next stale period all the old series will be flushed, and you should end up with much better load balancing. With this flag enabled in the queriers, reads will always catch all the data from all ingesters.
 
diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go
index 1428176445..411f66d336 100644
--- a/pkg/compactor/blocks_cleaner.go
+++ b/pkg/compactor/blocks_cleaner.go
@@ -7,11 +7,13 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
 	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/objstore"
 
@@ -158,7 +160,8 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string) error {
 
 	// Runs a bucket scan to get a fresh list of all blocks and populate
 	// the list of deleted blocks in filter.
-	if _, _, err = fetcher.Fetch(ctx); err != nil {
+	_, partials, err := fetcher.Fetch(ctx)
+	if err != nil {
 		return errors.Wrap(err, "error fetching metadata")
 	}
 
@@ -174,5 +177,43 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string) error {
 		return errors.Wrap(err, "error cleaning blocks")
 	}
 
+	// Partial blocks with a deletion mark can be cleaned up. This is a best effort, so we don't return
+	// error if the cleanup of partial blocks fail.
+	if len(partials) > 0 {
+		level.Info(userLogger).Log("msg", "started cleaning of partial blocks marked for deletion")
+		c.cleanUserPartialBlocks(ctx, partials, userBucket, userLogger)
+		level.Info(userLogger).Log("msg", "cleaning of partial blocks marked for deletion done")
+	}
+
 	return nil
 }
+
+func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map[ulid.ULID]error, userBucket *cortex_tsdb.UserBucketClient, userLogger log.Logger) {
+	for blockID, blockErr := range partials {
+		// We can safely delete only blocks which are partial because the meta.json is missing.
+		if blockErr != block.ErrorSyncMetaNotFound {
+			continue
+		}
+
+		// We can safely delete only partial blocks with a deletion mark.
+		_, err := metadata.ReadDeletionMark(ctx, userBucket, userLogger, blockID.String())
+		if err == metadata.ErrorDeletionMarkNotFound {
+			continue
+		}
+		if err != nil {
+			level.Warn(userLogger).Log("msg", "error reading partial block deletion mark", "block", blockID, "err", err)
+			continue
+		}
+
+		// Hard-delete partial blocks having a deletion mark, even if the deletion threshold has not
+		// been reached yet.
+		if err := block.Delete(ctx, userLogger, userBucket, blockID); err != nil {
+			c.blocksFailedTotal.Inc()
+			level.Warn(userLogger).Log("msg", "error deleting partial block marked for deletion", "block", blockID, "err", err)
+			continue
+		}
+
+		c.blocksCleanedTotal.Inc()
+		level.Info(userLogger).Log("msg", "deleted partial block marked for deletion", "block", blockID)
+	}
+}
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index bbc3ea0032..c4426b1cef 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -239,7 +239,7 @@ func (c *Compactor) starting(ctx context.Context) error {
 		DataDir:             c.compactorCfg.DataDir,
 		MetaSyncConcurrency: c.compactorCfg.MetaSyncConcurrency,
 		DeletionDelay:       c.compactorCfg.DeletionDelay,
-		CleanupInterval:     c.compactorCfg.CompactionInterval,
+		CleanupInterval:     util.DurationWithJitter(c.compactorCfg.CompactionInterval, 0.05),
 	}, c.bucketClient, c.usersScanner, c.parentLogger, c.registerer)
 
 	// Ensure an initial cleanup occurred before starting the compactor.
@@ -265,7 +265,7 @@ func (c *Compactor) running(ctx context.Context) error {
 	// Run an initial compaction before starting the interval.
 	c.compactUsersWithRetries(ctx)
 
-	ticker := time.NewTicker(c.compactorCfg.CompactionInterval)
+	ticker := time.NewTicker(util.DurationWithJitter(c.compactorCfg.CompactionInterval, 0.05))
 	defer ticker.Stop()
 
 	for {


Test patch:
diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go
index 126b72be5a..cd4137bf20 100644
--- a/pkg/compactor/blocks_cleaner_test.go
+++ b/pkg/compactor/blocks_cleaner_test.go
@@ -2,6 +2,7 @@ package compactor
 
 import (
 	"context"
+	"crypto/rand"
 	"io/ioutil"
 	"os"
 	"path"
@@ -10,6 +11,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log"
+	"github.com/oklog/ulid"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -30,18 +32,25 @@ func TestBlocksCleaner(t *testing.T) {
 	require.NoError(t, err)
 	defer os.RemoveAll(dataDir) //nolint:errcheck
 
+	// Create a bucket client on the local storage.
+	bucketClient, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir})
+	require.NoError(t, err)
+
 	// Create blocks.
+	ctx := context.Background()
 	now := time.Now()
 	deletionDelay := 12 * time.Hour
 	block1 := createTSDBBlock(t, filepath.Join(storageDir, "user-1"), 10, 20, nil)
 	block2 := createTSDBBlock(t, filepath.Join(storageDir, "user-1"), 20, 30, nil)
 	block3 := createTSDBBlock(t, filepath.Join(storageDir, "user-1"), 30, 40, nil)
-	createDeletionMark(t, filepath.Join(storageDir, "user-1"), block2, now.Add(-deletionDelay).Add(time.Hour))  // Hasn't reached the deletion threshold yet.
-	createDeletionMark(t, filepath.Join(storageDir, "user-1"), block3, now.Add(-deletionDelay).Add(-time.Hour)) // Reached the deletion threshold.
-
-	// Create a bucket client on the local storage.
-	bucketClient, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir})
-	require.NoError(t, err)
+	block4 := ulid.MustNew(4, rand.Reader)
+	block5 := ulid.MustNew(5, rand.Reader)
+	block6 := createTSDBBlock(t, filepath.Join(storageDir, "user-1"), 40, 50, nil)
+	createDeletionMark(t, filepath.Join(storageDir, "user-1"), block2, now.Add(-deletionDelay).Add(time.Hour))  // Block hasn't reached the deletion threshold yet.
+	createDeletionMark(t, filepath.Join(storageDir, "user-1"), block3, now.Add(-deletionDelay).Add(-time.Hour)) // Block reached the deletion threshold.
+	createDeletionMark(t, filepath.Join(storageDir, "user-1"), block4, now.Add(-deletionDelay).Add(time.Hour))  // Partial block hasn't reached the deletion threshold yet.
+	createDeletionMark(t, filepath.Join(storageDir, "user-1"), block5, now.Add(-deletionDelay).Add(-time.Hour)) // Partial block reached the deletion threshold.
+	require.NoError(t, bucketClient.Delete(ctx, path.Join("user-1", block6.String(), metadata.MetaFilename)))   // Partial block without deletion mark.
 
 	cfg := BlocksCleanerConfig{
 		DataDir:             dataDir,
@@ -50,7 +59,6 @@ func TestBlocksCleaner(t *testing.T) {
 		CleanupInterval:     time.Minute,
 	}
 
-	ctx := context.Background()
 	logger := log.NewNopLogger()
 	scanner := NewUsersScanner(bucketClient, func(_ string) (bool, error) { return true, nil }, logger)
 
@@ -72,9 +80,24 @@ func TestBlocksCleaner(t *testing.T) {
 	require.NoError(t, err)
 	assert.False(t, exists)
 
+	// Should delete a partial block with deletion mark who hasn't reached the deletion threshold yet.
+	exists, err = bucketClient.Exists(ctx, path.Join("user-1", block4.String(), metadata.DeletionMarkFilename))
+	require.NoError(t, err)
+	assert.False(t, exists)
+
+	// Should delete a partial block with deletion mark who has reached the deletion threshold.
+	exists, err = bucketClient.Exists(ctx, path.Join("user-1", block5.String(), metadata.DeletionMarkFilename))
+	require.NoError(t, err)
+	assert.False(t, exists)
+
+	// Should not delete a partial block without deletion mark.
+	exists, err = bucketClient.Exists(ctx, path.Join("user-1", block6.String(), "index"))
+	require.NoError(t, err)
+	assert.True(t, exists)
+
 	assert.Equal(t, float64(1), testutil.ToFloat64(cleaner.runsStarted))
 	assert.Equal(t, float64(1), testutil.ToFloat64(cleaner.runsCompleted))
 	assert.Equal(t, float64(0), testutil.ToFloat64(cleaner.runsFailed))
-	assert.Equal(t, float64(1), testutil.ToFloat64(cleaner.blocksCleanedTotal))
+	assert.Equal(t, float64(3), testutil.ToFloat64(cleaner.blocksCleanedTotal))
 	assert.Equal(t, float64(0), testutil.ToFloat64(cleaner.blocksFailedTotal))
 }
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index ee85c1ea5b..ff362a18ea 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -799,9 +799,13 @@ func createTSDBBlock(t *testing.T, dir string, minT, maxT int64, externalLabels
 
 func createDeletionMark(t *testing.T, dir string, blockID ulid.ULID, deletionTime time.Time) {
 	content := mockDeletionMarkJSON(blockID.String(), deletionTime)
-	path := filepath.Join(dir, blockID.String(), metadata.DeletionMarkFilename)
+	blockPath := filepath.Join(dir, blockID.String())
+	markPath := filepath.Join(blockPath, metadata.DeletionMarkFilename)
 
-	require.NoError(t, ioutil.WriteFile(path, []byte(content), os.ModePerm))
+	// Ensure the block directory exists.
+	require.NoError(t, os.MkdirAll(blockPath, os.ModePerm))
+
+	require.NoError(t, ioutil.WriteFile(markPath, []byte(content), os.ModePerm))
 }
 
 func findCompactorByUserID(compactors []*Compactor, logs []*bytes.Buffer, userID string) (*Compactor, *bytes.Buffer, error) {