feat: store details of processed streams while processing delete requests (#16825)

Author: sandeepsukhani

pkg/compactor/compactor.go

@@ -13,14 +13,14 @@ import (
 	"unsafe"
 
 	"github.com/go-kit/log/level"
-	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/common/model"
-
 	"github.com/grafana/dskit/backoff"
 	"github.com/grafana/dskit/kv"
 	"github.com/grafana/dskit/ring"
 	"github.com/grafana/dskit/services"
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
 
 	"github.com/grafana/loki/v3/pkg/analytics"
 	"github.com/grafana/loki/v3/pkg/compactor/deletion"
@@ -406,13 +406,10 @@ func (c *Compactor) initDeletes(objectClient client.ObjectClient, indexUpdatePro
 
 	c.DeleteRequestsGRPCHandler = deletion.NewGRPCRequestHandler(c.deleteRequestsStore, limits)
 
-	c.deleteRequestsManager = deletion.NewDeleteRequestsManager(
-		c.deleteRequestsStore,
-		c.cfg.DeleteRequestCancelPeriod,
-		c.cfg.DeleteBatchSize,
-		limits,
-		r,
-	)
+	c.deleteRequestsManager, err = deletion.NewDeleteRequestsManager(deletionWorkDir, c.deleteRequestsStore, c.cfg.DeleteRequestCancelPeriod, c.cfg.DeleteBatchSize, limits, r)
+	if err != nil {
+		return err
+	}
 
 	c.expirationChecker = newExpirationChecker(retention.NewExpirationChecker(limits), c.deleteRequestsManager)
 	return nil
@@ -853,12 +850,12 @@ func newExpirationChecker(retentionExpiryChecker, deletionExpiryChecker retentio
 	return &expirationChecker{retentionExpiryChecker, deletionExpiryChecker}
 }
 
-func (e *expirationChecker) Expired(ref retention.ChunkEntry, now model.Time) (bool, filter.Func) {
-	if expired, nonDeletedIntervals := e.retentionExpiryChecker.Expired(ref, now); expired {
+func (e *expirationChecker) Expired(userID []byte, chk retention.Chunk, lbls labels.Labels, seriesID []byte, tableName string, now model.Time) (bool, filter.Func) {
+	if expired, nonDeletedIntervals := e.retentionExpiryChecker.Expired(userID, chk, lbls, seriesID, tableName, now); expired {
 		return expired, nonDeletedIntervals
 	}
 
-	return e.deletionExpiryChecker.Expired(ref, now)
+	return e.deletionExpiryChecker.Expired(userID, chk, lbls, seriesID, tableName, now)
 }
 
 func (e *expirationChecker) MarkPhaseStarted() {
@@ -885,8 +882,12 @@ func (e *expirationChecker) IntervalMayHaveExpiredChunks(interval model.Interval
 	return e.retentionExpiryChecker.IntervalMayHaveExpiredChunks(interval, userID) || e.deletionExpiryChecker.IntervalMayHaveExpiredChunks(interval, userID)
 }
 
-func (e *expirationChecker) DropFromIndex(ref retention.ChunkEntry, tableEndTime model.Time, now model.Time) bool {
-	return e.retentionExpiryChecker.DropFromIndex(ref, tableEndTime, now) || e.deletionExpiryChecker.DropFromIndex(ref, tableEndTime, now)
+func (e *expirationChecker) DropFromIndex(userID []byte, chk retention.Chunk, labels labels.Labels, tableEndTime model.Time, now model.Time) bool {
+	return e.retentionExpiryChecker.DropFromIndex(userID, chk, labels, tableEndTime, now) || e.deletionExpiryChecker.DropFromIndex(userID, chk, labels, tableEndTime, now)
+}
+
+func (e *expirationChecker) CanSkipSeries(userID []byte, lbls labels.Labels, seriesID []byte, seriesStart model.Time, tableName string, now model.Time) bool {
+	return e.retentionExpiryChecker.CanSkipSeries(userID, lbls, seriesID, seriesStart, tableName, now) && e.deletionExpiryChecker.CanSkipSeries(userID, lbls, seriesID, seriesStart, tableName, now)
 }
 
 func (c *Compactor) OnRingInstanceRegister(_ *ring.BasicLifecycler, ringDesc ring.Desc, instanceExists bool, _ string, instanceDesc ring.InstanceDesc) (ring.InstanceState, ring.Tokens) {

pkg/compactor/deletion/delete_request.go

@@ -109,14 +109,14 @@ func allMatch(matchers []*labels.Matcher, labels labels.Labels) bool {
 // IsDeleted checks if the given ChunkEntry will be deleted by this DeleteRequest.
 // It returns a filter.Func if the chunk is supposed to be deleted partially or the delete request contains line filters.
 // If the filter.Func is nil, the whole chunk is supposed to be deleted.
-func (d *DeleteRequest) IsDeleted(entry retention.ChunkEntry) (bool, filter.Func) {
-	if d.UserID != unsafeGetString(entry.UserID) {
+func (d *DeleteRequest) IsDeleted(userID []byte, lbls labels.Labels, chunk retention.Chunk) (bool, filter.Func) {
+	if d.UserID != unsafeGetString(userID) {
 		return false, nil
 	}
 
 	if !intervalsOverlap(model.Interval{
-		Start: entry.From,
-		End:   entry.Through,
+		Start: chunk.From,
+		End:   chunk.Through,
 	}, model.Interval{
 		Start: d.StartTime,
 		End:   d.EndTime,
@@ -137,16 +137,16 @@ func (d *DeleteRequest) IsDeleted(entry retention.ChunkEntry) (bool, filter.Func
 		}
 	}
 
-	if !labels.Selector(d.matchers).Matches(entry.Labels) {
+	if !labels.Selector(d.matchers).Matches(lbls) {
 		return false, nil
 	}
 
-	if d.StartTime <= entry.From && d.EndTime >= entry.Through && !d.logSelectorExpr.HasFilter() {
+	if d.StartTime <= chunk.From && d.EndTime >= chunk.Through && !d.logSelectorExpr.HasFilter() {
 		// Delete request covers the whole chunk and there are no line filters in the logSelectorExpr so the whole chunk will be deleted
 		return true, nil
 	}
 
-	ff, err := d.FilterFunction(entry.Labels)
+	ff, err := d.FilterFunction(lbls)
 	if err != nil {
 		// The query in the delete request is checked when added to the table.
 		// So this error should not occur.

pkg/compactor/deletion/delete_request_test.go

@@ -33,13 +33,9 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 	lblWithStructuredMetadataFilter := `{foo="bar", fizz="buzz"} | ping="pong"`
 	lblWithLineAndStructuredMetadataFilter := `{foo="bar", fizz="buzz"} | ping="pong" |= "filter"`
 
-	chunkEntry := retention.ChunkEntry{
-		ChunkRef: retention.ChunkRef{
-			UserID:  []byte(user1),
-			From:    now.Add(-3 * time.Hour),
-			Through: now.Add(-time.Hour),
-		},
-		Labels: mustParseLabel(lbl),
+	chunkEntry := retention.Chunk{
+		From:    now.Add(-3 * time.Hour),
+		Through: now.Add(-time.Hour),
 	}
 
 	type resp struct {
@@ -275,7 +271,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			require.NoError(t, tc.deleteRequest.SetQuery(tc.deleteRequest.Query))
 			tc.deleteRequest.Metrics = newDeleteRequestsManagerMetrics(nil)
-			isExpired, filterFunc := tc.deleteRequest.IsDeleted(chunkEntry)
+			isExpired, filterFunc := tc.deleteRequest.IsDeleted([]byte(user1), mustParseLabel(lbl), chunkEntry)
 			require.Equal(t, tc.expectedResp.isDeleted, isExpired)
 			if tc.expectedResp.expectedFilter == nil {
 				require.Nil(t, filterFunc)