Add a DeletionMode config variable

CHANGELOG.md

@@ -12,6 +12,7 @@
 * [5536](https://github.com/grafana/loki/pull/5536) **jiachengxu**: Loki mixin: make labelsSelector in loki chunks dashboards configurable
 * [5535](https://github.com/grafana/loki/pull/5535) **jiachengxu**: Loki mixins: use labels selector for loki chunks dashboard
 * [5507](https://github.com/grafana/loki/pull/5507) **MichelHollands**: Remove extra param in call for inflightRequests metric.
+* [5481](https://github.com/grafana/loki/pull/5481) **MichelHollands**: Add a DeletionMode config variable to specify the delete mode and validate match parameters.
 * [5356](https://github.com/grafana/loki/pull/5356) **jbschami**: Enhance lambda-promtail to support adding extra labels from an environment variable value
 * [5409](https://github.com/grafana/loki/pull/5409) **ldb**: Enable best effort parsing for Syslog messages
 * [5392](https://github.com/grafana/loki/pull/5392) **MichelHollands**: Etcd credentials are parsed as secrets instead of plain text now.

pkg/loki/modules.go

@@ -12,8 +12,6 @@ import (
 
 	"github.com/grafana/dskit/tenant"
 
-	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor/deletion"
-
 	"github.com/NYTimes/gziphandler"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -52,6 +50,7 @@ import (
 	chunk_util "github.com/grafana/loki/pkg/storage/chunk/util"
 	"github.com/grafana/loki/pkg/storage/stores/shipper"
 	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor"
+	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor/deletion"
 	"github.com/grafana/loki/pkg/storage/stores/shipper/indexgateway"
 	"github.com/grafana/loki/pkg/storage/stores/shipper/indexgateway/indexgatewaypb"
 	"github.com/grafana/loki/pkg/storage/stores/shipper/uploads"
@@ -749,10 +748,12 @@ func (t *Loki) initCompactor() (services.Service, error) {
 	}
 
 	t.Server.HTTP.Path("/compactor/ring").Methods("GET", "POST").Handler(t.compactor)
-	if t.Cfg.CompactorConfig.RetentionEnabled {
-		t.Server.HTTP.Path("/loki/api/admin/delete").Methods("PUT", "POST").Handler(t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.compactor.DeleteRequestsHandler.AddDeleteRequestHandler)))
-		t.Server.HTTP.Path("/loki/api/admin/delete").Methods("GET").Handler(t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.compactor.DeleteRequestsHandler.GetAllDeleteRequestsHandler)))
-		t.Server.HTTP.Path("/loki/api/admin/cancel_delete_request").Methods("PUT", "POST").Handler(t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.compactor.DeleteRequestsHandler.CancelDeleteRequestHandler)))
+
+	// TODO: update this when the other deletion modes are available
+	if t.Cfg.CompactorConfig.RetentionEnabled && t.compactor.DeleteMode() == deletion.WholeStreamDeletion {
+		t.Server.HTTP.Path("/loki/api/v1/delete").Methods("PUT", "POST").Handler(t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.compactor.DeleteRequestsHandler.AddDeleteRequestHandler)))
+		t.Server.HTTP.Path("/loki/api/v1/delete").Methods("GET").Handler(t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.compactor.DeleteRequestsHandler.GetAllDeleteRequestsHandler)))
+		t.Server.HTTP.Path("/loki/api/v1/delete").Methods("DELETE").Handler(t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.compactor.DeleteRequestsHandler.CancelDeleteRequestHandler)))
 	}
 
 	return t.compactor, nil

pkg/querier/querier.go

