addresses pr comments

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

Author: owen-d

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 ## master / unreleased
 
+* [FEATURE] Fan out parallelizable queries to backend queriers concurrently. #1878
+  * `-querier.sum-shards` (bool)
+  * Requires a shard-compatible schema (v10+)
+  * This causes the number of traces to increase accordingly.
+  * The query-frontend now requires a schema config to determine how/when to shard queries, either from a file or from flags (i.e. by the `config-yaml` CLI flag). This is the same schema config the queriers consume. The schema is only required to use this option.
+  * It's also advised to increase downstream concurrency controls as well:
+    * `querier.max-outstanding-requests-per-tenant`
+    * `querier.max-query-parallelism`
+    * `querier.max-concurrent`
+    * `server.grpc-max-concurrent-streams` (for both query-frontends and queriers)
 * [CHANGE] The frontend http server will now send 502 in case of deadline exceeded and 499 if the user requested cancellation. #2156
 * [CHANGE] Config file changed to remove top level `config_store` field in favor of a nested `configdb` field. #2125
 * [CHANGE] Removed unnecessary `frontend.cache-split-interval` in favor of `querier.split-queries-by-interval` both to reduce configuration complexity and guarantee alignment of these two configs. Starting from now, `-querier.cache-results` may only be enabled in conjunction with `-querier.split-queries-by-interval` (previously the cache interval default was `24h` so if you want to preserve the same behaviour you should set `-querier.split-queries-by-interval=24h`). #2040
@@ -62,17 +72,6 @@ Note that the ruler flags need to be changed in this upgrade. You're moving from
 Further, if you're using the configs service, we've upgraded the migration library and this requires some manual intervention. See full instructions below to upgrade your PostgreSQL.
 
 * [CHANGE] The frontend component now does not cache results if it finds a `Cache-Control` header and if one of its values is `no-store`. #1974
-* [FEATURE] Fan out parallelizable queries to backend queriers concurrently.
-  * `-querier.sum-shards` (bool)
-  * Requires a shard-compatible schema (v10+)
-  * This causes the number of traces to increase accordingly.
-  * The query-frontend now requires a schema config to determine how/when to shard queries, either from a file or from flags (i.e. by the `config-yaml` CLI flag). This is the same schema config the queriers consume.
-  * It's also advised to increase downstream concurrency controls as well:
-    * `querier.max-outstanding-requests-per-tenant`
-    * `querier.max-query-parallelism`
-    * `querier.max-concurrent`
-    * `server.grpc-max-concurrent-streams` (for both query-frontends and queriers)
-* [ENHANCEMENT] metric `cortex_ingester_flush_reasons` gets a new `reason` value: `Spread`, when `-ingester.spread-flushes` option is enabled.
 * [CHANGE] Flags changed with transition to upstream Prometheus rules manager:
   * `-ruler.client-timeout` is now `ruler.configs.client-timeout` in order to match `ruler.configs.url`.
   * `-ruler.group-timeout`has been removed.

docs/configuration/arguments.md

@@ -86,9 +86,11 @@ The ingester query API was improved over time, but defaults to the old behaviour
    - `querier.max-outstanding-requests-per-tenant`
    - `querier.max-query-parallelism`
    - `querier.max-concurrent`
+   - `server.grpc-max-concurrent-streams` (for both query-frontends and queriers)
+
    Furthermore, both querier and query-frontend components require the `querier.query-ingesters-within` parameter to know when to start sharding requests (ingester queries are not sharded). It's recommended to align this with `ingester.max-chunk-age`.
 
-   Instrumentation (traces) also scale with the number of sharded queries and it's suggested to account for increased throughput there as well.
+   Instrumentation (traces) also scale with the number of sharded queries and it's suggested to account for increased throughput there as well (for instance via `JAEGER_REPORTER_MAX_QUEUE_SIZE`).
 
 - `-querier.align-querier-with-step`
 

pkg/chunk/chunk_store_utils.go

@@ -10,7 +10,6 @@ import (
 	"github.com/prometheus/prometheus/promql"
 
 	"github.com/cortexproject/cortex/pkg/chunk/cache"
-	"github.com/cortexproject/cortex/pkg/querier/astmapper"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
@@ -256,10 +255,3 @@ func (c *Fetcher) processCacheResponse(ctx context.Context, chunks []Chunk, keys
 	}
 	return found, missing, err
 }
-
-func injectShardLabels(chunks []Chunk, shard astmapper.ShardAnnotation) {
-	for i, chunk := range chunks {
-		chunk.Metric = append(chunk.Metric, shard.Label())
-		chunks[i] = chunk
-	}
-}

pkg/chunk/schema_config.go

@@ -317,14 +317,6 @@ func (cfg *SchemaConfig) Load() error {
 		return err
 	}
 
