fix(ingest-limits): Get usage from owing instances only

[periklis]

What this PR does / why we need it:
This PR updates the stream usage retrieval routine in the ingest-limits-frontend component to query only ingest-limits instances that own the requested streams. The purpose of this optimization is to streamline traffic between these components to necessary minimum per ExceedLimits request.

Which issue(s) this PR fixes:
Fixes #

Special notes for your reviewer:

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

[github-actions]

💻 Deploy preview deleted.

[grobinson-grafana]

I am a bit worried that someone configures this for ingest-limits and forgets to configure it for ingest-limits-frontend. Perhaps we could share the configuration here? I'm not sure what's best...

[grobinson-grafana]

Can we remove partitions from GetStreamUsageRequest in logproto.proto? Does it still serve a use?

[periklis]

Can we remove partitions from GetStreamUsageRequest in logproto.proto? Does it still serve a use?

Yes it does because we want to check the requested patitions vs. the assigned ones. Between a the two RPC calls we still can get a re-assignment.

[grobinson-grafana]

What does this do (the whole block of code rather than the specific line)? I'm not sure I follow here

[periklis]

Glad that you point that out. The whole loop needs an overhaul here. This setup was written when the two requests types matched one to one.

[grobinson-grafana]

If I delete this, the tests still pass 😕

clients[i] = &mockIngestLimitsClient{
					expectedAssignedPartitionsRequest: test.expectedAssignedPartitionsRequest[i],
					getAssignedPartitionsResponse:     test.getAssignedPartitionsResponses[i],
					expectedStreamUsageRequest:        test.expectedStreamUsageRequest[i],
					getStreamUsageResponse:            test.getStreamUsageResponses[i],
					t:                                 t,
				}

[grobinson-grafana]

One of the tests is panicking in frontend.go it seems:

// Sum the number of active streams and rates of all responses.
	for _, resp := range resps {
		activeStreamsTotal += resp.Response.ActiveStreams
		rateTotal += float64(resp.Response.Rate)
	}

[grobinson-grafana]

One of the tests is panicking in frontend.go it seems:

// Sum the number of active streams and rates of all responses.
	for _, resp := range resps {
		activeStreamsTotal += resp.Response.ActiveStreams
		rateTotal += float64(resp.Response.Rate)
	}

It's because of j := 0 instead of j := i.

[grobinson-grafana]

Sorry but I still don't understand this? The tests pass without it. If you want to put in nils, wouldn't this work?

diff --git a/pkg/limits/frontend/ring_test.go b/pkg/limits/frontend/ring_test.go
index e7621d0cf..b3f2b7a48 100644
--- a/pkg/limits/frontend/ring_test.go
+++ b/pkg/limits/frontend/ring_test.go
@@ -106,7 +106,7 @@ func TestRingStreamUsageGatherer_GetStreamUsage(t *testing.T) {
                        Tenant:       "test",
                        StreamHashes: []uint64{1}, // Hash 1 maps to partition 1
                },
-               expectedAssignedPartitionsRequest: []*logproto.GetAssignedPartitionsRequest{{}, {}},
+               expectedAssignedPartitionsRequest: []*logproto.GetAssignedPartitionsRequest{nil, nil},
                getAssignedPartitionsResponses: []*logproto.GetAssignedPartitionsResponse{{
                        AssignedPartitions: map[int32]int64{
                                1: time.Now().Add(-time.Second).UnixNano(),
@@ -116,15 +116,17 @@ func TestRingStreamUsageGatherer_GetStreamUsage(t *testing.T) {
                                1: time.Now().UnixNano(),
                        },
                }},
-               expectedStreamUsageRequest: []*logproto.GetStreamUsageRequest{{}, {
-                       Tenant:       "test",
-                       StreamHashes: []uint64{1},
-               }},
-               getStreamUsageResponses: []*logproto.GetStreamUsageResponse{{}, {
-                       Tenant:        "test",
-                       ActiveStreams: 1,
-                       Rate:          10,
-               }},
+               expectedStreamUsageRequest: []*logproto.GetStreamUsageRequest{
+                       nil, {
+                               Tenant:       "test",
+                               StreamHashes: []uint64{1},
+                       }},
+               getStreamUsageResponses: []*logproto.GetStreamUsageResponse{
+                       nil, {
+                               Tenant:        "test",
+                               ActiveStreams: 1,
+                               Rate:          10,
+                       }},
                expectedResponses: []GetStreamUsageResponse{{
                        Addr: "instance-1",
                        Response: &logproto.GetStreamUsageResponse{
@@ -148,17 +150,11 @@ func TestRingStreamUsageGatherer_GetStreamUsage(t *testing.T) {
                                        t:                                 t,
                                        expectedAssignedPartitionsRequest: test.expectedAssignedPartitionsRequest[i],
                                        getAssignedPartitionsResponse:     test.getAssignedPartitionsResponses[i],
+                                       expectedStreamUsageRequest:        test.expectedStreamUsageRequest[i],
+                                       getStreamUsageResponse:            test.getStreamUsageResponses[i],
                                }
                        }

-                       // Set up the mock clients for the stream usage requests.
-                       // Note: Not every client will have a stream usage request because the
-                       // requested stream hashes are owned only by a subset of partition consumers.
-                       for i := range test.expectedStreamUsageRequest {
-                               clients[i].(*mockIngestLimitsClient).expectedStreamUsageRequest = test.expectedStreamUsageRequest[i]
-                               clients[i].(*mockIngestLimitsClient).getStreamUsageResponse = test.getStreamUsageResponses[i]
-                       }
-
                        // Set up the instances for the ring.
                        for i := range len(clients) {
                                instances[i] = ring.InstanceDesc{