Cancel abandoned operations in ReplicationSet.Do() (#3178)

Generally it will start a set of operations in parallel, and return
once enough of them have succeeded. Pass down a context to each one,
and cancel that context when ReplicationSet.Do() returns.

Signed-off-by: Bryan Boreham <[email protected]>

Author: bboreham

CHANGELOG.md

@@ -82,6 +82,7 @@
 * [BUGFIX] Configs: prevent validation of templates to fail when using template functions. #3157
 * [BUGFIX] Configuring the S3 URL with an `@` but without username and password doesn't enable the AWS static credentials anymore. #3170
 * [BUGFIX] Limit errors on ranged queries (`api/v1/query_range`) no longer return a status code `500` but `422` instead. #3167
+* [BUGFIX] No-longer-needed ingester operations from queries are now canceled. #3178
 
 ## 1.3.0 / 2020-08-21
 

pkg/distributor/distributor.go

@@ -604,7 +604,7 @@ func (d *Distributor) send(ctx context.Context, ingester ring.IngesterDesc, time
 }
 
 // ForAllIngesters runs f, in parallel, for all ingesters
-func (d *Distributor) ForAllIngesters(ctx context.Context, reallyAll bool, f func(client.IngesterClient) (interface{}, error)) ([]interface{}, error) {
+func (d *Distributor) ForAllIngesters(ctx context.Context, reallyAll bool, f func(context.Context, client.IngesterClient) (interface{}, error)) ([]interface{}, error) {
 	replicationSet, err := d.ingestersRing.GetAll(ring.Read)
 	if err != nil {
 		return nil, err
@@ -613,13 +613,13 @@ func (d *Distributor) ForAllIngesters(ctx context.Context, reallyAll bool, f fun
 		replicationSet.MaxErrors = 0
 	}
 
-	return replicationSet.Do(ctx, d.cfg.ExtraQueryDelay, func(ing *ring.IngesterDesc) (interface{}, error) {
+	return replicationSet.Do(ctx, d.cfg.ExtraQueryDelay, func(ctx context.Context, ing *ring.IngesterDesc) (interface{}, error) {
 		client, err := d.ingesterPool.GetClientFor(ing.Addr)
 		if err != nil {
 			return nil, err
 		}
 
-		return f(client.(ingester_client.IngesterClient))
+		return f(ctx, client.(ingester_client.IngesterClient))
 	})
 }
 
@@ -628,7 +628,7 @@ func (d *Distributor) LabelValuesForLabelName(ctx context.Context, labelName mod
 	req := &client.LabelValuesRequest{
 		LabelName: string(labelName),
 	}
-	resps, err := d.ForAllIngesters(ctx, false, func(client client.IngesterClient) (interface{}, error) {
+	resps, err := d.ForAllIngesters(ctx, false, func(ctx context.Context, client client.IngesterClient) (interface{}, error) {
 		return client.LabelValues(ctx, req)
 	})
 	if err != nil {
@@ -652,7 +652,7 @@ func (d *Distributor) LabelValuesForLabelName(ctx context.Context, labelName mod
 // LabelNames returns all of the label names.
 func (d *Distributor) LabelNames(ctx context.Context) ([]string, error) {
 	req := &client.LabelNamesRequest{}
-	resps, err := d.ForAllIngesters(ctx, false, func(client client.IngesterClient) (interface{}, error) {
+	resps, err := d.ForAllIngesters(ctx, false, func(ctx context.Context, client client.IngesterClient) (interface{}, error) {
 		return client.LabelNames(ctx, req)
 	})
 	if err != nil {
@@ -684,7 +684,7 @@ func (d *Distributor) MetricsForLabelMatchers(ctx context.Context, from, through
 		return nil, err
 	}
 
-	resps, err := d.ForAllIngesters(ctx, false, func(client client.IngesterClient) (interface{}, error) {
+	resps, err := d.ForAllIngesters(ctx, false, func(ctx context.Context, client client.IngesterClient) (interface{}, error) {
 		return client.MetricsForLabelMatchers(ctx, req)
 	})
 	if err != nil {
@@ -712,7 +712,7 @@ func (d *Distributor) MetricsForLabelMatchers(ctx context.Context, from, through
 func (d *Distributor) MetricsMetadata(ctx context.Context) ([]scrape.MetricMetadata, error) {
 	req := &ingester_client.MetricsMetadataRequest{}
 	// TODO(gotjosh): We only need to look in all the ingesters if shardByAllLabels is enabled.
-	resps, err := d.ForAllIngesters(ctx, false, func(client client.IngesterClient) (interface{}, error) {
+	resps, err := d.ForAllIngesters(ctx, false, func(ctx context.Context, client client.IngesterClient) (interface{}, error) {
 		return client.MetricsMetadata(ctx, req)
 	})
 	if err != nil {
@@ -746,7 +746,7 @@ func (d *Distributor) MetricsMetadata(ctx context.Context) ([]scrape.MetricMetad
 // UserStats returns statistics about the current user.
 func (d *Distributor) UserStats(ctx context.Context) (*UserStats, error) {
 	req := &client.UserStatsRequest{}
-	resps, err := d.ForAllIngesters(ctx, true, func(client client.IngesterClient) (interface{}, error) {
+	resps, err := d.ForAllIngesters(ctx, true, func(ctx context.Context, client client.IngesterClient) (interface{}, error) {
 		return client.UserStats(ctx, req)
 	})
 	if err != nil {

pkg/distributor/query.go

@@ -88,7 +88,7 @@ func (d *Distributor) queryPrep(ctx context.Context, from, to model.Time, matche
 func (d *Distributor) queryIngesters(ctx context.Context, replicationSet ring.ReplicationSet, req *client.QueryRequest) (model.Matrix, error) {
 	// Fetch samples from multiple ingesters in parallel, using the replicationSet
 	// to deal with consistency.
-	results, err := replicationSet.Do(ctx, d.cfg.ExtraQueryDelay, func(ing *ring.IngesterDesc) (interface{}, error) {
+	results, err := replicationSet.Do(ctx, d.cfg.ExtraQueryDelay, func(ctx context.Context, ing *ring.IngesterDesc) (interface{}, error) {
 		client, err := d.ingesterPool.GetClientFor(ing.Addr)
 		if err != nil {
 			return nil, err
@@ -133,7 +133,7 @@ func (d *Distributor) queryIngesters(ctx context.Context, replicationSet ring.Re
 // queryIngesterStream queries the ingesters using the new streaming API.
 func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ring.ReplicationSet, req *client.QueryRequest) (*ingester_client.QueryStreamResponse, error) {
 	// Fetch samples from multiple ingesters
-	results, err := replicationSet.Do(ctx, d.cfg.ExtraQueryDelay, func(ing *ring.IngesterDesc) (interface{}, error) {
+	results, err := replicationSet.Do(ctx, d.cfg.ExtraQueryDelay, func(ctx context.Context, ing *ring.IngesterDesc) (interface{}, error) {
 		client, err := d.ingesterPool.GetClientFor(ing.Addr)
 		if err != nil {
 			return nil, err

pkg/ring/replication_set.go

@@ -14,17 +14,15 @@ type ReplicationSet struct {
 
 // Do function f in parallel for all replicas in the set, erroring is we exceed
 // MaxErrors and returning early otherwise.
-func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, f func(*IngesterDesc) (interface{}, error)) ([]interface{}, error) {
+func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, f func(context.Context, *IngesterDesc) (interface{}, error)) ([]interface{}, error) {
 	var (
 		errs        = make(chan error, len(r.Ingesters))
 		resultsChan = make(chan interface{}, len(r.Ingesters))
 		minSuccess  = len(r.Ingesters) - r.MaxErrors
-		done        = make(chan struct{})
 		forceStart  = make(chan struct{}, r.MaxErrors)
 	)
-	defer func() {
-		close(done)
-	}()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 
 	for i := range r.Ingesters {
 		go func(i int, ing *IngesterDesc) {
@@ -33,13 +31,13 @@ func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, f func(*Ing
 				after := time.NewTimer(delay)
 				defer after.Stop()
 				select {
-				case <-done:
+				case <-ctx.Done():
 					return
 				case <-forceStart:
 				case <-after.C:
 				}
 			}
-			result, err := f(ing)
+			result, err := f(ctx, ing)
 			if err != nil {
 				errs <- err
 			} else {