fix: RangeQuery goroutine leak

[yincongcyincong]

every time https://github.com/grafana/loki/blob/main/pkg/querier/queryrange/downstreamer.go#L197 this function return. there is no receiver for this channel.

https://github.com/grafana/loki/blob/main/pkg/querier/queryrange/downstreamer.go#L187. but there still have some error send to this channel. which result in goroutine leakage.

we can get this leakage through pprof/goroutine

[cyriltovena]

Yes that's correct what do you think of doing instead this

if err != nil {
                 select {
			case <-ctx.Done():
			case ch <- logql.Resp{
				I:   -1,
				Err: err,
			}:
                }
		}

Technically the context will be done because the main function returns.

[chaudum]

IMO there are two problems:

1. We send a response error twice:
   • First time on line 182 when the downstream function call returns an error https://github.com/grafana/loki/blob/f2fc0c26c2/pkg/querier/queryrange/downstreamer.go#L172-L177
   • Second, https://github.com/grafana/loki/blob/f2fc0c26c2/pkg/querier/queryrange/downstreamer.go#L186-L191 after the concurrency.ForEachJob() function returns an error on line 171
2. We do not drain the channel ch when accumulation returns an error but instead return https://github.com/grafana/loki/blob/f2fc0c26c2/pkg/querier/queryrange/downstreamer.go#L199-L201 This leaves remaining results in the channel.

diff --git a/pkg/querier/queryrange/downstreamer.go b/pkg/querier/queryrange/downstreamer.go
index c6fba0fbf4..490a63150d 100644
--- a/pkg/querier/queryrange/downstreamer.go
+++ b/pkg/querier/queryrange/downstreamer.go
@@ -170,18 +170,19 @@ func (in instance) For(
 	go func() {
 		err := concurrency.ForEachJob(ctx, len(queries), in.parallelism, func(ctx context.Context, i int) error {
 			res, err := fn(queries[i])
+			if err != nil {
+				return err
+			}
 			response := logql.Resp{
 				I:   i,
 				Res: res,
-				Err: err,
 			}
-
 			// Feed the result into the channel unless the work has completed.
 			select {
 			case <-ctx.Done():
 			case ch <- response:
 			}
-			return err
+			return nil
 		})
 		if err != nil {
 			ch <- logql.Resp{
@@ -192,15 +193,15 @@ func (in instance) For(
 		close(ch)
 	}()
 
+	var err error
 	for resp := range ch {
-		if resp.Err != nil {
-			return nil, resp.Err
-		}
-		if err := acc.Accumulate(ctx, resp.Res, resp.I); err != nil {
-			return nil, err
+		if err != nil || resp.Err != nil {
+			err = resp.Err
+			continue
 		}
+		err = acc.Accumulate(ctx, resp.Res, resp.I)
 	}
-	return acc.Result(), nil
+	return acc.Result(), err
 }
 
 // convert to matrix

[yincongcyincong]

@cyriltovena @chaudum thanks for your review. i modified these code.
but i think this part has something wrong. when acc.Accumulate have an error and resp.Err is nil, err will replace by nil

[cyriltovena]

LGTM

[chaudum]

@cyriltovena @chaudum thanks for your review. i modified these code. but i think this part has something wrong. when acc.Accumulate have an error and resp.Err is nil, err will replace by nil

You're right. Both errors need to be checked separately.

[chaudum]

LGMT. Thanks!