Merge remote-tracking branch 'upstream/master' into query-metering

Author: Tyler Reid

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## master / unreleased
 
+* [CHANGE] Enable strict JSON unmarshal for `pkg/util/validation.Limits` struct. The custom `UnmarshalJSON()` will now fail if the input has unknown fields. #4298
 * [CHANGE] Cortex chunks storage has been deprecated and it's now in maintenance mode: all Cortex users are encouraged to migrate to the blocks storage. No new features will be added to the chunks storage. The default Cortex configuration still runs the chunks engine; please check out the [blocks storage doc](https://cortexmetrics.io/docs/blocks-storage/) on how to configure Cortex to run with the blocks storage.  #4268
 * [CHANGE] The example Kubernetes manifests (stored at `k8s/`) have been removed due to a lack of proper support and maintenance. #4268
 * [CHANGE] Querier / ruler: deprecated `-store.query-chunk-limit` CLI flag (and its respective YAML config option `max_chunks_per_query`) in favour of `-querier.max-fetched-chunks-per-query` (and its respective YAML config option `max_fetched_chunks_per_query`). The new limit specifies the maximum number of chunks that can be fetched in a single query from ingesters and long-term storage: the total number of actual fetched chunks could be 2x the limit, being independently applied when querying ingesters and long-term storage. #4125

pkg/api/api.go

@@ -351,11 +351,16 @@ func (a *API) RegisterCompactor(c *compactor.Compactor) {
 	a.RegisterRoute("/compactor/ring", http.HandlerFunc(c.RingHandler), false, "GET", "POST")
 }
 
