Add walqueue sharding

CHANGELOG.md

@@ -10,6 +10,10 @@ internal API changes are not present.
 Main (unreleased)
 -----------------
 
+### Breaking changes
+
+- (_Experimental_) Changed `parallelism` from attribute to block to allow dynamic scaling of `prometheus.write.queue`. (@mattdurham)
+
 ### Features
 
 - (_Experimental_) Add a `stage.windowsevent` block in the `loki.process` component. This aims to replace the existing `stage.eventlogmessage`. (@wildum)

docs/sources/reference/components/prometheus/prometheus.write.queue.md

@@ -48,13 +48,13 @@ The following arguments are supported:
 The following blocks are supported inside the definition of
 `prometheus.write.queue`:
 
- Hierarchy             | Block            | Description                                              | Required 
------------------------|------------------|----------------------------------------------------------|----------
- persistence           | [persistence][]  | Configuration for persistence                            | no       
- endpoint              | [endpoint][]     | Location to send metrics to.                             | no       
- endpoint > basic_auth | [basic_auth][]   | Configure basic_auth for authenticating to the endpoint. | no       
- endpoint > tls_config | [tls_config][]   | Configure TLS settings for connecting to the endpoint.   | no
-
+ Hierarchy             | Block           | Description                                               | Required 
+-----------------------|-----------------|-----------------------------------------------------------|----------
+ persistence           | [persistence][] | Configuration for persistence                             | no       
+ endpoint              | [endpoint][]    | Location to send metrics to.                              | no       
+ endpoint > basic_auth | [basic_auth][]  | Configure basic_auth for authenticating to the endpoint.  | no       
+ endpoint > tls_config | [tls_config][]  | Configure TLS settings for connecting to the endpoint.    | no
+ endpoint > parallelism | [parallelism][] | Configure parralelism for the endpoint.                   | no
 
 The `>` symbol indicates deeper levels of nesting. For example, `endpoint >
 basic_auth` refers to a `basic_auth` block defined inside an
