Fix queriers shuffle-sharding blast radius containment

CHANGELOG.md

@@ -6,6 +6,7 @@
 * [ENHANCEMENT] Ruler: added the following metrics when ruler sharding is enabled: #3916
   * `cortex_ruler_clients`
   * `cortex_ruler_client_request_duration_seconds`
+* [ENHANCEMENT] Query-frontend/scheduler: added querier forget delay (`-query-frontend.querier-forget-delay` and `-query-scheduler.querier-forget-delay`) to mitigate the blast radius in the event queriers crash because of a repeatedly sent "query of death" when shuffle-sharding is enabled. #3901
 
 ## 1.8.0 in progress
 

docs/configuration/config-file-reference.md

@@ -589,6 +589,13 @@ query_scheduler:
   # CLI flag: -query-scheduler.max-outstanding-requests-per-tenant
   [max_outstanding_requests_per_tenant: <int> | default = 100]
 
+  # If a querier disconnects without sending notification about graceful
+  # shutdown, the query-scheduler will keep the querier in the tenant's shard
+  # until the forget delay has passed. This feature is useful to reduce the
+  # blast radius when shuffle-sharding is enabled.
+  # CLI flag: -query-scheduler.querier-forget-delay
+  [querier_forget_delay: <duration> | default = 0s]
+
   # This configures the gRPC client used to report errors back to the
   # query-frontend.
   grpc_client_config:
@@ -1322,6 +1329,13 @@ The `query_frontend_config` configures the Cortex query-frontend.
 # CLI flag: -querier.max-outstanding-requests-per-tenant
 [max_outstanding_per_tenant: <int> | default = 100]
 
+# If a querier disconnects without sending notification about graceful shutdown,
+# the query-frontend will keep the querier in the tenant's shard until the
+# forget delay has passed. This feature is useful to reduce the blast radius
+# when shuffle-sharding is enabled.
+# CLI flag: -query-frontend.querier-forget-delay
+[querier_forget_delay: <duration> | default = 0s]
+
 # DNS hostname used for finding query-schedulers.
 # CLI flag: -frontend.scheduler-address
 [scheduler_address: <string> | default = ""]

docs/guides/shuffle-sharding.md

@@ -125,6 +125,15 @@ Note that this distribution happens in query-frontend, or query-scheduler if use
 
 _The maximum number of queriers can be overridden on a per-tenant basis in the limits overrides configuration._
 
+#### The impact of "query of death"
+
+In the event a tenant is repeatedly sending a "query of death" which leads the querier to crash or getting killed because of out-of-memory, the crashed querier will get disconnected from the query-frontend or query-scheduler and a new querier will be immediately assigned to the tenant's shard. This practically invalidates the assumption that shuffle-sharding can be used to contain the blast radius in case of a query of death.
+
+To mitigate it, Cortex allows to configure a delay between when a querier disconnects because of a crash and when the crashed querier is actually removed from the tenant's shard (and another healthy querier is added as replacement). A delay of 1 minute may be a reasonable trade-off:
+
+- Query-frontend: `-query-frontend.querier-forget-delay=1m`
+- Query-scheduler: `-query-scheduler.querier-forget-delay=1m`
+
 ### Store-gateway shuffle sharding
 
 The Cortex store-gateway -- used by the [blocks storage](../blocks-storage/_index.md) -- by default spreads each tenant's blocks across all running store-gateways.

pkg/frontend/config.go

@@ -70,7 +70,10 @@ func InitFrontend(cfg CombinedFrontendConfig, limits v1.Limits, grpcListenPort i
 
 	default:
 		// No scheduler = use original frontend.
-		fr := v1.New(cfg.FrontendV1, limits, log, reg)
+		fr, err := v1.New(cfg.FrontendV1, limits, log, reg)
+		if err != nil {
+			return nil, nil, nil, err
+		}
 		return transport.AdaptGrpcRoundTripperToHTTPRoundTripper(fr), fr, nil, nil
 	}
 }

pkg/frontend/v1/frontend.go

