Re-add IsDeleted method

Signed-off-by: Michel Hollands <michel.hollands@grafana.com>

pkg/storage/stores/shipper/compactor/deletion/delete_request.go

@@ -1,6 +1,7 @@
 package deletion
 
 import (
+	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor/retention"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
 )
@@ -13,6 +14,66 @@ type DeleteRequest struct {
 	Status       DeleteRequestStatus `json:"status"`
 	CreatedAt    model.Time          `json:"created_at"`
 
-	UserID   string              `json:"-"`
-	Matchers [][]*labels.Matcher `json:"-"`
+	UserID   string            `json:"-"`
+	matchers []*labels.Matcher `json:"-"`
+}
+
+func (d *DeleteRequest) AddLogQL(logQL string) error {
+	d.LogQLRequest = logQL
+	matchers, err := parseLogQLExpressionForDeletion(logQL)
+	if err != nil {
+		return err
+	}
+	d.matchers = matchers
+	return nil
+}
+
+func (d *DeleteRequest) IsDeleted(entry retention.ChunkEntry) (bool, []model.Interval) {
+	if d.UserID != unsafeGetString(entry.UserID) {
+		return false, nil
+	}
+
+	if !intervalsOverlap(model.Interval{
+		Start: entry.From,
+		End:   entry.Through,
+	}, model.Interval{
+		Start: d.StartTime,
+		End:   d.EndTime,
+	}) {
+		return false, nil
+	}
+
+	if !labels.Selector(d.matchers).Matches(entry.Labels) {
+		return false, nil
+	}
+
+	if d.StartTime <= entry.From && d.EndTime >= entry.Through {
+		return true, nil
+	}
+
+	intervals := make([]model.Interval, 0, 2)
+
+	if d.StartTime > entry.From {
+		intervals = append(intervals, model.Interval{
+			Start: entry.From,
+			End:   d.StartTime - 1,
+		})
+	}
+
+	if d.EndTime < entry.Through {
+		intervals = append(intervals, model.Interval{
+			Start: d.EndTime + 1,
+			End:   entry.Through,
+		})
+	}
+
+	return true, intervals
+}
+
+func intervalsOverlap(interval1, interval2 model.Interval) bool {
+	if interval1.Start > interval2.End || interval2.Start > interval1.End {
+		return false
+	}
+
+	return true
 }

pkg/storage/stores/shipper/compactor/deletion/delete_requests_manager.go

@@ -14,6 +14,11 @@ import (
 	util_log "github.com/grafana/loki/pkg/util/log"
 )
 