@@ -64,7 +64,7 @@ basic_auth` refers to a `basic_auth` block defined inside an
 [basic_auth]: #basic_auth-block
 [persistence]: #persistence-block
 [tls_config]: #tls_config-block
-
+[parallelism]: #parralelism
 
 ### persistence block
 
@@ -96,7 +96,6 @@ The following arguments are supported:
  `max_retry_attempts` | `uint`        | Maximum number of retries before dropping the batch.                                        | `0`     | no      
  `batch_count`        | `uint`        | How many series to queue in each queue.                                                     | `1000`  | no       
  `flush_interval`     | `duration`    | How long to wait until sending if `batch_count` is not trigger.                             | `1s`    | no       
- `parallelism`        | `uint`        | How many parallel batches to write.                                                         | 10      | no       
  `external_labels`    | `map(string)` | Labels to add to metrics sent over the network.                                             |         | no       
  `enable_round_robin` | `bool`        | Use round robin load balancing when there are multiple IPs for a given endpoint. | `false` | no       
 
@@ -117,6 +116,31 @@ Name                   | Type     | Description
 `insecure_skip_verify` | `bool`   | Disables validation of the server certificate.           |         | no
 `key_pem`              | `secret` | Key PEM-encoded text for client authentication.          |         | no
 
+
+### parallelism block
+
+Name            | Type       | Description                                                                                                                        | Default | Required
+----------------|------------|------------------------------------------------------------------------------------------------------------------------------------|---------|---------
+`drift_scale_up_seconds`      | `int`      | The maximum amount of time between the timestamps of incoming signals and outgoing signals before increasing desired connections.  | `60`    | no
+`drift_scale_down_seconds`    | `int`      | The minimum amount of time between the timestamps of incoming signals and outgoing signals before descreasing desired connections. | `30`    | no
+`max_connections`    | `uint`     | The maximum number of desired connections.                                                                                         | `50`    | no
+`min_connections`    | `uint`     | The minimum number of desired connections.                                                                                         | `2`     | no
+`network_flush_interval`    | `duration` | The length of time that network successes and failures are kept for determining desired connections.                               | `1m`    | no
+`desired_connections_lookback`    | `duration` | The length of time that previous desired connections are kept for determining desired connections.                                 | `5m`    | no
+`desired_check_interval`    | `duration` | The length of time that between checking for desired connections.                                                                  | `5s`    | no
+`allowed_network_error_percent`    | `float`    | The allowed error rate before scaling down.                                                                                        | `0.50`   | no
+
+Parralelism determines when to scale up or down the number of desired connections. This is accomplished by a variety of inputs.
+By determining the drift between the incoming and outgoing timestamps that will determine whether to increase or decrease the
+desired connections. This is represented by `drift_scale_up_seconds` and `drift_scale_down_seconds`, if the drift is between these
+two values then the value will stay the same. Network success and failures are recorded and kept in memory, this helps determine
+the nature of the drift. For instance if the drift is increasing but the network failures are increasing we should not increase
+desired connections since that would only increase load on the endpoint. The last major part is to prevent flapping of desired connections.
+This is accomplished with the `desired_check_interval`, each time a desired connection is calculated it is added to a list, before actually changing the
+desired connection the system will choose the highest value in the lookback. Example; for the past 5 minutes desired connections have been: [2,1,1] the check runs
+and determines that the desired connections are 1, but will not change the value since the value 2 is still in the lookback. On the next check we have [1,1,1],
+now it will change to 1. In general the system is fast to increase and slow to decrease.
+
 ## Exported fields
 
 The following fields are exported and can be referenced by other components:

go.mod

@@ -73,7 +73,7 @@ require (
 	github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc
 	github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0
 	github.com/grafana/vmware_exporter v0.0.5-beta
-	github.com/grafana/walqueue v0.0.0-20250113171943-e5fe545d1408
+	github.com/grafana/walqueue v0.0.0-20250210172748-0dc5bd134b25
 	github.com/hashicorp/consul/api v1.30.0
 	github.com/hashicorp/go-discover v0.0.0-20230724184603-e89ebd1b2f65
 	github.com/hashicorp/go-multierror v1.1.1
@@ -499,7 +499,7 @@ require (
 	github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
 	github.com/go-git/go-billy/v5 v5.6.0 // indirect
 	github.com/go-jose/go-jose/v3 v3.0.3 // indirect
-	github.com/go-kit/kit v0.13.0 // indirect
+	github.com/go-kit/kit v0.13.0
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
@@ -883,9 +883,8 @@ require (
 	github.com/containers/common v0.61.0 // indirect
 	github.com/deneonet/benc v1.1.2 // indirect
 	github.com/itchyny/timefmt-go v0.1.6 // indirect
-	github.com/onsi/ginkgo/v2 v2.21.0 // indirect
-	github.com/onsi/gomega v1.35.1 // indirect
-	go.etcd.io/bbolt v1.3.11 // indirect
+	github.com/panjf2000/ants/v2 v2.11.0 // indirect
+	golang.design/x/chann v0.1.2 // indirect
 )
 
 // NOTE: replace directives below must always be *temporary*.

go.sum

@@ -1852,17 +1852,11 @@ github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoA
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/gosnmp/gosnmp v1.37.0 h1:/Tf8D3b9wrnNuf/SfbvO+44mPrjVphBhRtcGg22V07Y=
-github.com/gosnmp/gosnmp v1.37.0/go.mod h1:GDH9vNqpsD7f2HvZhKs5dlqSEcAS6s6Qp099oZRCR+M=
 github.com/gosnmp/gosnmp v1.38.0 h1:I5ZOMR8kb0DXAFg/88ACurnuwGwYkXWq3eLpJPHMEYc=
 github.com/gosnmp/gosnmp v1.38.0/go.mod h1:FE+PEZvKrFz9afP9ii1W3cprXuVZ17ypCcyyfYuu5LY=
 github.com/gotestyourself/gotestyourself v2.2.0+incompatible/go.mod h1:zZKM6oeNM8k+FRljX1mnzVYeS8wiGgQyvST1/GafPbY=
 github.com/grafana/alloy-remote-config v0.0.9 h1:gy34SxZ8Iq/HrDTIFZi80+8BlT+FnJhKiP9mryHNEUE=
 github.com/grafana/alloy-remote-config v0.0.9/go.mod h1:kHE1usYo2WAVCikQkIXuoG1Clz8BSdiz3kF+DZSCQ4k=
-github.com/grafana/beyla v0.0.0-20250108110233-3f1b9b55c6dc h1:oY8yQB8IG0dBo1UrLlLC2CspxbiVtSWWExMxXOnfWgk=
-github.com/grafana/beyla v0.0.0-20250108110233-3f1b9b55c6dc/go.mod h1:hpk185gTeIQXjxV/so9vAxhZtSEgm8ODanWXZNVnH2M=
-github.com/grafana/beyla v1.9.1-0.20250122195759-1117708def46 h1:/aw+Ze9lUluE1hNZ0fAtwhmf2CKP0VbsLFumpN8xztY=
-github.com/grafana/beyla v1.9.1-0.20250122195759-1117708def46/go.mod h1:CRWu15fkScScSYBlYUtdJu2Ak8ojGvnuwHToGGkaOXY=
 github.com/grafana/beyla v1.10.0-alloy h1:kGyZtBSS/Br2qdhbvzu8sVYZHuE9a3OzWpbp6gN55EY=
 github.com/grafana/beyla v1.10.0-alloy/go.mod h1:CRWu15fkScScSYBlYUtdJu2Ak8ojGvnuwHToGGkaOXY=
 github.com/grafana/cadvisor v0.0.0-20240729082359-1f04a91701e2 h1:ju6EcY2aEobeBg185ETtFCKj5WzaQ48qfkbsSRRQrF4=
@@ -1920,8 +1914,8 @@ github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0 h1:bjh0PVYSVVFxzINqPF
 github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0/go.mod h1:7t5XR+2IA8P2qggOAHTj/GCZfoLBle3OvNSYh1VkRBU=
 github.com/grafana/vmware_exporter v0.0.5-beta h1:2JCqzIWJzns8FN78wPsueC9rT3e3kZo2OUoL5kGMjdM=
 github.com/grafana/vmware_exporter v0.0.5-beta/go.mod h1:1CecUZII0zVsVcHtNfNeTTcxK7EksqAsAn/TCCB0Mh4=
-github.com/grafana/walqueue v0.0.0-20250113171943-e5fe545d1408 h1:TGoFWafEVwzy4Wg0L2PPQonwTKYaqa8kMpiAxiEvXW8=
-github.com/grafana/walqueue v0.0.0-20250113171943-e5fe545d1408/go.mod h1:Hisxv1n+PxFQEkayynKy+B4AiJiJVRKHKT/8ng6jgOM=
+github.com/grafana/walqueue v0.0.0-20250210172748-0dc5bd134b25 h1:p8plAhxyEWfDXRuYCfeZgrGS1GtJU5GnnOxsOBLgi9Q=
+github.com/grafana/walqueue v0.0.0-20250210172748-0dc5bd134b25/go.mod h1:rnU7r397nvQCTyVbcODlD3P6DIbQlidxPDweV+4ab2M=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grobie/gomemcache v0.0.0-20230213081705-239240bbc445 h1:FlKQKUYPZ5yDCN248M3R7x8yu2E3yEZ0H7aLomE4EoE=
 github.com/grobie/gomemcache v0.0.0-20230213081705-239240bbc445/go.mod h1:L69/dBlPQlWkcnU76WgcppK5e4rrxzQdi6LhLnK/ytA=
@@ -2126,8 +2120,6 @@ github.com/iancoleman/strcase v0.3.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w=
-github.com/ianlancetaylor/demangle v0.0.0-20240312041847-bd984b5ce465 h1:KwWnWVWCNtNq/ewIX7HIKnELmEx2nDP42yskD/pi7QE=
-github.com/ianlancetaylor/demangle v0.0.0-20240312041847-bd984b5ce465/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw=
 github.com/ianlancetaylor/demangle v0.0.0-20240912202439-0a2b6291aafd h1:EVX1s+XNss9jkRW9K6XGJn2jL2lB1h5H804oKPsxOec=
 github.com/ianlancetaylor/demangle v0.0.0-20240912202439-0a2b6291aafd/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw=
 github.com/illumos/go-kstat v0.0.0-20210513183136-173c9b0a9973 h1:hk4LPqXIY/c9XzRbe7dA6qQxaT6Axcbny0L/G5a4owQ=
@@ -2807,6 +2799,8 @@ github.com/ovh/go-ovh v1.6.0/go.mod h1:cTVDnl94z4tl8pP1uZ/8jlVxntjSIf09bNcQ5TJSC
 github.com/packethost/packngo v0.1.1-0.20180711074735-b9cb5096f54c h1:vwpFWvAO8DeIZfFeqASzZfsxuWPno9ncAebBEP0N3uE=
 github.com/packethost/packngo v0.1.1-0.20180711074735-b9cb5096f54c/go.mod h1:otzZQXgoO96RTzDB/Hycg0qZcXZsWJGJRSXbmEIJ+4M=
 github.com/pact-foundation/pact-go v1.0.4/go.mod h1:uExwJY4kCzNPcHRj+hCR/HBbOOIwwtUjcrb0b5/5kLM=
+github.com/panjf2000/ants/v2 v2.11.0 h1:sHrqEwTBQTQ2w6PMvbMfvBtVUuhsaYPzUmAYDLYmJPg=
+github.com/panjf2000/ants/v2 v2.11.0/go.mod h1:V9HhTupTWxcaRmIglJvGwvzqXUTnIZW9uO6q4hAfApw=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
 github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
@@ -3559,6 +3553,8 @@ go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 go4.org/netipx v0.0.0-20230125063823-8449b0a6169f h1:ketMxHg+vWm3yccyYiq+uK8D3fRmna2Fcj+awpQp84s=
 go4.org/netipx v0.0.0-20230125063823-8449b0a6169f/go.mod h1:tgPU4N2u9RByaTN3NC2p9xOzyFpte4jYwsIIRF7XlSc=
+golang.design/x/chann v0.1.2 h1:eHF9wjuQnpp+j4ryWhyxC/pFuYzbvMAkudA/I5ALovY=
+golang.design/x/chann v0.1.2/go.mod h1:Rh5KhCAp+0qu9+FfKPymHpu8onmjl89sFwMeiw3SK14=
 golang.org/x/arch v0.7.0 h1:pskyeJh/3AmoQ8CPE95vxHLqp1G1GfGNXTmcl9NEKTc=
 golang.org/x/arch v0.7.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

internal/component/prometheus/write/queue/types.go

@@ -50,7 +50,16 @@ func defaultEndpointConfig() EndpointConfig {
 		MaxRetryAttempts: 0,
 		BatchCount:       1_000,
 		FlushInterval:    1 * time.Second,
-		Parallelism:      4,
+		Parallelism: ParralelismConfig{
+			DriftScaleUpSeconds:        60,
+			DriftScaleDownSeconds:      30,
+			MaxConnections:             50,
+			MinConnections:             2,
+			NetworkFlushInterval:       1 * time.Minute,
+			DesiredConnectionsLookback: 5 * time.Minute,
+			DesiredCheckInterval:       5 * time.Second,
+			AllowedNetworkErrorPercent: 0.50,
+		},
 	}
 }
 
@@ -66,6 +75,23 @@ func (r *Arguments) Validate() error {
 		if conn.FlushInterval < 1*time.Second {
 			return fmt.Errorf("flush_interval must be greater or equal to 1s, the internal timers resolution is 1s")
 		}
+		if conn.Parallelism.MaxConnections < conn.Parallelism.MinConnections {
+			return fmt.Errorf("max_connections less than min_connections")
+		}
+		if conn.Parallelism.MinConnections == 0 {
+			return fmt.Errorf("min_connections must be greater than 0")
+		}
+		if conn.Parallelism.DriftScaleUpSeconds <= conn.Parallelism.DriftScaleDownSeconds {
+			return fmt.Errorf("drift_scale_up_seconds less than or equal drift_scale_down_seconds")
+		}
+		// Any lower than 1 second and you spend a fair amount of time churning on the draining and
+		// refilling the write buffers.
+		if conn.Parallelism.DesiredCheckInterval < 1*time.Second {
+			return fmt.Errorf("desired_check_interval must be greater than or equal to 1 second")
+		}
+		if conn.Parallelism.AllowedNetworkErrorPercent < 0 || conn.Parallelism.AllowedNetworkErrorPercent > 1 {
+			return fmt.Errorf("allowed_network_error_percent must be between 0.00 and 1.00")
+		}
 	}
 
 	return nil
@@ -87,7 +113,7 @@ type EndpointConfig struct {
 	// How long to wait before sending regardless of batch count.
 	FlushInterval time.Duration `alloy:"flush_interval,attr,optional"`
 	// How many concurrent queues to have.
-	Parallelism    uint              `alloy:"parallelism,attr,optional"`
+	Parallelism    ParralelismConfig `alloy:"parallelism,block,optional"`
 	ExternalLabels map[string]string `alloy:"external_labels,attr,optional"`
 	TLSConfig      *TLSConfig        `alloy:"tls_config,block,optional"`
 	RoundRobin     bool              `alloy:"enable_round_robin,attr,optional"`
@@ -100,6 +126,32 @@ type TLSConfig struct {
 	InsecureSkipVerify bool              `alloy:"insecure_skip_verify,attr,optional"`
 }
 
+type ParralelismConfig struct {
+	// DriftScaleUpSeconds is the maximum amount of seconds that is allowed for the Newest Timestamp Serializer - Newest Timestamp Sent via Network before the connections scales up.
+	// Using non unix timestamp numbers. If Newest TS In Serializer sees 100s and Newest TS Out Network sees 20s then we have a drift of 80s. If AllowDriftSeconds is 60s that would
+	// trigger a scaling up event.
+	DriftScaleUpSeconds int64 `alloy:"drift_scale_up_seconds,attr,optional"`
+	// DriftScaleDownSeconds is the amount if we go below that we can scale down. Using the above if In is 100s and Out is 70s and DriftScaleDownSeconds is 30 then we wont scale
+	// down even though we are below the 60s. This is to keep the number of connections from flapping. In practice we should consider 30s DriftScaleDownSeconds and 60s DriftScaleUpSeconds to be a sweet spot
+	// for general usage.
+	DriftScaleDownSeconds int64 `alloy:"drift_scale_down_seconds,attr,optional"`
+	// MaxConnections is the maximum number of concurrent connections to use.
+	MaxConnections uint `alloy:"max_connections,attr,optional"`
+	// MinConnections is the minimum number of concurrent connections to use.
+	MinConnections uint `alloy:"min_connections,attr,optional"`
+	// NetworkFlushInterval is how long to keep network successes and errors in memory for calculations.
+	NetworkFlushInterval time.Duration `alloy:"network_flush_interval,attr,optional"`
+	// DesiredConnectionsLookback is how far to lookback for previous desired values. This is to prevent flapping.
+	// In a situation where in the past 5 minutes you have desired [1,2,1,1] and desired is 1 it will
+	// choose 2 since that was the greatest. This determines how fast you can scale down.
+	DesiredConnectionsLookback time.Duration `alloy:"desired_connections_lookback,attr,optional"`
+	// DesiredCheckInterval is how long to check for desired values.
+	DesiredCheckInterval time.Duration `alloy:"desired_check_interval,attr,optional"`
+	// AllowedNetworkErrorPercent is the percentage of failed network requests that are allowable. This will
+	// trigger a decrease in connections if exceeded.
+	AllowedNetworkErrorPercent float64 `alloy:"allowed_network_error_percent,attr,optional"`
+}
+
 var UserAgent = fmt.Sprintf("Alloy/%s", version.Version)
 
 func (cc EndpointConfig) ToNativeType() types.ConnectionConfig {
@@ -113,8 +165,17 @@ func (cc EndpointConfig) ToNativeType() types.ConnectionConfig {
 		BatchCount:       cc.BatchCount,
 		FlushInterval:    cc.FlushInterval,
 		ExternalLabels:   cc.ExternalLabels,
-		Connections:      cc.Parallelism,
 		UseRoundRobin:    cc.RoundRobin,
+		Parralelism: types.ParralelismConfig{
+			AllowedDriftSeconds:          cc.Parallelism.DriftScaleUpSeconds,
+			MinimumScaleDownDriftSeconds: cc.Parallelism.DriftScaleDownSeconds,
+			MaxConnections:               cc.Parallelism.MaxConnections,
+			MinConnections:               cc.Parallelism.MinConnections,
+			ResetInterval:                cc.Parallelism.NetworkFlushInterval,
+			Lookback:                     cc.Parallelism.DesiredConnectionsLookback,
+			CheckInterval:                cc.Parallelism.DesiredCheckInterval,
+			AllowedNetworkErrorPercent:   cc.Parallelism.AllowedNetworkErrorPercent,
+		},
 	}
 	if cc.BasicAuth != nil {
 		tcc.BasicAuth = &types.BasicAuth{