Only delete data when WholeStreamDeletion or FilterAndDelete (#6286) (#…

…6292) * Only delete data when WholeStreamDeletion or FilterAndDelete * remove broken snyn workflow (cherry picked from commit 1c5e094) Co-authored-by: Travis Patterson <travis.patterson@grafana.com>
grafana · Jun 2, 2022 · f48ec39 · f48ec39
1 parent 8f179f7
commit f48ec39
Show file tree

Hide file tree

Showing 2 changed files with 91 additions and 9 deletions.
diff --git a/pkg/storage/stores/shipper/compactor/deletion/delete_requests_manager.go b/pkg/storage/stores/shipper/compactor/deletion/delete_requests_manager.go
@@ -133,6 +133,11 @@ func (d *DeleteRequestsManager) Expired(ref retention.ChunkEntry, _ model.Time)
 		return false, nil
 	}
 
+	if d.deletionMode == Disabled || d.deletionMode == FilterOnly {
+		// Don't process deletes
+		return false, nil
+	}
+
 	d.chunkIntervalsToRetain = d.chunkIntervalsToRetain[:0]
 	d.chunkIntervalsToRetain = append(d.chunkIntervalsToRetain, retention.IntervalFilter{
 		Interval: model.Interval{

diff --git a/pkg/storage/stores/shipper/compactor/deletion/delete_requests_manager_test.go b/pkg/storage/stores/shipper/compactor/deletion/delete_requests_manager_test.go
@@ -35,18 +35,21 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 
 	for _, tc := range []struct {
 		name                    string
+		deletionMode            Mode
 		deleteRequestsFromStore []DeleteRequest
 		expectedResp            resp
 	}{
 		{
-			name: "no delete requests",
+			name:         "no delete requests",
+			deletionMode: WholeStreamDeletion,
 			expectedResp: resp{
 				isExpired:           false,
 				nonDeletedIntervals: nil,
 			},
 		},
 		{
-			name: "no relevant delete requests",
+			name:         "no relevant delete requests",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    "different-user",
@@ -61,7 +64,8 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			},
 		},
 		{
-			name: "whole chunk deleted by single request",
+			name:         "whole chunk deleted by single request",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    testUserID,
@@ -76,7 +80,8 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			},
 		},
 		{
-			name: "deleted interval out of range",
+			name:         "deleted interval out of range",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    testUserID,
@@ -91,7 +96,8 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			},
 		},
 		{
-			name: "multiple delete requests with one deleting the whole chunk",
+			name:         "multiple delete requests with one deleting the whole chunk",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    testUserID,
@@ -112,7 +118,8 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			},
 		},
 		{
-			name: "multiple delete requests causing multiple holes",
+			name:         "multiple delete requests causing multiple holes",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    testUserID,
@@ -164,7 +171,8 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			},
 		},
 		{
-			name: "multiple overlapping requests deleting the whole chunk",
+			name:         "multiple overlapping requests deleting the whole chunk",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    testUserID,
@@ -185,7 +193,8 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 			},
 		},
 		{
-			name: "multiple non-overlapping requests deleting the whole chunk",
+			name:         "multiple non-overlapping requests deleting the whole chunk",
+			deletionMode: WholeStreamDeletion,
 			deleteRequestsFromStore: []DeleteRequest{
 				{
 					UserID:    testUserID,
@@ -211,9 +220,77 @@ func TestDeleteRequestsManager_Expired(t *testing.T) {
 				nonDeletedIntervals: nil,
 			},
 		},
+		{
+			name:         "deletes are disabled",
+			deletionMode: Disabled,
+			deleteRequestsFromStore: []DeleteRequest{
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-13 * time.Hour),
+					EndTime:   now.Add(-11 * time.Hour),
+				},
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-10 * time.Hour),
+					EndTime:   now.Add(-8 * time.Hour),
+				},
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-6 * time.Hour),
+					EndTime:   now.Add(-5 * time.Hour),
+				},
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-2 * time.Hour),
+					EndTime:   now,
+				},
+			},
+			expectedResp: resp{
+				isExpired:           false,
+				nonDeletedIntervals: nil,
+			},
+		},
+		{
+			name:         "deletes are `filter-only`",
+			deletionMode: FilterOnly,
+			deleteRequestsFromStore: []DeleteRequest{
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-13 * time.Hour),
+					EndTime:   now.Add(-11 * time.Hour),
+				},
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-10 * time.Hour),
+					EndTime:   now.Add(-8 * time.Hour),
+				},
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-6 * time.Hour),
+					EndTime:   now.Add(-5 * time.Hour),
+				},
+				{
+					UserID:    testUserID,
+					Query:     lblFoo.String(),
+					StartTime: now.Add(-2 * time.Hour),
+					EndTime:   now,
+				},
+			},
+			expectedResp: resp{
+				isExpired:           false,
+				nonDeletedIntervals: nil,
+			},
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			mgr := NewDeleteRequestsManager(mockDeleteRequestsStore{deleteRequests: tc.deleteRequestsFromStore}, time.Hour, nil, WholeStreamDeletion)
+			mgr := NewDeleteRequestsManager(mockDeleteRequestsStore{deleteRequests: tc.deleteRequestsFromStore}, time.Hour, nil, tc.deletionMode)
 			require.NoError(t, mgr.loadDeleteRequestsToProcess())
 
 			isExpired, nonDeletedIntervals := mgr.Expired(chunkEntry, model.Now())