Refactor queueManager into generic queue.Queue (#1796)

* Refactor queueManager into generic queue.Queue

* Update CHANGELOG.md

* Create shared processFun method and a helper function for queue creation

* Remove returning error from ProcessFunc and update chan *request[T] to chan T.

* Fix unit tests

* Replace go.uber.org/atomic with sync/atomic

* Remove redundant accessor

* Fix processFunc definition

* Add breaking change to CHANGELOG.md

* Define metrics on the package level to overcome Prometheus DefaultRegistry limitations

* Fix tests in queue_test.go

* Deprecate generatorForwarder metrics in favor of queue.Queue metrics

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## main / unreleased
 
+* [ENHANCEMENT] Refactor queueManager into generic queue.Queue [#1796](https://github.com/grafana/tempo/pull/1796) (@Blinkuu)
+  - **BREAKING CHANGE** Rename `tempo_distributor_forwarder_queue_length` metric to `tempo_distributor_queue_length`. New metric has two custom labels: `name` and  `tenant`.
+  - Deprecated `tempo_distributor_forwarder_pushes_total` metric in favor of `tempo_distributor_queue_pushes_total`.
+  - Deprecated `tempo_distributor_forwarder_pushes_failures_total` metric in favor of `tempo_distributor_queue_pushes_failures_total`.
 * [ENHANCEMENT] Filter namespace by cluster in tempo dashboards variables [#1771](https://github.com/grafana/tempo/pull/1771) (@electron0zero)
 * [ENHANCEMENT] Exit early from sharded search requests [#1742](https://github.com/grafana/tempo/pull/1742) (@electron0zero)
 * [CHANGE] Identify bloom that could not be retrieved from backend block [#1737](https://github.com/grafana/tempo/pull/1737) (@AlexDHoffer)

modules/distributor/distributor.go

@@ -131,7 +131,7 @@ type Distributor struct {
 	generatorClientCfg      generator_client.Config
 	generatorsRing          ring.ReadRing
 	generatorsPool          *ring_client.Pool
-	generatorForwarder      *forwarder
+	generatorForwarder      *generatorForwarder
 
 	// Per-user rate limiter.
 	ingestionRateLimiter *limiter.RateLimiter
@@ -223,7 +223,7 @@ func New(cfg Config, clientCfg ingester_client.Config, ingestersRing ring.ReadRi
 
 		subservices = append(subservices, d.generatorsPool)
 
-		d.generatorForwarder = newForwarder(d.sendToGenerators, o)
+		d.generatorForwarder = newGeneratorForwarder(logger, reg, d.sendToGenerators, o)
 		subservices = append(subservices, d.generatorForwarder)
 	}