+const (
+	statusSuccess = "success"
+	statusFail    = "fail"
+)
+
 type DeleteRequestsManager struct {
 	deleteRequestsStore       DeleteRequestsStore
 	deleteRequestCancelPeriod time.Duration

pkg/storage/stores/shipper/compactor/deletion/delete_requests_manager_test.go

@@ -26,7 +26,7 @@ func (m mockDeleteRequestsStore) UpdateStatus(ctx context.Context, userID, reque
 	return nil
 }
 
-func (m mockDeleteRequestsStore) AddDeleteRequest(ctx context.Context, userID string, startTime, endTime model.Time, selectors []string) error {
+func (m mockDeleteRequestsStore) AddDeleteRequest(ctx context.Context, userID string, startTime, endTime model.Time, LogQLRequest string) error {
 	panic("implement me")
 }
 
@@ -89,10 +89,10 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "no relevant delete requests",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    "different-user",
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-24 * time.Hour),
-					EndTime:   now,
+					UserID:       "different-user",
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-24 * time.Hour),
+					EndTime:      now,
 				},
 			},
 			expectedResp: resp{
@@ -104,10 +104,10 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "whole chunk deleted by single request",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-24 * time.Hour),
-					EndTime:   now,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-24 * time.Hour),
+					EndTime:      now,
 				},
 			},
 			expectedResp: resp{
@@ -119,10 +119,10 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "deleted interval out of range",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-48 * time.Hour),
-					EndTime:   now.Add(-24 * time.Hour),
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-48 * time.Hour),
+					EndTime:      now.Add(-24 * time.Hour),
 				},
 			},
 			expectedResp: resp{
@@ -134,16 +134,16 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "multiple delete requests with one deleting the whole chunk",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-48 * time.Hour),
-					EndTime:   now.Add(-24 * time.Hour),
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-48 * time.Hour),
+					EndTime:      now.Add(-24 * time.Hour),
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-12 * time.Hour),
-					EndTime:   now,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-12 * time.Hour),
+					EndTime:      now,
 				},
 			},
 			expectedResp: resp{
@@ -155,28 +155,28 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "multiple delete requests causing multiple holes",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-13 * time.Hour),
-					EndTime:   now.Add(-11 * time.Hour),
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-13 * time.Hour),
+					EndTime:      now.Add(-11 * time.Hour),
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-10 * time.Hour),
-					EndTime:   now.Add(-8 * time.Hour),
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-10 * time.Hour),
+					EndTime:      now.Add(-8 * time.Hour),
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-6 * time.Hour),
-					EndTime:   now.Add(-5 * time.Hour),
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-6 * time.Hour),
+					EndTime:      now.Add(-5 * time.Hour),
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-2 * time.Hour),
-					EndTime:   now,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-2 * time.Hour),
+					EndTime:      now,
 				},
 			},
 			expectedResp: resp{
@@ -201,16 +201,16 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "multiple overlapping requests deleting the whole chunk",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-13 * time.Hour),
-					EndTime:   now.Add(-6 * time.Hour),
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-13 * time.Hour),
+					EndTime:      now.Add(-6 * time.Hour),
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-8 * time.Hour),
-					EndTime:   now,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-8 * time.Hour),
+					EndTime:      now,
 				},
 			},
 			expectedResp: resp{
@@ -222,22 +222,22 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			name: "multiple non-overlapping requests deleting the whole chunk",
 			deleteRequestsFromStore: []DeleteRequest{
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-12 * time.Hour),
-					EndTime:   now.Add(-6*time.Hour) - 1,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-12 * time.Hour),
+					EndTime:      now.Add(-6*time.Hour) - 1,
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-6 * time.Hour),
-					EndTime:   now.Add(-4*time.Hour) - 1,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-6 * time.Hour),
+					EndTime:      now.Add(-4*time.Hour) - 1,
 				},
 				{
-					UserID:    testUserID,
-					Selectors: []string{lblFoo.String()},
-					StartTime: now.Add(-4 * time.Hour),
-					EndTime:   now,
+					UserID:       testUserID,
+					LogQLRequest: lblFoo.String(),
+					StartTime:    now.Add(-4 * time.Hour),
+					EndTime:      now,
 				},
 			},
 			expectedResp: resp{

pkg/storage/stores/shipper/compactor/deletion/delete_requests_store.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"time"
+	"unsafe"
 
 	"github.com/prometheus/common/model"
 
@@ -200,7 +201,7 @@ func (ds *deleteRequestsStore) queryDeleteRequests(ctx context.Context, deleteQu
 				return false
 			}
 
-			deleteRequest.LogQLRequest = string(itr.Value())
+			deleteRequest.AddLogQL(string(itr.Value()))
 			deleteRequests[i] = deleteRequest
 
 			return true
@@ -289,3 +290,8 @@ func splitUserIDAndRequestID(rangeValue string) (userID, requestID string) {
 
 	return
 }
+
+// unsafeGetString is like yolostring but with a meaningful name
+func unsafeGetString(buf []byte) string {
+	return *((*string)(unsafe.Pointer(&buf)))
+}

pkg/storage/stores/shipper/compactor/deletion/request_handler.go

@@ -50,7 +50,7 @@ func (dm *DeleteRequestHandler) AddDeleteRequestHandler(w http.ResponseWriter, r
 		return
 	}
 
-	err = checkLogQLExpressionForDeletion(logQLStatement)
+	_, err = parseLogQLExpressionForDeletion(logQLStatement)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return

pkg/storage/stores/shipper/compactor/deletion/validation.go

@@ -4,32 +4,24 @@ import (
 	"errors"
 
 	"github.com/grafana/loki/pkg/logql"
+	"github.com/prometheus/prometheus/model/labels"
 )
 
 var (
 	errInvalidLogQL     = errors.New("invalid LogQL expression")
 	errUnsupportedLogQL = errors.New("unsupported LogQL expression")
 )
 
-// checkLogQLExpressionForDeletion checks if the given logQL is valid for deletions
-func checkLogQLExpressionForDeletion(logQL string) error {
+// parseLogQLExpressionForDeletion checks if the given logQL is valid for deletions
+func parseLogQLExpressionForDeletion(logQL string) ([]*labels.Matcher, error) {
 	expr, err := logql.ParseExpr(logQL)
 	if err != nil {
-		return errInvalidLogQL
+		return nil, errInvalidLogQL
 	}
 
-	if _, ok := expr.(*logql.MatchersExpr); ok {
-		return nil
-	}
-	if pipelineExpr, ok := expr.(*logql.PipelineExpr); ok {
-		// Only support line filters for now
-		for _, stage := range pipelineExpr.MultiStages {
-			if _, ok := stage.(*logql.LineFilterExpr); !ok {
-				return errUnsupportedLogQL
-			}
-		}
-		return nil
+	if matchersExpr, ok := expr.(*logql.MatchersExpr); ok {
+		return matchersExpr.Matchers(), nil
 	}
 
-	return errUnsupportedLogQL
+	return nil, errUnsupportedLogQL
 }

pkg/storage/stores/shipper/compactor/deletion/validation_test.go

@@ -6,29 +6,34 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestInvalidLogQLExpressionForDeletion(t *testing.T) {
+func TestParseLogQLExpressionForDeletion(t *testing.T) {
 	t.Run("invalid logql", func(t *testing.T) {
-		err := checkLogQLExpressionForDeletion("gjgjg ggj")
+		matchers, err := parseLogQLExpressionForDeletion("gjgjg ggj")
+		require.Nil(t, matchers)
 		require.ErrorIs(t, err, errInvalidLogQL)
 	})
 
 	t.Run("matcher expression", func(t *testing.T) {
-		err := checkLogQLExpressionForDeletion(`{env="dev", secret="true"}`)
+		matchers, err := parseLogQLExpressionForDeletion(`{env="dev", secret="true"}`)
+		require.NotNil(t, matchers)
 		require.NoError(t, err)
 	})
 
 	t.Run("pipeline expression with line filter", func(t *testing.T) {
-		err := checkLogQLExpressionForDeletion(`{env="dev", secret="true"} |= "social sec number"`)
-		require.NoError(t, err)
+		matchers, err := parseLogQLExpressionForDeletion(`{env="dev", secret="true"} |= "social sec number"`)
+		require.Nil(t, matchers)
+		require.ErrorIs(t, err, errUnsupportedLogQL)
 	})
 
 	t.Run("pipeline expression with label filter ", func(t *testing.T) {
-		err := checkLogQLExpressionForDeletion(`{env="dev", secret="true"} | json bob="top.params[0]"`)
+		matchers, err := parseLogQLExpressionForDeletion(`{env="dev", secret="true"} | json bob="top.params[0]"`)
+		require.Nil(t, matchers)
 		require.ErrorIs(t, err, errUnsupportedLogQL)
 	})
 
-	t.Run("pipeline expression with ", func(t *testing.T) {
-		err := checkLogQLExpressionForDeletion(`count_over_time({job="mysql"}[5m])`)
+	t.Run("metrics query", func(t *testing.T) {
+		matchers, err := parseLogQLExpressionForDeletion(`count_over_time({job="mysql"}[5m])`)
+		require.Nil(t, matchers)
 		require.ErrorIs(t, err, errUnsupportedLogQL)
 	})
 }