TSDB-46: Prometheus MIN_OVER_TIME function omits the first value in a series (#182)

* improve debug log

* add more logs

* fix compilation issue and initialize logger

* Apply fix to IterSortMerger. See also PR 174

* fix problem when querying multi partitions and getting data labelsets OOO

* fix skipping first element issue

* Add integration test for MIN, MAX aggregations

* Revert "Add integration test for MIN, MAX aggregations"

This reverts commit f8d0bfd.

* cosmetic fixes; see comments in PR

Author: Igor Makhlin

pkg/querier/querier.go

@@ -31,6 +31,7 @@ import (
 	"github.com/v3io/v3io-tsdb/pkg/utils"
 	"sort"
 	"strings"
+	"time"
 )
 
 // Create a new Querier interface
@@ -86,16 +87,18 @@ func (q *V3ioQuerier) SelectOverlap(name, functions string, step int64, windows
 func (q *V3ioQuerier) selectQry(
 	name, functions string, step int64, windows []int, filter string) (set SeriesSet, err error) {
 
-	set = nullSeriesSet{}
-
-	filter = strings.Replace(filter, "__name__", "_name", -1)
-	q.logger.DebugWith("Select query", "metric", name, "func", functions, "step", step, "filter", filter, "disableAllAggr", q.disableAllAggr, "disableClientAggr", q.disableClientAggr, "window", windows)
 	err = q.partitionMngr.ReadAndUpdateSchema()
-
 	if err != nil {
 		return nullSeriesSet{}, errors.Wrap(err, "Failed to read/update the TSDB schema.")
 	}
 
+	set = nullSeriesSet{}
+
+	q.logger.Debug("Select query:\n\tMetric: %s\n\tStart Time: %s (%d)\n\tEnd Time: %s (%d)\n\tFunction: %s\n\t"+
+		"Step: %d\n\tFilter: %s\n\tWindows: %v\n\tDisable All Aggr: %t\n\tDisable Client Aggr: %t",
+		name, time.Unix(q.mint/1000, 0).String(), q.mint, time.Unix(q.maxt/1000, 0).String(), q.maxt, functions, step,
+		filter, windows, q.disableAllAggr, q.disableClientAggr)
+
 	q.performanceReporter.WithTimer("QueryTimer", func() {
 		filter = strings.Replace(filter, "__name__", "_name", -1)
 

pkg/querier/series.go

@@ -147,9 +147,18 @@ func (it *v3ioSeriesIterator) Seek(t int64) bool {
 		t = it.mint
 	}
 
+	// Check the first element
+	t0, _ := it.iter.At()
+	if t0 > it.maxt {
+		return false
+	}
+	if t <= t0 {
+		return true
+	}
+
 	for {
 		if it.iter.Next() {
-			t0, _ := it.At()
+			t0, _ := it.iter.At()
 			if t0 > it.maxt {
 				return false
 			}

promtsdb/promtsdb.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/nuclio/logger"
+	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/storage"
 	"github.com/v3io/v3io-go-http"
@@ -12,17 +13,27 @@ import (
 	"github.com/v3io/v3io-tsdb/pkg/config"
 	"github.com/v3io/v3io-tsdb/pkg/querier"
 	"github.com/v3io/v3io-tsdb/pkg/tsdb"
+	"github.com/v3io/v3io-tsdb/pkg/utils"
 	"strings"
 )
 
 type V3ioPromAdapter struct {
-	db *tsdb.V3ioAdapter
+	db     *tsdb.V3ioAdapter
+	logger logger.Logger
 }
 
 func NewV3ioProm(cfg *config.V3ioConfig, container *v3io.Container, logger logger.Logger) (*V3ioPromAdapter, error) {
 
+	if logger == nil {
+		newLogger, err := utils.NewLogger(cfg.LogLevel)
+		if err != nil {
+			return nil, errors.Wrap(err, "Unable to initialize logger.")
+		}
+		logger = newLogger
+	}
+
 	adapter, err := tsdb.NewV3ioAdapter(cfg, container, logger)
-	newAdapter := V3ioPromAdapter{db: adapter}
+	newAdapter := V3ioPromAdapter{db: adapter, logger: logger.GetChild("v3io-prom-adapter")}
 	return &newAdapter, err
 }
 
@@ -46,19 +57,22 @@ func (a *V3ioPromAdapter) Close() error {
 
 func (a *V3ioPromAdapter) Querier(_ context.Context, mint, maxt int64) (storage.Querier, error) {
 	v3ioQuerier, err := a.db.Querier(nil, mint, maxt)
-	querier := V3ioPromQuerier{q: v3ioQuerier}
-	return &querier, err
+	promQuerier := V3ioPromQuerier{v3ioQuerier: v3ioQuerier, logger: a.logger.GetChild("v3io-prom-query")}
+	return &promQuerier, err
 }
 
 type V3ioPromQuerier struct {
-	q *querier.V3ioQuerier
+	v3ioQuerier *querier.V3ioQuerier
+	logger      logger.Logger
 }
 
 // Select returns a set of series that matches the given label matchers.
-func (q *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*labels.Matcher) (storage.SeriesSet, error) {
-	name, filter, functions := match2filter(oms)
+func (promQuery *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*labels.Matcher) (storage.SeriesSet, error) {
+	name, filter, functions := match2filter(oms, promQuery.logger)
 	noAggr := false
 
+	promQuery.logger.Debug("SelectParams: %+v", params)
+
 	if params.Func != "" {
 		// only pass xx_over_time functions (just the xx part)
 		// TODO: support count/stdxx, require changes in Prometheus: promql/functions.go, not calc aggregate twice
@@ -71,26 +85,27 @@ func (q *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*labels.Ma
 			}
 		}
 	}
-	set, err := q.q.SelectProm(name, functions, params.Step, filter, noAggr)
+	set, err := promQuery.v3ioQuerier.SelectProm(name, functions, params.Step, filter, noAggr)
 	return &V3ioPromSeriesSet{s: set}, err
 }
 
 // LabelValues returns all potential values for a label name.
-func (q *V3ioPromQuerier) LabelValues(name string) ([]string, error) {
-	return q.q.LabelValues(name)
+func (promQuery *V3ioPromQuerier) LabelValues(name string) ([]string, error) {
+	return promQuery.v3ioQuerier.LabelValues(name)
 }
 
 // Close releases the resources of the Querier.
-func (q *V3ioPromQuerier) Close() error {
+func (promQuery *V3ioPromQuerier) Close() error {
 	return nil
 }
 
-func match2filter(oms []*labels.Matcher) (string, string, string) {
-	filter := []string{}
+func match2filter(oms []*labels.Matcher, logger logger.Logger) (string, string, string) {
+	var filter []string
 	agg := ""
 	name := ""
 
 	for _, matcher := range oms {
+		logger.Debug("Matcher: %+v", matcher)
 		if matcher.Name == aggregate.AggregateLabel {
 			agg = matcher.Value
 		} else if matcher.Name == "__name__" && matcher.Type == labels.MatchEqual {
@@ -109,7 +124,8 @@ func match2filter(oms []*labels.Matcher) (string, string, string) {
 			}
 		}
 	}
-	return name, strings.Join(filter, " and "), agg
+	filterExp := strings.Join(filter, " and ")
+	return name, filterExp, agg
 }
 
 type V3ioPromSeriesSet struct {

promtsdb/promtsdb_test.go

@@ -23,18 +23,22 @@ such restriction.
 package promtsdb
 
 import (
+	"github.com/nuclio/logger"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/stretchr/testify/suite"
+	"github.com/v3io/v3io-tsdb/pkg/config"
+	"github.com/v3io/v3io-tsdb/pkg/utils"
 	"testing"
 )
 
 type testPromTsdbSuite struct {
 	suite.Suite
+	logger logger.Logger
 }
 
 func (suite *testPromTsdbSuite) TestMatch2filterEmpty() {
 
-	name, filter, aggr := match2filter(nil)
+	name, filter, aggr := match2filter(nil, suite.logger)
 
 	suite.Require().Equal("", name)
 	suite.Require().Equal("", filter)
@@ -46,7 +50,7 @@ func (suite *testPromTsdbSuite) TestMatch2filterEqual() {
 	matchers := []*labels.Matcher{
 		{Type: labels.MatchEqual, Name: "field", Value: "literal"},
 	}
-	name, filter, aggr := match2filter(matchers)
+	name, filter, aggr := match2filter(matchers, suite.logger)
 
 	suite.Require().Equal("", name)
 	suite.Require().Equal("field=='literal'", filter)
@@ -59,7 +63,7 @@ func (suite *testPromTsdbSuite) TestMatch2filterMultiple() {
 		{Type: labels.MatchEqual, Name: "field1", Value: "literal1"},
 		{Type: labels.MatchNotEqual, Name: "field2", Value: "literal2"},
 	}
-	name, filter, aggr := match2filter(matchers)
+	name, filter, aggr := match2filter(matchers, suite.logger)
 
 	suite.Require().Equal("", name)
 	suite.Require().Equal("field1=='literal1' and field2!='literal2'", filter)
@@ -72,7 +76,7 @@ func (suite *testPromTsdbSuite) TestMatch2filterMultipleWithName() {
 		{Type: labels.MatchEqual, Name: "__name__", Value: "literal1"},
 		{Type: labels.MatchNotEqual, Name: "field2", Value: "literal2"},
 	}
-	name, filter, aggr := match2filter(matchers)
+	name, filter, aggr := match2filter(matchers, suite.logger)
 
 	suite.Require().Equal("literal1", name)
 	suite.Require().Equal("field2!='literal2'", filter)
@@ -84,7 +88,7 @@ func (suite *testPromTsdbSuite) TestMatch2filterRegex() {
 	matchers := []*labels.Matcher{
 		{Type: labels.MatchRegexp, Name: "field", Value: ".*"},
 	}
-	name, filter, aggr := match2filter(matchers)
+	name, filter, aggr := match2filter(matchers, suite.logger)
 
 	suite.Require().Equal("", name)
 	suite.Require().Equal(`regexp_instr(field,'.*') == 0`, filter)
@@ -97,13 +101,20 @@ func (suite *testPromTsdbSuite) TestMatch2filterRegexMultiple() {
 		{Type: labels.MatchRegexp, Name: "field1", Value: ".*"},
 		{Type: labels.MatchNotRegexp, Name: "field2", Value: "..."},
 	}
-	name, filter, aggr := match2filter(matchers)
+	name, filter, aggr := match2filter(matchers, suite.logger)
 
 	suite.Require().Equal("", name)
 	suite.Require().Equal(`regexp_instr(field1,'.*') == 0 and regexp_instr(field2,'...') != 0`, filter)
 	suite.Require().Equal("", aggr)
 }
 
 func TestPromTsdbSuite(t *testing.T) {
-	suite.Run(t, new(testPromTsdbSuite))
+	log, err := utils.NewLogger(config.DefaultLogLevel)
+	if err != nil {
+		t.Fatalf("Unable to initialize logger. Error: %v", err)
+	}
+
+	testSuit := new(testPromTsdbSuite)
+	testSuit.logger = log
+	suite.Run(t, testSuit)
 }