feat: support per policy overriden stream limits on limits service by salvacorts · Pull Request #19403 · grafana/loki

salvacorts · 2025-10-07T11:51:54Z

What this PR does / why we need it:

This is a followup for #18994. Here we update the new limits service to support the policy-overridable stream limits.

We modified the tenantUsage to add a new map of policies so we can track the number of streams per policy independently. This way, the stream of a policy with a stream limit override does not account for the global stream limit and vice-versa.

Special notes for reviewer:

I see two ways to get the policy for each stream on the stream service:

Option 1 - What's implemented in this PR:

Update: @grobinson-grafana and I agreed on this approach

We resolve the policy for each stream on the distributor, and pass the policy (if any) on the proto.StreamMetadata that is part of the proto.ExceedsLimitsRequest distributor sends to the limits frontend and this to the limits service.

Resolving the policies for a stream can be somewhat expensive as we need to check the stream matchers from the mappings against the stream labels. This way we only resolve the policies once on the distributors which is already done anyways.

Option 2

Resolve the policies again on the limits service. This would make a bit of more work, but will reduce the request size. Plus we can decide to only resolve the policy for those policies that are overriding the stream limits.

We are considering allowing to set the policy via a header on the gateway, so this option may end up being a bit more complicated since we'd need to pass the header from the distributors down to the limits service.

Checklist

Reviewed the CONTRIBUTING.md guide (required)
Documentation added
Tests updated
Title matches the required conventional commits format, see here
- Note that Promtail is considered to be feature complete, and future development for logs collection will be in Grafana Alloy. As such, feat PRs are unlikely to be accepted unless a case can be made for the feature actually being a bug fix to existing behavior.
Changes that require user attention or interaction to upgrade are documented in docs/sources/setup/upgrade/_index.md
If the change is deprecating or removing a configuration option, update the deprecated-config.yaml and deleted-config.yaml files respectively in the tools/deprecated-config-checker directory. Example PR

salvacorts · 2025-10-10T09:37:29Z

I tested it in dev and works fine

Given these tenant overrides

I pushed 5k streams for source=app, and 1k for source=synthetic-monitoring. I modified the endpoint on the limits-service to return the per policy streams and as can be seen, the sum of total streams is indeed over the configured max_global_streams_per_user: 5000, but the no-poliocy bucket is below 5k and the synthetic-monitoring policy bucket is also below 1k.

…s-service

grobinson-grafana · 2025-10-09T09:56:11Z

  uint64 streamHash = 1;
  uint64 totalSize = 2;
+  // The resolved ingestion policy for this stream if any. May be used to override some ingestion limits for this stream such as the max streams allowed.
+  string ingestionPolicy = 3;


Is this an optional field? If so, I think it should have optional, this means the field is not serialized to the wire if there is no value.

Doesn't look like our current proto version supports optionals.

$ make protos ... pkg/limits/proto/limits.proto: is a proto3 file that contains optional fields, but code generator protoc-gen-gogoslick hasn't been updated to support optional fields in proto3. Please ask the owner of this code generator to support proto3 optional.--gogoslick_out:

grobinson-grafana · 2025-10-09T09:56:22Z

 	client           kafkaConsumer
 	partitionManager *partitionManager
 	usage            *usageStore
+	limits           Limits


Not used, can be removed?

grobinson-grafana · 2025-10-09T10:44:45Z

 // tenantUsage contains the per-partition stream usage for a tenant.
-type tenantUsage map[int32]map[uint64]streamUsage
+// The structure is: partition -> policy -> streamHash -> streamUsage
+// Policy "" represents streams that don't match any specific policy.


Normally I wouldn't bother, but let's define a const NoPolicy as empty string?

Agree, I think it makes it easier to read and reason about

grobinson-grafana · 2025-10-09T10:48:04Z

 	// implementing rate limits to a later date in the future.
 	totalSize   uint64
 	rateBuckets []rateBucket
+	policy      string // The ingestion policy for this stream if any


I did some quick napkin math and if we assume 1 million streams, an average of ~100 bytes per string, this would consume an extra 95MB of memory. If we use a uint64 hash, we consume an extra 7.5MB of memory. Not sure this is worth optimizing for now.

I think this is a valid concern, but on the other hand the amount of streams that will have this field populated will be smaller. Only the streams of policies (i.e. products in our usecase) we override the stream limits for.

My vote goes to leave is as it now: simpler to reason about and less compute to get the hash of the policy. I will leave a comment on how to optimize this if necessary.

