Test old block compaction. Bunch of sorting, infinite loop and other fixes.

Signed-off-by: Ganesh Vernekar <[email protected]>

Author: codesome

pkg/ingester/ingester_v2.go

@@ -132,7 +132,6 @@ func (u *userTSDB) getShippedBlocksULID() ([]ulid.ULID, error) {
 	if err := json.Unmarshal(b, &shipperMeta); err != nil {
 		return nil, errors.Wrap(err, "unmarshal shipper meta file to json")
 	}
-
 	return shipperMeta.Uploaded, nil
 }
 

pkg/ingester/ingester_v2_backfill.go

@@ -176,7 +176,7 @@ func (i *Ingester) getOrCreateBackfillTSDB(userBuckets *tsdbBuckets, userID stri
 	}
 	userBuckets.buckets = append(userBuckets.buckets, bucket)
 	sort.Slice(userBuckets.buckets, func(i, j int) bool {
-		return userBuckets.buckets[i].bucketStart < userBuckets.buckets[i].bucketEnd
+		return userBuckets.buckets[i].bucketStart < userBuckets.buckets[j].bucketStart
 	})
 
 	return bucket, nil
@@ -286,7 +286,7 @@ func (i *Ingester) openExistingBackfillTSDB(ctx context.Context) error {
 	for _, buckets := range i.TSDBState.backfillDBs.tsdbs {
 		buckets.Lock()
 		sort.Slice(buckets.buckets, func(i, j int) bool {
-			return buckets.buckets[i].bucketStart < buckets.buckets[i].bucketEnd
+			return buckets.buckets[i].bucketStart < buckets.buckets[j].bucketStart
 		})
 		buckets.Unlock()
 	}
@@ -494,63 +494,72 @@ func (i *Ingester) closeOldBackfillTSDBsAndDelete(gracePeriod int64) error {
 }
 
 func (i *Ingester) runOnBucketsBefore(deleteBucket bool, gracePeriodFunc func(t int64) int64, f func(db *userTSDB) error) error {
-	i.TSDBState.backfillDBs.closeAndShipMtx.Lock()
-	defer i.TSDBState.backfillDBs.closeAndShipMtx.Unlock()
+	i.TSDBState.backfillDBs.compactShipDeleteMtx.Lock()
+	defer i.TSDBState.backfillDBs.compactShipDeleteMtx.Unlock()
 
 	type tempType struct {
-		userID  string
-		buckets *tsdbBuckets
+		userID     string
+		cutoffTime int64
+		buckets    *tsdbBuckets
 	}
 
 	var usersHavingOldTSDBs []tempType
 
 	// Collecting users who have old TSDBs.
 	i.TSDBState.backfillDBs.tsdbsMtx.RLock()
 	for userID, userBuckets := range i.TSDBState.backfillDBs.tsdbs {
+		cutoffTime := int64(0)
+		i.userStatesMtx.RLock()
+		mainDB := i.TSDBState.dbs[userID]
+		i.userStatesMtx.RUnlock()
 		userBuckets.RLock()
+		if mainDB != nil {
+			cutoffTime = gracePeriodFunc(mainDB.Head().MaxTime())
+		} else {
+			// There is no main TSDB. So use the maxt of the last bucket.
+			cutoffTime = gracePeriodFunc(userBuckets.buckets[len(userBuckets.buckets)-1].db.Head().MaxTime())
+		}
 		for _, bucket := range userBuckets.buckets {
-			if bucket.bucketEnd < gracePeriodFunc(bucket.db.Head().MaxTime()) {
-				usersHavingOldTSDBs = append(usersHavingOldTSDBs, tempType{
-					userID:  userID,
-					buckets: userBuckets,
-				})
+			if bucket.bucketEnd >= cutoffTime {
 				break
 			}
+			usersHavingOldTSDBs = append(usersHavingOldTSDBs, tempType{
+				userID:     userID,
+				cutoffTime: cutoffTime,
+				buckets:    userBuckets,
+			})
 		}
 		userBuckets.RUnlock()
 	}
 	i.TSDBState.backfillDBs.tsdbsMtx.RUnlock()
 
 	var merr tsdb_errors.MultiError
 	for _, user := range usersHavingOldTSDBs {
+		idx := 0
 		for {
-			user.buckets.Lock()
-			if len(user.buckets.buckets) == 0 {
-				user.buckets.Unlock()
+			user.buckets.RLock()
+			if len(user.buckets.buckets) == 0 || idx == len(user.buckets.buckets) {
+				user.buckets.RUnlock()
 				break
 			}
-			bucket := user.buckets.buckets[0]
-			if bucket.bucketEnd >= gracePeriodFunc(bucket.db.Head().MaxTime()) {
-				user.buckets.Unlock()
+			bucket := user.buckets.buckets[idx]
+			if bucket.bucketEnd >= user.cutoffTime {
+				user.buckets.RUnlock()
 				break
 			}
-			if deleteBucket {
-				user.buckets.buckets = user.buckets.buckets[1:]
-			}
-			user.buckets.Unlock()
+			user.buckets.RUnlock()
 
 			if err := f(bucket.db); err != nil {
 				merr.Add(err)
-				if deleteBucket {
-					// Restore back the old slice to attempt running later.
-					// As the slice would be small and the failure would be rare,
-					// we are not caring about efficient slice management.
-					user.buckets.Lock()
-					user.buckets.buckets = append([]*tsdbBucket{bucket}, user.buckets.buckets...)
-					user.buckets.Unlock()
-				}
 				break
 			}
+			idx++
+			if deleteBucket {
+				user.buckets.Lock()
+				user.buckets.buckets = user.buckets.buckets[1:]
+				user.buckets.Unlock()
+				idx--
+			}
 		}
 
 		if deleteBucket {
@@ -570,9 +579,9 @@ func (i *Ingester) runOnBucketsBefore(deleteBucket bool, gracePeriodFunc func(t
 
 // Assumes 1h bucket range for . TODO(codesome): protect stuff with locks.
 type backfillTSDBs struct {
-	tsdbsMtx        sync.RWMutex
-	closeAndShipMtx sync.Mutex
-	tsdbs           map[string]*tsdbBuckets
+	tsdbsMtx             sync.RWMutex
+	compactShipDeleteMtx sync.Mutex
+	tsdbs                map[string]*tsdbBuckets
 }
 
 func newBackfillTSDBs() *backfillTSDBs {
@@ -620,8 +629,8 @@ func overlapsOpenInterval(mint1, maxt1, mint2, maxt2 int64) bool {
 // getBucketName returns the string representation of the bucket.
 // YYYY_MM_DD_HH_YYYY_MM_DD_HH
 func getBucketName(start, end int64) string {
-	startTime := model.Time(start).Time()
-	endTime := model.Time(end).Time()
+	startTime := model.Time(start).Time().UTC()
+	endTime := model.Time(end).Time().UTC()
 
 	return fmt.Sprintf(
 		"%04d_%02d_%02d_%02d_%04d_%02d_%02d_%02d",

pkg/ingester/ingester_v2_test.go

@@ -1821,6 +1821,17 @@ func TestIngesterV2BackfillPushAndQuery(t *testing.T) {
 		}
 	}
 
+	testQuery := func() {
+		res, err := i.v2Query(ctx, &client.QueryRequest{
+			StartTimestampMs: math.MinInt64,
+			EndTimestampMs:   math.MaxInt64,
+			Matchers:         []*client.LabelMatcher{{Type: client.REGEX_MATCH, Name: labels.MetricName, Value: ".*"}},
+		})
+		require.NoError(t, err)
+		require.NotNil(t, res)
+		assert.Equal(t, expectedIngested, res.Timeseries)
+	}
+
 	// Samples for 100h. The main tsdb will be able to handle samples only
 	// upto 99h after this.
 	ts := 100 * time.Hour.Milliseconds()
@@ -1863,14 +1874,7 @@ func TestIngesterV2BackfillPushAndQuery(t *testing.T) {
 	ingestSample(ts, 4, true)
 
 	// Query back all the samples.
-	res, err := i.v2Query(ctx, &client.QueryRequest{
-		StartTimestampMs: math.MinInt64,
-		EndTimestampMs:   math.MaxInt64,
-		Matchers:         []*client.LabelMatcher{{Type: client.REGEX_MATCH, Name: labels.MetricName, Value: ".*"}},
-	})
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	assert.Equal(t, expectedIngested, res.Timeseries)
+	testQuery()
 
 	// Restart to check if we can still query backfill TSDBs.
 
@@ -1890,12 +1894,31 @@ func TestIngesterV2BackfillPushAndQuery(t *testing.T) {
 	})
 
 	// Query back all the samples.
-	res, err = i.v2Query(ctx, &client.QueryRequest{
-		StartTimestampMs: math.MinInt64,
-		EndTimestampMs:   math.MaxInt64,
-		Matchers:         []*client.LabelMatcher{{Type: client.REGEX_MATCH, Name: labels.MetricName, Value: ".*"}},
-	})
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	assert.Equal(t, expectedIngested, res.Timeseries)
+	testQuery()
+
+	// Compact old blocks partially to check
+	// * Compaction is happening properly.
+	// * We can still query it.
+
+	// Check there are no blocks yet.
+	userBuckets := i.TSDBState.backfillDBs.getBucketsForUser(userID)
+	for _, bucket := range userBuckets.buckets {
+		require.Equal(t, 0, len(bucket.db.Blocks()))
+		m := &shipperMock{}
+		m.On("Sync", mock.Anything).Return(0, nil)
+		bucket.db.shipper = m
+	}
+
+	// Compacting the oldest 2 buckets. They are <=97h, so compacting buckets upto 97.5h (current 100h minus 2.5h).
+	require.NoError(t, i.compactOldBackfillTSDBsAndShip(2*time.Hour.Milliseconds()+30*time.Minute.Milliseconds()))
+	for idx, bucket := range userBuckets.buckets {
+		if idx < 2 {
+			require.Equal(t, 1, len(bucket.db.Blocks()), "bucket index %d", idx)
+		} else {
+			require.Equal(t, 0, len(bucket.db.Blocks()), "bucket index %d", idx)
+		}
+	}
+
+	// We can still query compacted blocks.
+	testQuery()
 }