feat: when using sqlite for storing delete requests, use the stored completion time to reduce the requests considered for querytime filtering (#16600)

Author: sandeepsukhani

pkg/compactor/client/grpc.go

@@ -67,7 +67,7 @@ func (s *compactorGRPCClient) Stop() {
 
 func (s *compactorGRPCClient) GetAllDeleteRequestsForUser(ctx context.Context, userID string) ([]deletion.DeleteRequest, error) {
 	ctx = user.InjectOrgID(ctx, userID)
-	grpcResp, err := s.grpcClient.GetDeleteRequests(ctx, &deletion_grpc.GetDeleteRequestsRequest{})
+	grpcResp, err := s.grpcClient.GetDeleteRequests(ctx, &deletion_grpc.GetDeleteRequestsRequest{ForQuerytimeFiltering: true})
 	if err != nil {
 		return nil, err
 	}

pkg/compactor/client/grpc/grpc.pb.go

@@ -7,7 +7,9 @@ service Compactor {
   rpc GetCacheGenNumbers(GetCacheGenNumbersRequest) returns (GetCacheGenNumbersResponse);
 }
 
-message GetDeleteRequestsRequest {}
+message GetDeleteRequestsRequest {
+  bool forQuerytimeFiltering = 1;
+}
 
 message GetDeleteRequestsResponse {
   repeated DeleteRequest deleteRequests = 1;

pkg/compactor/client/grpc/grpc.proto

@@ -51,7 +51,11 @@ func NewHTTPClient(addr string, cfg HTTPConfig) (deletion.CompactorClient, error
 		level.Error(log.Logger).Log("msg", "error parsing url", "err", err)
 		return nil, err
 	}
+
 	u.Path = getDeletePath
+	q := u.Query()
+	q.Set(deletion.ForQuerytimeFilteringQueryParam, "true")
+	u.RawQuery = q.Encode()
 	deleteRequestsURL := u.String()
 
 	u.Path = cacheGenNumPath

pkg/compactor/client/http.go

@@ -210,7 +210,16 @@ type Limits interface {
 	DefaultLimits() *validation.Limits
 }
 
-func NewCompactor(cfg Config, objectStoreClients map[config.DayTime]client.ObjectClient, deleteStoreClient client.ObjectClient, schemaConfig config.SchemaConfig, limits Limits, r prometheus.Registerer, metricsNamespace string) (*Compactor, error) {
+func NewCompactor(
+	cfg Config,
+	objectStoreClients map[config.DayTime]client.ObjectClient,
+	deleteStoreClient client.ObjectClient,
+	schemaConfig config.SchemaConfig,
+	limits Limits,
+	indexUpdatePropagationMaxDelay time.Duration,
+	r prometheus.Registerer,
+	metricsNamespace string,
+) (*Compactor, error) {
 	retentionEnabledStats.Set("false")
 	if cfg.RetentionEnabled {
 		retentionEnabledStats.Set("true")
@@ -267,15 +276,22 @@ func NewCompactor(cfg Config, objectStoreClients map[config.DayTime]client.Objec
 	compactor.subservicesWatcher = services.NewFailureWatcher()
 	compactor.subservicesWatcher.WatchManager(compactor.subservices)
 
-	if err := compactor.init(objectStoreClients, deleteStoreClient, schemaConfig, limits, r); err != nil {
+	if err := compactor.init(objectStoreClients, deleteStoreClient, schemaConfig, indexUpdatePropagationMaxDelay, limits, r); err != nil {
 		return nil, fmt.Errorf("init compactor: %w", err)
 	}
 
 	compactor.Service = services.NewBasicService(compactor.starting, compactor.loop, compactor.stopping)
 	return compactor, nil
 }
 
-func (c *Compactor) init(objectStoreClients map[config.DayTime]client.ObjectClient, deleteStoreClient client.ObjectClient, schemaConfig config.SchemaConfig, limits Limits, r prometheus.Registerer) error {
+func (c *Compactor) init(
+	objectStoreClients map[config.DayTime]client.ObjectClient,
+	deleteStoreClient client.ObjectClient,
+	schemaConfig config.SchemaConfig,
+	indexUpdatePropagationMaxDelay time.Duration,
+	limits Limits,
+	r prometheus.Registerer,
+) error {
 	err := chunk_util.EnsureDirectory(c.cfg.WorkingDirectory)
 	if err != nil {
 		return err
@@ -286,7 +302,7 @@ func (c *Compactor) init(objectStoreClients map[config.DayTime]client.ObjectClie
 			return fmt.Errorf("delete store client not initialised when retention is enabled")
 		}
 
-		if err := c.initDeletes(deleteStoreClient, r, limits); err != nil {
+		if err := c.initDeletes(deleteStoreClient, indexUpdatePropagationMaxDelay, r, limits); err != nil {
 			return fmt.Errorf("failed to init delete store: %w", err)
 		}
 	}
@@ -365,10 +381,16 @@ func (c *Compactor) init(objectStoreClients map[config.DayTime]client.ObjectClie
 	return nil
 }
 
-func (c *Compactor) initDeletes(objectClient client.ObjectClient, r prometheus.Registerer, limits Limits) error {
+func (c *Compactor) initDeletes(objectClient client.ObjectClient, indexUpdatePropagationMaxDelay time.Duration, r prometheus.Registerer, limits Limits) error {
 	deletionWorkDir := filepath.Join(c.cfg.WorkingDirectory, "deletion")
 	indexStorageClient := storage.NewIndexStorageClient(objectClient, c.cfg.DeleteRequestStoreKeyPrefix)
-	store, err := deletion.NewDeleteRequestsStore(deletion.DeleteRequestsStoreDBType(c.cfg.DeleteRequestStoreDBType), deletionWorkDir, indexStorageClient, deletion.DeleteRequestsStoreDBType(c.cfg.BackupDeleteRequestStoreDBType))
+	store, err := deletion.NewDeleteRequestsStore(
+		deletion.DeleteRequestsStoreDBType(c.cfg.DeleteRequestStoreDBType),
+		deletionWorkDir,
+		indexStorageClient,
+		deletion.DeleteRequestsStoreDBType(c.cfg.BackupDeleteRequestStoreDBType),
+		indexUpdatePropagationMaxDelay,
+	)
 	if err != nil {
 		return err
 	}

pkg/compactor/compactor.go

@@ -63,7 +63,7 @@ func setupTestCompactor(t *testing.T, objectClients map[config.DayTime]client.Ob
 
 	c, err := NewCompactor(cfg, objectClients, objectClients[periodConfigs[len(periodConfigs)-1].From], config.SchemaConfig{
 		Configs: periodConfigs,
-	}, overrides, prometheus.NewPedanticRegistry(), constants.Loki)
+	}, overrides, 0, prometheus.NewPedanticRegistry(), constants.Loki)
 	require.NoError(t, err)
 
 	c.RegisterIndexCompactor("dummy", testIndexCompactor{})

pkg/compactor/compactor_test.go

@@ -137,3 +137,15 @@ func (c *deleteRequestsClient) currentUserIDs() []string {
 
 	return userIDs
 }
+
+func NewNoOpDeleteRequestsClient() DeleteRequestsClient {
+	return &noOpDeleteRequestsClient{}
+}
+
+type noOpDeleteRequestsClient struct{}
+
+func (n noOpDeleteRequestsClient) GetAllDeleteRequestsForUser(_ context.Context, _ string) ([]DeleteRequest, error) {
+	return nil, nil
+}
+
+func (n noOpDeleteRequestsClient) Stop() {}

pkg/compactor/deletion/delete_requests_client.go

@@ -1040,9 +1040,10 @@ type mockDeleteRequestsStore struct {
 	getResult []DeleteRequest
 	getErr    error
 
-	getAllUser   string
-	getAllResult []DeleteRequest
-	getAllErr    error
+	getAllUser                           string
+	getAllResult                         []DeleteRequest
+	getAllErr                            error
+	getAllRequestedForQuerytimeFiltering bool
 
 	genNumber string
 }
@@ -1093,8 +1094,9 @@ func (m *mockDeleteRequestsStore) GetDeleteRequest(_ context.Context, userID, re
 	return m.getResult[0], m.getErr
 }
 
-func (m *mockDeleteRequestsStore) GetAllDeleteRequestsForUser(_ context.Context, userID string) ([]DeleteRequest, error) {
+func (m *mockDeleteRequestsStore) GetAllDeleteRequestsForUser(_ context.Context, userID string, forQuerytimeFiltering bool) ([]DeleteRequest, error) {
 	m.getAllUser = userID
+	m.getAllRequestedForQuerytimeFiltering = forQuerytimeFiltering
 	return m.getAllResult, m.getAllErr
 }