Do not split requests around ingester boundaries (#2766)

* only shards requests which dont touch ingesters
Signed-off-by: Owen Diehl <[email protected]>

* updates changelog
Signed-off-by: Owen Diehl <[email protected]>

Co-authored-by: Tom Wilkie <[email protected]>

Author: owen-d

CHANGELOG.md

@@ -24,6 +24,7 @@
 * [BUGFIX] Fixed a bug in the index intersect code causing storage to return more chunks/series than required. #2796
 * [BUGFIX] Fixed the number of reported keys in the background cache queue. #2764
 * [BUGFIX] Fix race in processing of headers in sharded queries. #2762
+* [BUGFIX] Query Frontend: Do not re-split sharded requests around ingester boundaries. #2766
 
 ## 1.2.0 / 2020-06-xx
 (in progress of release: current release candidate is https://github.com/cortexproject/cortex/releases/tag/v1.2.0-rc.0)

pkg/querier/queryrange/querysharding.go

@@ -252,73 +252,10 @@ type shardSplitter struct {
 
 func (splitter *shardSplitter) Do(ctx context.Context, r Request) (Response, error) {
 	cutoff := splitter.now().Add(-splitter.MinShardingLookback)
-	sharded, nonsharded := partitionRequest(r, cutoff)
 
-	return splitter.parallel(ctx, sharded, nonsharded)
-
-}
-
-func (splitter *shardSplitter) parallel(ctx context.Context, sharded, nonsharded Request) (Response, error) {
-	if sharded == nil {
-		return splitter.next.Do(ctx, nonsharded)
+	// Only attempt to shard queries which are older than the sharding lookback (the period for which ingesters are also queried).
+	if !cutoff.After(util.TimeFromMillis(r.GetEnd())) {
+		return splitter.next.Do(ctx, r)
 	}
-
-	if nonsharded == nil {
-		return splitter.shardingware.Do(ctx, sharded)
-	}
-
-	nonshardCh := make(chan Response, 1)
-	shardCh := make(chan Response, 1)
-	errCh := make(chan error, 2)
-
-	go func() {
-		res, err := splitter.next.Do(ctx, nonsharded)
-		if err != nil {
-			errCh <- err
-			return
-		}
-		nonshardCh <- res
-
-	}()
-
-	go func() {
-		res, err := splitter.shardingware.Do(ctx, sharded)
-		if err != nil {
-			errCh <- err
-			return
-		}
-		shardCh <- res
-	}()
-
-	resps := make([]Response, 0, 2)
-	for i := 0; i < 2; i++ {
-		select {
-		case r := <-nonshardCh:
-			resps = append(resps, r)
-		case r := <-shardCh:
-			resps = append(resps, r)
-		case err := <-errCh:
-			return nil, err
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		}
-
-	}
-
-	return splitter.codec.MergeResponse(resps...)
-}
-
-// partitionQuery splits a request into potentially multiple requests, one including the request's time range
-// [0,t). The other will include [t,inf)
-func partitionRequest(r Request, t time.Time) (before Request, after Request) {
-	boundary := util.TimeToMillis(t)
-	if r.GetStart() >= boundary {
-		return nil, r
-	}
-
-	if r.GetEnd() < boundary {
-		return r, nil
-	}
-
-	return r.WithStartEnd(r.GetStart(), boundary), r.WithStartEnd(boundary, r.GetEnd())
+	return splitter.shardingware.Do(ctx, r)
 }

pkg/querier/queryrange/querysharding_test.go

@@ -425,45 +425,81 @@ func TestQueryshardingCorrectness(t *testing.T) {
 
 func TestShardSplitting(t *testing.T) {
 
-	req := &PrometheusRequest{
-		Path:  "/query_range",
-		Start: util.TimeToMillis(start),
-		End:   util.TimeToMillis(end),
-		Step:  int64(step) / int64(time.Second),
-		Query: "sum(rate(bar1[1m]))",
-	}
-
-	shardingware := NewQueryShardMiddleware(
-		log.NewNopLogger(),
-		engine,
-		// ensure that all requests are shard compatbile
-		ShardingConfigs{
-			chunk.PeriodConfig{
-				Schema:    "v10",
-				RowShards: uint32(2),
-			},
+	for _, tc := range []struct {
+		desc        string
+		lookback    time.Duration
+		shouldShard bool
+	}{
+		{
+			desc:        "older than lookback",
+			lookback:    -1, // a negative lookback will ensure the entire query doesn't cross the sharding boundary & can safely be sharded.
+			shouldShard: true,
 		},
-		PrometheusCodec,
-		end.Sub(start)/2, // shard 1/2 of the req
-		nil,
-		nil,
-	)
+		{
+			desc:        "overlaps lookback",
+			lookback:    end.Sub(start) / 2, // intersect the request causing it to avoid sharding
+			shouldShard: false,
+		},
+		{
+			desc:        "newer than lookback",
+			lookback:    end.Sub(start) + 1,
+			shouldShard: false,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			req := &PrometheusRequest{
+				Path:  "/query_range",
+				Start: util.TimeToMillis(start),
+				End:   util.TimeToMillis(end),
+				Step:  int64(step) / int64(time.Second),
+				Query: "sum(rate(bar1[1m]))",
+			}
 
-	downstream := &downstreamHandler{
-		engine:    engine,
-		queryable: shardAwareQueryable,
-	}
+			shardingware := NewQueryShardMiddleware(
+				log.NewNopLogger(),
+				engine,
+				// ensure that all requests are shard compatbile
+				ShardingConfigs{
+					chunk.PeriodConfig{
+						Schema:    "v10",
+						RowShards: uint32(2),
+					},
+				},
+				PrometheusCodec,
+				tc.lookback,
+				nil,
+				nil,
+			)
 
-	handler := shardingware.Wrap(downstream).(*shardSplitter)
-	handler.now = func() time.Time { return end } // make the split cut the request in half (don't use time.Now)
+			downstream := &downstreamHandler{
+				engine:    engine,
+				queryable: shardAwareQueryable,
+			}
 
-	resp, err := handler.Do(context.Background(), req)
-	require.Nil(t, err)
+			handler := shardingware.Wrap(downstream).(*shardSplitter)
+			handler.now = func() time.Time { return end } // make the split cut the request in half (don't use time.Now)
 
-	unaltered, err := downstream.Do(context.Background(), req)
-	require.Nil(t, err)
+			var didShard bool
+
+			old := handler.shardingware
+			handler.shardingware = HandlerFunc(func(ctx context.Context, req Request) (Response, error) {
+				didShard = true
+				return old.Do(ctx, req)
+			})
+
+			resp, err := handler.Do(context.Background(), req)
+			require.Nil(t, err)
+
+			require.Equal(t, tc.shouldShard, didShard)
+
+			unaltered, err := downstream.Do(context.Background(), req)
+			require.Nil(t, err)
+
+			approximatelyEquals(t, unaltered.(*PrometheusResponse), resp.(*PrometheusResponse))
+
+		})
+	}
 
-	approximatelyEquals(t, unaltered.(*PrometheusResponse), resp.(*PrometheusResponse))
 }
 
 func BenchmarkQuerySharding(b *testing.B) {