@@ -226,13 +226,11 @@ func (q *SingleTenantQuerier) deletesForUser(ctx context.Context, startT, endT t
 	var deletes []*logproto.Delete
 	for _, del := range d {
 		if int64(del.StartTime) <= end && int64(del.EndTime) >= start {
-			for _, selector := range del.Selectors {
-				deletes = append(deletes, &logproto.Delete{
-					Selector: selector,
-					Start:    int64(del.StartTime),
-					End:      int64(del.EndTime),
-				})
-			}
+			deletes = append(deletes, &logproto.Delete{
+				Selector: del.Query,
+				Start:    int64(del.StartTime),
+				End:      int64(del.EndTime),
+			})
 		}
 	}
 

pkg/querier/querier_test.go

@@ -740,11 +740,11 @@ func TestQuerier_SelectLogWithDeletes(t *testing.T) {
 
 	delGetter := &mockDeleteGettter{
 		results: []deletion.DeleteRequest{
-			{Selectors: []string{`0`}, StartTime: 0, EndTime: 100},
-			{Selectors: []string{`1`}, StartTime: 200, EndTime: 400},
-			{Selectors: []string{`2`}, StartTime: 400, EndTime: 500},
-			{Selectors: []string{`3`}, StartTime: 500, EndTime: 700},
-			{Selectors: []string{`4`}, StartTime: 700, EndTime: 900},
+			{Query: `0`, StartTime: 0, EndTime: 100},
+			{Query: `1`, StartTime: 200, EndTime: 400},
+			{Query: `2`, StartTime: 400, EndTime: 500},
+			{Query: `3`, StartTime: 500, EndTime: 700},
+			{Query: `4`, StartTime: 700, EndTime: 900},
 		},
 	}
 
@@ -802,11 +802,11 @@ func TestQuerier_SelectSamplesWithDeletes(t *testing.T) {
 
 	delGetter := &mockDeleteGettter{
 		results: []deletion.DeleteRequest{
-			{Selectors: []string{`0`}, StartTime: 0, EndTime: 100},
-			{Selectors: []string{`1`}, StartTime: 200, EndTime: 400},
-			{Selectors: []string{`2`}, StartTime: 400, EndTime: 500},
-			{Selectors: []string{`3`}, StartTime: 500, EndTime: 700},
-			{Selectors: []string{`4`}, StartTime: 700, EndTime: 900},
+			{Query: `0`, StartTime: 0, EndTime: 100},
+			{Query: `1`, StartTime: 200, EndTime: 400},
+			{Query: `2`, StartTime: 400, EndTime: 500},
+			{Query: `3`, StartTime: 500, EndTime: 700},
+			{Query: `4`, StartTime: 700, EndTime: 900},
 		},
 	}
 

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

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 
@@ -67,6 +68,7 @@ type Config struct {
 	RetentionEnabled          bool            `yaml:"retention_enabled"`
 	RetentionDeleteDelay      time.Duration   `yaml:"retention_delete_delay"`
 	RetentionDeleteWorkCount  int             `yaml:"retention_delete_worker_count"`
+	DeletionMode              string          `yaml:"deletion_enabled"`
 	DeleteRequestCancelPeriod time.Duration   `yaml:"delete_request_cancel_period"`
 	MaxCompactionParallelism  int             `yaml:"max_compaction_parallelism"`
 	CompactorRing             util.RingConfig `yaml:"compactor_ring,omitempty"`
@@ -84,6 +86,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.RetentionDeleteWorkCount, "boltdb.shipper.compactor.retention-delete-worker-count", 150, "The total amount of worker to use to delete chunks.")
 	f.DurationVar(&cfg.DeleteRequestCancelPeriod, "boltdb.shipper.compactor.delete-request-cancel-period", 24*time.Hour, "Allow cancellation of delete request until duration after they are created. Data would be deleted only after delete requests have been older than this duration. Ideally this should be set to at least 24h.")
 	f.IntVar(&cfg.MaxCompactionParallelism, "boltdb.shipper.compactor.max-compaction-parallelism", 1, "Maximum number of tables to compact in parallel. While increasing this value, please make sure compactor has enough disk space allocated to be able to store and compact as many tables.")
+	f.StringVar(&cfg.DeletionMode, "boltdb.shipper.compactor.deletion-mode", "whole-stream-deletion", fmt.Sprintf("(Experimental) Deletion mode. Can be one of %v", strings.Join(deletion.AllModes(), "|")))
 	cfg.CompactorRing.RegisterFlagsWithPrefix("boltdb.shipper.compactor.", "collectors/", f)
 }
 
@@ -96,6 +99,10 @@ func (cfg *Config) Validate() error {
 		return errors.New("interval for applying retention should either be set to a 0 or a multiple of compaction interval")
 	}
 
+	if _, err := deletion.ParseMode(cfg.DeletionMode); err != nil {
+		return err
+	}
+
 	return shipper_util.ValidateSharedStoreKeyPrefix(cfg.SharedStoreKeyPrefix)
 }
 
@@ -113,6 +120,7 @@ type Compactor struct {
 	metrics               *metrics
 	running               bool
 	wg                    sync.WaitGroup
+	deleteMode            deletion.Mode
 
 	// Ring used for running a single compactor
 	ringLifecycler *ring.BasicLifecycler
@@ -180,6 +188,12 @@ func NewCompactor(cfg Config, storageConfig storage.Config, schemaConfig loki_st
 	compactor.subservicesWatcher = services.NewFailureWatcher()
 	compactor.subservicesWatcher.WatchManager(compactor.subservices)
 
+	mode, err := deletion.ParseMode(cfg.DeletionMode)
+	if err != nil {
+		return nil, err
+	}
+	compactor.deleteMode = mode
+
 	if err := compactor.init(storageConfig, schemaConfig, limits, clientMetrics, r); err != nil {
 		return nil, err
 	}
@@ -215,18 +229,24 @@ func (c *Compactor) init(storageConfig storage.Config, schemaConfig loki_storage
 			return err
 		}
 
-		deletionWorkDir := filepath.Join(c.cfg.WorkingDirectory, "deletion")
+		if c.deleteMode == deletion.WholeStreamDeletion {
+			deletionWorkDir := filepath.Join(c.cfg.WorkingDirectory, "deletion")
 
-		c.deleteRequestsStore, err = deletion.NewDeleteStore(deletionWorkDir, c.indexStorageClient)
-		if err != nil {
-			return err
+			c.deleteRequestsStore, err = deletion.NewDeleteStore(deletionWorkDir, c.indexStorageClient)
+			if err != nil {
+				return err
+			}
+			c.DeleteRequestsHandler = deletion.NewDeleteRequestHandler(c.deleteRequestsStore, time.Hour, r)
+			c.deleteRequestsManager = deletion.NewDeleteRequestsManager(c.deleteRequestsStore, c.cfg.DeleteRequestCancelPeriod, r)
+			c.expirationChecker = newExpirationChecker(retention.NewExpirationChecker(limits), c.deleteRequestsManager)
+		} else {
+			c.expirationChecker = newExpirationChecker(
+				retention.NewExpirationChecker(limits),
+				// This is a dummy deletion ExpirationChecker that never expires anything
+				retention.NeverExpiringExpirationChecker(limits),
+			)
 		}
 
-		c.DeleteRequestsHandler = deletion.NewDeleteRequestHandler(c.deleteRequestsStore, time.Hour, r)
-		c.deleteRequestsManager = deletion.NewDeleteRequestsManager(c.deleteRequestsStore, c.cfg.DeleteRequestCancelPeriod, r)
-
-		c.expirationChecker = newExpirationChecker(retention.NewExpirationChecker(limits), c.deleteRequestsManager)
-
 		c.tableMarker, err = retention.NewMarker(retentionWorkDir, schemaConfig, c.expirationChecker, chunkClient, r)
 		if err != nil {
 			return err
@@ -286,8 +306,12 @@ func (c *Compactor) starting(ctx context.Context) (err error) {
 
 func (c *Compactor) loop(ctx context.Context) error {
 	if c.cfg.RetentionEnabled {
-		defer c.deleteRequestsStore.Stop()
-		defer c.deleteRequestsManager.Stop()
+		if c.deleteRequestsStore != nil {
+			defer c.deleteRequestsStore.Stop()
+		}
+		if c.deleteRequestsManager != nil {
+			defer c.deleteRequestsManager.Stop()
+		}
 	}
 
 	syncTicker := time.NewTicker(c.ringPollPeriod)
@@ -530,6 +554,10 @@ func (c *Compactor) RunCompaction(ctx context.Context, applyRetention bool) erro
 	return firstErr
 }
 
+func (c *Compactor) DeleteMode() deletion.Mode {
+	return c.deleteMode
+}
+
 type expirationChecker struct {
 	retentionExpiryChecker retention.ExpirationChecker
 	deletionExpiryChecker  retention.ExpirationChecker

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

@@ -3,22 +3,30 @@ package deletion
 import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
-	"github.com/prometheus/prometheus/promql/parser"
 
 	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor/retention"
 )
 
-// DeleteRequest holds all the details about a delete request.
 type DeleteRequest struct {
 	RequestID string              `json:"request_id"`
 	StartTime model.Time          `json:"start_time"`
 	EndTime   model.Time          `json:"end_time"`
-	Selectors []string            `json:"selectors"`
+	Query     string              `json:"query"`
 	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) AddQuery(logQL string) error {
+	d.Query = logQL
+	matchers, err := parseDeletionQuery(logQL)
+	if err != nil {
+		return err
+	}
+	d.matchers = matchers
+	return nil
 }
 
 func (d *DeleteRequest) IsDeleted(entry retention.ChunkEntry) (bool, []model.Interval) {
@@ -36,26 +44,7 @@ func (d *DeleteRequest) IsDeleted(entry retention.ChunkEntry) (bool, []model.Int
 		return false, nil
 	}
 
-	matchers := make([][]*labels.Matcher, len(d.Selectors))
-	var err error
-
-	for j, selector := range d.Selectors {
-		matchers[j], err = parser.ParseMetricSelector(selector)
-
-		if err != nil {
-			return false, nil
-		}
-	}
-
-	matches := false
-	for _, matchers := range matchers {
-		if labels.Selector(matchers).Matches(entry.Labels) {
-			matches = true
-			break
-		}
-	}
-
-	if !matches {
+	if !labels.Selector(d.matchers).Matches(entry.Labels) {
 		return false, nil
 	}
 

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

@@ -16,15 +16,15 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 	now := model.Now()
 	user1 := "user1"
 
-	lbls := `{foo="bar", fizz="buzz"}`
+	lbl := `{foo="bar", fizz="buzz"}`
 
 	chunkEntry := retention.ChunkEntry{
 		ChunkRef: retention.ChunkRef{
 			UserID:  []byte(user1),
 			From:    now.Add(-3 * time.Hour),
 			Through: now.Add(-time.Hour),
 		},
-		Labels: mustParseLabel(lbls),
+		Labels: mustParseLabel(lbl),
 	}
 
 	type resp struct {
@@ -43,7 +43,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    user1,
 				StartTime: now.Add(-3 * time.Hour),
 				EndTime:   now.Add(-time.Hour),
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted:           true,
@@ -56,7 +56,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    user1,
 				StartTime: now.Add(-3 * time.Hour),
 				EndTime:   now.Add(-2 * time.Hour),
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted: true,
@@ -74,7 +74,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    user1,
 				StartTime: now.Add(-2 * time.Hour),
 				EndTime:   now,
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted: true,
@@ -92,7 +92,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    user1,
 				StartTime: now.Add(-2 * time.Hour),
 				EndTime:   now,
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted: true,
@@ -110,7 +110,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    user1,
 				StartTime: now.Add(-(2*time.Hour + 30*time.Minute)),
 				EndTime:   now.Add(-(time.Hour + 30*time.Minute)),
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted: true,
@@ -132,7 +132,7 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    user1,
 				StartTime: now.Add(-12 * time.Hour),
 				EndTime:   now.Add(-10 * time.Hour),
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted: false,
@@ -141,10 +141,10 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 		{
 			name: "request not matching due to matchers",
 			deleteRequest: DeleteRequest{
-				UserID:    user1,
+				UserID:    "user1",
 				StartTime: now.Add(-3 * time.Hour),
 				EndTime:   now.Add(-time.Hour),
-				Selectors: []string{`{foo1="bar"}`, `{fizz1="buzz"}`},
+				Query:     `{foo1="bar"}`,
 			},
 			expectedResp: resp{
 				isDeleted: false,
@@ -156,14 +156,15 @@ func TestDeleteRequest_IsDeleted(t *testing.T) {
 				UserID:    "user2",
 				StartTime: now.Add(-3 * time.Hour),
 				EndTime:   now.Add(-time.Hour),
-				Selectors: []string{lbls},
+				Query:     lbl,
 			},
 			expectedResp: resp{
 				isDeleted: false,
 			},
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
+			require.NoError(t, tc.deleteRequest.AddQuery(tc.deleteRequest.Query))
 			isDeleted, nonDeletedIntervals := tc.deleteRequest.IsDeleted(chunkEntry)
 			require.Equal(t, tc.expectedResp.isDeleted, isDeleted)
 			require.Equal(t, tc.expectedResp.nonDeletedIntervals, nonDeletedIntervals)