From e9388db7fb509bf6c305ce7cabfcf47950459f9e Mon Sep 17 00:00:00 2001
From: David Leadbeater <David.Leadbeater@gresearch.co.uk>
Date: Mon, 30 Nov 2020 11:36:04 +0000
Subject: [PATCH] Handle empty series correctly (#86)

Behaviour of this changed in thanos-io/thanos#3010 -- this now generates an error, as it probably should, but we sent back a series with no datapoints in it previously.
---
 pkg/store/store.go      | 17 +++++++++++++----
 pkg/store/store_test.go | 27 +++++++++++++++++----------
 2 files changed, 30 insertions(+), 14 deletions(-)

diff --git a/pkg/store/store.go b/pkg/store/store.go
index 6c89b28..86ca9c2 100644
--- a/pkg/store/store.go
+++ b/pkg/store/store.go
@@ -248,6 +248,11 @@ func (store *OpenTSDBStore) Series(
 			if err != nil {
 				return err
 			}
+			if res == nil {
+				// OpenTSDB can return a series with no datapoints, in Prometheus this
+				// is just not returned.
+				continue
+			}
 			if err := server.Send(res); err != nil {
 				return err
 			}
@@ -650,10 +655,14 @@ func (store *OpenTSDBStore) convertOpenTSDBResultsToSeriesResponse(respI *opents
 		}
 		chunks = append(chunks, aggrChunk)
 	}
-	return storepb.NewSeriesResponse(&storepb.Series{
-		Labels: seriesLabels,
-		Chunks: chunks,
-	}), len(dps), nil
+	var response *storepb.SeriesResponse
+	if len(chunks) > 0 {
+		response = storepb.NewSeriesResponse(&storepb.Series{
+			Labels: seriesLabels,
+			Chunks: chunks,
+		})
+	}
+	return response, len(dps), nil
 }
 
 func convertPromQLMatcherToFilter(matcher storepb.LabelMatcher) (opentsdb.Filter, error) {
diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go
index c408eec..dd5d700 100644
--- a/pkg/store/store_test.go
+++ b/pkg/store/store_test.go
@@ -979,23 +979,21 @@ func TestConvertOpenTSDBResultsToSeriesResponse(t *testing.T) {
 				Tags:   map[string]string{},
 				Dps:    newDps(map[string]interface{}{}),
 			},
-			expectedOutput: storepb.NewSeriesResponse(&storepb.Series{
-				Labels: []storepb.Label{{Name: "__name__", Value: "metric"}},
-				Chunks: []storepb.AggrChunk{},
-			}),
-			expectedChunkTypes: []storepb.Aggr{},
+			expectedOutput: nil,
 		},
 		{
 			input: opentsdb.QueryRespItem{
 				Metric: "metric.with.dot.and-dash",
 				Tags:   map[string]string{},
-				Dps:    newDps(map[string]interface{}{}),
+				Dps: newDps(map[string]interface{}{
+					"1": 1.0,
+				}),
 			},
 			expectedOutput: storepb.NewSeriesResponse(&storepb.Series{
 				Labels: []storepb.Label{{Name: "__name__", Value: "metric:with:dot:and_dash"}},
-				Chunks: []storepb.AggrChunk{},
+				Chunks: []storepb.AggrChunk{{MinTime: 1, MaxTime: 1}},
 			}),
-			expectedChunkTypes: []storepb.Aggr{},
+			expectedChunkTypes: []storepb.Aggr{storepb.Aggr_RAW},
 		},
 		{
 			input: opentsdb.QueryRespItem{
@@ -1187,13 +1185,22 @@ func TestConvertOpenTSDBResultsToSeriesResponse(t *testing.T) {
 			expectedChunkTypes: []storepb.Aggr{storepb.Aggr_RAW},
 		},
 	}
-	for _, test := range testCases {
+	for i, test := range testCases {
 		store := NewOpenTSDBStore(
 			log.NewJSONLogger(&testLogger{t}), nil, nil, time.Duration(0), 1*time.Minute, test.storeLabels, nil, nil, false, true, "foo")
 		converted, _, err := store.convertOpenTSDBResultsToSeriesResponse(&test.input)
 		if err != nil {
 			t.Errorf("unexpected error: %s", err.Error())
 		}
+		if converted == nil && test.expectedOutput == nil {
+			continue // Expected nothing, got nothing.
+		}
+		if test.expectedOutput == nil {
+			t.Errorf("expected nil, got %v", converted)
+		}
+		if converted == nil {
+			t.Errorf("%d: expected response, got nil", i)
+		}
 		if len(converted.GetSeries().Labels) == len(test.expectedOutput.GetSeries().Labels) {
 			// Labels must be alphabetically sorted
 			for i, label := range converted.GetSeries().Labels {
@@ -1206,7 +1213,7 @@ func TestConvertOpenTSDBResultsToSeriesResponse(t *testing.T) {
 			t.Errorf("number of tags does not match")
 		}
 		if len(test.expectedOutput.GetSeries().Chunks) != len(converted.GetSeries().Chunks) {
-			t.Error("number of chunks does not match")
+			t.Errorf("%d: number of chunks does not match", i)
 		}
 		for ci, chunk := range test.expectedOutput.GetSeries().Chunks {
 			convertedChunk := converted.GetSeries().Chunks[ci]