Add global ingestion rate limiter to distributors

[pracucci]

Following up this design doc, in this PR I'm proposing to introduce a global ingestion rate limiter implemented as a local rate limiter configured with limit / N, where N is the actual number of distributors.

Fixes #1090

Notes about the distributors ring

We currently don't have a ring of distributors, so I had to introduce it.

For this purpose, in this PR I'm proposing to introduce a generic simple service discovery based on the KVStore, instead of the ring itself. It's an opinionated change. If there's no consensus, I can rollback to reuse the ring implementation for distributors too.

From my perspective, a generic lightweight internal service discovery/registry would be just easier to plug compared to the current ring, because of:

• No need to have a real ring (ie. no need for tokens)
• No need to deal with IngesterState
• No need to have FlushTransferer

Notes about alternatives

• I've excluded to use Kubernetes API to figure out the current number of distributors replica, to not introduce Kubernetes as an hard dependency.

[gouthamve]

For this purpose, in this PR I'm proposing to introduce a generic simple service discovery based on the KVStore, instead of the ring itself. It's an opinionated change. If there's no consensus, I can rollback to reuse the ring implementation for distributors too.

I don't want to maintain another way of doing service discovery tbh. I'd would just use the current one again. There is no proof that the current method is limiting us in a anyway and my guess would be that people would be inclined to use the gossip backend for rate-limiting SD which should be even lesser IO.

If we find that the ring based implementation is too IO intensive / becoming a bottleneck, we can then add a new one.

[pstibrany]

Nice work. This is a straightforward extension of ring idea. I like that.

[pstibrany]

I don't want to maintain another way of doing service discovery tbh. I'd would just use the current one again. There is no proof that the current method is limiting us in a anyway and my guess would be that people would be inclined to use the gossip backend for rate-limiting SD which should be even lesser IO.

This code is reusing existing KVStore, but is storing a different structure into it, under a different key. I think that makes sense, instead of making ring structure bigger. It's written in a generic way (I don't see a need for that, for now, but is harmless otherwise) and called "service discovery", but idea is the same as in the ring (without tokens).

[pracucci]

@pstibrany and @gouthamve thanks for your comments. I've addressed the feedback:

1. Removed the service discovery in favour of the ring
2. The FlushTransferer in the ring is no more required (ie. distributors don't need it)
3. I've generalized the ring.Lifecycler logs, removing any reference to "ingester" or "consul" (from the logs) because misleading (not necessarily a ring of ingesters, or backed by consul)
4. I've masked the LifecyclerConfig behind a custom ring config for the distributors, so that we expose only the config options / CLI flags which make sense for the distributor (see distributor_ring.go)

May you do another round of review, please?

[pstibrany]

Looks good! Nice work. (I like fixed error messages 👍 )

[pstibrany]

Still looks good to me :-)

[tomwilkie]

Ping @gouthamve?

[jtlisi]

LGTM
One nit and a suggestion if you have a chance. It would be nice to reuse the NoopFlushTransferer in the ruler. Currently the ruler basically implements NoopFlushTransferer itself.

[jtlisi]

Something similar for this would be nice for the ruler. The main difference being the number of tokens. However, that can probably be figured out at another time. Maybe the ring package could use a refactor to make it more friendly to simpler use cases.

[pracucci]

Right. I would suggest to keep ruler's refactoring as a separate PR, given it's not related to this work. I will add it to my backlog.

[jtlisi]

The ruler has identical functionality in pkg/ruler/lifecycle.go it would be cool to delete that and reuse this instead.

[pracucci]

The ruler Flush() is not a noop. Currently the NoopFlushTransferer is a adopt all or nothing: how would you see reusing it in the ruler?

[jtlisi]

True, I forgot the change to make the ruler flush has not been merged yet:
https://github.com/cortexproject/cortex/pull/1571/files#diff-95b5a43683ab83429e93fef5c5daf87fL26

[pracucci]

No problem. We can use the NoopFlushTransferer in the ruler as soon as the PR 1571 will be merged.

[owen-d]

Nice work -- I like the ring re-use.

[gouthamve]

Looks good, but have some minor nits!

[gouthamve]

Can be simplified to:

	if !canJoinRing {
		ingestionRateStrategy = newLocalIngestionRateStrategy(limits)
	} else if limits.IngestionRateStrategy() == validation.GlobalIngestionRateStrategy {
		distributorsRing, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ring.DistributorRingKey)
		if err != nil {
			return nil, err
		}

		distributorsRing.Start()

		ingestionRateStrategy = newGlobalIngestionRateStrategy(limits, distributorsRing)
	}

[gouthamve]

I think this means that it's never going to be infinite, maybe double the logic once?

[pracucci]

Not sure. I think the current logic is safer and the intention is more clear (when it's an internal dependency, do not configure any rate limit at all).

[codesome]

TestDistributor_PushIngestionRateLimiter fails in my local machine because none of these interfaces match :/ I will open a PR to get it from the system so that tests pass on all machines.

[codesome]

#1915