From fc7406b486252b75cb9dc4da6f53244bb25801df Mon Sep 17 00:00:00 2001 From: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> Date: Fri, 21 Aug 2026 09:55:17 -0700 Subject: [PATCH] Ingester client: close conn and cancel stream ctx when Run() fails Motivation: MakeIngesterClient dials the ingester connection before starting the 100 stream-push workers used by -distributor.use-stream-push=true. If starting those workers fails, the function returned the error without closing the connection or cancelling the stream context. The grpc.ClientConn was left alive and unreferenced, with its addrConn retrying forever, and any job-processing goroutines started by workers that did succeed were left running with nothing to stop them. This happens whenever PushStream fails fast (no WaitForReady) against an address that is still in the ring but no longer reachable, e.g. during an ingester rollout, and it is also reachable from pure read traffic since the querier builds an ingester client pool through the same factory. Run() also wrote its workerErr result from up to 100 goroutines with no synchronization, a data race flagged in the same report. Approach: On MakeIngesterClient's Run() error path, cancel the stream context and close the connection before returning, matching the fix suggested in the report. Cancelling the context unblocks the job-processing goroutines of any workers that had already started successfully. Change Run()'s workerErr from a plain error to a go.uber.org/atomic.Error (already imported in this file), using Store/Load instead of an unsynchronized assignment, preserving the existing last-error-wins behavior with defined semantics under the race detector. Validation: Added TestMakeIngesterClient_CleansUpOnRunFailure, which points MakeIngesterClient (useStreamConnection=true) at a TCP port nothing listens on, so every stream-push worker fails fast with "connection refused" - the same failure mode described in the report. The test takes a runtime.NumGoroutine() baseline and polls for the goroutine count to return to it after the call returns. Ran: go test -race -tags "netgo slicelabels" ./pkg/ingester/client/... -count=1 -v all tests pass. Verified the new test fails without the fix (reverted client.go via git stash) and passes with it restored, repeated several times with no flakiness. Also ran golangci-lint run ./pkg/ingester/client/... (0 issues) and goimports -local github.com/cortexproject/cortex -l on both changed files (no output). Not run: integration tests and a live cluster reproduction - this failure mode isn't reachable through the integration harness without simulating an ingester rollout, which the unit test reproduces more directly. Checked that upstream/master's own CI is currently green. Report: https://github.com/cortexproject/cortex/issues/7759 Signed-off-by: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> Assisted-by: claude-sonnet-5 (via Claude Code) --- CHANGELOG.md | 1 + pkg/ingester/client/client.go | 11 ++++++--- pkg/ingester/client/client_test.go | 37 ++++++++++++++++++++++++++++++ 3 files changed, 46 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 16148eeae63..9de2e84343d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -92,6 +92,7 @@ * [BUGFIX] Parquet Converter: Fix `auto_forget_delay` having no effect. The ring lifecycler was created without the auto-forget delegate, so unhealthy instances were never automatically removed from the ring. #7752 * [BUGFIX] Alertmanager: Reject the global `mattermost_webhook_url_file` setting in per-tenant configs, consistent with every other global `*_file` setting. #7768 * [BUGFIX] Alertmanager: Tighten per-tenant config validation to reject additional file-based settings. #7767 +* [BUGFIX] Ingester Client: Fix `grpc.ClientConn` and goroutine leak in `MakeIngesterClient` when `-distributor.use-stream-push=true` and starting the stream-push workers fails (e.g. the ingester address is in the ring but unreachable). The connection is now closed and the stream context cancelled on that error path. #7759 ## 1.21.1 2026-06-04 diff --git a/pkg/ingester/client/client.go b/pkg/ingester/client/client.go index 0604b0de7f7..2f2da724120 100644 --- a/pkg/ingester/client/client.go +++ b/pkg/ingester/client/client.go @@ -168,6 +168,11 @@ 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 { + // Cancelling the stream context unblocks the job-processing goroutines + // started by any workers that did succeed, and closing conn tears down + // the underlying ClientConn. Without this, both are leaked forever. + streamCancel() + _ = conn.Close() return nil, err } } @@ -210,7 +215,7 @@ func (c *closableHealthAndIngesterClient) Run(streamPushChan chan *streamWriteJo c.streamCtx = streamCtx c.streamCancel = streamCancel - var workerErr error + workerErr := atomic.NewError(nil) var wg sync.WaitGroup // Sanitize addr: colons (from host:port) are not allowed in tenant IDs. sanitizedAddr := strings.ReplaceAll(c.addr, ":", "-") @@ -220,12 +225,12 @@ func (c *closableHealthAndIngesterClient) Run(streamPushChan chan *streamWriteJo workerCtx := user.InjectOrgID(streamCtx, workerName) err := c.worker(workerCtx) if err != nil { - workerErr = err + workerErr.Store(err) } }) } wg.Wait() - return workerErr + return workerErr.Load() } func (c *closableHealthAndIngesterClient) worker(ctx context.Context) error { diff --git a/pkg/ingester/client/client_test.go b/pkg/ingester/client/client_test.go index 65b46d6ca24..b57a8bd7496 100644 --- a/pkg/ingester/client/client_test.go +++ b/pkg/ingester/client/client_test.go @@ -2,8 +2,11 @@ package client import ( "context" + "flag" "fmt" + "net" "net/http/httptest" + "runtime" "strconv" "testing" "time" @@ -17,6 +20,7 @@ import ( "github.com/cortexproject/cortex/pkg/cortexpb" "github.com/cortexproject/cortex/pkg/util" + "github.com/cortexproject/cortex/pkg/util/test" ) // TestMarshall is useful to try out various optimisation on the unmarshalling code. @@ -393,3 +397,36 @@ func TestClosableHealthAndIngesterClient_ShouldNotPanicWhenClose(t *testing.T) { time.Sleep(100 * time.Millisecond) } + +// TestMakeIngesterClient_CleansUpOnRunFailure is a regression test for a +// leaked grpc.ClientConn (and its background reconnect goroutines) when +// Run() fails to start the stream-push workers. Using an address nothing +// listens on reproduces the reported failure mode: PushStream is created +// without WaitForReady, so it fails fast once the addrConn reaches +// TRANSIENT_FAILURE, and every worker fails the same way. +func TestMakeIngesterClient_CleansUpOnRunFailure(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := ln.Addr().String() + require.NoError(t, ln.Close()) + + cfg := Config{} + cfg.RegisterFlags(flag.NewFlagSet("test", flag.ContinueOnError)) + + // Let goroutines from earlier tests settle before taking the baseline. + test.Poll(t, time.Second, true, func() any { + before := runtime.NumGoroutine() + time.Sleep(10 * time.Millisecond) + return runtime.NumGoroutine() == before + }) + baseline := runtime.NumGoroutine() + + _, err = MakeIngesterClient(addr, cfg, true) + require.Error(t, err) + + // Without the fix, the ClientConn's addrConn keeps retrying in the + // background and its goroutines never return to baseline. + test.Poll(t, time.Second, true, func() any { + return runtime.NumGoroutine() <= baseline + }) +}