integration test for deletion of series (#2946)

* integration test for deletion of series

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

* update changelog

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

* Reuse existing timeout

Signed-off-by: Marco Pracucci <[email protected]>

* Removed Println()

Signed-off-by: Marco Pracucci <[email protected]>

* Assert on Purger metrics too

Signed-off-by: Marco Pracucci <[email protected]>

* Removed time.Sleep() from delete series integration test

Signed-off-by: Marco Pracucci <[email protected]>

* Moved CHANGELOG entry

Signed-off-by: Marco Pracucci <[email protected]>

Co-authored-by: Marco Pracucci <[email protected]>

Author: sandeepsukhani

CHANGELOG.md

@@ -7,6 +7,7 @@
   * `cortex_ingester_received_files`
   * `cortex_ingester_received_bytes_total`
   * `cortex_ingester_sent_bytes_total`
+* [CHANGE] Experimental Delete Series: `/api/v1/admin/tsdb/delete_series` and `/api/v1/admin/tsdb/cancel_delete_request` purger APIs to return status code `204` instead of `200` for success. #2946
 * [ENHANCEMENT] Add support for azure storage in China, German and US Government environments. #2988
 * [ENHANCEMENT] Query-tee: added a small tolerance to floating point sample values comparison. #2994
 * [BUGFIX] Query-frontend: Fixed rounding for incoming query timestamps, to be 100% Prometheus compatible. #2990

integration/asserts.go

@@ -23,6 +23,7 @@ const (
 	AlertManager
 	Ruler
 	StoreGateway
+	Purger
 )
 
 var (
@@ -36,6 +37,7 @@ var (
 		AlertManager:  {"cortex_alertmanager"},
 		Ruler:         {},
 		StoreGateway:  {"!cortex_storegateway_client", "cortex_storegateway"}, // The metrics prefix cortex_storegateway_client may be used by other components so we ignore it.
+		Purger:        {"cortex_purger"},
 	}
 
 	// Blacklisted metrics prefixes across any Cortex service.

integration/chunks_delete_series_test.go

@@ -0,0 +1,221 @@
+// +build requires_docker
+
+package main
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/prometheus/prometheus/prompb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/cortexproject/cortex/integration/e2ecortex"
+)
+
+func TestDeleteSeriesAllIndexBackends(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Start dependencies.
+	dynamo := e2edb.NewDynamoDB()
+	bigtable := e2edb.NewBigtable()
+	cassandra := e2edb.NewCassandra()
+
+	stores := []string{"aws-dynamo", "bigtable", "cassandra"}
+	perStoreDuration := 7 * 24 * time.Hour
+
+	consul := e2edb.NewConsul()
+	require.NoError(t, s.StartAndWaitReady(cassandra, dynamo, bigtable, consul))
+
+	// lets build config for each type of Index Store.
+	now := time.Now()
+	oldestStoreStartTime := now.Add(time.Duration(-len(stores)) * perStoreDuration)
+
+	storeConfigs := make([]storeConfig, len(stores))
+	for i, store := range stores {
+		storeConfigs[i] = storeConfig{From: oldestStoreStartTime.Add(time.Duration(i) * perStoreDuration).UTC().Format("2006-01-02"), IndexStore: store}
+	}
+
+	flags := mergeFlags(ChunksStorageFlags, map[string]string{
+		"-cassandra.addresses": cassandra.NetworkHTTPEndpoint(),
+		"-cassandra.keyspace":  "tests", // keyspace gets created on startup if it does not exist
+		"-purger.enable":       "true",
+		"-deletes.store":       "bigtable",
+	})
+
+	// bigtable client needs to set an environment variable when connecting to an emulator.
+	bigtableFlag := map[string]string{"BIGTABLE_EMULATOR_HOST": bigtable.NetworkHTTPEndpoint()}
+
+	// here we are starting and stopping table manager for each index store
+	// this is a workaround to make table manager create tables for each config since it considers only latest schema config while creating tables
+	for i := range storeConfigs {
+		require.NoError(t, writeFileToSharedDir(s, cortexSchemaConfigFile, []byte(buildSchemaConfigWith(storeConfigs[i:i+1]))))
+
+		tableManager := e2ecortex.NewTableManager("table-manager", mergeFlags(flags, map[string]string{
+			"-table-manager.retention-period": "2520h", // setting retention high enough
+			"-log.level":                      "warn",
+		}), "")
+		tableManager.HTTPService.SetEnvVars(bigtableFlag)
+		require.NoError(t, s.StartAndWaitReady(tableManager))
+
+		// Wait until the first table-manager sync has completed, so that we're
+		// sure the tables have been created.
+		require.NoError(t, tableManager.WaitSumMetrics(e2e.Greater(0), "cortex_table_manager_sync_success_timestamp_seconds"))
+		require.NoError(t, s.Stop(tableManager))
+	}
+
+	// Start rest of the Cortex components.
+	require.NoError(t, writeFileToSharedDir(s, cortexSchemaConfigFile, []byte(buildSchemaConfigWith(storeConfigs))))
+
+	ingester := e2ecortex.NewIngester("ingester", consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
+		"-ingester.retain-period": "0s", // we want to make ingester not retain any chunks in memory after they are flushed so that queries get data only from the store
+		"-log.level":              "warn",
+	}), "")
+	ingester.HTTPService.SetEnvVars(bigtableFlag)
+
+	distributor := e2ecortex.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), flags, "")
+	querier := e2ecortex.NewQuerier("querier", consul.NetworkHTTPEndpoint(), flags, "")
+	querier.HTTPService.SetEnvVars(bigtableFlag)
+
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester, querier))
+
+	// Wait until both the distributor and querier have updated the ring.
+	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	client, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+
+	seriesToPush := []struct {
+		name          string
+		lables        []prompb.Label
+		pushedVectors []model.Vector
+	}{
+		{
+			name:   "series_1",
+			lables: []prompb.Label{{Name: "common", Value: "label"}, {Name: "distinct", Value: "label1"}},
+		},
+		{
+			name:   "series_2",
+			lables: []prompb.Label{{Name: "common", Value: "label"}, {Name: "distinct", Value: "label2"}},
+		},
+		{
+			name:   "delete_series",
+			lables: []prompb.Label{{Name: "common", Value: "label"}, {Name: "distinct", Value: "label3"}},
+		},
+	}
+
+	// Push some series for each day starting from oldest start time from configs until now so that we test all the Index Stores.
+	for ts := oldestStoreStartTime; ts.Before(now); ts = ts.Add(24 * time.Hour) {
+		for i, s := range seriesToPush {
+			series, expectedVector := generateSeries(s.name, ts, s.lables...)
+
+			res, err := client.Push(series)
+			require.NoError(t, err)
+			require.Equal(t, 200, res.StatusCode)
+
+			seriesToPush[i].pushedVectors = append(seriesToPush[i].pushedVectors, expectedVector)
+		}
+
+		// lets make ingester flush the chunks immediately to the store.
+		res, err := e2e.GetRequest("http://" + ingester.HTTPEndpoint() + "/flush")
+		require.NoError(t, err)
+		require.Equal(t, 204, res.StatusCode)
+
+		// Let's wait until all chunks are flushed.
+		require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(0), "cortex_ingester_flush_queue_length"))
+		require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(0), "cortex_ingester_flush_series_in_progress"))
+	}
+
+	// call flush again because chunks are sometimes still retained in memory due to chunks flush operation being async while chunk cleanup from memory is not.
+	res, err := e2e.GetRequest("http://" + ingester.HTTPEndpoint() + "/flush")
+	require.NoError(t, err)
+	require.Equal(t, 204, res.StatusCode)
+
+	require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(0), "cortex_ingester_memory_chunks"))
+
+	// start a purger
+	purger := e2ecortex.NewPurger("purger", flags, "")
+	purger.HTTPService.SetEnvVars(bigtableFlag)
+
+	require.NoError(t, s.StartAndWaitReady(purger))
+
+	// perform deletion for 2 days interval for each week covering all the Index Stores.
+	deletionDuration := 2 * 24 * time.Hour
+	deletedIntervals := intervals{}
+	endpoint := "http://" + purger.HTTPEndpoint() + "/prometheus/api/v1/admin/tsdb/delete_series?match[]=delete_series{common=\"label\"}&start=%f&end=%f"
+	for ts := now; ts.After(oldestStoreStartTime); ts = ts.Add(-perStoreDuration) {
+		// expand the interval by a second on both the ends since the requests are getting aligned with pushed samples at ms precision which
+		// sometimes does not match exactly in purger during parsing time from a string.
+		deletedInterval := interval{e2e.TimeToMilliseconds(ts.Add(-(deletionDuration + time.Second))), e2e.TimeToMilliseconds(ts.Add(time.Second))}
+		deletedIntervals = append(deletedIntervals, deletedInterval)
+
+		res, err := client.PostRequest(fmt.Sprintf(endpoint, float64(deletedInterval.start)/float64(1000), float64(deletedInterval.end)/float64(1000)), nil)
+		require.NoError(t, err)
+		require.Equal(t, 204, res.StatusCode)
+	}
+
+	// check whether purger has received expected number of delete requests.
+	require.NoError(t, purger.WaitSumMetrics(e2e.Equals(float64(len(deletedIntervals))), "cortex_purger_delete_requests_received_total"))
+
+	// stop the purger and recreate it since we load requests for deletion every hour.
+	require.NoError(t, s.Stop(purger))
+
+	purger = e2ecortex.NewPurger("purger", mergeFlags(flags, map[string]string{
+		"-purger.delete-request-cancel-period": "-1m", // we retain delete requests for a day(default)+1m, changing it to a negative value so that delete requests get picked up immediately.
+	}), "")
+	purger.HTTPService.SetEnvVars(bigtableFlag)
+	require.NoError(t, s.StartAndWaitReady(purger))
+
+	require.NoError(t, purger.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_purger_load_pending_requests_attempts_total"},
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "status", "success")),
+		e2e.WaitMissingMetrics))
+
+	require.NoError(t, purger.WaitSumMetricsWithOptions(e2e.Equals(float64(len(deletedIntervals))), []string{"cortex_purger_delete_requests_processed_total"},
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "user", "user-1")),
+		e2e.WaitMissingMetrics))
+
+	// query and verify that only delete series for delete interval are gone.
+	for _, s := range seriesToPush {
+		for _, expectedVector := range s.pushedVectors {
+			result, err := client.Query(s.name, expectedVector[0].Timestamp.Time())
+			require.NoError(t, err)
+
+			require.Equal(t, model.ValVector, result.Type())
+			if s.name == "delete_series" && deletedIntervals.includes(e2e.TimeToMilliseconds(expectedVector[0].Timestamp.Time())) {
+				require.Len(t, result.(model.Vector), 0)
+			} else {
+				assert.Equal(t, expectedVector, result.(model.Vector))
+			}
+		}
+	}
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Distributor, distributor)
+	assertServiceMetricsPrefixes(t, Ingester, ingester)
+	assertServiceMetricsPrefixes(t, Querier, querier)
+	assertServiceMetricsPrefixes(t, Purger, purger)
+}
+
+type interval struct {
+	start, end int64
+}
+
+type intervals []interval
+
+func (i intervals) includes(ts int64) bool {
+	for _, interval := range i {
+		if ts < interval.start || ts > interval.end {
+			continue
+		}
+		return true
+	}
+	return false
+}