@@ -2,7 +2,6 @@ package v1
 
 import (
 	"context"
-	"errors"
 	"flag"
 	"fmt"
 	"net/http"
@@ -11,6 +10,7 @@ import (
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/opentracing/opentracing-go"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/weaveworks/common/httpgrpc"
@@ -31,12 +31,14 @@ var (
 
 // Config for a Frontend.
 type Config struct {
-	MaxOutstandingPerTenant int `yaml:"max_outstanding_per_tenant"`
+	MaxOutstandingPerTenant int           `yaml:"max_outstanding_per_tenant"`
+	QuerierForgetDelay      time.Duration `yaml:"querier_forget_delay"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.MaxOutstandingPerTenant, "querier.max-outstanding-requests-per-tenant", 100, "Maximum number of outstanding requests per tenant per frontend; requests beyond this error with HTTP 429.")
+	f.DurationVar(&cfg.QuerierForgetDelay, "query-frontend.querier-forget-delay", 0, "If a querier disconnects without sending notification about graceful shutdown, the query-frontend will keep the querier in the tenant's shard until the forget delay has passed. This feature is useful to reduce the blast radius when shuffle-sharding is enabled.")
 }
 
 type Limits interface {
@@ -56,6 +58,10 @@ type Frontend struct {
 	requestQueue *queue.RequestQueue
 	activeUsers  *util.ActiveUsersCleanupService
 
+	// Subservices manager.
+	subservices        *services.Manager
+	subservicesWatcher *services.FailureWatcher
+
 	// Metrics.
 	queueLength       *prometheus.GaugeVec
 	discardedRequests *prometheus.CounterVec
@@ -74,8 +80,7 @@ type request struct {
 }
 
 // New creates a new frontend. Frontend implements service, and must be started and stopped.
-func New(cfg Config, limits Limits, log log.Logger, registerer prometheus.Registerer) *Frontend {
-
+func New(cfg Config, limits Limits, log log.Logger, registerer prometheus.Registerer) (*Frontend, error) {
 	f := &Frontend{
 		cfg:    cfg,
 		log:    log,
@@ -95,26 +100,48 @@ func New(cfg Config, limits Limits, log log.Logger, registerer prometheus.Regist
 		}),
 	}
 
-	f.requestQueue = queue.NewRequestQueue(cfg.MaxOutstandingPerTenant, f.queueLength, f.discardedRequests)
+	f.requestQueue = queue.NewRequestQueue(cfg.MaxOutstandingPerTenant, cfg.QuerierForgetDelay, f.queueLength, f.discardedRequests)
 	f.activeUsers = util.NewActiveUsersCleanupWithDefaultValues(f.cleanupInactiveUserMetrics)
 
+	var err error
+	f.subservices, err = services.NewManager(f.requestQueue, f.activeUsers)
+	if err != nil {
+		return nil, err
+	}
+
 	f.numClients = promauto.With(registerer).NewGaugeFunc(prometheus.GaugeOpts{
 		Name: "cortex_query_frontend_connected_clients",
 		Help: "Number of worker clients currently connected to the frontend.",
 	}, f.requestQueue.GetConnectedQuerierWorkersMetric)
 
-	f.Service = services.NewIdleService(f.starting, f.stopping)
-	return f
+	f.Service = services.NewBasicService(f.starting, f.running, f.stopping)
+	return f, nil
 }
 
 func (f *Frontend) starting(ctx context.Context) error {
-	return services.StartAndAwaitRunning(ctx, f.activeUsers)
+	f.subservicesWatcher.WatchManager(f.subservices)
+
+	if err := services.StartManagerAndAwaitHealthy(ctx, f.subservices); err != nil {
+		return errors.Wrap(err, "unable to start frontend subservices")
+	}
+
+	return nil
+}
+
+func (f *Frontend) running(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case err := <-f.subservicesWatcher.Chan():
+			return errors.Wrap(err, "frontend subservice failed")
+		}
+	}
 }
 
 func (f *Frontend) stopping(_ error) error {
-	// Stops new requests and errors out any pending requests.
-	f.requestQueue.Stop()
-	return services.StopAndAwaitTerminated(context.Background(), f.activeUsers)
+	// This will also stop the requests queue, which stop accepting new requests and errors out any pending requests.
+	return services.StopManagerAndAwaitStopped(context.Background(), f.subservices)
 }
 
 func (f *Frontend) cleanupInactiveUserMetrics(user string) {
@@ -258,6 +285,13 @@ func (f *Frontend) Process(server frontendv1pb.Frontend_ProcessServer) error {
 	}
 }
 
+func (f *Frontend) NotifyClientShutdown(_ context.Context, req *frontendv1pb.NotifyClientShutdownRequest) (*frontendv1pb.NotifyClientShutdownResponse, error) {
+	level.Info(f.log).Log("msg", "received shutdown notification from querier", "querier", req.GetClientID())
+	f.requestQueue.NotifyQuerierShutdown(req.GetClientID())
+
+	return &frontendv1pb.NotifyClientShutdownResponse{}, nil
+}
+
 func getQuerierID(server frontendv1pb.Frontend_ProcessServer) (string, error) {
 	err := server.Send(&frontendv1pb.FrontendToClient{
 		Type: frontendv1pb.GET_ID,

pkg/frontend/v1/frontend_test.go

@@ -128,7 +128,7 @@ func TestFrontendCheckReady(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			f := &Frontend{
 				log: log.NewNopLogger(),
-				requestQueue: queue.NewRequestQueue(5,
+				requestQueue: queue.NewRequestQueue(5, 0,
 					prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user"}),
 					prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"}),
 				),
@@ -243,7 +243,8 @@ func testFrontend(t *testing.T, config Config, handler http.Handler, test func(a
 	httpListen, err := net.Listen("tcp", "localhost:0")
 	require.NoError(t, err)
 
-	v1 := New(config, limits{}, logger, reg)
+	v1, err := New(config, limits{}, logger, reg)
+	require.NoError(t, err)
 	require.NotNil(t, v1)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), v1))
 	defer func() {