do direct deletion instead of delayed deletion for tenant retention policy to clean backlogs (#121)

e2e tests are flaky

Author: jnyi

cmd/thanos/compact.go

@@ -463,6 +463,12 @@ func runCompact(
 
 	compactMainFn := func() error {
 		// this should happen before any compaction to remove unnecessary process on backlogs beyond retention.
+		if len(retentionByTenant) != 0 && len(sy.Metas()) == 0 {
+			level.Info(logger).Log("msg", "sync before tenant retention due to no blocks")
+			if err := sy.SyncMetas(ctx); err != nil {
+				return errors.Wrap(err, "sync before tenant retention")
+			}
+		}
 		if err := compact.ApplyRetentionPolicyByTenant(ctx, logger, insBkt, sy.Metas(), retentionByTenant, compactMetrics.blocksMarked.WithLabelValues(metadata.DeletionMarkFilename, metadata.TenantRetentionExpired)); err != nil {
 			return errors.Wrap(err, "retention by tenant failed")
 		}

pkg/compact/retention.go

@@ -114,20 +114,28 @@ func ApplyRetentionPolicyByTenant(
 		level.Info(logger).Log("msg", "tenant retention is disabled due to no policy")
 		return nil
 	}
-	level.Info(logger).Log("msg", "start tenant retention")
+	level.Info(logger).Log("msg", "start tenant retention", "total", len(metas))
+	deleted, skipped, notExpired := 0, 0, 0
 	for id, m := range metas {
 		policy, ok := retentionByTenant[m.Thanos.GetTenant()]
 		if !ok {
+			skipped++
 			continue
 		}
 		maxTime := time.Unix(m.MaxTime/1000, 0)
 		if policy.isExpired(maxTime) {
 			level.Info(logger).Log("msg", "applying retention: marking block for deletion", "id", id, "maxTime", maxTime.String())
-			if err := block.MarkForDeletion(ctx, logger, bkt, id, fmt.Sprintf("block exceeding retention of %v", policy), blocksMarkedForDeletion); err != nil {
-				return errors.Wrap(err, "delete block")
+			if err := block.Delete(ctx, logger, bkt, id); err != nil {
+				level.Error(logger).Log("msg", "failed to delete block", "id", id, "err", err)
+				continue // continue to next block to clean up backlogs
+			} else {
+				blocksMarkedForDeletion.Inc()
+				deleted++
 			}
+		} else {
+			notExpired++
 		}
 	}
-	level.Info(logger).Log("msg", "tenant retention apply done")
+	level.Info(logger).Log("msg", "tenant retention apply done", "deleted", deleted, "skipped", skipped, "notExpired", notExpired)
 	return nil
 }

pkg/compact/retention_test.go

@@ -525,7 +525,6 @@ func TestApplyRetentionPolicyByTenant(t *testing.T) {
 			}
 
 			got := []string{}
-			gotMarkedBlocksCount := 0.0
 			testutil.Ok(t, bkt.Iter(context.TODO(), "", func(name string) error {
 				exists, err := bkt.Exists(ctx, filepath.Join(name, metadata.DeletionMarkFilename))
 				if err != nil {
@@ -535,12 +534,12 @@ func TestApplyRetentionPolicyByTenant(t *testing.T) {
 					got = append(got, name)
 					return nil
 				}
-				gotMarkedBlocksCount += 1.0
 				return nil
 			}))
+			deleted := float64(len(tt.blocks) - len(got))
 
 			testutil.Equals(t, got, tt.want)
-			testutil.Equals(t, gotMarkedBlocksCount, promtest.ToFloat64(blocksMarkedForDeletion))
+			testutil.Equals(t, deleted, promtest.ToFloat64(blocksMarkedForDeletion))
 		})
 	}
 }