Cherry pick changes into rc4 (#2815)

* Cherry pick changes into rc4

* Cherry pick changes into rc4

Author: mattdurham

CHANGELOG.md

@@ -7,6 +7,14 @@ This document contains a historical list of changes between releases. Only
 changes that impact end-user behavior are listed; changes to documentation or
 internal API changes are not present.
 
+
+v1.7.0-rc.4
+-----------------
+
+### Bugfixes
+
+- Fix issue with `prometheus.write.queue` that lead to excessive connections. (@mattdurham)
+
 v1.7.0-rc.3
 -----------------
 

go.mod

@@ -75,7 +75,7 @@ require (
 	github.com/grafana/snowflake-prometheus-exporter v0.0.0-20240813124544-9995e8354548
 	github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0
 	github.com/grafana/vmware_exporter v0.0.5-beta.0.20250218170317-73398ba08329
-	github.com/grafana/walqueue v0.0.0-20250211154548-6435b3242458
+	github.com/grafana/walqueue v0.0.0-20250222022458-49d08775d0f4
 	github.com/hashicorp/consul/api v1.31.0
 	github.com/hashicorp/go-discover v0.0.0-20230724184603-e89ebd1b2f65
 	github.com/hashicorp/go-multierror v1.1.1
@@ -524,7 +524,6 @@ require (
 	github.com/form3tech-oss/jwt-go v3.2.5+incompatible // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
-	github.com/gammazero/deque v0.2.1 // indirect
 	github.com/gavv/monotime v0.0.0-20190418164738-30dba4353424 // indirect
 	github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
 	github.com/go-git/go-billy/v5 v5.6.0 // indirect
@@ -801,7 +800,6 @@ require (
 	github.com/vertica/vertica-sql-go v1.3.3 // indirect
 	github.com/vishvananda/netlink v1.3.0 // indirect
 	github.com/vishvananda/netns v0.0.4 // indirect
-	github.com/vladopajic/go-actor v0.9.1-0.20241115212052-39d92aec6093 // indirect
 	github.com/vmware/govmomi v0.48.0 // indirect
 	github.com/vultr/govultr/v2 v2.17.2 // indirect
 	github.com/willf/bitset v1.1.11 // indirect

go.sum

@@ -933,8 +933,6 @@ github.com/fxamacker/cbor/v2 v2.7.0 h1:iM5WgngdRBanHcxugY4JySA0nk1wZorNOpTgCMedv
 github.com/fxamacker/cbor/v2 v2.7.0/go.mod h1:pxXPTn3joSm21Gbwsv0w9OSA2y1HFR9qXEeXQVeNoDQ=
 github.com/gabriel-vasile/mimetype v1.4.3 h1:in2uUcidCuFcDKtdcBxlR0rJ1+fsokWf+uqxgUFjbI0=
 github.com/gabriel-vasile/mimetype v1.4.3/go.mod h1:d8uq/6HKRL6CGdk+aubisF/M5GcPfT7nKyLpA0lbSSk=
-github.com/gammazero/deque v0.2.1 h1:qSdsbG6pgp6nL7A0+K/B7s12mcCY/5l5SIUpMOl+dC0=
-github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=
 github.com/gavv/monotime v0.0.0-20190418164738-30dba4353424 h1:Vh7rylVZRZCj6W41lRlP17xPk4Nq260H4Xo/DDYmEZk=
 github.com/gavv/monotime v0.0.0-20190418164738-30dba4353424/go.mod h1:vmp8DIyckQMXOPl0AQVHt+7n5h7Gb7hS6CUydiV8QeA=
 github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -1305,8 +1303,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.0.20250218170317-73398ba08329 h1:Rs4H1yv2Abk3xE82qpyhMGGA8rswAOA0HQZde/BYkFo=
 github.com/grafana/vmware_exporter v0.0.5-beta.0.20250218170317-73398ba08329/go.mod h1:Z28219aViNlsLlPvuCnlgHDagRdZBAZ7JOnQg1b3eWg=
-github.com/grafana/walqueue v0.0.0-20250211154548-6435b3242458 h1:KUa/teNk/VYBDSSjsgdLI37gnFXTz1+h513CmNWJeU4=
-github.com/grafana/walqueue v0.0.0-20250211154548-6435b3242458/go.mod h1:rnU7r397nvQCTyVbcODlD3P6DIbQlidxPDweV+4ab2M=
+github.com/grafana/walqueue v0.0.0-20250222022458-49d08775d0f4 h1:oEDouKyuO62LKR46nb/4Z1nKx2tcxLK9eLMmHJtNdwQ=
+github.com/grafana/walqueue v0.0.0-20250222022458-49d08775d0f4/go.mod h1:++oxbVcfnIzbSKHNxpO3WpQJG5ufmnwtBJqh5rvEmVo=
 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=
@@ -2590,8 +2588,6 @@ github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17
 github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=
 github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
 github.com/vjeantet/grok v1.0.0/go.mod h1:/FWYEVYekkm+2VjcFmO9PufDU5FgXHUz9oy2EGqmQBo=
-github.com/vladopajic/go-actor v0.9.1-0.20241115212052-39d92aec6093 h1:wZ2W9ei2PSGeeUWb0dZOnX7s+hIDLqwkRnpU30ejEpw=
-github.com/vladopajic/go-actor v0.9.1-0.20241115212052-39d92aec6093/go.mod h1:qcIpcfAXGBaWKyBtzyrhEhPjlKpw4nK83rbyZONymcg=
 github.com/vmihailenco/msgpack/v4 v4.3.13 h1:A2wsiTbvp63ilDaWmsk2wjx6xZdxQOvpiNlKBGKKXKI=
 github.com/vmihailenco/msgpack/v4 v4.3.13/go.mod h1:gborTTJjAo/GWTqqRjrLCn9pgNN+NXzzngzBKDPIqw4=
 github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=

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

@@ -33,16 +33,11 @@ func NewComponent(opts component.Options, args Arguments) (*Queue, error) {
 		endpoints: map[string]promqueue.Queue{},
 	}
 
+	s.opts.OnStateChange(Exports{Receiver: s})
 	err := s.createEndpoints()
 	if err != nil {
 		return nil, err
 	}
-	// This needs to be started before we export the onstatechange so that it can accept
-	// signals.
-	for _, ep := range s.endpoints {
-		ep.Start()
-	}
-	s.opts.OnStateChange(Exports{Receiver: s})
 
 	return s, nil
 }
@@ -55,12 +50,14 @@ type Queue struct {
 	opts      component.Options
 	log       log.Logger
 	endpoints map[string]promqueue.Queue
+	ctx       context.Context
 }
 
 // Run starts the component, blocking until ctx is canceled or the component
 // suffers a fatal error. Run is guaranteed to be called exactly once per
 // Component.
 func (s *Queue) Run(ctx context.Context) error {
+	s.ctx = ctx
 	defer func() {
 		s.mut.Lock()
 		defer s.mut.Unlock()
@@ -69,6 +66,9 @@ func (s *Queue) Run(ctx context.Context) error {
 			ep.Stop()
 		}
 	}()
+	for _, ep := range s.endpoints {
+		ep.Start(ctx)
+	}
 
 	<-ctx.Done()
 	return nil
@@ -113,7 +113,7 @@ func (s *Queue) Update(args component.Arguments) error {
 		if err != nil {
 			return err
 		}
-		end.Start()
+		end.Start(s.ctx)
 		s.endpoints[epCfg.Name] = end
 
 	}