integration/e2ecortex/client.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -368,3 +369,15 @@ func (c *Client) DeleteAlertmanagerConfig(ctx context.Context) error {
 
 	return nil
 }
+
+func (c *Client) PostRequest(url string, body io.Reader) (*http.Response, error) {
+	req, err := http.NewRequest("POST", url, body)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("X-Scope-OrgID", c.orgID)
+
+	client := &http.Client{Timeout: c.timeout}
+	return client.Do(req)
+}

integration/e2ecortex/services.go

@@ -302,3 +302,31 @@ func NewRuler(name string, flags map[string]string, image string) *CortexService
 		grpcPort,
 	)
 }
+
+func NewPurger(name string, flags map[string]string, image string) *CortexService {
+	return NewPurgerWithConfigFile(name, "", flags, image)
+}
+
+func NewPurgerWithConfigFile(name, configFile string, flags map[string]string, image string) *CortexService {
+	if configFile != "" {
+		flags["-config.file"] = filepath.Join(e2e.ContainerSharedDir, configFile)
+	}
+
+	if image == "" {
+		image = GetDefaultImage()
+	}
+
+	return NewCortexService(
+		name,
+		image,
+		e2e.NewCommandWithoutEntrypoint("cortex", e2e.BuildArgs(e2e.MergeFlags(map[string]string{
+			"-target":                   "purger",
+			"-log.level":                "warn",
+			"-purger.object-store-type": "filesystem",
+			"-local.chunk-directory":    e2e.ContainerSharedDir,
+		}, flags))...),
+		e2e.NewHTTPReadinessProbe(httpPort, "/ready", 200, 299),
+		httpPort,
+		grpcPort,
+	)
+}

pkg/chunk/purger/request_handler.go

@@ -113,6 +113,7 @@ func (dm *DeleteRequestHandler) AddDeleteRequestHandler(w http.ResponseWriter, r
 	}
 
 	dm.metrics.deleteRequestsReceivedTotal.WithLabelValues(userID).Inc()
+	w.WriteHeader(http.StatusNoContent)
 }
 
 // GetAllDeleteRequestsHandler handles get all delete requests
@@ -177,5 +178,5 @@ func (dm *DeleteRequestHandler) CancelDeleteRequestHandler(w http.ResponseWriter
 		return
 	}
 
-	w.WriteHeader(http.StatusOK)
+	w.WriteHeader(http.StatusNoContent)
 }

pkg/testexporter/correctness/delete_series.go

@@ -239,7 +239,7 @@ func (d *DeleteSeriesTest) sendDeleteRequest() (err error) {
 		return
 	}
 
-	if resp.StatusCode != 200 {
+	if resp.StatusCode != 204 {
 		return fmt.Errorf("unexpected status code %d", resp.StatusCode)
 	}