Merge pull request #184 from v3io/development

Development --> Master (0.0.14)

Author: gshatz73

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
 			}
@@ -251,7 +260,7 @@ type aggrSeriesIterator struct {
 // Advance an iterator to the specified time (t)
 func (s *aggrSeriesIterator) Seek(t int64) bool {
 	if t <= s.set.baseTime {
-		s.index = 0
+		s.index = s.getNextValidCell(-1)
 		return true
 	}
 
@@ -266,14 +275,16 @@ func (s *aggrSeriesIterator) Seek(t int64) bool {
 // Advance an iterator to the next time interval/bucket
 func (s *aggrSeriesIterator) Next() bool {
 	// Advance the index to the next non-empty cell
-	var nextIndex int
-	for nextIndex = s.index + 1; nextIndex <= s.aggrSet.GetMaxCell() && !s.aggrSet.DoesCellHaveData(nextIndex); nextIndex++ {
-	}
-
-	s.index = nextIndex
+	s.index = s.getNextValidCell(s.index)
 	return s.index <= s.aggrSet.GetMaxCell()
 }
 
+func (s *aggrSeriesIterator) getNextValidCell(from int) (nextIndex int) {
+	for nextIndex = from + 1; nextIndex <= s.aggrSet.GetMaxCell() && !s.aggrSet.DoesCellHaveData(nextIndex); nextIndex++ {
+	}
+	return
+}
+
 // Return the time and value at the current bucket
 func (s *aggrSeriesIterator) At() (t int64, v float64) {
 	val, _ := s.aggrSet.GetCellValue(s.aggrType, s.index)

pkg/tsdb/v3iotsdb.go

@@ -151,7 +151,6 @@ func (a *V3ioAdapter) connect() error {
 		return errors.Wrapf(err, "Failed to initialize the TSDB partition manager at: %s", fullpath)
 	}
 
-	a.logger.Info("Starting the V3IO TSDB client for the TSDB instance at '%s'", fullpath)
 	a.logger.Debug("Running with the following TSDB configuration: %+v\n", a.cfg)
 
 	return nil

pkg/tsdb/v3iotsdb_integration_test.go

@@ -471,6 +471,81 @@ func testQueryDataOverlappingWindowCase(test *testing.T, v3ioConfig *config.V3io
 	}
 }
 
+// Calling Seek instead of next for the first time while iterating over data (TSDB-43)
+func TestIgnoreNaNWhenSeekingAggSeries(t *testing.T) {
+	v3ioConfig, err := tsdbtest.LoadV3ioConfig()
+	if err != nil {
+		t.Fatalf("unable to load configuration. Error: %v", err)
+	}
+	metricsName := "cpu"
+	baseTime := int64(1532940510000)
+	userLabels := utils.LabelsFromStrings("os", "linux", "iguaz", "yesplease")
+	data := []tsdbtest.DataPoint{{Time: baseTime, Value: 300.3},
+		{Time: baseTime + minuteInMillis, Value: 300.3},
+		{Time: baseTime + 2*minuteInMillis, Value: 100.4},
+		{Time: baseTime + 5*minuteInMillis, Value: 200.0}}
+	from := int64(baseTime - 60*minuteInMillis)
+	to := int64(baseTime + 6*minuteInMillis)
+	step := int64(2 * minuteInMillis)
+	agg := "avg"
+	expected := map[string][]tsdbtest.DataPoint{
+		"avg": {{baseTime, 300.3},
+			{baseTime + step, 100.4},
+			{baseTime + 2*step, 200}}}
+
+	adapter, teardown := tsdbtest.SetUpWithData(t, v3ioConfig, metricsName, data, userLabels)
+	defer teardown()
+
+	qry, err := adapter.Querier(nil, from, to)
+	if err != nil {
+		t.Fatalf("Failed to create Querier. reason: %v", err)
+	}
+
+	set, err := qry.Select(metricsName, agg, step, "")
+	if err != nil {
+		t.Fatalf("Failed to run Select. reason: %v", err)
+	}
+
+	var counter int
+	for counter = 0; set.Next(); counter++ {
+		if set.Err() != nil {
+			t.Fatalf("Failed to query metric. reason: %v", set.Err())
+		}
+
+		series := set.At()
+		agg := series.Labels().Get(aggregate.AggregateLabel)
+		iter := series.Iterator()
+		if iter.Err() != nil {
+			t.Fatalf("Failed to query data series. reason: %v", iter.Err())
+		}
+		if !iter.Seek(0) {
+			t.Fatal("Seek time returned false, iterator error:", iter.Err())
+		}
+		var actual []tsdbtest.DataPoint
+		t0, v0 := iter.At()
+		if iter.Err() != nil {
+			t.Fatal("error iterating over series", iter.Err())
+		}
+		actual = append(actual, tsdbtest.DataPoint{Time: t0, Value: v0})
+		for iter.Next() {
+			t1, v1 := iter.At()
+
+			if iter.Err() != nil {
+				t.Fatal("error iterating over series", iter.Err())
+			}
+			actual = append(actual, tsdbtest.DataPoint{Time: t1, Value: v1})
+		}
+		assert.ElementsMatch(t, expected[agg], actual)
+	}
+
+	if set.Err() != nil {
+		t.Fatalf("Failed to query metric. reason: %v", set.Err())
+	}
+	if counter == 0 && len(expected) > 0 {
+		t.Fatalf("No data was received")
+	}
+}
+
 func TestCreateTSDB(t *testing.T) {
 	v3ioConfig, err := tsdbtest.LoadV3ioConfig()
 	if err != nil {

pkg/tsdbctl/info.go

@@ -53,9 +53,9 @@ func newInfoCommandeer(rootCommandeer *RootCommandeer) *infoCommandeer {
 	}
 
 	cmd.Flags().BoolVarP(&commandeer.getNames, "names", "n", false,
-		"Display the names of the metrics contained in the TSDB.")
+		"Display the metric names in the TSDB.")
 	cmd.Flags().BoolVarP(&commandeer.getCount, "performance", "m", false,
-		"Display a count of the number of metric objects contained in the TSDB.")
+		"Display the number of metric items in the TSDB.")
 
 	commandeer.cmd = cmd
 

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)
 }