Add global ingestion rate limiter to distributors (#1766)

* Upgraded golang.org/x/time introducing rate.Limit.SetBurst()

Signed-off-by: Marco Pracucci <[email protected]>

* Added a global ingestion rate limiter

Signed-off-by: Marco Pracucci <[email protected]>

* Commented ReadRegistry mock to fix linter errors

Signed-off-by: Marco Pracucci <[email protected]>

* Added PR number to changelog entry

Signed-off-by: Marco Pracucci <[email protected]>

* Replaced distributors service discovery with the ring

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed linter errors

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed TSDB ingester initialization after rebasing with master

Signed-off-by: Marco Pracucci <[email protected]>

* Tested distributor.RingConfig

Signed-off-by: Marco Pracucci <[email protected]>

* Enriched ring.Lifecycler tests

Signed-off-by: Marco Pracucci <[email protected]>

* Renamed distributor's internalDependency param into canJoinRing to make it more clear

Signed-off-by: Marco Pracucci <[email protected]>

* Removed 'consul' and 'ingester' from ring logs, because it's not necessarily a ring of ingesters backed by consul

Signed-off-by: Marco Pracucci <[email protected]>

* Fixes after rebasing with master

Signed-off-by: Marco Pracucci <[email protected]>

* Changed ring key reference in the ingester, to keep code consistency

Signed-off-by: Marco Pracucci <[email protected]>

* Moved ReadLifecycler interface definition at the place of consumption

Signed-off-by: Marco Pracucci <[email protected]>

* Re-introduced -distributor.limiter-reload-period but deprecated

Signed-off-by: Marco Pracucci <[email protected]>

* Removed spurious bmizerany/assert dependency

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed issues after rebasing master

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed issues after rebasing master

Signed-off-by: Marco Pracucci <[email protected]>

* Updated changed with a CHANGE entry too

Signed-off-by: Marco Pracucci <[email protected]>

* Updated doc

Signed-off-by: Marco Pracucci <[email protected]>

* Renamed variable in distributor.New()

Signed-off-by: Marco Pracucci <[email protected]>

* Fix distributor tests when running on a system without eth0 and en0

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

CHANGELOG.md

@@ -10,7 +10,9 @@
   * `ruler.storage.type` has beem added to specify the rule store backend type, currently only the configdb.
   * `ruler.poll-interval` has been added to specify the interval in which to poll new rule groups.
 * [CHANGE] Use relative links from /ring page to make it work when used behind reverse proxy. #1896
+* [CHANGE] Deprecated `-distributor.limiter-reload-period` flag. #1766
 * [FEATURE] The distributor can now drop labels from samples (similar to the removal of the replica label for HA ingestion) per user via the `distributor.drop-label` flag. #1726
+* [FEATURE] Added `global` ingestion rate limiter strategy. Deprecated `-distributor.limiter-reload-period` flag. #1766
 * [BUGFIX] Fixed unnecessary CAS operations done by the HA tracker when the jitter is enabled. #1861
 
 ## 0.4.0 / 2019-12-02

docs/arguments.md

@@ -291,12 +291,17 @@ When running Cortex on Kubernetes, store this file in a config map and mount it
 
 Valid fields are (with their corresponding flags for default values):
 
+- `ingestion_rate_strategy` / `-distributor.ingestion-rate-limit-strategy`
 - `ingestion_rate` / `-distributor.ingestion-rate-limit`
 - `ingestion_burst_size` / `-distributor.ingestion-burst-size`
 
-  The per-tenant rate limit (and burst size), in samples per second. Enforced on a per distributor basis, actual effective rate limit will be N times higher, where N is the number of distributor replicas.
+  The per-tenant rate limit (and burst size), in samples per second. It supports two strategies: `local` (default) and `global`.
 
-  **NB** Limits are reset every `-distributor.limiter-reload-period`, as such if you set a very high burst limit it will never be hit.
+  The `local` strategy enforces the limit on a per distributor basis, actual effective rate limit will be N times higher, where N is the number of distributor replicas.
+
+  The `global` strategy enforces the limit globally, configuring a per-distributor local rate limiter as `ingestion_rate / N`, where N is the number of distributor replicas (it's automatically adjusted if the number of replicas change). The `ingestion_burst_size` refers to the per-distributor local rate limiter (even in the case of the `global` strategy) and should be set at least to the maximum number of samples expected in a single push request.
+
+  The `global` strategy requires the distributors to form their own ring, which is used to keep track of the current number of healthy distributor replicas. The ring is configured by `distributor: { ring: {}}` / `-distributor.ring.*`.
 
 - `max_label_name_length` / `-validation.max-length-label-name`
 - `max_label_value_length` / `-validation.max-length-label-value`

go.mod

@@ -67,7 +67,7 @@ require (
 	go.etcd.io/bbolt v1.3.3
 	go.etcd.io/etcd v0.0.0-20190709142735-eb7dd97135a5
 	golang.org/x/net v0.0.0-20190923162816-aa69164e4478
-	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4
+	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
 	google.golang.org/api v0.11.0
 	google.golang.org/grpc v1.25.1
 	gopkg.in/yaml.v2 v2.2.2

go.sum

@@ -794,6 +794,8 @@ golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxb
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20191024005414-555d28b269f0 h1:/5xXl8Y5W96D+TtHSlonuFqGHIWVuyCkGJLwGh9JJFs=
+golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

pkg/cortex/modules.go

@@ -151,7 +151,7 @@ func (t *Cortex) stopServer() (err error) {
 }
 
 func (t *Cortex) initRing(cfg *Config) (err error) {
-	t.ring, err = ring.New(cfg.Ingester.LifecyclerConfig.RingConfig, "ingester")
+	t.ring, err = ring.New(cfg.Ingester.LifecyclerConfig.RingConfig, "ingester", ring.IngesterRingKey)
 	if err != nil {
 		return
 	}
@@ -171,7 +171,14 @@ func (t *Cortex) stopOverrides() error {
 }
 
 func (t *Cortex) initDistributor(cfg *Config) (err error) {
-	t.distributor, err = distributor.New(cfg.Distributor, cfg.IngesterClient, t.overrides, t.ring)
+	cfg.Distributor.DistributorRing.ListenPort = cfg.Server.GRPCListenPort
+
+	// Check whether the distributor can join the distributors ring, which is
+	// whenever it's not running as an internal dependency (ie. querier or
+	// ruler's dependency)
+	canJoinDistributorsRing := (cfg.Target == All || cfg.Target == Distributor)
+
+	t.distributor, err = distributor.New(cfg.Distributor, cfg.IngesterClient, t.overrides, t.ring, canJoinDistributorsRing)
 	if err != nil {
 		return
 	}

pkg/distributor/distributor.go

@@ -7,10 +7,8 @@ import (
 	"net/http"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
-	"golang.org/x/time/rate"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
 	opentracing "github.com/opentracing/opentracing-go"
@@ -25,6 +23,8 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/extract"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/limiter"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 	billing "github.com/weaveworks/billing-client"
 	"github.com/weaveworks/common/httpgrpc"
@@ -97,18 +97,20 @@ var (
 // forwards appends and queries to individual ingesters.
 type Distributor struct {
 	cfg           Config
-	ring          ring.ReadRing
+	ingestersRing ring.ReadRing
 	ingesterPool  *ingester_client.Pool
 	limits        *validation.Overrides
 	billingClient *billing.Client
 
+	// The global rate limiter requires a distributors ring to count
+	// the number of healthy instances
+	distributorsRing *ring.Lifecycler
+
 	// For handling HA replicas.
 	Replicas *haTracker
 
-	// Per-user rate limiters.
-	ingestLimitersMtx sync.RWMutex
-	ingestLimiters    map[string]*rate.Limiter
-	quit              chan struct{}
+	// Per-user rate limiter.
+	ingestionRateLimiter *limiter.RateLimiter
 }
 
 // Config contains the configuration require to
@@ -120,28 +122,31 @@ type Config struct {
 
 	HATrackerConfig HATrackerConfig `yaml:"ha_tracker,omitempty"`
 
-	MaxRecvMsgSize      int           `yaml:"max_recv_msg_size"`
-	RemoteTimeout       time.Duration `yaml:"remote_timeout,omitempty"`
-	ExtraQueryDelay     time.Duration `yaml:"extra_queue_delay,omitempty"`
-	LimiterReloadPeriod time.Duration `yaml:"limiter_reload_period,omitempty"`
+	MaxRecvMsgSize  int           `yaml:"max_recv_msg_size"`
+	RemoteTimeout   time.Duration `yaml:"remote_timeout,omitempty"`
+	ExtraQueryDelay time.Duration `yaml:"extra_queue_delay,omitempty"`
 
 	ShardByAllLabels bool `yaml:"shard_by_all_labels,omitempty"`
 
+	// Distributors ring
+	DistributorRing RingConfig `yaml:"ring,omitempty"`
+
 	// for testing
-	ingesterClientFactory client.Factory
+	ingesterClientFactory client.Factory `yaml:"-"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.BillingConfig.RegisterFlags(f)
 	cfg.PoolConfig.RegisterFlags(f)
 	cfg.HATrackerConfig.RegisterFlags(f)
+	cfg.DistributorRing.RegisterFlags(f)
 
 	f.BoolVar(&cfg.EnableBilling, "distributor.enable-billing", false, "Report number of ingested samples to billing system.")
 	f.IntVar(&cfg.MaxRecvMsgSize, "distributor.max-recv-msg-size", 100<<20, "remote_write API max receive message size (bytes).")
 	f.DurationVar(&cfg.RemoteTimeout, "distributor.remote-timeout", 2*time.Second, "Timeout for downstream ingesters.")
 	f.DurationVar(&cfg.ExtraQueryDelay, "distributor.extra-query-delay", 0, "Time to wait before sending more than the minimum successful query requests.")
-	f.DurationVar(&cfg.LimiterReloadPeriod, "distributor.limiter-reload-period", 5*time.Minute, "Period at which to reload user ingestion limits.")
+	flagext.DeprecatedFlag(f, "distributor.limiter-reload-period", "DEPRECATED. No more required because the local limiter is reconfigured as soon as the overrides change.")
 	f.BoolVar(&cfg.ShardByAllLabels, "distributor.shard-by-all-labels", false, "Distribute samples based on all labels, as opposed to solely by user and metric name.")
 }
 
@@ -151,7 +156,7 @@ func (cfg *Config) Validate() error {
 }
 
 // New constructs a new Distributor
-func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Overrides, ring ring.ReadRing) (*Distributor, error) {
+func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Overrides, ingestersRing ring.ReadRing, canJoinDistributorsRing bool) (*Distributor, error) {
 	if cfg.ingesterClientFactory == nil {
 		cfg.ingesterClientFactory = func(addr string) (grpc_health_v1.HealthClient, error) {
 			return ingester_client.MakeIngesterClient(addr, clientConfig)
@@ -167,55 +172,57 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 		}
 	}
 
-	replicationFactor.Set(float64(ring.ReplicationFactor()))
+	replicationFactor.Set(float64(ingestersRing.ReplicationFactor()))
 	cfg.PoolConfig.RemoteTimeout = cfg.RemoteTimeout
 
 	replicas, err := newClusterTracker(cfg.HATrackerConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	d := &Distributor{
-		cfg:            cfg,
-		ring:           ring,
-		ingesterPool:   ingester_client.NewPool(cfg.PoolConfig, ring, cfg.ingesterClientFactory, util.Logger),
-		billingClient:  billingClient,
-		limits:         limits,
-		ingestLimiters: map[string]*rate.Limiter{},
-		quit:           make(chan struct{}),
-		Replicas:       replicas,
-	}
-	go d.loop()
-
-	return d, nil
-}
+	// Create the configured ingestion rate limit strategy (local or global). In case
+	// it's an internal dependency and can't join the distributors ring, we skip rate
+	// limiting.
+	var ingestionRateStrategy limiter.RateLimiterStrategy
+	var distributorsRing *ring.Lifecycler
 
-func (d *Distributor) loop() {
-	if d.cfg.LimiterReloadPeriod == 0 {
-		return
-	}
+	if !canJoinDistributorsRing {
+		ingestionRateStrategy = newInfiniteIngestionRateStrategy()
+	} else if limits.IngestionRateStrategy() == validation.GlobalIngestionRateStrategy {
+		distributorsRing, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ring.DistributorRingKey)
+		if err != nil {
+			return nil, err
+		}
 
-	ticker := time.NewTicker(d.cfg.LimiterReloadPeriod)
-	defer ticker.Stop()
+		distributorsRing.Start()
 
-	for {
-		select {
-		case <-ticker.C:
-			d.ingestLimitersMtx.Lock()
-			d.ingestLimiters = make(map[string]*rate.Limiter, len(d.ingestLimiters))
-			d.ingestLimitersMtx.Unlock()
+		ingestionRateStrategy = newGlobalIngestionRateStrategy(limits, distributorsRing)
+	} else {
+		ingestionRateStrategy = newLocalIngestionRateStrategy(limits)
+	}
 
-		case <-d.quit:
-			return
-		}
+	d := &Distributor{
+		cfg:                  cfg,
+		ingestersRing:        ingestersRing,
+		ingesterPool:         ingester_client.NewPool(cfg.PoolConfig, ingestersRing, cfg.ingesterClientFactory, util.Logger),
+		billingClient:        billingClient,
+		distributorsRing:     distributorsRing,
+		limits:               limits,
+		ingestionRateLimiter: limiter.NewRateLimiter(ingestionRateStrategy, 10*time.Second),
+		Replicas:             replicas,
 	}
+
+	return d, nil
 }
 
 // Stop stops the distributor's maintenance loop.
 func (d *Distributor) Stop() {
-	close(d.quit)
 	d.ingesterPool.Stop()
 	d.Replicas.stop()
+
+	if d.distributorsRing != nil {
+		d.distributorsRing.Shutdown()
+	}
 }
 
 func (d *Distributor) tokenForLabels(userID string, labels []client.LabelAdapter) (uint32, error) {
@@ -410,18 +417,18 @@ func (d *Distributor) Push(ctx context.Context, req *client.WriteRequest) (*clie
 		return &client.WriteResponse{}, lastPartialErr
 	}
 
-	limiter := d.getOrCreateIngestLimiter(userID)
-	if !limiter.AllowN(time.Now(), validatedSamples) {
+	now := time.Now()
+	if !d.ingestionRateLimiter.AllowN(now, userID, validatedSamples) {
 		// Ensure the request slice is reused if the request is rate limited.
 		client.ReuseSlice(req.Timeseries)
 
 		// Return a 4xx here to have the client discard the data and not retry. If a client
 		// is sending too much data consistently we will unlikely ever catch up otherwise.
 		validation.DiscardedSamples.WithLabelValues(validation.RateLimited, userID).Add(float64(validatedSamples))
-		return nil, httpgrpc.Errorf(http.StatusTooManyRequests, "ingestion rate limit (%v) exceeded while adding %d samples", limiter.Limit(), numSamples)
+		return nil, httpgrpc.Errorf(http.StatusTooManyRequests, "ingestion rate limit (%v) exceeded while adding %d samples", d.ingestionRateLimiter.Limit(now, userID), numSamples)
 	}
 
-	err = ring.DoBatch(ctx, d.ring, keys, func(ingester ring.IngesterDesc, indexes []int) error {
+	err = ring.DoBatch(ctx, d.ingestersRing, keys, func(ingester ring.IngesterDesc, indexes []int) error {
 		timeseries := make([]client.PreallocTimeseries, 0, len(indexes))
 		for _, i := range indexes {
 			timeseries = append(timeseries, validatedTimeseries[i])
@@ -442,24 +449,6 @@ func (d *Distributor) Push(ctx context.Context, req *client.WriteRequest) (*clie
 	return &client.WriteResponse{}, lastPartialErr
 }
 
-func (d *Distributor) getOrCreateIngestLimiter(userID string) *rate.Limiter {
-	d.ingestLimitersMtx.RLock()
-	limiter, ok := d.ingestLimiters[userID]
-	d.ingestLimitersMtx.RUnlock()
-
-	if ok {
-		return limiter
-	}
-
-	limiter = rate.NewLimiter(rate.Limit(d.limits.IngestionRate(userID)), d.limits.IngestionBurstSize(userID))
-
-	d.ingestLimitersMtx.Lock()
-	d.ingestLimiters[userID] = limiter
-	d.ingestLimitersMtx.Unlock()
-
-	return limiter
-}
-
 func (d *Distributor) sendSamples(ctx context.Context, ingester ring.IngesterDesc, timeseries []client.PreallocTimeseries) error {
 	h, err := d.ingesterPool.GetClientFor(ingester.Addr)
 	if err != nil {
@@ -481,7 +470,7 @@ func (d *Distributor) sendSamples(ctx context.Context, ingester ring.IngesterDes
 
 // 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) {
-	replicationSet, err := d.ring.GetAll()
+	replicationSet, err := d.ingestersRing.GetAll()
 	if err != nil {
 		return nil, err
 	}
@@ -603,8 +592,8 @@ func (d *Distributor) UserStats(ctx context.Context) (*UserStats, error) {
 		totalStats.NumSeries += r.NumSeries
 	}
 
-	totalStats.IngestionRate /= float64(d.ring.ReplicationFactor())
-	totalStats.NumSeries /= uint64(d.ring.ReplicationFactor())
+	totalStats.IngestionRate /= float64(d.ingestersRing.ReplicationFactor())
+	totalStats.NumSeries /= uint64(d.ingestersRing.ReplicationFactor())
 
 	return totalStats, nil
 }
@@ -624,7 +613,7 @@ func (d *Distributor) AllUserStats(ctx context.Context) ([]UserIDStats, error) {
 	req := &client.UserStatsRequest{}
 	ctx = user.InjectOrgID(ctx, "1") // fake: ingester insists on having an org ID
 	// Not using d.forAllIngesters(), so we can fail after first error.
-	replicationSet, err := d.ring.GetAll()
+	replicationSet, err := d.ingestersRing.GetAll()
 	if err != nil {
 		return nil, err
 	}