Add shuffle sharding grouper/planner

[ac1214]

Signed-off-by: Albert ac1214@users.noreply.github.com

What this PR does:

Implements generation of parallelize plans for the proposal outlined in #4272 using a shuffle sharding grouper and planner. Currently the parallelizable plans are generated but every compactor runs every planned compaction, the actual sharding will happen in a subsequent PR.

Checklist

• Tests updated
• Documentation added
• CHANGELOG.md updated - the order of entries should be [CHANGE], [FEATURE], [ENHANCEMENT], [BUGFIX]

[ac1214]

I wanted to discuss a change in the block compaction behavior that this PR would introduce.

The current implementation of the Thanos compactor will always compact the first set of overlapping blocks if there exists such a set. This means that if the most recently ingested set of blocks from multiple ingesters are overlapping, the blocks will be compacted. If this happens, there is potentially a “missing” block in the compaction with the Thanos planner since there may be an ingester that has not fully uploaded the block when the compaction begins. So if there are 3 overlapping blocks when the compaction begins, and they are the latest blocks passed to the Thanos planner, the planner will plan a compaction of those 3 blocks even if there is a potential fourth ingester that has yet to upload a block.

With this PR, overlapping blocks will not be compacted if they are the last set of blocks meaning in the example above, the 3 blocks won’t be compacted if there are the latest ones and they don’t cover a full range.

In a real-world situation, this would only have an impact on customers who stop ingesting blocks. The impact will be that the last group of n blocks, where n is the number of ingesters will remain uncompacted for as long as they are the latest blocks. The impact of leaving the last n blocks uncompacted would be increased storage size as well as query time (if they continue to query even after stopping ingesting blocks). One thing to note with the Thanos approach, there can be duplicate work if the blocks are compacted, and another block that overlaps is uploaded after the compaction begins.

A couple of different approaches I considered were adding grouping overlapping blocks before grouping by compactable ranges, this results in the compaction behavior being the same using these changes compared to Thanos. Another approach is if there are no new blocks after the time defined by the smallest block range passes from the max time of all the blocks, the block which are overlapping can be compacted, even if they are the latest blocks.

Something else that I considered is making this a toggle to allow the user to define their own preference, but I think that this isn't ideal as it would either lead to having to support the toggle indefinitely or eventually having to have users switch to a single behavior.

Small example illustrating what’s mentioned above

4 total blocks with 1 block incoming (not yet uploaded)

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2: {
  MinTime: 21, MaxTime: 40
}
block 3: {
  MinTime: 21, MaxTime: 40
}
block 4: {
  MinTime: 21, MaxTime: 40
}
block 5 (incoming): {
  MinTime: 21, MaxTime: 40
}

Thanos compaction

The above blocks with the current (Thanos) compaction with time ranges [20, 120, 240] would result in blocks:

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2/3/4: {
  MinTime: 21, MaxTime: 40
}

Afterwards, once block 5 is fully uploaded the final resulting blocks from a single run of the compaction will be

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2/3/4: {
  MinTime: 21, MaxTime: 40
}
block 5: {
  MinTime: 21, MaxTime: 40
}

With these blocks, another compaction will need to be done to fully compact the overlapping blocks 2-5.

New compaction behavior

With this PR and the shuffle-sharding strategy, the blocks would remain uncompacted. And would wait until a more recent block than 2-5 is uploaded. Once that block is uploaded blocks 2-5 would be impacted in 1 compaction.
If there is a block 6 uploaded with MinTime: 41, MaxTime: 60 after block 5 is fully uploaded, then the resulting blocks after a single compaction would be.

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2/3/4/5: {
  MinTime: 21, MaxTime: 40
}
block 6: {
  MinTime: 41, MaxTime: 60
}

The downside with this approach is that the uncompacted blocks 2-5 were stored for a longer time compared to the current (Thanos) approach as it was waiting for a more recent block to be uploaded before compacting the blocks. In the above with this PR if block 6 didn't exist, then blocks 2-5 would never be compacted as they would remain as the most recent blocks.

I was wondering what your thoughts were about which approach would be preferable?

[ac1214]

This PR replaces and implements the changes recommended in #4318

[ac1214]

I wanted to discuss a change in the block compaction behavior that this PR would introduce.

The current implementation of the Thanos compactor will always compact the first set of overlapping blocks if there exists such a set. This means that if the most recently ingested set of blocks from multiple ingesters are overlapping, the blocks will be compacted. If this happens, there is potentially a “missing” block in the compaction with the Thanos planner since there may be an ingester that has not fully uploaded the block when the compaction begins. So if there are 3 overlapping blocks when the compaction begins, and they are the latest blocks passed to the Thanos planner, the planner will plan a compaction of those 3 blocks even if there is a potential fourth ingester that has yet to upload a block.

