feat: Add limit on max buffered bytes in pattern-tee

[grobinson-grafana]

What this PR does / why we need it:

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
• 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

---

Note

Medium Risk
Adds a new buffering limit that can drop streams/flushes when the tee’s in-memory/queued payloads exceed a configured size, which changes backpressure/dropping behavior under load. The logic touches concurrency and accounting across buffering, batching, and send paths, so miscounting could lead to unexpected drops or memory growth.

Overview
Adds a new tee_config.max_buffered_bytes/-pattern-ingester.tee.max-buffered-bytes setting to cap how many log bytes the pattern tee can buffer (disabled by default).

TeeService now tracks total buffered bytes (in-memory buffer + queued flush requests) using atomic accounting; streams/flush requests are dropped when the cap would be exceeded, and bytes are released when requests are sent or dropped. A new max-sample metric pattern_ingester_tee_buffered_bytes is exported, and tests are updated/added to cover empty streams and the new limit behavior.

^{Reviewed by Cursor Bugbot for commit 94cfbdd. Bugbot is set up for automated code reviews on this repo. Configure here.}

[github-actions]

💻 Deploy preview deleted (feat: Add limit on max buffered bytes in pattern-tee).

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: New Inc and Dec methods are unused
  • Removed the unused MaxSampleCollector Inc and Dec convenience methods while leaving the Add/Sub API used by callers intact.

Or push these changes by commenting:

@cursor push 0c77582240

Preview (0c77582240)

diff --git a/pkg/util/metric/max_sample_collector.go b/pkg/util/metric/max_sample_collector.go
--- a/pkg/util/metric/max_sample_collector.go
+++ b/pkg/util/metric/max_sample_collector.go
@@ -40,16 +40,6 @@
 	c.val.Add(-delta)
 }
 
-// Inc increments the counter.
-func (c *MaxSampleCollector) Inc() {
-	c.val.Add(1)
-}
-
-// Dec decrements the counter.
-func (c *MaxSampleCollector) Dec() {
-	c.val.Add(-1)
-}
-
 // Describe implements [prometheus.Collector].
 func (c *MaxSampleCollector) Describe(descs chan<- *prometheus.Desc) {
 	descs <- c.desc

_{You can send follow-ups to the cloud agent here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Test uses require.Contains incorrectly, missing second element check
  • Replaced each multi-value Contains assertion with separate checks for s1 and s3 so both buffered streams are verified.

Or push these changes by commenting:

@cursor push 8c120028cc

Preview (8c120028cc)

diff --git a/pkg/pattern/tee_service_test.go b/pkg/pattern/tee_service_test.go
--- a/pkg/pattern/tee_service_test.go
+++ b/pkg/pattern/tee_service_test.go
@@ -247,7 +247,8 @@
 	// tenantBuf should contain s1 and s3.
 	tenantBuf, ok = tee.buf["test"]
 	require.True(t, ok)
-	require.Contains(t, tenantBuf, s1, s3)
+	require.Contains(t, tenantBuf, s1)
+	require.Contains(t, tenantBuf, s3)
 
 	// Stream should be rejected, total of s1, s3 and s4 is more than 1KB.
 	s4 := distributor.KeyedStream{
@@ -265,7 +266,8 @@
 	// tenantBuf should contain s1 and s3.
 	tenantBuf, ok = tee.buf["test"]
 	require.True(t, ok)
-	require.Contains(t, tenantBuf, s1, s3)
+	require.Contains(t, tenantBuf, s1)
+	require.Contains(t, tenantBuf, s3)
 
 	// Flush s1 and s3, s4 should be accepted.
 	tee.flush()

_{You can send follow-ups to the cloud agent here.}

^{Reviewed by Cursor Bugbot for commit 7adeba9. Configure here.}

The base branch was changed.