[ES-1393569] fix concurrency apply retention and deletion (#148)

Author: jnyi

pkg/compact/blocks_cleaner.go

@@ -5,6 +5,7 @@ package compact
 
 import (
 	"context"
+	"runtime"
 	"sync"
 	"time"
 
@@ -46,7 +47,7 @@ func (s *BlocksCleaner) DeleteMarkedBlocks(ctx context.Context) error {
 
 	deletionMarkMap := s.ignoreDeletionMarkFilter.DeletionMarkBlocks()
 	wg := &sync.WaitGroup{}
-	sem := make(chan struct{}, ParallelLimit)
+	sem := make(chan struct{}, runtime.NumCPU())
 	for _, deletionMark := range deletionMarkMap {
 		if time.Since(time.Unix(deletionMark.DeletionTime, 0)).Seconds() > s.deleteDelay.Seconds() {
 			sem <- struct{}{} // acquire BEFORE spawning goroutine

pkg/compact/clean.go

@@ -5,6 +5,7 @@ package compact
 
 import (
 	"context"
+	"runtime"
 	"sync"
 	"time"
 
@@ -41,7 +42,7 @@ func BestEffortCleanAbortedPartialUploads(
 	// can be assumed in this case. Keep partialUploadThresholdAge long for now.
 	// Mitigate this by adding ModifiedTime to bkt and check that instead of ULID (block creation time).
 	wg := &sync.WaitGroup{}
-	sem := make(chan struct{}, ParallelLimit)
+	sem := make(chan struct{}, runtime.NumCPU())
 	for id := range partial {
 		if ulid.Now()-id.Time() <= uint64(PartialUploadThresholdAge/time.Millisecond) {
 			// Minimum delay has not expired, ignore for now.

pkg/compact/overlapping.go

@@ -23,7 +23,7 @@ const (
 	overlappingReason = "blocks-overlapping"
 
 	symbolTableSizeExceedsError = "symbol table size exceeds"
-	symbolTableSizeLimit        = 128 * 1024
+	symbolTableSizeLimit        = 1024 * 1024
 )
 
 type OverlappingCompactionLifecycleCallback struct {

pkg/compact/overlapping_test.go

@@ -175,9 +175,9 @@ func TestHandleError(t *testing.T) {
 			input: []*metadata.Meta{
 				createCustomBlockMeta(1, 1, 2, metadata.ReceiveSource, 1),
 				createCustomBlockMeta(2, 1, 6, metadata.CompactorSource, 1),
-				createCustomBlockMeta(3, 1, 6, metadata.ReceiveSource, 1024*1024),
+				createCustomBlockMeta(3, 1, 6, metadata.ReceiveSource, 2*1024*1024),
 			},
-			err:         errors.New(symbolTableSizeExceedsError + " 1024*1024"),
+			err:         errors.New(symbolTableSizeExceedsError + " 2*1024*1024"),
 			handledErrs: 1,
 			errBlockIdx: 2,
 		},

pkg/compact/progress.go

@@ -10,8 +10,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-const ParallelLimit = 1024
-
 type CompactorState int64
 
 // Use a gauge to track the state of a compactor pod.

pkg/compact/retention.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"runtime"
 	"sync"
 	"time"
 
@@ -40,7 +41,7 @@ func ApplyRetentionPolicyByResolution(
 ) error {
 	level.Info(logger).Log("msg", "start optional retention")
 	wg := &sync.WaitGroup{}
-	sem := make(chan struct{}, ParallelLimit)
+	sem := make(chan struct{}, runtime.NumCPU())
 	for id, m := range metas {
 		retentionDuration := retentionByResolution[ResolutionLevel(m.Thanos.Downsample.Resolution)]
 		if retentionDuration.Seconds() == 0 {

pkg/receive/handler.go

@@ -1031,6 +1031,7 @@ func (h *Handler) sendRemoteWrite(
 		// Increment replica since on-the-wire format is 1-indexed and 0 indicates un-replicated.
 		Replica: realReplicationIndex,
 	}, endpointReplica, trackedSeries.seriesIDs, responses, func(err error) {
+		defer wg.Done()
 		if err == nil {
 			h.forwardRequests.WithLabelValues(labelSuccess).Inc()
 			if !alreadyReplicated {
@@ -1045,7 +1046,6 @@ func (h *Handler) sendRemoteWrite(
 				}
 			}
 		}
-		wg.Done()
 	})
 }