fix: lint and format

docs/sources/shared/configuration.md

@@ -612,6 +612,166 @@ pattern_ingester:
   # CLI flag: -pattern-ingester.max-eviction-ratio
   [max_eviction_ratio: <float> | default = 0.25]
 
+  # Configures the metric aggregation and storage behavior of the pattern
+  # ingester.
+  metric_aggregation:
+    # Whether the pattern ingester metric aggregation is enabled.
+    # CLI flag: -pattern-ingester.metric-aggregation.enabled
+    [enabled: <boolean> | default = false]
+
+    # How often to downsample metrics from raw push observations.
+    # CLI flag: -pattern-ingester.metric-aggregation.downsample-period
+    [downsample_period: <duration> | default = 10s]
+
+    # The address of the Loki instance to push aggregated metrics to.
+    # CLI flag: -pattern-ingester.metric-aggregation.loki-address
+    [loki_address: <string> | default = ""]
+
+    # The timeout for writing to Loki.
+    # CLI flag: -pattern-ingester.metric-aggregation.timeout
+    [timeout: <duration> | default = 10s]
+
+    # How long to wait in between pushes to Loki.
+    # CLI flag: -pattern-ingester.metric-aggregation.push-period
+    [push_period: <duration> | default = 1m]
+
+    # The HTTP client configuration for pushing metrics to Loki.
+    http_client_config:
+      basic_auth:
+        [username: <string> | default = ""]
+
+        [username_file: <string> | default = ""]
+
+        [password: <string> | default = ""]
+
+        [password_file: <string> | default = ""]
+
+      authorization:
+        [type: <string> | default = ""]
+
+        [credentials: <string> | default = ""]
+
+        [credentials_file: <string> | default = ""]
+
+      oauth2:
+        [client_id: <string> | default = ""]
+
+        [client_secret: <string> | default = ""]
+
+        [client_secret_file: <string> | default = ""]
+
+        [scopes: <list of strings>]
+
+        [token_url: <string> | default = ""]
+
+        [endpoint_params: <map of string to string>]
+
+        tls_config:
+          [ca: <string> | default = ""]
+
+          [cert: <string> | default = ""]
+
+          [key: <string> | default = ""]
+
+          [ca_file: <string> | default = ""]
+
+          [cert_file: <string> | default = ""]
+
+          [key_file: <string> | default = ""]
+
+          [server_name: <string> | default = ""]
+
+          [insecure_skip_verify: <boolean>]
+
+          [min_version: <int>]
+
+          [max_version: <int>]
+
+        proxy_url:
+          [url: <url>]
+
+        [no_proxy: <string> | default = ""]
+
+        [proxy_from_environment: <boolean>]
+
+        [proxy_connect_header: <map of string to list of strings>]
+
+      [bearer_token: <string> | default = ""]
+
+      [bearer_token_file: <string> | default = ""]
+
+      tls_config:
+        [ca: <string> | default = ""]
+
+        [cert: <string> | default = ""]
+
+        [key: <string> | default = ""]
+
+        [ca_file: <string> | default = ""]
+
+        [cert_file: <string> | default = ""]
+
+        [key_file: <string> | default = ""]
+
+        [server_name: <string> | default = ""]
+
+        [insecure_skip_verify: <boolean>]
+
+        [min_version: <int>]
+
+        [max_version: <int>]
+
+      [follow_redirects: <boolean>]
+
+      [enable_http2: <boolean>]
+
+      proxy_url:
+        [url: <url>]
+
+      [no_proxy: <string> | default = ""]
+
+      [proxy_from_environment: <boolean>]
+
+      [proxy_connect_header: <map of string to list of strings>]
+
+    # Whether to use TLS for pushing metrics to Loki.
+    # CLI flag: -pattern-ingester.metric-aggregation.tls
+    [use_tls: <boolean> | default = false]
+
+    # The basic auth configuration for pushing metrics to Loki.
+    basic_auth:
+      # Basic auth username for sending aggregations back to Loki.
+      # CLI flag: -pattern-ingester.metric-aggregation.basic-auth.username
+      [username: <string> | default = ""]
+
+      # Basic auth password for sending aggregations back to Loki.
+      # CLI flag: -pattern-ingester.metric-aggregation.basic-auth.password
+      [password: <string> | default = ""]
+
+    # The backoff configuration for pushing metrics to Loki.
+    backoff_config:
+      # Minimum delay when backing off.
+      # CLI flag: -pattern-ingester.metric-aggregation.backoff-min-period
+      [min_period: <duration> | default = 100ms]
+
+      # Maximum delay when backing off.
+      # CLI flag: -pattern-ingester.metric-aggregation.backoff-max-period
+      [max_period: <duration> | default = 10s]
+
+      # Number of times to backoff and retry before failing.
+      # CLI flag: -pattern-ingester.metric-aggregation.backoff-retries
+      [max_retries: <int> | default = 10]
+
+  # The number of parallel goroutines to use for forwarding requests to the
+  # pattern ingester.
+  # CLI flag: -pattern-ingester.tee-parallelism
+  [tee_parallelism: <int> | default = 5]
+
+  # Maxiumum number of pending teed request to pattern ingesters. If the buffer
+  # is full the request is dropped.
+  # CLI flag: -pattern-ingester.tee-buffer-size
+  [tee_buffer_size: <int> | default = 100]
+
 # The index_gateway block configures the Loki index gateway server, responsible
 # for serving index queries without the need to constantly interact with the
 # object store.