+type Distributor interface {
+	querier.Distributor
+	UserStatsHandler(w http.ResponseWriter, r *http.Request)
+}
+
 // RegisterQueryable registers the the default routes associated with the querier
 // module.
 func (a *API) RegisterQueryable(
 	queryable storage.SampleAndChunkQueryable,
-	distributor *distributor.Distributor,
+	distributor Distributor,
 ) {
 	// these routes are always registered to the default server
 	a.RegisterRoute("/api/v1/user_stats", http.HandlerFunc(distributor.UserStatsHandler), true, "GET")

pkg/api/handlers.go

@@ -23,7 +23,6 @@ import (
 	"github.com/weaveworks/common/middleware"
 
 	"github.com/cortexproject/cortex/pkg/chunk/purger"
-	"github.com/cortexproject/cortex/pkg/distributor"
 	"github.com/cortexproject/cortex/pkg/querier"
 	"github.com/cortexproject/cortex/pkg/querier/stats"
 	"github.com/cortexproject/cortex/pkg/util"
@@ -160,7 +159,7 @@ func NewQuerierHandler(
 	queryable storage.SampleAndChunkQueryable,
 	exemplarQueryable storage.ExemplarQueryable,
 	engine *promql.Engine,
-	distributor *distributor.Distributor,
+	distributor Distributor,
 	tombstonesLoader *purger.TombstonesLoader,
 	reg prometheus.Registerer,
 	logger log.Logger,

pkg/compactor/compactor.go

@@ -408,9 +408,9 @@ func (c *Compactor) starting(ctx context.Context) error {
 
 			level.Info(c.logger).Log("msg", "waiting until compactor ring topology is stable", "min_waiting", minWaiting.String(), "max_waiting", maxWaiting.String())
 			if err := ring.WaitRingStability(ctx, c.ring, RingOp, minWaiting, maxWaiting); err != nil {
-				level.Warn(c.logger).Log("msg", "compactor is ring topology is not stable after the max waiting time, proceeding anyway")
+				level.Warn(c.logger).Log("msg", "compactor ring topology is not stable after the max waiting time, proceeding anyway")
 			} else {
-				level.Info(c.logger).Log("msg", "compactor is ring topology is stable")
+				level.Info(c.logger).Log("msg", "compactor ring topology is stable")
 			}
 		}
 	}

pkg/distributor/distributor_test.go

@@ -1012,12 +1012,15 @@ func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunkBytesPerQueryLimitIs
 	flagext.DefaultValues(limits)
 
 	// Prepare distributors.
+	// Use replication factor of 2 to always read all the chunks from both ingesters,
+	// this guarantees us to always read the same chunks and have a stable test.
 	ds, _, r, _ := prepare(t, prepConfig{
-		numIngesters:     3,
-		happyIngesters:   3,
-		numDistributors:  1,
-		shardByAllLabels: true,
-		limits:           limits,
+		numIngesters:      2,
+		happyIngesters:    2,
+		numDistributors:   1,
+		shardByAllLabels:  true,
+		limits:            limits,
+		replicationFactor: 2,
 	})
 	defer stopAll(ds, r)
 
@@ -1893,6 +1896,7 @@ type prepConfig struct {
 	skipLabelNameValidation      bool
 	maxInflightRequests          int
 	maxIngestionRate             float64
+	replicationFactor            int
 }
 
 func prepare(t *testing.T, cfg prepConfig) ([]*Distributor, []mockIngester, *ring.Ring, []*prometheus.Registry) {
@@ -1935,12 +1939,18 @@ func prepare(t *testing.T, cfg prepConfig) ([]*Distributor, []mockIngester, *rin
 	)
 	require.NoError(t, err)
 
+	// Use a default replication factor of 3 if there isn't a provided replication factor.
+	rf := cfg.replicationFactor
+	if rf == 0 {
+		rf = 3
+	}
+
 	ingestersRing, err := ring.New(ring.Config{
 		KVStore: kv.Config{
 			Mock: kvStore,
 		},
 		HeartbeatTimeout:  60 * time.Minute,
-		ReplicationFactor: 3,
+		ReplicationFactor: rf,
 	}, ring.IngesterRingKey, ring.IngesterRingKey, nil)
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ingestersRing))

pkg/ruler/compat.go

@@ -27,7 +27,7 @@ type Pusher interface {
 	Push(context.Context, *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error)
 }
 
-type pusherAppender struct {
+type PusherAppender struct {
 	failedWrites prometheus.Counter
 	totalWrites  prometheus.Counter
 
@@ -39,7 +39,7 @@ type pusherAppender struct {
 	evaluationDelay time.Duration
 }
 
-func (a *pusherAppender) Append(_ uint64, l labels.Labels, t int64, v float64) (uint64, error) {
+func (a *PusherAppender) Append(_ uint64, l labels.Labels, t int64, v float64) (uint64, error) {
 	a.labels = append(a.labels, l)
 
 	// Adapt staleness markers for ruler evaluation delay. As the upstream code
@@ -61,11 +61,11 @@ func (a *pusherAppender) Append(_ uint64, l labels.Labels, t int64, v float64) (
 	return 0, nil
 }
 
-func (a *pusherAppender) AppendExemplar(_ uint64, _ labels.Labels, _ exemplar.Exemplar) (uint64, error) {
+func (a *PusherAppender) AppendExemplar(_ uint64, _ labels.Labels, _ exemplar.Exemplar) (uint64, error) {
 	return 0, errors.New("exemplars are unsupported")
 }
 
-func (a *pusherAppender) Commit() error {
+func (a *PusherAppender) Commit() error {
 	a.totalWrites.Inc()
 
 	// Since a.pusher is distributor, client.ReuseSlice will be called in a.pusher.Push.
@@ -84,7 +84,7 @@ func (a *pusherAppender) Commit() error {
 	return err
 }
 
-func (a *pusherAppender) Rollback() error {
+func (a *PusherAppender) Rollback() error {
 	a.labels = nil
 	a.samples = nil
 	return nil
@@ -112,7 +112,7 @@ func NewPusherAppendable(pusher Pusher, userID string, limits RulesLimits, total
 
 // Appender returns a storage.Appender
 func (t *PusherAppendable) Appender(ctx context.Context) storage.Appender {
-	return &pusherAppender{
+	return &PusherAppender{
 		failedWrites: t.failedWrites,
 		totalWrites:  t.totalWrites,
 
@@ -131,9 +131,9 @@ type RulesLimits interface {
 	RulerMaxRulesPerRuleGroup(userID string) int
 }
 
-// engineQueryFunc returns a new query function using the rules.EngineQueryFunc function
+// EngineQueryFunc returns a new query function using the rules.EngineQueryFunc function
 // and passing an altered timestamp.
-func engineQueryFunc(engine *promql.Engine, q storage.Queryable, overrides RulesLimits, userID string) rules.QueryFunc {
+func EngineQueryFunc(engine *promql.Engine, q storage.Queryable, overrides RulesLimits, userID string) rules.QueryFunc {
 	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
 		orig := rules.EngineQueryFunc(engine, q)
 		// Delay the evaluation of all rules by a set interval to give a buffer
@@ -143,7 +143,7 @@ func engineQueryFunc(engine *promql.Engine, q storage.Queryable, overrides Rules
 	}
 }
 
-func metricsQueryFunc(qf rules.QueryFunc, queries, failedQueries prometheus.Counter, queryTime *prometheus.CounterVec, userID string) rules.QueryFunc {
+func MetricsQueryFunc(qf rules.QueryFunc, queries, failedQueries prometheus.Counter, queryTime *prometheus.CounterVec, userID string) rules.QueryFunc {
 	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
 		queries.Inc()
 
@@ -219,7 +219,7 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 		return rules.NewManager(&rules.ManagerOptions{
 			Appendable:      NewPusherAppendable(p, userID, overrides, totalWrites, failedWrites),
 			Queryable:       q,
-			QueryFunc:       metricsQueryFunc(engineQueryFunc(engine, q, overrides, userID), totalQueries, failedQueries, rulerQuerySeconds, userID),
+			QueryFunc:       MetricsQueryFunc(EngineQueryFunc(engine, q, overrides, userID), totalQueries, failedQueries, rulerQuerySeconds, userID),
 			Context:         user.InjectOrgID(ctx, userID),
 			ExternalURL:     cfg.ExternalURL.URL,
 			NotifyFunc:      SendAlerts(notifier, cfg.ExternalURL.URL.String()),
@@ -230,4 +230,4 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 			ResendDelay:     cfg.ResendDelay,
 		})
 	}
-}
+}

pkg/ruler/compat_test.go

@@ -232,7 +232,7 @@ func TestMetricsQueryFuncErrors(t *testing.T) {
 				return promql.Vector{}, tc.returnedError
 			}
 
-			qf := metricsQueryFunc(mockFunc, queries, failures, queryTime, "user")
+			qf := MetricsQueryFunc(mockFunc, queries, failures, queryTime, "user")
 
 			_, err := qf(context.Background(), "test", time.Now())
 			require.Equal(t, tc.returnedError, err)
@@ -241,4 +241,4 @@ func TestMetricsQueryFuncErrors(t *testing.T) {
 			require.Equal(t, tc.expectedFailedQueries, int(testutil.ToFloat64(failures)))
 		})
 	}
-}
+}

pkg/storegateway/gateway.go

@@ -242,9 +242,9 @@ func (g *StoreGateway) starting(ctx context.Context) (err error) {
 
 			level.Info(g.logger).Log("msg", "waiting until store-gateway ring topology is stable", "min_waiting", minWaiting.String(), "max_waiting", maxWaiting.String())
 			if err := ring.WaitRingStability(ctx, g.ring, BlocksOwnerSync, minWaiting, maxWaiting); err != nil {
-				level.Warn(g.logger).Log("msg", "store-gateway is ring topology is not stable after the max waiting time, proceeding anyway")
+				level.Warn(g.logger).Log("msg", "store-gateway ring topology is not stable after the max waiting time, proceeding anyway")
 			} else {
-				level.Info(g.logger).Log("msg", "store-gateway is ring topology is stable")
+				level.Info(g.logger).Log("msg", "store-gateway ring topology is stable")
 			}
 		}
 	}

pkg/storegateway/gateway_test.go

@@ -300,7 +300,10 @@ func TestStoreGateway_InitialSyncWithWaitRingStability(t *testing.T) {
 				t.Log("random generator seed:", seed)
 
 				ctx := context.Background()
-				ringStore := consul.NewInMemoryClient(ring.GetCodec())
+				ringStore := consul.NewInMemoryClientWithConfig(ring.GetCodec(), consul.Config{
+					MaxCasRetries: 20,
+					CasRetryDelay: 500 * time.Millisecond,
+				})
 
 				// Create the configured number of gateways.
 				var gateways []*StoreGateway

pkg/util/validation/limits.go

@@ -1,6 +1,7 @@
 package validation
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"flag"
@@ -229,7 +230,10 @@ func (l *Limits) UnmarshalJSON(data []byte) error {
 	}
 
 	type plain Limits
-	return json.Unmarshal(data, (*plain)(l))
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.DisallowUnknownFields()
+
+	return dec.Decode((*plain)(l))
 }
 
 func (l *Limits) copyNotificationIntegrationLimits(defaults NotificationRateLimitMap) {

pkg/util/validation/limits_test.go

@@ -3,6 +3,7 @@ package validation
 import (
 	"encoding/json"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -165,6 +166,15 @@ func TestLimitsLoadingFromJson(t *testing.T) {
 
 	assert.Equal(t, 0.5, l.IngestionRate, "from json")
 	assert.Equal(t, 100, l.MaxLabelNameLength, "from defaults")
+
+	// Unmarshal should fail if input contains unknown struct fields and
+	// the decoder flag `json.Decoder.DisallowUnknownFields()` is set
+	inp = `{"unknown_fields": 100}`
+	l = Limits{}
+	dec := json.NewDecoder(strings.NewReader(inp))
+	dec.DisallowUnknownFields()
+	err = dec.Decode(&l)
+	assert.Error(t, err)
 }
 
 func TestLimitsTagsYamlMatchJson(t *testing.T) {