MakeIngesterClient leaks grpc.ClientConn and goroutines on failure

[alex-berger]

Describe the bug

When -distributor.use-stream-push=true, MakeIngesterClient() creates the
grpc.ClientConn before starting the push-stream workers. If starting those
workers fails, the function returns the error without closing the connection or
cancelling the stream context. The ClientConn is now unreferenced but still
alive: its addrConn reconnect loop keeps dialing the (usually dead) ingester
address forever, with gRPC's 120s maximum backoff.

Because ring/client.Pool.GetClientFor() only records the client after the
factory succeeds, a failed factory call leaves nothing in Pool.clients. Neither
removeStaleClients() nor cleanUnhealthy() can ever reach these connections,
and cortex_distributor_ingester_clients keeps reporting the correct (low)
number while orphans accumulate invisibly. The pool itself is fine — the leak is
entirely outside its bookkeeping.

Every failed factory call leaks one connection permanently, and failures cluster
during ingester rollouts. The window that matters is the one where an ingester's
address is still in the ring but no longer reachable (pod terminating or
already gone). In that window grpc.NewClient still succeeds — it is lazy — and
the first PushStream then fails fast, because the addrConn reaches
TRANSIENT_FAILURE and the stream is created without WaitForReady. So every
GetClientFor for that address leaks one connection, at whatever rate the
component happens to be calling it. The leak grows by a burst on each rollout and
never shrinks until the process restarts.

(Streams can also fail while the ingester is gracefully shutting down. Note the
server-side PushStream handler does not reject at stream-open time — it has no
checkRunning of its own and only reaches one per message via i.Push — so a
draining-but-alive ingester generally accepts the stream. Unreachability is the
dominant trigger.)

Why this also hits components that never push

