distributor: prevent panics when concurrently calling to forwarder's… (…

…#1422)

* distributor: prevent panics when concurrently calling  to forwarder's queueManager

* Update PR number

* Pin k6 version to v0.37.0

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 * [BUGFIX] metrics-generator: don't inject X-Scope-OrgID header for single-tenant setups [1417](https://github.com/grafana/tempo/pull/1417) (@kvrhdn)
 * [BUGFIX] compactor: populate `compaction_objects_combined_total` and `tempo_discarded_spans_total{reason="trace_too_large_to_compact"}` metrics again [1420](https://github.com/grafana/tempo/pull/1420) (@mdisibio)
+* [BUGFIX] distributor: prevent panics when concurrently calling `shutdown` to forwarder's queueManager [1422](https://github.com/grafana/tempo/pull/1422) (@mapno)
 
 ## v1.4.0 / 2022-04-28
 

integration/bench/load_test.go

@@ -13,7 +13,7 @@ import (
 )
 
 const (
-	k6Image = "loadimpact/k6:latest"
+	k6Image = "loadimpact/k6:0.37.0"
 )
 
 func TestAllInOne(t *testing.T) {

modules/distributor/forwarder.go

@@ -278,17 +278,15 @@ func (m *queueManager) forwardRequest(ctx context.Context, req *request) {
 }
 
 func (m *queueManager) shutdown() error {
-	// Already being shutdown
-	if m.readOnly.Load() {
-		return nil
-	}
+	// Call to stopWorkers only once
+	if m.readOnly.CAS(false, true) {
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+		defer cancel()
 
-	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
-	defer cancel()
-
-	m.readOnly.Store(true)
+		return m.stopWorkers(ctx)
+	}
 
-	return m.stopWorkers(ctx)
+	return nil
 }
 
 func (m *queueManager) stopWorkers(ctx context.Context) error {