pkg/distributor/distributor_test.go

@@ -1380,6 +1380,10 @@ type mockTee struct {
 	tenant     string
 }
 
+func (m *mockTee) Stop() {
+	// no-op
+}
+
 func (mt *mockTee) Duplicate(tenant string, streams []KeyedStream) {
 	mt.mu.Lock()
 	defer mt.mu.Unlock()

pkg/distributor/tee_test.go

@@ -12,6 +12,10 @@ type mockedTee struct {
 	mock.Mock
 }
 
+func (m *mockedTee) Stop() {
+	// no-op
+}
+
 func (m *mockedTee) Duplicate(tenant string, streams []KeyedStream) {
 	m.Called(tenant, streams)
 }

pkg/pattern/aggregation/push_test.go

@@ -121,7 +121,6 @@ func Test_Push(t *testing.T) {
 			password:    "secret",
 			logger:      log.NewNopLogger(),
 			quit:        make(chan struct{}),
-			done:        make(chan struct{}),
 			backoff:     &backoff,
 			entries:     entries{},
 		}

pkg/pattern/ingester_test.go

@@ -20,8 +20,9 @@ import (
 
 	"github.com/grafana/loki/v3/pkg/pattern/drain"
 
-	"github.com/grafana/loki/pkg/push"
 	loghttp_push "github.com/grafana/loki/v3/pkg/loghttp/push"
+
+	"github.com/grafana/loki/pkg/push"
 )
 
 func TestInstancePushQuery(t *testing.T) {

pkg/pattern/tee.go

@@ -84,15 +84,15 @@ func (t *Tee) run() {
 			)
 			defer cancel()
 
-			if err := t.sendStream(ctx, req.tenant, req.stream); err != nil {
+			if err := t.sendStream(ctx, req.stream); err != nil {
 				level.Error(t.logger).Log("msg", "failed to send stream to pattern ingester", "err", err)
 			}
 		}
 	}
 }
 
-func (t *Tee) sendStream(ctx context.Context, tenant string, stream distributor.KeyedStream) error {
-	err := t.sendOwnedStream(ctx, tenant, stream)
+func (t *Tee) sendStream(ctx context.Context, stream distributor.KeyedStream) error {
+	err := t.sendOwnedStream(ctx, stream)
 	if err == nil {
 		// Success, return early
 		return nil
@@ -131,7 +131,7 @@ func (t *Tee) sendStream(ctx context.Context, tenant string, stream distributor.
 	return err
 }
 
-func (t *Tee) sendOwnedStream(ctx context.Context, tenant string, stream distributor.KeyedStream) error {
+func (t *Tee) sendOwnedStream(ctx context.Context, stream distributor.KeyedStream) error {
 	var descs [1]ring.InstanceDesc
 	replicationSet, err := t.ringClient.Ring().Get(stream.HashKey, ring.WriteNoExtend, descs[:0], nil, nil)
 	if err != nil {