Repository: cortexproject/cortex
Pull request #4921 (2022-10-18 16:52:02)

treat interrupted compactions different than failures
<!--  Thanks for sending a pull request!  Before submitting:

1. Read our CONTRIBUTING.md guide
2. Rebase your PR if it gets out of sync with master
-->

**What this PR does**:

This PR adds "interruption" as a new category of compaction outcome.  Currently, compaction starts and then it either fails or succeeds according to metrics.  In the event of a graceful shutdown, an ongoing compaction is recorded as a failure and `cortex_compactor_runs_failed_total` and `cortex_compactor_tenants_processing_failed` are incremented.  This makes it cumbersome to monitor for compaction failures.  After this PR, compactions interrupted by shutdown will not be recorded as failures but rather as "interruptions".

**Notes**:
Ideally, we could also verify in unit tests that `cortex_compactor_tenants_processing_failed` is not incremented, but this would require more complicated refactoring because `cortex_compactor_tenants_processing_*` metrics are reset at the end of compaction runs.

**Checklist**
- [x] Tests updated
- [x] 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 359cae53ac..d0c1e779d7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -41,6 +41,7 @@
 * [CHANGE] Disables TSDB isolation. #4825
 * [CHANGE] Drops support Prometheus 1.x rule format on configdb. #4826
 * [CHANGE] Removes `-ingester.stream-chunks-when-using-blocks` experimental flag and stream chunks by default when `querier.ingester-streaming` is enabled. #4864
+* [CHANGE] Compactor: Added `cortex_compactor_runs_interrupted_total` to separate compaction interruptions from failures
 * [ENHANCEMENT] AlertManager: Retrying AlertManager Get Requests (Get Alertmanager status, Get Alertmanager Receivers) on next replica on error #4840
 * [ENHANCEMENT] Querier/Ruler: Retry store-gateway in case of unexpected failure, instead of failing the query. #4532 #4839
 * [ENHANCEMENT] Ring: DoBatch prioritize 4xx errors when failing. #4783
diff --git a/docs/proposals/timeseries-partitioning-in-compactor.md b/docs/proposals/timeseries-partitioning-in-compactor.md
index f61e25192c..5b733717b3 100644
--- a/docs/proposals/timeseries-partitioning-in-compactor.md
+++ b/docs/proposals/timeseries-partitioning-in-compactor.md
@@ -122,6 +122,37 @@ Now that the planner has produced a compaction plan for the T1-T3 compaction gro
 ![Grouping](/images/proposals/timeseries-partitioning-in-compactor-compact.png)
 T1 - Partition 1-2 was created with hash % 2 == 0, and in order to avoid having duplication information in blocks produced by compaction group 3-8, compaction group 5-8, and compaction group 7-8, we need apply the filter the `%8 == 0` hash, as that’s the hash of the highest partition count.
 
