Merge timeseries and chunkseries from ingesters (#3013)

* Merge incoming chunks and timeseries.

Signed-off-by: Peter Štibraný <[email protected]>

* Added CHANGELOG.md

Signed-off-by: Peter Štibraný <[email protected]>

* Update CHANGELOG.md

Signed-off-by: Peter Štibraný <[email protected]>

* Remove obsolete comments.

Signed-off-by: Peter Štibraný <[email protected]>

* Mention that disabling streaming is only necessary in Cortex 1.3.0.

Signed-off-by: Peter Štibraný <[email protected]>

* Added integration test, and fixed panic when mixing chunk-based series and other series.

Signed-off-by: Peter Štibraný <[email protected]>

* Fix lint warnings.

Signed-off-by: Peter Štibraný <[email protected]>

Co-authored-by: Marco Pracucci <[email protected]>

Author: pstibrany

CHANGELOG.md

@@ -19,6 +19,7 @@
 * [ENHANCEMENT] Added `cortex_request_message_bytes` and `cortex_response_message_bytes` histograms to track received and sent gRPC message and HTTP request/response sizes. Added `cortex_inflight_requests` gauge to track number of inflight gRPC and HTTP requests. #3064
 * [ENHANCEMENT] Add config validation to the experimental Alertmanager API. Invalid configs are no longer accepted. #3053
 * [BUGFIX] Query-frontend: Fixed rounding for incoming query timestamps, to be 100% Prometheus compatible. #2990
+* [BUGFIX] Querier: Merge results from chunks and blocks ingesters when using streaming of results. #3013
 * [BUGFIX] Querier: query /series from ingesters regardless the `-querier.query-ingesters-within` setting. #3035
 * [BUGFIX] Experimental blocks storage: Ingester is less likely to hit gRPC message size limit when streaming data to queriers. #3015
 * [BUGFIX] Fix configuration for TLS server validation, TLS skip verify was hardcoded to true for all TLS configurations and prevented validation of server certificates. #3030

docs/blocks-storage/migrate-from-chunks-to-blocks.md

@@ -18,6 +18,7 @@ This article **assumes** that:
 - Cortex cluster is managed by Kubernetes
 - Cortex is using chunks storage
 - Ingesters are using WAL
+- Cortex version 1.3.0 or later.
 
 _If your ingesters are not using WAL, the documented procedure will still apply, but the presented migration script will not work properly without changes, as it assumes that ingesters are managed via StatefulSet._
 
@@ -59,9 +60,9 @@ and need to use storage instead.
 
 #### `-querier.ingester-streaming=false`
 
-Querier (and ruler) has a [bug](https://github.com/cortexproject/cortex/issues/2935) and doesn't properly
-merge streamed results from chunks and blocks-based ingesters. Instead it only returns data from blocks- instesters.
-To avoid this problem, we need to temporarily disable this feature by setting `-querier.ingester-streaming=false`.
+Querier (and ruler) in Cortex version 1.3.0 has a [bug](https://github.com/cortexproject/cortex/issues/2935) and doesn't properly
+merge streamed results from chunks and blocks-based ingesters. Instead it only returns data from blocks instesters.
+To avoid this problem we can use newer Cortex release, or temporarily disable this feature by setting `-querier.ingester-streaming=false`.
 After migration is complete (i.e. all ingesters are running blocks only), this can be turned back to true, which is the default value.
 
 ### Query-frontend
@@ -132,7 +133,7 @@ This flag can be set to a timestamp when migration has finished, and it avoids q
 
 #### `-querier.ingester-streaming=true`
 
-Querier can be configured to make use of streamed responses from ingester at this point (`-querier.ingester-streaming=true`).
+If querier was configured to disable ingester streaming during migration (required for Cortex 1.3.0), Querier can be configured to make use of streamed responses from ingester at this point (`-querier.ingester-streaming=true`).
 
 ## Rollback
 

integration/querier_streaming_mixed_ingester_test.go

@@ -0,0 +1,154 @@
+// +build requires_docker
+
+package main
+
+import (
+	"context"
+	"flag"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/common/user"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/cortexproject/cortex/integration/e2ecortex"
+	client2 "github.com/cortexproject/cortex/pkg/ingester/client"
+)
+
+func TestQuerierWithStreamingBlocksAndChunksIngesters(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	require.NoError(t, writeFileToSharedDir(s, cortexSchemaConfigFile, []byte(cortexSchemaConfigYaml)))
+	chunksFlags := mergeFlags(ChunksStorageFlags, map[string]string{})
+
+	blockFlags := mergeFlags(BlocksStorageFlags, map[string]string{
+		"-experimental.blocks-storage.tsdb.block-ranges-period":      "1h",
+		"-experimental.blocks-storage.tsdb.head-compaction-interval": "1m",
+		"-experimental.store-gateway.sharding-enabled":               "false",
+		"-querier.ingester-streaming":                                "true",
+	})
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, blockFlags["-experimental.blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	// Start Cortex components.
+	ingesterBlocks := e2ecortex.NewIngester("ingester-blocks", consul.NetworkHTTPEndpoint(), blockFlags, "")
+	ingesterChunks := e2ecortex.NewIngester("ingester-chunks", consul.NetworkHTTPEndpoint(), chunksFlags, "")
+	storeGateway := e2ecortex.NewStoreGateway("store-gateway", consul.NetworkHTTPEndpoint(), blockFlags, "")
+	require.NoError(t, s.StartAndWaitReady(ingesterBlocks, ingesterChunks, storeGateway))
+
+	// Sharding is disabled, pass gateway address.
+	querierFlags := mergeFlags(blockFlags, map[string]string{
+		"-experimental.querier.store-gateway-addresses": strings.Join([]string{storeGateway.NetworkGRPCEndpoint()}, ","),
+		"-distributor.shard-by-all-labels":              "true",
+	})
+	querier := e2ecortex.NewQuerier("querier", consul.NetworkHTTPEndpoint(), querierFlags, "")
+	require.NoError(t, s.StartAndWaitReady(querier))
+
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(1024), "cortex_ring_tokens_total"))
+
+	s1 := []client2.Sample{
+		{Value: 1, TimestampMs: 1000},
+		{Value: 2, TimestampMs: 2000},
+		{Value: 3, TimestampMs: 3000},
+		{Value: 4, TimestampMs: 4000},
+		{Value: 5, TimestampMs: 5000},
+	}
+
+	s2 := []client2.Sample{
+		{Value: 1, TimestampMs: 1000},
+		{Value: 2.5, TimestampMs: 2500},
+		{Value: 3, TimestampMs: 3000},
+		{Value: 5.5, TimestampMs: 5500},
+	}
+
+	clientConfig := client2.Config{}
+	clientConfig.RegisterFlags(flag.NewFlagSet("unused", flag.ContinueOnError)) // registers default values
+
+	// Push data to chunks ingester.
+	{
+		ingesterChunksClient, err := client2.MakeIngesterClient(ingesterChunks.GRPCEndpoint(), clientConfig)
+		require.NoError(t, err)
+		defer ingesterChunksClient.Close()
+
+		_, err = ingesterChunksClient.Push(user.InjectOrgID(context.Background(), "user"), &client2.WriteRequest{
+			Timeseries: []client2.PreallocTimeseries{
+				{TimeSeries: &client2.TimeSeries{Labels: []client2.LabelAdapter{{Name: labels.MetricName, Value: "s"}, {Name: "l", Value: "1"}}, Samples: s1}},
+				{TimeSeries: &client2.TimeSeries{Labels: []client2.LabelAdapter{{Name: labels.MetricName, Value: "s"}, {Name: "l", Value: "2"}}, Samples: s1}}},
+			Source: client2.API,
+		})
+		require.NoError(t, err)
+	}
+
+	// Push data to blocks ingester.
+	{
+		ingesterBlocksClient, err := client2.MakeIngesterClient(ingesterBlocks.GRPCEndpoint(), clientConfig)
+		require.NoError(t, err)
+		defer ingesterBlocksClient.Close()
+
+		_, err = ingesterBlocksClient.Push(user.InjectOrgID(context.Background(), "user"), &client2.WriteRequest{
+			Timeseries: []client2.PreallocTimeseries{
+				{TimeSeries: &client2.TimeSeries{Labels: []client2.LabelAdapter{{Name: labels.MetricName, Value: "s"}, {Name: "l", Value: "2"}}, Samples: s2}},
+				{TimeSeries: &client2.TimeSeries{Labels: []client2.LabelAdapter{{Name: labels.MetricName, Value: "s"}, {Name: "l", Value: "3"}}, Samples: s1}}},
+			Source: client2.API,
+		})
+		require.NoError(t, err)
+	}
+
+	c, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", "", "user")
+	require.NoError(t, err)
+
+	// Query back the series (1 only in the storage, 1 only in the ingesters, 1 on both).
+	result, err := c.Query("s[1m]", time.Unix(10, 0))
+	require.NoError(t, err)
+
+	s1Values := []model.SamplePair{
+		{Value: 1, Timestamp: 1000},
+		{Value: 2, Timestamp: 2000},
+		{Value: 3, Timestamp: 3000},
+		{Value: 4, Timestamp: 4000},
+		{Value: 5, Timestamp: 5000},
+	}
+
+	s1AndS2ValuesMerged := []model.SamplePair{
+		{Value: 1, Timestamp: 1000},
+		{Value: 2, Timestamp: 2000},
+		{Value: 2.5, Timestamp: 2500},
+		{Value: 3, Timestamp: 3000},
+		{Value: 4, Timestamp: 4000},
+		{Value: 5, Timestamp: 5000},
+		{Value: 5.5, Timestamp: 5500},
+	}
+
+	expectedMatrix := model.Matrix{
+		// From chunks ingester only.
+		&model.SampleStream{
+			Metric: model.Metric{labels.MetricName: "s", "l": "1"},
+			Values: s1Values,
+		},
+
+		// From blocks ingester only.
+		&model.SampleStream{
+			Metric: model.Metric{labels.MetricName: "s", "l": "3"},
+			Values: s1Values,
+		},
+
+		// Merged from both ingesters.
+		&model.SampleStream{
+			Metric: model.Metric{labels.MetricName: "s", "l": "2"},
+			Values: s1AndS2ValuesMerged,
+		},
+	}
+
+	require.Equal(t, model.ValMatrix, result.Type())
+	require.ElementsMatch(t, expectedMatrix, result.(model.Matrix))
+}

pkg/querier/distributor_queryable.go

@@ -137,8 +137,9 @@ func (q *distributorQuerier) streamingSelect(minT, maxT int64, matchers []*label
 		return storage.ErrSeriesSet(err)
 	}
 
-	if len(results.Timeseries) != 0 {
-		return newTimeSeriesSeriesSet(results.Timeseries)
+	sets := []storage.SeriesSet(nil)
+	if len(results.Timeseries) > 0 {
+		sets = append(sets, newTimeSeriesSeriesSet(results.Timeseries))
 	}
 
 	serieses := make([]storage.Series, 0, len(results.Chunkseries))
@@ -156,15 +157,27 @@ func (q *distributorQuerier) streamingSelect(minT, maxT int64, matchers []*label
 			return storage.ErrSeriesSet(err)
 		}
 
-		series := &chunkSeries{
+		serieses = append(serieses, &chunkSeries{
 			labels:            ls,
 			chunks:            chunks,
 			chunkIteratorFunc: q.chunkIterFn,
-		}
-		serieses = append(serieses, series)
+			mint:              minT,
+			maxt:              maxT,
+		})
+	}
+
+	if len(serieses) > 0 {
+		sets = append(sets, series.NewConcreteSeriesSet(serieses))
 	}
 
-	return series.NewConcreteSeriesSet(serieses)
+	if len(sets) == 0 {
+		return storage.EmptySeriesSet()
+	}
+	if len(sets) == 1 {
+		return sets[0]
+	}
+	// Sets need to be sorted. Both series.NewConcreteSeriesSet and newTimeSeriesSeriesSet take care of that.
+	return storage.NewMergeSeriesSet(sets, storage.ChainedSeriesMerge)
 }
 
 func (q *distributorQuerier) LabelValues(name string) ([]string, storage.Warnings, error) {

pkg/querier/distributor_queryable_test.go

@@ -214,6 +214,118 @@ func TestIngesterStreaming(t *testing.T) {
 	require.NoError(t, seriesSet.Err())
 }
 
+func TestIngesterStreamingMixedResults(t *testing.T) {
+	const (
+		mint = 0
+		maxt = 10000
+	)
+	s1 := []client.Sample{
+		{Value: 1, TimestampMs: 1000},
+		{Value: 2, TimestampMs: 2000},
+		{Value: 3, TimestampMs: 3000},
+		{Value: 4, TimestampMs: 4000},
+		{Value: 5, TimestampMs: 5000},
+	}
+	s2 := []client.Sample{
+		{Value: 1, TimestampMs: 1000},
+		{Value: 2.5, TimestampMs: 2500},
+		{Value: 3, TimestampMs: 3000},
+		{Value: 5.5, TimestampMs: 5500},
+	}
+
+	mergedSamplesS1S2 := []client.Sample{
+		{Value: 1, TimestampMs: 1000},
+		{Value: 2, TimestampMs: 2000},
+		{Value: 2.5, TimestampMs: 2500},
+		{Value: 3, TimestampMs: 3000},
+		{Value: 4, TimestampMs: 4000},
+		{Value: 5, TimestampMs: 5000},
+		{Value: 5.5, TimestampMs: 5500},
+	}
+
+	d := &mockDistributor{}
+	d.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(
+		&client.QueryStreamResponse{
+			Chunkseries: []client.TimeSeriesChunk{
+				{
+					Labels: []client.LabelAdapter{{Name: labels.MetricName, Value: "one"}},
+					Chunks: convertToChunks(t, s1),
+				},
+				{
+					Labels: []client.LabelAdapter{{Name: labels.MetricName, Value: "two"}},
+					Chunks: convertToChunks(t, s1),
+				},
+			},
+
+			Timeseries: []client.TimeSeries{
+				{
+					Labels:  []client.LabelAdapter{{Name: labels.MetricName, Value: "two"}},
+					Samples: s2,
+				},
+				{
+					Labels:  []client.LabelAdapter{{Name: labels.MetricName, Value: "three"}},
+					Samples: s1,
+				},
+			},
+		},
+		nil)
+
+	ctx := user.InjectOrgID(context.Background(), "0")
+	queryable := newDistributorQueryable(d, true, mergeChunks, 0)
+	querier, err := queryable.Querier(ctx, mint, maxt)
+	require.NoError(t, err)
+
+	seriesSet := querier.Select(true, &storage.SelectHints{Start: mint, End: maxt}, labels.MustNewMatcher(labels.MatchRegexp, labels.MetricName, ".*"))
+	require.NoError(t, seriesSet.Err())
+
+	require.True(t, seriesSet.Next())
+	verifySeries(t, seriesSet.At(), labels.Labels{{Name: labels.MetricName, Value: "one"}}, s1)
+
+	require.True(t, seriesSet.Next())
+	verifySeries(t, seriesSet.At(), labels.Labels{{Name: labels.MetricName, Value: "three"}}, s1)
+
+	require.True(t, seriesSet.Next())
+	verifySeries(t, seriesSet.At(), labels.Labels{{Name: labels.MetricName, Value: "two"}}, mergedSamplesS1S2)
+
+	require.False(t, seriesSet.Next())
+	require.NoError(t, seriesSet.Err())
+}
+
+func verifySeries(t *testing.T, series storage.Series, l labels.Labels, samples []client.Sample) {
+	require.Equal(t, l, series.Labels())
+
+	it := series.Iterator()
+	for _, s := range samples {
+		require.True(t, it.Next())
+		require.Nil(t, it.Err())
+		ts, v := it.At()
+		require.Equal(t, s.Value, v)
+		require.Equal(t, s.TimestampMs, ts)
+	}
+	require.False(t, it.Next())
+	require.Nil(t, it.Err())
+}
+
+func convertToChunks(t *testing.T, samples []client.Sample) []client.Chunk {
+	// We need to make sure that there is atleast one chunk present,
+	// else no series will be selected.
+	promChunk, err := encoding.NewForEncoding(encoding.Bigchunk)
+	require.NoError(t, err)
+
+	for _, s := range samples {
+		c, err := promChunk.Add(model.SamplePair{Value: model.SampleValue(s.Value), Timestamp: model.Time(s.TimestampMs)})
+		require.NoError(t, err)
+		require.Nil(t, c)
+	}
+
+	clientChunks, err := chunkcompat.ToChunks([]chunk.Chunk{
+		chunk.NewChunk("", 0, nil, promChunk, model.Time(samples[0].TimestampMs), model.Time(samples[len(samples)-1].TimestampMs)),
+	})
+	require.NoError(t, err)
+
+	return clientChunks
+}
+
 type mockDistributor struct {
 	mock.Mock
 }