-	for i, periodCfg := range cfg.Configs {
-		// apply default row shards
-		if periodCfg.RowShards == 0 {
-			periodCfg.RowShards = defaultRowShards(periodCfg.Schema)
-			cfg.Configs[i] = periodCfg
-		}
-	}
-
 	return cfg.Validate()
 }
 

pkg/chunk/series_store.go

@@ -534,3 +534,14 @@ func (c *seriesStore) calculateIndexEntries(ctx context.Context, from, through m
 
 	return result, missing, nil
 }
+
+func injectShardLabels(chunks []Chunk, shard astmapper.ShardAnnotation) {
+	for i, chunk := range chunks {
+
+		b := labels.NewBuilder(chunk.Metric)
+		l := shard.Label()
+		b.Set(l.Name, l.Value)
+		chunk.Metric = b.Labels()
+		chunks[i] = chunk
+	}
+}

pkg/querier/astmapper/embedded.go

@@ -21,8 +21,8 @@ be remapped into vector or matrix selectors utilizing a reserved label containin
 const (
 	// QueryLabel is a reserved label containing an embedded query
 	QueryLabel = "__cortex_queries__"
-	// EmbeddedQueryFlag is a reserved label (metric name) denoting an embedded query
-	EmbeddedQueryFlag = "__embedded_queries__"
+	// EmbeddedQueriesMetricName is a reserved label (metric name) denoting an embedded query
+	EmbeddedQueriesMetricName = "__embedded_queries__"
 )
 
 // EmbeddedQueries is a wrapper type for encoding queries
@@ -35,17 +35,12 @@ var JSONCodec jsonCodec
 
 type jsonCodec struct{}
 
-func (c jsonCodec) Encode(queries []string) string {
+func (c jsonCodec) Encode(queries []string) (string, error) {
 	embedded := EmbeddedQueries{
 		Concat: queries,
 	}
 	b, err := json.Marshal(embedded)
-
-	if err != nil {
-		panic(err)
-	}
-
-	return string(b)
+	return string(b), err
 }
 
 func (c jsonCodec) Decode(encoded string) (queries []string, err error) {
@@ -69,7 +64,10 @@ func VectorSquasher(nodes ...promql.Node) (promql.Expr, error) {
 		strs = append(strs, node.String())
 	}
 
-	encoded := JSONCodec.Encode(strs)
+	encoded, err := JSONCodec.Encode(strs)
+	if err != nil {
+		return nil, err
+	}
 
 	embeddedQuery, err := labels.NewMatcher(labels.MatchEqual, QueryLabel, encoded)
 
@@ -78,7 +76,7 @@ func VectorSquasher(nodes ...promql.Node) (promql.Expr, error) {
 	}
 
 	return &promql.VectorSelector{
-		Name:          EmbeddedQueryFlag,
+		Name:          EmbeddedQueriesMetricName,
 		LabelMatchers: []*labels.Matcher{embeddedQuery},
 	}, nil
 

pkg/querier/astmapper/parallel.go

@@ -1,7 +1,10 @@
 package astmapper
 
 import (
-	"github.com/pkg/errors"
+	"fmt"
+
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/prometheus/promql"
 )
 
@@ -76,7 +79,8 @@ func CanParallel(node promql.Node) bool {
 		return true
 
 	default:
-		panic(errors.Errorf("CanParallel: unhandled node type %T", node))
+		level.Error(util.Logger).Log("err", fmt.Sprintf("CanParallel: unhandled node type %T", node))
+		return false
 	}
 
 }

pkg/querier/astmapper/subtree_folder.go

@@ -42,12 +42,12 @@ func (f *subtreeFolder) MapNode(node promql.Node) (promql.Node, bool, error) {
 func isEmbedded(node promql.Node) (bool, error) {
 	switch n := node.(type) {
 	case *promql.VectorSelector:
-		if n.Name == EmbeddedQueryFlag {
+		if n.Name == EmbeddedQueriesMetricName {
 			return true, nil
 		}
 
 	case *promql.MatrixSelector:
-		if n.Name == EmbeddedQueryFlag {
+		if n.Name == EmbeddedQueriesMetricName {
 			return true, nil
 		}
 

pkg/querier/queryrange/instrumentation.go

@@ -25,7 +25,7 @@ var (
 	splitByCounter = promauto.NewCounter(prometheus.CounterOpts{
 		Namespace: "cortex",
 		Name:      "frontend_split_queries_total",
-		Help:      "Total number of split (parallelized) request segments",
+		Help:      "Total number of underlying query requests after the split by interval is applied",
 	})
 )
 

pkg/querier/queryrange/queryable.go

@@ -41,7 +41,7 @@ func (q *DownstreamQuerier) Select(
 	var embeddedQuery string
 	var isEmbedded bool
 	for _, matcher := range matchers {
-		if matcher.Name == labels.MetricName && matcher.Value == astmapper.EmbeddedQueryFlag {
+		if matcher.Name == labels.MetricName && matcher.Value == astmapper.EmbeddedQueriesMetricName {
 			isEmbedded = true
 		}