On a related note, I fixed this line to use the policyBucket instead. That way we only populate it for streams for which we have a stream limit override.

loki/pkg/limits/store.go

Lines 388 to 390 in 40d8600

stream.hash = streamHash

stream.totalSize = 0

stream.policy = metadata.IngestionPolicy

grobinson-grafana · 2025-10-23T15:22:22Z

-	Rate    float64 `json:"rate"`
+	Tenant           string            `json:"tenant"`
+	Streams          uint64            `json:"streams"`
+	PerPolicyStreams map[string]uint64 `json:"per_policy_streams"`


Suggested change

PerPolicyStreams map[string]uint64 `json:"per_policy_streams"`

StreamsByPolicy map[string]uint64 `json:"streams_by_policy"`

grobinson-grafana · 2025-10-23T15:23:51Z

-	Streams uint64  `json:"streams"`
-	Rate    float64 `json:"rate"`
+	Tenant           string            `json:"tenant"`
+	Streams          uint64            `json:"streams"`


Suggested change

Streams uint64 `json:"streams"`

StreamsTotal uint64 `json:"streams_total"`

grobinson-grafana · 2025-10-23T15:31:28Z

+// for a given tenant and policy. Returns the policy bucket name and the max streams limit.
+// The policy bucket will be the input policy name only if the max streams limit is overriden for the policy.
+func (s *usageStore) getPolicyBucketAndLimit(tenant, policy string) (policyBucket string, maxStreams uint64) {
+	defaultMaxStreams := uint64(math.Max(float64(s.limits.MaxGlobalStreamsPerUser(tenant)/s.numPartitions), 1))


I think the use of math.Max is incorrect. 0 should disable it (which I think is also missing from the current implementation – 0 is interpreted as a literal 0).

I added this because if the stream limit is small (e.g. 50) and the numParitions is bigger (e.g. 64) this would return 0.

That's not a reasonable stream limit tho 😂 but it's what I configured when I was testing this and I found the edge-case. This will not happen on the real world where stream limits are >thousands.

Still I wonder if a ceil would make more sense here. 10000/64=156.25, should we ceil it to 157?
Tbf I don't think this will solve any actual problem, so I think we can skip the ceiling compute.

So there are two problems here:

Limit is set to 0

Limit < number of partitions, such that limit / num_partitions is less than 1

The first one (limit is set to 0) is broken in both main and this PR. Looking at how these limits work in the ingesters, limit set to 0 is supposed to mean unlimited.

I suggest we remove the math.Max for now, and we solve both 1 and 2 in a future PR.

Agree. My latest commit reverted that:

loki/pkg/limits/store.go

Lines 44 to 53 in 5133697

func (s *usageStore) getPolicyBucketAndStreamsLimit(tenant, policy string) (policyBucket string, maxStreams uint64) {

defaultMaxStreams := uint64(s.limits.MaxGlobalStreamsPerUser(tenant) / s.numPartitions)

if policy != noPolicy {

if policyMaxStreams := s.limits.PolicyMaxGlobalStreamsPerUser(tenant, policy); policyMaxStreams > 0 {

return policy, uint64(policyMaxStreams / s.numPartitions) // Use policy-specific bucket

}

}

return "", defaultMaxStreams // Use default bucket

}

grobinson-grafana · 2025-10-23T15:32:12Z

-			streams := s.stripes[i][tenant][partition]
+
+			// Determine which policy bucket to use and the max streams limit
+			policyBucket, maxStreamsForStream := s.getPolicyBucketAndLimit(tenant, m.IngestionPolicy)


Suggested change

policyBucket, maxStreamsForStream := s.getPolicyBucketAndLimit(tenant, m.IngestionPolicy)

policyBucket, maxStreamsForPolicy := s.getPolicyBucketAndLimit(tenant, m.IngestionPolicy)

I assume?

Actually I meant it to be like that. I'm changing it back to maxStreams

grobinson-grafana · 2025-10-23T15:33:44Z

+	// Search across all policies for this stream
+	// Note that in most cases, there will only be one item on this list (empty policy, "").
+	// and at most just a few items. One idea to speed this up would be to keep a cache of tenant-partition-streamHash -> policy for those streams that have a matching policy.
+	for _, streams := range policies {


I think the issue here is that if a stream is present in two policies at the same time, this is non-deterministic now because iterating maps is non-deterministic in Go.

I think we can just remove the Get method for now, it isn't used.

I removed the Get method.

In the future if we add this back, I think the stream should be in both the noPolicy ("") bucket and a named policy bucket when we strat configuring those streams to be assigned to a particular policy. In which case I think we could remove the stream form the noPolicy bucket when it is added for the first time.

…s-service

…rafana#19403)

feat: support per policy overriden stream limits on limits service

63813d2

pull-request-size Bot added the size/L label Oct 7, 2025

track overriden policies separately

9c4442f

salvacorts force-pushed the salvacorts/per-policy-stream-limits-on-limits-service branch from 7c5cd05 to 40895ee Compare October 8, 2025 10:17

add test

5f1990f

salvacorts force-pushed the salvacorts/per-policy-stream-limits-on-limits-service branch from 40895ee to 5f1990f Compare October 8, 2025 10:30

salvacorts added 2 commits October 10, 2025 09:52

debug

bb41bc9

undo debug prints

6241b34

pull-request-size Bot added size/XL and removed size/L labels Oct 10, 2025

salvacorts marked this pull request as ready for review October 10, 2025 09:37

salvacorts requested a review from a team as a code owner October 10, 2025 09:37

Merge branch 'main' into salvacorts/per-policy-stream-limits-on-limit…

40d8600

…s-service

grobinson-grafana reviewed Oct 23, 2025

View reviewed changes

salvacorts added 2 commits October 24, 2025 11:59

CR feedback

1dae462

Merge branch 'main' into salvacorts/per-policy-stream-limits-on-limit…

5133697

…s-service

grobinson-grafana approved these changes Oct 27, 2025

View reviewed changes

salvacorts added 2 commits October 27, 2025 11:08

Honor 0 stream limit and unlimited

ca2cc25

unlimited check

d1bcb09

salvacorts merged commit 8486acb into main Oct 27, 2025
63 checks passed

salvacorts deleted the salvacorts/per-policy-stream-limits-on-limits-service branch October 27, 2025 12:01

felix0102 pushed a commit to felix0102/loki that referenced this pull request Aug 2, 2026

feat: support per policy overriden stream limits on limits service (g…

d91e5b4

…rafana#19403)

Navigation Menu

Uh oh!

feat: support per policy overriden stream limits on limits service - #19403

feat: support per policy overriden stream limits on limits service#19403
salvacorts merged 10 commits into
mainfrom
salvacorts/per-policy-stream-limits-on-limits-service

salvacorts commented Oct 7, 2025 •

edited

Loading

salvacorts commented Oct 10, 2025

grobinson-grafana Oct 9, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 9, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 9, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 9, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 23, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 23, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 23, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 24, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 23, 2025

salvacorts Oct 24, 2025

grobinson-grafana Oct 23, 2025

grobinson-grafana Oct 24, 2025

salvacorts Oct 24, 2025

Uh oh!

Labels

2 participants

	stream.hash = streamHash
	stream.totalSize = 0
	stream.policy = metadata.IngestionPolicy

	PerPolicyStreams map[string]uint64 `json:"per_policy_streams"`
	StreamsByPolicy map[string]uint64 `json:"streams_by_policy"`

	Streams uint64 `json:"streams"`
	StreamsTotal uint64 `json:"streams_total"`

	func (s *usageStore) getPolicyBucketAndStreamsLimit(tenant, policy string) (policyBucket string, maxStreams uint64) {
	defaultMaxStreams := uint64(s.limits.MaxGlobalStreamsPerUser(tenant) / s.numPartitions)

	if policy != noPolicy {
	if policyMaxStreams := s.limits.PolicyMaxGlobalStreamsPerUser(tenant, policy); policyMaxStreams > 0 {
	return policy, uint64(policyMaxStreams / s.numPartitions) // Use policy-specific bucket
	}
	}
	return "", defaultMaxStreams // Use default bucket
	}

	policyBucket, maxStreamsForStream := s.getPolicyBucketAndLimit(tenant, m.IngestionPolicy)
	policyBucket, maxStreamsForPolicy := s.getPolicyBucketAndLimit(tenant, m.IngestionPolicy)

Uh oh!

Conversation

salvacorts commented Oct 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What this PR does / why we need it:

Special notes for reviewer:

Option 1 - What's implemented in this PR:

Option 2

salvacorts commented Oct 10, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Uh oh!

Labels

2 participants

salvacorts commented Oct 7, 2025 •

edited

Loading