diff --git a/CHANGELOG.md b/CHANGELOG.md index e399bc413b8..2aaa0f77438 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/pkg/ingester/client/client.go b/pkg/ingester/client/client.go index 0604b0de7f7..c8cbad30b77 100644 --- a/pkg/ingester/client/client.go +++ b/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 { diff --git a/pkg/ingester/client/client_test.go b/pkg/ingester/client/client_test.go index 65b46d6ca24..a454ce6fe51 100644 --- a/pkg/ingester/client/client_test.go +++ b/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") +}