+### Compaction Workflow
+
+1. Compactor initializes Grouper and Planner.
+2. Compactor retrieves block's meta.json and call Grouper to group blocks for compaction.
+3. Grouper generates partitioned compaction groups:
+   1. Grouper groups source blocks into unpartitioned groups.
+   2. For each unpartitioned group:
+      1. Generates partitioned compaction group ID which is hash of min and max time of result block.
+      2. If the ID exists under the tenant directory in block storage, continue on next unpartitioned group.
+      3. Calculates number of partitions. Number of partitions indicates how many partitions one unpartitioned group would be partitioned into based on the total size of indices and number of time series from each source blocks in the unpartitioned group.
+      4. Assign source blocks into each partition with partition ID (value is in range from 0 to number_of_partitions - 1). Note that one source block could be used in multiple partitions (explanation in [Planning the compaction](#planning-the-compaction) and [Compaction](#compaction)). So multiple partition ID could be assigned to same source block. Check more partitioning examples in [Compaction Partitioning Examples](#compaction-partitioning-examples)
+      5. Generates partitioned compaction group that indicates which partition ID each blocks got assigned.
+      6. Partitioned compaction group information would be stored in block storage under the tenant directory it belongs to and the stored file can be picked up by cleaner later. Partitioned compaction group information contains partitioned compaction group ID, number of partitions, list of partitions which has partition ID and list of source blocks.
+      7. Store partitioned compaction group ID in block storage under each blocks' directory that are used by the generated partitioned compaction group.
+4. Grouper returns partitioned compaction groups to Compactor. Each returned group would have partition ID, number of partitions, and list of source blocks in memory.
+5. Compactor iterates over each partitioned compaction group. For each iteration, calls Planner to make sure the group is ready for compaction.
+6. Planner collects partitioned compaction group which is ready for compaction.
+   1. For each partitions in the group and for each blocks in the partition:
+      1. Make sure all source blocks fit within the time range of the group.
+      2. Make sure each source block with assigned partition IDs is currently not used by another ongoing compaction. This could utilize visit marker file that is introduced in #4805 by expanding it for each partition ID of the source block.
+      3. If all blocks in the partition are ready to be compacted,
+         1. mark status of those blocks with assigned partition ID as `pending`.
+         2. The status information of each partition ID would be stored in block storage under the corresponding block directory in order for cleaner to pick it up later.
+      4. If not all blocks in the partition are ready, continue on next partition
+7. Return all ready partitions to Compactor.
+8. Compactor starts compacting partitioned blocks. Once compaction completed, Compactor would mark status of all blocks along with assigned partition ID in the group as `completed`. Compactor should use partitioned compaction group ID to retrieve partitioned compaction group information from block storage to get all partition IDs assigned to each block. Then, retrieve status information of each partition ID this assigned to block under current block directory in block storage. If all assigned partition ID of the block have status set to `completed`, upload deletion marker for this block. Otherwise, no deletion marker would be uploaded.
+
+### Clean up Workflow
+
+Cleaner would periodically check any tenants having deletion marker. If there is a deletion marker for the tenant, Cleaner should remove all blocks and then clean up other files including partitioned group information files after tenant clean up delay. If there is no deletion marker for tenant, Clean should scan any source blocks having a deletion marker. If there is a deletion marker for the block, Cleaner should delete it.
+
 ## Performance
 
 Currently a 400M timeseries takes 12 hours to compact, without taking block download into consideration. If we have a partition count of 2, we can reduce this down to 6 hours, and a partition count of 10 is 3 hours. The scaling is not linear, and I’m still attempting to find out why. The initial result is promising enough to continue though.
@@ -264,4 +295,160 @@ T2 partition 4 - Hash(timeseries label) % 4 == 3 && % 8 == 7
 T3 partition 8 - Hash(timeseries label) % 8 == 7
 ```
 
+### Compaction Partitioning Examples
+
+#### Scenario: All source blocks were compacted by partitioning compaction (Idea case)
+
+All source blocks were previously compacted through partitioning compaction. In this case for each time range, the number of blocks belong to same time range would be 2^x if multiplier is set to 2.
+
+```
+Time ranges:
+T1, T2, T3
+
+Source blocks:
+T1: B1, B2
+T2: B3, B4, B5, B6
+T3: B7, B8, B9, B10, B11, B12, B13, B14
+
+Total indices size of all source blocks:
+200G
+```
+
+Number of Partitions = (200G / 64G = 3.125) => round up to next 2^x = 4
+
+Partitioning:
+* For T1, there are only 2 blocks which is < 4. So
+    * B1 (index 0 in the time range) can be grouped with other blocks having N % 4 == 0 or 2. Because 0 % 2 == 0.
+    * B2 (index 1 in the time range) can be grouped with other blocks having N % 4 == 1 or 3. Because 1 % 2 == 1.
+* For T2,
+    * B3 (index 0 in the time range) can be grouped with other blocks having N % 4 == 0.
+    * B4 (index 1 in the time range) can be grouped with other blocks having N % 4 == 1.
+    * B5 (index 2 in the time range) can be grouped with other blocks having N % 4 == 2.
+    * B6 (index 3 in the time range) can be grouped with other blocks having N % 4 == 3.
+* For T3,
+    * B7 (index 0 in the time range) can be grouped with other blocks having N % 4 == 0.
+    * B8 (index 1 in the time range) can be grouped with other blocks having N % 4 == 1.
+    * B9 (index 2 in the time range) can be grouped with other blocks having N % 4 == 2.
+    * B10 (index 3 in the time range) can be grouped with other blocks having N % 4 == 3.
+    * B11 (index 4 in the time range) can be grouped with other blocks having N % 4 == 0.
+    * B12 (index 5 in the time range) can be grouped with other blocks having N % 4 == 1.
+    * B13 (index 6 in the time range) can be grouped with other blocks having N % 4 == 2.
+    * B14 (index 7 in the time range) can be grouped with other blocks having N % 4 == 3.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 4 \
+  Blocks: B1, B3, B7, B11
+* Partition ID: 1 \
+  Number of Partitions: 4 \
+  Blocks: B2, B4, B8, B12
+* Partition ID: 2 \
+  Number of Partitions: 4 \
+  Blocks: B1, B5, B9, B13
+* Partition ID: 3 \
+  Number of Partitions: 4 \
+  Blocks: B2, B6, B10, B14
+
+---
+
+#### Scenario: All source blocks are level 1 blocks
+
+All source blocks are level 1 blocks. Since number of level 1 blocks in one time range is not guaranteed to be 2^x, all blocks need to be included in each partition.
+
+```
+Time ranges:
+T1
+
+Source blocks:
+T1: B1, B2, B3
+
+Total indices size of all source blocks:
+100G
+```
+
+Number of Partitions = (100G / 64G = 1.5625) => round up to next 2^x = 2
+
+Partitioning: There is only one time range from all source blocks which means it is compacting level 1 blocks. Partitioning needs to include all source blocks in each partition.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+* Partition ID: 1 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+
+---
+
+#### Scenario: All source blocks are with compaction level > 1 and were generated by compactor without partitioning compaction
+
+If source block was generated by compactor without partitioning compaction, there should be only one block per time range. Since there is only one block in one time range, that one block would be included in all partitions.
+
+```
+Time ranges:
+T1, T2, T3
+
+Source blocks:
+T1: B1
+T2: B2
+T3: B3
+
+Total indices size of all source blocks:
+100G
+```
+
+Number of Partitions = (100G / 64G = 1.5625) => round up to next 2^x = 2
+
+Partitioning:
+* For T1, there is only one source block. Include B1 in all partitions.
+* For T2, there is only one source block. Include B2 in all partitions.
+* For T3, there is only one source block. Include B3 in all partitions.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+* Partition ID: 1 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B3
+
+---
+
+#### Scenario: All source blocks are with compaction level > 1 and some of them were generated by compactor with partitioning compaction
+
+Blocks generated by compactor without partitioning compaction would be included in all partitions. Blocks generated with partitioning compaction would be partitioned based on multiplier.
+
+```
+Time ranges:
+T1, T2, T3
+
+Source blocks:
+T1: B1 (unpartitioned)
+T2: B2, B3
+T3: B4, B5, B6, B7
+
+Total indices size of all source blocks:
+100G
+```
+
+Number of Partitions = (100G / 64G = 1.5625) => round up to next 2^x = 2
+
+Partitioning:
+* For T1, there is only one source block. Include B1 in all partitions.
+* For T2,
+    * B2 (index 0 in the time range) can be grouped with other blocks having N % 2 == 0.
+    * B3 (index 1 in the time range) can be grouped with other blocks having N % 2 == 1.
+* For T3,
+    * B4 (index 0 in the time range) can be grouped with other blocks having N % 2 == 0.
+    * B5 (index 1 in the time range) can be grouped with other blocks having N % 2 == 1.
+    * B6 (index 2 in the time range) can be grouped with other blocks having N % 2 == 0.
+    * B7 (index 3 in the time range) can be grouped with other blocks having N % 2 == 1.
+
+Partitions in Partitioned Compaction Group:
+* Partition ID: 0 \
+  Number of Partitions: 2 \
+  Blocks: B1, B2, B4, B6
+* Partition ID: 1 \
+  Number of Partitions: 2 \
+  Blocks: B1, B3, B5, B7
 
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index c36ee8224d..75732e2200 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -313,6 +313,7 @@ type Compactor struct {
 
 	// Metrics.
 	compactionRunsStarted          prometheus.Counter
+	compactionRunsInterrupted      prometheus.Counter
 	compactionRunsCompleted        prometheus.Counter
 	compactionRunsFailed           prometheus.Counter
 	compactionRunsLastSuccess      prometheus.Gauge
@@ -399,6 +400,10 @@ func newCompactor(
 			Name: "cortex_compactor_runs_started_total",
 			Help: "Total number of compaction runs started.",
 		}),
+		compactionRunsInterrupted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
+			Name: "cortex_compactor_runs_interrupted_total",
+			Help: "Total number of compaction runs interrupted.",
+		}),
 		compactionRunsCompleted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_runs_completed_total",
 			Help: "Total number of compaction runs successfully completed.",
@@ -598,16 +603,23 @@ func (c *Compactor) running(ctx context.Context) error {
 }
 
 func (c *Compactor) compactUsers(ctx context.Context) {
-	succeeded := false
-	compactionErrorCount := 0
+	failed := false
+	interrupted := false
 
 	c.compactionRunsStarted.Inc()
 
 	defer func() {
-		if succeeded && compactionErrorCount == 0 {
+		// interruptions and successful runs are considered
+		// mutually exclusive but we consider a run failed if any
+		// tenant runs failed even if later runs are interrupted
+		if !interrupted && !failed {
 			c.compactionRunsCompleted.Inc()
 			c.compactionRunsLastSuccess.SetToCurrentTime()
-		} else {
+		}
+		if interrupted {
+			c.compactionRunsInterrupted.Inc()
+		}
+		if failed {
 			c.compactionRunsFailed.Inc()
 		}
 
@@ -621,6 +633,7 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 	level.Info(c.logger).Log("msg", "discovering users from bucket")
 	users, err := c.discoverUsersWithRetries(ctx)
 	if err != nil {
+		failed = true
 		level.Error(c.logger).Log("msg", "failed to discover users from bucket", "err", err)
 		return
 	}
@@ -640,7 +653,8 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 	for _, userID := range users {
 		// Ensure the context has not been canceled (ie. compactor shutdown has been triggered).
 		if ctx.Err() != nil {
-			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "err", err)
+			interrupted = true
+			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID)
 			return
 		}
 
@@ -670,8 +684,15 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 		level.Info(c.logger).Log("msg", "starting compaction of user blocks", "user", userID)
 
 		if err = c.compactUserWithRetries(ctx, userID); err != nil {
+			// TODO: patch thanos error types to support errors.Is(err, context.Canceled) here
+			if ctx.Err() != nil && ctx.Err() == context.Canceled {
+				interrupted = true
+				level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID)
+				return
+			}
+
 			c.compactionRunFailedTenants.Inc()
-			compactionErrorCount++
+			failed = true
 			level.Error(c.logger).Log("msg", "failed to compact user blocks", "user", userID, "err", err)
 			continue
 		}
@@ -706,8 +727,6 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 			}
 		}
 	}
-
-	succeeded = true
 }
 
 func (c *Compactor) compactUserWithRetries(ctx context.Context, userID string) error {


Test patch:
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 62e815cb4e..52e4cf3828 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -1796,3 +1796,64 @@ func TestCompactor_ShouldFailCompactionOnTimeout(t *testing.T) {
 		`level=error component=compactor msg="compactor failed to become ACTIVE in the ring" err="context deadline exceeded"`,
 	}, removeIgnoredLogs(strings.Split(strings.TrimSpace(logs.String()), "\n")))
 }
+
+func TestCompactor_ShouldNotTreatInterruptionsAsErrors(t *testing.T) {
+	bucketClient := objstore.NewInMemBucket()
+	id := ulid.MustNew(ulid.Now(), rand.Reader)
+	require.NoError(t, bucketClient.Upload(context.Background(), "user-1/"+id.String()+"/meta.json", strings.NewReader(mockBlockMetaJSON(id.String()))))
+
+	b1 := createTSDBBlock(t, bucketClient, "user-1", 10, 20, map[string]string{"__name__": "Teste"})
+	b2 := createTSDBBlock(t, bucketClient, "user-1", 20, 30, map[string]string{"__name__": "Teste"})
+
+	c, tsdbCompactor, tsdbPlanner, logs, registry := prepare(t, prepareConfig(), bucketClient, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	tsdbCompactor.On("Compact", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ulid.ULID{}, context.Canceled).Run(func(args mock.Arguments) {
+		cancel()
+	})
+	tsdbPlanner.On("Plan", mock.Anything, mock.Anything).Return([]*metadata.Meta{
+		{
+			BlockMeta: tsdb.BlockMeta{
+				ULID:    b1,
+				MinTime: 10,
+				MaxTime: 20,
+			},
+		},
+		{
+			BlockMeta: tsdb.BlockMeta{
+				ULID:    b2,
+				MinTime: 20,
+				MaxTime: 30,
+			},
+		},
+	}, nil)
+	require.NoError(t, services.StartAndAwaitRunning(ctx, c))
+
+	cortex_testutil.Poll(t, 1*time.Second, 1.0, func() interface{} {
+		return prom_testutil.ToFloat64(c.compactionRunsInterrupted)
+	})
+
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
+
+	assert.NoError(t, prom_testutil.GatherAndCompare(registry, strings.NewReader(`
+		# TYPE cortex_compactor_runs_completed_total counter
+		# HELP cortex_compactor_runs_completed_total Total number of compaction runs successfully completed.
+		cortex_compactor_runs_completed_total 0
+
+		# TYPE cortex_compactor_runs_interrupted_total counter
+		# HELP cortex_compactor_runs_interrupted_total Total number of compaction runs interrupted.
+		cortex_compactor_runs_interrupted_total 1
+
+		# TYPE cortex_compactor_runs_failed_total counter
+		# HELP cortex_compactor_runs_failed_total Total number of compaction runs failed.
+		cortex_compactor_runs_failed_total 0
+	`),
+		"cortex_compactor_runs_completed_total",
+		"cortex_compactor_runs_interrupted_total",
+		"cortex_compactor_runs_failed_total",
+	))
+
+	lines := strings.Split(logs.String(), "\n")
+	require.Contains(t, lines, `level=info component=compactor msg="interrupting compaction of user blocks" user=user-1`)
+	require.NotContains(t, logs.String(), `level=error`)
+}