Run() opens all 100 PushStreams eagerly, at construction time, not lazily
on the first push. Every component that builds an ingester client pool through
the distributor gets this — including the querier, whose embedded distributor
creates the pool with the same factory and reaches GetClientFor from the read
path (ForReplicationSet,
distributor.go#L1364).

Two consequences:

• Ordinary query traffic is sufficient to trigger this leak. No write traffic
  to the affected process is required.
• A querier permanently maintains 100 idle push streams per live ingester that it
  will never use. That is a separate design question from this bug, but it is why
  the goroutine numbers below are as large as they are.

The code

pkg/ingester/client/client.go#L146-L175
(master e450d52; identical in v1.21.1):

conn, err := grpc.NewClient(addr, dialOpts...)
if err != nil {
	return nil, err
}
c := &closableHealthAndIngesterClient{ /* ... */ conn: conn /* ... */ }
if useStreamConnection {
	streamCtx, streamCancel := context.WithCancel(context.Background())
	err = c.Run(make(chan *streamWriteJob, INGESTER_CLIENT_STREAM_WORKER_COUNT), streamCtx, streamCancel)
	if err != nil {
		return nil, err // <-- conn and streamCtx are both leaked here
	}
}
return c, nil

Run() starts INGESTER_CLIENT_STREAM_WORKER_COUNT (100) workers, each of which
opens a PushStream. A single worker failing is enough to make Run() return
non-nil.

Three secondary problems in the same path

1. Run() does not stop the workers that already started. It waits for all
   100 to return, keeps only the last error, and leaves the successful ones
   running even though the client is about to be discarded.
2. Goroutine leak. worker() spawns a job-processing goroutine for every
   stream it opens successfully. Those goroutines only exit on ctx.Done() or on
   streamPushChan closing. When Run() returns an error, streamCancel() is
   never called and Close() is never reached, so they leak alongside the
   connection. In the fully-unreachable case above all 100 workers fail, so
   little or nothing leaks here — this matters in the partial-failure case, where
   up to 99 goroutines and their streams survive a discarded client.
3. Data race. In
   Run(),
   workerErr is written from all 100 worker goroutines with no
   synchronisation. It should be an atomic.Error, a mutex-guarded value, or a
   multierror collected through a channel.

To Reproduce

1. Run Cortex v1.21.1 (or master) with -distributor.use-stream-push=true.
2. Roll the ingester StatefulSet, so that one ingester's address remains in the
   ring while its pod is unreachable.
3. During that window, keep traffic flowing so Pool.GetClientFor() is called
   repeatedly for that address. Read traffic is enough — a querier reaches the
   same factory through ForReplicationSet.
4. After the ingester's pod is gone, observe that the process keeps opening TCP
   connections to the now-dead pod IP indefinitely.

Useful confirmations:

# Reconnect loops far exceed the number of real peers
$ curl -s localhost:8080/debug/pprof/goroutine?debug=1 \
    | awk '/^[0-9]+ @/{n=$1} /addrConn.resetTransportAndUnlock/{print n}'

(Frame name as of grpc-go v1.82.1, the version pinned in go.mod.)

Note: grep -c on the pprof text output gives 1, not the goroutine count —
aggregated pprof prints one stack per unique trace with an N @ prefix.

Expected behavior

A failed MakeIngesterClient() should leave nothing behind. The early return
needs to undo what was already set up:

 	if useStreamConnection {
 		streamCtx, streamCancel := context.WithCancel(context.Background())
 		err = c.Run(make(chan *streamWriteJob, INGESTER_CLIENT_STREAM_WORKER_COUNT), streamCtx, streamCancel)
 		if err != nil {
+			streamCancel()
+			_ = conn.Close()
 			return nil, err
 		}
 	}

Cancelling streamCtx also unblocks the job-processing goroutines started by the
workers that did succeed.

Environment

• Infrastructure: Kubernetes on AWS EKS, Cilium CNI
• Deployment tool: Helm / FluxCD
• Cortex version: v1.21.1 (defect verified unchanged on master e450d52)
• grpc-go: v1.82.1
• Relevant config: distributor.use_stream_push: true

Storage Engine

• Blocks
• Chunks

Additional Context

Observed on a production querier, which receives no write traffic at all —
its embedded distributor builds the ingester pool through the same factory, and
the read path calls GetClientFor. One ingester was replaced; ~12.8h later the
querier still held roughly 165 orphaned ClientConns to the dead pod's IP,
producing a sustained ~1.5 SYN/s that the CNI dropped — after a pod is
deleted its IP no longer maps to a pod identity, so the network policy that
allows querier→ingester traffic no longer matches and the packets are denied.

Signal	Observed	Expected
goroutines in addrConn.resetTransportAndUnlock	180	~15 (6 ingesters + 3 store-gateways + 3 frontends + 3 schedulers)
goroutines in newClientStreamWithParams	620	~600 (6 live ingesters × 100 eagerly opened, permanently idle streams)
cortex_distributor_ingester_clients	6	6 — correct, and blind to the orphans
SYNs to the dead ingester IP	~1.5/s	0

The orphan count is derived from the reconnect-goroutine delta (180 observed − 15
legitimate peers). The measured SYN rate is an independent corroboration:
165 orphans ÷ gRPC's 120s maximum backoff ≈ 1.4 attempts/s, against ~1.5/s
actually observed on the wire.

The goroutine counts are consistent with one leaked ClientConn per orphan and
no accompanying goroutine leak, which is what the unreachable-address path
predicts: all 100 workers fail, so only the connection survives.

In the interest of precision: the code defect and these measurements are certain,
but the attribution of these specific orphans to that early return is inference
from the arithmetic and the timeline rather than a heap dump. I could not find
another path in Cortex that creates an ingester ClientConn without registering
it in the pool.

Queries were unaffected in steady state; the cost is leaked sockets, leaked
goroutines and continuous dropped packets, all of which persist for the lifetime
of the process and grow with every ingester rollout.

Rings, DNS and EndpointSlices were all verified clean — the dead address exists
nowhere in Cortex's own service discovery, which is what pointed at connections
that no component is tracking.