Fix MakeIngesterClient leaking grpc.ClientConn and goroutines on Run() failure

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## master / unreleased
+* [BUGFIX] Ingester Client: Fix `MakeIngesterClient` leaking the `grpc.ClientConn` and stream-push worker goroutines when starting stream workers fails (`-distributor.use-stream-push=true`). Also fix a data race on the collected worker error, and stop already-started workers instead of leaving them running when a sibling worker fails. #7839
 * [ENHANCEMENT] Query Frontend: Log `X-Grafana-User` header in query stats, slow query, and query request logs when Grafana's `send_user_header` is enabled. #7799
 * [FEATURE] Engine: Add `-querier.selector-batch-size` and `-ruler.selector-batch-size` flags to configure series batching in the Thanos promQL engine. 0 disables batching. #7763
 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792

pkg/ingester/client/client.go

@@ -168,6 +168,12 @@ func MakeIngesterClient(addr string, cfg Config, useStreamConnection bool) (Heal
 		streamCtx, streamCancel := context.WithCancel(context.Background())
 		err = c.Run(make(chan *streamWriteJob, INGESTER_CLIENT_STREAM_WORKER_COUNT), streamCtx, streamCancel)
 		if err != nil {
+			// Run() failed to start all stream-push workers. Undo what was
+			// already set up: cancel the stream context (so any workers that
+			// did start their job-processing goroutine successfully stop),
+			// and close the connection so it isn't leaked.
+			streamCancel()
+			_ = conn.Close()
 			return nil, err
 		}
 	}
@@ -210,7 +216,9 @@ func (c *closableHealthAndIngesterClient) Run(streamPushChan chan *streamWriteJo
 	c.streamCtx = streamCtx
 	c.streamCancel = streamCancel
 
-	var workerErr error
+	// Buffered so every worker can report its error without blocking, even
+	// though we only ever consume the first one.
+	errCh := make(chan error, INGESTER_CLIENT_STREAM_WORKER_COUNT)
 	var wg sync.WaitGroup
 	// Sanitize addr: colons (from host:port) are not allowed in tenant IDs.
 	sanitizedAddr := strings.ReplaceAll(c.addr, ":", "-")
@@ -220,12 +228,24 @@ func (c *closableHealthAndIngesterClient) Run(streamPushChan chan *streamWriteJo
 			workerCtx := user.InjectOrgID(streamCtx, workerName)
 			err := c.worker(workerCtx)
 			if err != nil {
-				workerErr = err
+				// A sibling worker failed to open its stream: cancel the
+				// shared stream context so the remaining workers stop
+				// opening new streams, and any job-processing goroutines
+				// that already started exit via ctx.Done().
+				streamCancel()
+				errCh <- err
 			}
 		})
 	}
 	wg.Wait()
