fix race in processing of headers in sharded queries (#2762)

* fix race in processing of headers in sharded queries

Signed-off-by: Sandeep Sukhani <[email protected]>

* put a lock for extra protection from panicing

Signed-off-by: Sandeep Sukhani <[email protected]>

* update changelog

Signed-off-by: Sandeep Sukhani <[email protected]>

Author: sandeepsukhani

CHANGELOG.md

@@ -140,6 +140,7 @@
 * [BUGFIX] Ingester: increment series per metric when recovering from WAL or transfer. #2674
 * [BUGFIX] Fixed `wrong number of arguments for 'mget' command` Redis error when a query has no chunks to lookup from storage. #2700
 * [BUGFIX] Ingester: Automatically remove old tmp checkpoints, fixing a potential disk space leak after an ingester crashes.
+* [BUGFIX] Fix race in processing of headers in sharded queries. #2762
 
 ## 1.1.0 / 2020-05-21
 

pkg/querier/queryrange/queryable.go

@@ -30,8 +30,11 @@ func (q *ShardedQueryable) Querier(ctx context.Context, mint, maxt int64) (stora
 	return q.sharededQuerier, nil
 }
 
-func (q *ShardedQueryable) getResponseHeaders() map[string][]string {
-	return q.sharededQuerier.ResponseHeaders
+func (q *ShardedQueryable) getResponseHeaders() []*PrometheusResponseHeader {
+	q.sharededQuerier.ResponseHeadersMtx.Lock()
+	defer q.sharededQuerier.ResponseHeadersMtx.Unlock()
+
+	return headersMapToPrometheusResponseHeaders(q.sharededQuerier.ResponseHeaders)
 }
 
 // ShardedQuerier is a an implementor of the Querier interface.
@@ -95,8 +98,8 @@ func (q *ShardedQuerier) handleEmbeddedQuery(encoded string) storage.SeriesSet {
 				errCh <- err
 				return
 			}
-			samplesCh <- streams
 			q.setResponseHeaders(resp.(*PrometheusResponse).Headers)
+			samplesCh <- streams
 		}(query)
 	}
 
@@ -141,3 +144,11 @@ func (q *ShardedQuerier) LabelNames() ([]string, storage.Warnings, error) {
 func (q *ShardedQuerier) Close() error {
 	return nil
 }
+
+func headersMapToPrometheusResponseHeaders(headersMap map[string][]string) (prs []*PrometheusResponseHeader) {
+	for h, v := range headersMap {
+		prs = append(prs, &PrometheusResponseHeader{Name: h, Values: v})
+	}
+
+	return
+}

pkg/querier/queryrange/querysharding.go

@@ -236,7 +236,7 @@ func (qs *queryShard) Do(ctx context.Context, r Request) (Response, error) {
 			ResultType: string(res.Value.Type()),
 			Result:     extracted,
 		},
-		Headers: headersMapToPrometheusResponseHeaders(shardedQueryable.getResponseHeaders()),
+		Headers: shardedQueryable.getResponseHeaders(),
 	}, nil
 }
 
@@ -322,11 +322,3 @@ func partitionRequest(r Request, t time.Time) (before Request, after Request) {
 
 	return r.WithStartEnd(r.GetStart(), boundary), r.WithStartEnd(boundary, r.GetEnd())
 }
-
-func headersMapToPrometheusResponseHeaders(headersMap map[string][]string) (prs []*PrometheusResponseHeader) {
-	for h, v := range headersMap {
-		prs = append(prs, &PrometheusResponseHeader{Name: h, Values: v})
-	}
-
-	return
-}