support for cancellation of delete requests (#2555)

* support for cancellation of delete requests

Signed-off-by: Sandeep Sukhani <[email protected]>

* having deadline to cancel delete request instead of synchronizing with mutex

Signed-off-by: Sandeep Sukhani <[email protected]>

* changes suggested from PR review

Signed-off-by: Sandeep Sukhani <[email protected]>

Author: sandeepsukhani

pkg/api/api.go

@@ -177,18 +177,20 @@ func (a *API) RegisterIngester(i *ingester.Ingester, pushConfig distributor.Conf
 	a.RegisterRoute("/push", push.Handler(pushConfig, i.Push), true) // For testing and debugging.
 }
 
-// RegisterPurger registers the endpoints associated with the Purger/DeleteStore. They do not exacty
+// RegisterPurger registers the endpoints associated with the Purger/DeleteStore. They do not exactly
 // match the Prometheus API but mirror it closely enough to justify their routing under the Prometheus
 // component/
 func (a *API) RegisterPurger(store *purger.DeleteStore) {
 	deleteRequestHandler := purger.NewDeleteRequestHandler(store, prometheus.DefaultRegisterer)
 
 	a.RegisterRoute(a.cfg.PrometheusHTTPPrefix+"/api/v1/admin/tsdb/delete_series", http.HandlerFunc(deleteRequestHandler.AddDeleteRequestHandler), true, "PUT", "POST")
 	a.RegisterRoute(a.cfg.PrometheusHTTPPrefix+"/api/v1/admin/tsdb/delete_series", http.HandlerFunc(deleteRequestHandler.GetAllDeleteRequestsHandler), true, "GET")
+	a.RegisterRoute(a.cfg.PrometheusHTTPPrefix+"/api/v1/admin/tsdb/cancel_delete_request", http.HandlerFunc(deleteRequestHandler.CancelDeleteRequestHandler), true, "PUT", "POST")
 
 	// Legacy Routes
 	a.RegisterRoute(a.cfg.LegacyHTTPPrefix+"/api/v1/admin/tsdb/delete_series", http.HandlerFunc(deleteRequestHandler.AddDeleteRequestHandler), true, "PUT", "POST")
 	a.RegisterRoute(a.cfg.LegacyHTTPPrefix+"/api/v1/admin/tsdb/delete_series", http.HandlerFunc(deleteRequestHandler.GetAllDeleteRequestsHandler), true, "GET")
+	a.RegisterRoute(a.cfg.LegacyHTTPPrefix+"/api/v1/admin/tsdb/cancel_delete_request", http.HandlerFunc(deleteRequestHandler.CancelDeleteRequestHandler), true, "PUT", "POST")
 }
 
 // RegisterRuler registers routes associated with the Ruler service. If the

pkg/chunk/purger/delete_requests_store.go

@@ -293,6 +293,25 @@ func (ds *DeleteStore) queryCacheGenerationNumber(ctx context.Context, userID st
 	return genNumber, nil
 }
 
+// RemoveDeleteRequest removes a delete request and increments cache gen number
+func (ds *DeleteStore) RemoveDeleteRequest(ctx context.Context, userID, requestID string, createdAt, startTime, endTime model.Time) error {
+	userIDAndRequestID := fmt.Sprintf("%s:%s", userID, requestID)
+
+	writeBatch := ds.indexClient.NewWriteBatch()
+	writeBatch.Delete(ds.cfg.RequestsTableName, string(deleteRequestID), []byte(userIDAndRequestID))
+
+	// Add another entry with additional details like creation time, time range of delete request and selectors in value
+	rangeValue := fmt.Sprintf("%x:%x:%x", int64(createdAt), int64(startTime), int64(endTime))
+	writeBatch.Delete(ds.cfg.RequestsTableName, fmt.Sprintf("%s:%s", deleteRequestDetails, userIDAndRequestID),
+		[]byte(rangeValue))
+
+	// we need to invalidate results cache since removal of delete request would cause query results to change
+	writeBatch.Add(ds.cfg.RequestsTableName, fmt.Sprintf("%s:%s:%s", cacheGenNum, userID, CacheKindResults),
+		nil, []byte(strconv.FormatInt(time.Now().Unix(), 10)))
+
+	return ds.indexClient.BatchWrite(ctx, writeBatch)
+}
+
 func parseDeleteRequestTimestamps(rangeValue []byte, deleteRequest DeleteRequest) (DeleteRequest, error) {
 	hexParts := strings.Split(string(rangeValue), ":")
 	if len(hexParts) != 3 {

pkg/chunk/purger/purger.go

@@ -24,7 +24,10 @@ import (
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
-const millisecondPerDay = int64(24 * time.Hour / time.Millisecond)
+const (
+	millisecondPerDay                 = int64(24 * time.Hour / time.Millisecond)
+	deleteRequestCancellationDeadline = 24 * time.Hour
+)
 
 type purgerMetrics struct {
 	deleteRequestsProcessedTotal      *prometheus.CounterVec
@@ -335,7 +338,8 @@ func (dp *DataPurger) pullDeleteRequestsToPlanDeletes() error {
 	}
 
 	for _, deleteRequest := range deleteRequests {
-		if deleteRequest.CreatedAt.Add(24 * time.Hour).After(model.Now()) {
+		// adding an extra minute here to avoid a race between cancellation of request and picking of the request for processing
+		if deleteRequest.CreatedAt.Add(deleteRequestCancellationDeadline).Add(time.Minute).After(model.Now()) {
 			continue
 		}
 

pkg/chunk/purger/request_handler.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"net/http"
 
+	"github.com/go-kit/kit/log/level"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
@@ -102,6 +104,7 @@ func (dm *DeleteRequestHandler) AddDeleteRequestHandler(w http.ResponseWriter, r
 	}
 
 	if err := dm.deleteStore.AddDeleteRequest(ctx, userID, model.Time(startTime), model.Time(endTime), match); err != nil {
+		level.Error(util.Logger).Log("msg", "error adding delete request to the store", "err", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
@@ -120,11 +123,56 @@ func (dm *DeleteRequestHandler) GetAllDeleteRequestsHandler(w http.ResponseWrite
 
 	deleteRequests, err := dm.deleteStore.GetAllDeleteRequestsForUser(ctx, userID)
 	if err != nil {
+		level.Error(util.Logger).Log("msg", "error getting delete requests from the store", "err", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
 	if err := json.NewEncoder(w).Encode(deleteRequests); err != nil {
+		level.Error(util.Logger).Log("msg", "error marshalling response", "err", err)
 		http.Error(w, fmt.Sprintf("Error marshalling response: %v", err), http.StatusInternalServerError)
 	}
 }
+
+// CancelDeleteRequestHandler handles delete request cancellation
+func (dm *DeleteRequestHandler) CancelDeleteRequestHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	userID, err := user.ExtractOrgID(ctx)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	params := r.URL.Query()
+	requestID := params.Get("request_id")
+
+	deleteRequest, err := dm.deleteStore.GetDeleteRequest(ctx, userID, requestID)
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "error getting delete request from the store", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if deleteRequest == nil {
+		http.Error(w, "could not find delete request with given id", http.StatusBadRequest)
+		return
+	}
+
+	if deleteRequest.Status != StatusReceived {
+		http.Error(w, "deletion of request which is in process or already processed is not allowed", http.StatusBadRequest)
+		return
+	}
+
+	if deleteRequest.CreatedAt.Add(deleteRequestCancellationDeadline).Before(model.Now()) {
+		http.Error(w, fmt.Sprintf("deletion of request past the deadline of %s since its creation is not allowed", deleteRequestCancellationDeadline.String()), http.StatusBadRequest)
+		return
+	}
+
+	if err := dm.deleteStore.RemoveDeleteRequest(ctx, userID, requestID, deleteRequest.CreatedAt, deleteRequest.StartTime, deleteRequest.EndTime); err != nil {
+		level.Error(util.Logger).Log("msg", "error cancelling the delete request", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}