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 + }) +}