-	return workerErr
+	close(errCh)
+
+	// Only the first error is returned; the rest are worker failures caused
+	// by the same cancellation and aren't useful on top of it.
+	for err := range errCh {
+		return err
+	}
+	return nil
 }
 
 func (c *closableHealthAndIngesterClient) worker(ctx context.Context) error {

pkg/ingester/client/client_test.go

@@ -3,16 +3,21 @@ package client
 import (
 	"context"
 	"fmt"
+	"net"
 	"net/http/httptest"
+	"runtime"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
+	"go.uber.org/atomic"
 	"google.golang.org/grpc"
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
@@ -127,7 +132,8 @@ func (m *mockIngester) Push(_ context.Context, _ *cortexpb.WriteRequest, _ ...gr
 
 func (m *mockIngester) PushStream(ctx context.Context, opts ...grpc.CallOption) (Ingester_PushStreamClient, error) {
 	args := m.Called(ctx, opts)
-	return args.Get(0).(Ingester_PushStreamClient), nil
+	stream, _ := args.Get(0).(Ingester_PushStreamClient)
+	return stream, args.Error(1)
 }
 
 type mockClientConn struct {
@@ -393,3 +399,104 @@ func TestClosableHealthAndIngesterClient_ShouldNotPanicWhenClose(t *testing.T) {
 
 	time.Sleep(100 * time.Millisecond)
 }
+
+// noopPushStreamClient is a minimal Ingester_PushStreamClient whose Send/Recv
+// are never expected to be called by these tests (no jobs are pushed).
+type noopPushStreamClient struct {
+	grpc.ClientStream
+}
+
+func (noopPushStreamClient) Send(*cortexpb.StreamWriteRequest) error { return nil }
+func (noopPushStreamClient) Recv() (*cortexpb.WriteResponse, error) {
+	return &cortexpb.WriteResponse{}, nil
+}
+
+// partialFailIngester simulates a subset of PushStream() calls failing, as
+// happens in production when an ingester address is still in the ring but no
+// longer reachable: some workers manage to open a stream, others don't.
+type partialFailIngester struct {
+	IngesterClient
+	calls atomic.Int32
+}
+
+func (p *partialFailIngester) PushStream(context.Context, ...grpc.CallOption) (Ingester_PushStreamClient, error) {
+	n := p.calls.Add(1)
+	if n%2 == 0 {
+		return nil, errors.New("injected PushStream failure")
+	}
+	return noopPushStreamClient{}, nil
+}
+
+// TestClosableHealthAndIngesterClient_Run_PartialFailureCancelsSiblingsAndReturnsError
+// is a regression test for #7759: when a subset of the stream-push workers
+// fail to open their PushStream, Run() must still return an error (safely,
+// without a data race on the collected error) and must cancel streamCtx so
+// that the job-processing goroutines started by the workers that *did*
+// succeed are not orphaned.
+func TestClosableHealthAndIngesterClient_Run_PartialFailureCancelsSiblingsAndReturnsError(t *testing.T) {
+	streamCtx, streamCancel := context.WithCancel(context.Background())
+	defer streamCancel()
+
+	client := &closableHealthAndIngesterClient{
+		IngesterClient:       &partialFailIngester{},
+		conn:                 &mockClientConn{},
+		addr:                 "test-addr",
+		inflightPushRequests: prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"ingester"}),
+	}
+
+	streamChan := make(chan *streamWriteJob, INGESTER_CLIENT_STREAM_WORKER_COUNT)
+	err := client.Run(streamChan, streamCtx, streamCancel)
+	require.Error(t, err)
+
+	// streamCtx must have been cancelled by the failing worker(s) so the
+	// job-processing goroutines spawned by the workers that succeeded exit
+	// instead of leaking forever.
+	select {
+	case <-streamCtx.Done():
+	default:
+		t.Fatal("expected streamCtx to be cancelled after a worker failure")
+	}
+}
+
+// TestMakeIngesterClient_StreamFailure_ClosesConnAndDoesNotLeak is an
+// end-to-end regression test for #7759. It reproduces the reported scenario:
+// the target address is dialable (grpc.NewClient is lazy and never errors up
+// front) but unreachable, so every eagerly-opened PushStream fails once the
+// connection reaches TRANSIENT_FAILURE. MakeIngesterClient must return the
+// error without leaking the grpc.ClientConn or its reconnect/stream
+// goroutines.
+func TestMakeIngesterClient_StreamFailure_ClosesConnAndDoesNotLeak(t *testing.T) {
+	// Reserve a local port and then stop listening on it, so connections to
+	// it are refused immediately (fast, deterministic "unreachable" target)
+	// instead of relying on an external address.
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	addr := l.Addr().String()
+	require.NoError(t, l.Close())
+
+	countRelevantGoroutines := func() int {
+		buf := make([]byte, 4<<20)
+		n := runtime.Stack(buf, true)
+		stacks := string(buf[:n])
+		count := 0
+		for frame := range strings.SplitSeq(stacks, "\n\n") {
+			// Same signatures used in the issue's own reproduction to detect
+			// leaked reconnect loops and leaked/idle push streams.
+			if strings.Contains(frame, "resetTransportAndUnlock") || strings.Contains(frame, "newClientStreamWithParams") {
+				count++
+			}
+		}
+		return count
+	}
+
+	baseline := countRelevantGoroutines()
+
+	var cfg Config
+	client, err := MakeIngesterClient(addr, cfg, true)
+	require.Error(t, err)
+	require.Nil(t, client)
+
+	require.Eventually(t, func() bool {
+		return countRelevantGoroutines() <= baseline
+	}, 10*time.Second, 50*time.Millisecond, "expected no leaked ingester-client goroutines (reconnect loop or idle streams) after MakeIngesterClient failed")
+}