With this PR, overlapping blocks will not be compacted if they are the last set of blocks meaning in the example above, the 3 blocks won’t be compacted if there are the latest ones and they don’t cover a full range.

In a real-world situation, this would only have an impact on customers who stop ingesting blocks. The impact will be that the last group of n blocks, where n is the number of ingesters will remain uncompacted for as long as they are the latest blocks. The impact of leaving the last n blocks uncompacted would be increased storage size as well as query time (if they continue to query even after stopping ingesting blocks). One thing to note with the Thanos approach, there can be duplicate work if the blocks are compacted, and another block that overlaps is uploaded after the compaction begins.

A couple of different approaches I considered were adding grouping overlapping blocks before grouping by compactable ranges, this results in the compaction behavior being the same using these changes compared to Thanos. Another approach is if there are no new blocks after the time defined by the smallest block range passes from the max time of all the blocks, the block which are overlapping can be compacted, even if they are the latest blocks.

Something else that I considered is making this a toggle to allow the user to define their own preference, but I think that this isn't ideal as it would either lead to having to support the toggle indefinitely or eventually having to have users switch to a single behavior.

Small example illustrating what’s mentioned above

4 total blocks with 1 block incoming (not yet uploaded)

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2: {
  MinTime: 21, MaxTime: 40
}
block 3: {
  MinTime: 21, MaxTime: 40
}
block 4: {
  MinTime: 21, MaxTime: 40
}
block 5 (incoming): {
  MinTime: 21, MaxTime: 40
}

Thanos compaction

The above blocks with the current (Thanos) compaction with time ranges [20, 120, 240] would result in blocks:

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2/3/4: {
  MinTime: 21, MaxTime: 40
}

Afterwards, once block 5 is fully uploaded the final resulting blocks from a single run of the compaction will be

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2/3/4: {
  MinTime: 21, MaxTime: 40
}
block 5: {
  MinTime: 21, MaxTime: 40
}

With these blocks, another compaction will need to be done to fully compact the overlapping blocks 2-5.

New compaction behavior

With this PR and the shuffle-sharding strategy, the blocks would remain uncompacted. And would wait until a more recent block than 2-5 is uploaded. Once that block is uploaded blocks 2-5 would be impacted in 1 compaction.
If there is a block 6 uploaded with MinTime: 41, MaxTime: 60 after block 5 is fully uploaded, then the resulting blocks after a single compaction would be.

block 1: {
  MinTime: 0, MaxTime: 20
}
block 2/3/4/5: {
  MinTime: 21, MaxTime: 40
}
block 6: {
  MinTime: 41, MaxTime: 60
}

The downside with this approach is that the uncompacted blocks 2-5 were stored for a longer time compared to the current (Thanos) approach as it was waiting for a more recent block to be uploaded before compacting the blocks. In the above with this PR if block 6 didn't exist, then blocks 2-5 would never be compacted as they would remain as the most recent blocks.

I was wondering what your thoughts were about which approach would be preferable?

Discussed in the community call and leaving the blocks uncompacted is okay.

[jeromeinsf]

@pracucci and/or @bboreham how can we help with the review of this?

[bboreham]

The code is long and I didn't read through every line. Broadly it looks ok.
I did wonder why the word "thanos" shows up so often - if the code is copied from Thanos it should say so, and if not can you just explain your thinking to me?

[bboreham]

Do we want to add new metrics in Cortex starting "thanos_"?

[ac1214]

Added a note where the metrics were copied from in Thanos. With these changes wouldn't the metrics in Cortex remain the same? They are only used when creating a new group using compact.NewGroup which is what is being done now (https://github.com/cortexproject/cortex/blob/master/vendor/github.com/thanos-io/thanos/pkg/compact/compact.go#L262-L312)

[yeya24]

Then it might be better to expose those metrics as another function in Thanos?

[thejosephstevens]

Any way I can help with this PR? We're running into limits in our compaction (we have about 25M active time series in a single-tenant cortex). I'd be happy to run pre-release compactor builds if this needs some kind of validation.

[alvinlin123]

error message from build is

go: updates to go.mod needed, disabled by -mod=vendor
	(Go version in go.mod is at least 1.14 and vendor directory exists.)
	to update it:
	go mod tidy
make: *** [Makefile:164: cmd/blocksconvert/blocksconvert] Error 1

Since this looks like a useful PR that we want to merge, but I don't won the original branch, I will create a new branch to work on resolving the error.

[alvinlin123]

Please see #4624 instead.