Merge pull request #115 from v3io/development

dev -> master (for v0.0.10)

Author: Gal Topper

Diff for: pkg/aggregate/iterator.go

@@ -69,8 +69,9 @@ func (as *AggregateSeries) CanAggregate(partitionAggr AggrType) bool {
 	// keep only real aggregators
 	aggrMask := 0x7f & as.aggrMask
 	// make sure the DB has all the aggregators we need (on bits in the mask)
-	// and that the aggregator resolution is greater/eq to requested interval
-	return ((aggrMask & partitionAggr) == aggrMask) && as.interval >= as.rollupTime
+	// and that the requested interval is greater/eq to aggregator resolution and is an even divisor
+	return ((aggrMask & partitionAggr) == aggrMask) &&
+		as.interval >= as.rollupTime && (as.interval%as.rollupTime == 0)
 }
 
 func (as *AggregateSeries) GetAggrMask() AggrType {
@@ -176,6 +177,11 @@ func (as *AggregateSeries) NewSetFromChunks(length int) *AggregateSet {
 	for _, aggr := range rawAggregators {
 		if aggr&as.aggrMask != 0 {
 			dataArrays[aggr] = make([]float64, length, length) // TODO: len/capacity & reuse (pool)
+			if aggr == aggrTypeMax || aggr == aggrTypeMin || aggr == aggrTypeLast {
+				for i := 0; i < length; i++ {
+					dataArrays[aggr][i] = math.NaN()
+				}
+			}
 		}
 	}
 
@@ -256,11 +262,11 @@ func (as *AggregateSet) updateCell(aggr AggrType, cell int, val float64) {
 	case aggrTypeSqr:
 		as.dataArrays[aggr][cell] += val * val
 	case aggrTypeMin:
-		if val < as.dataArrays[aggr][cell] {
+		if math.IsNaN(as.dataArrays[aggr][cell]) || val < as.dataArrays[aggr][cell] {
 			as.dataArrays[aggr][cell] = val
 		}
 	case aggrTypeMax:
-		if val > as.dataArrays[aggr][cell] {
+		if math.IsNaN(as.dataArrays[aggr][cell]) || val > as.dataArrays[aggr][cell] {
 			as.dataArrays[aggr][cell] = val
 		}
 	case aggrTypeLast:

Diff for: pkg/partmgr/partmgr.go

@@ -369,21 +369,11 @@ func (p *DBPartition) InRange(t int64) bool {
 }
 
 // return the mint and maxt for this partition, may need maxt for cyclic partition
-func (p *DBPartition) GetPartitionRange(maxt int64) (int64, int64) {
+func (p *DBPartition) GetPartitionRange() (int64, int64) {
 	// start p.days ago, rounded to next hour
 	return p.startTime, p.startTime + p.partitionInterval
 }
 
-// return the valid minimum time in a cyclic partition based on max time
-func (p *DBPartition) CyclicMinTime(mint, maxt int64) int64 {
-	// start p.days ago, rounded to next hour
-	newMin, _ := p.GetPartitionRange(maxt)
-	if mint > newMin {
-		return mint
-	}
-	return newMin
-}
-
 // Attribute name of a chunk
 func (p *DBPartition) ChunkID2Attr(col string, id int) string {
 	return fmt.Sprintf("_%s%d", col, id)

Diff for: pkg/querier/querier.go

@@ -106,14 +106,20 @@ func (q *V3ioQuerier) selectQry(name, functions string, step int64, windows []in
 func (q *V3ioQuerier) queryNumericPartition(
 	partition *partmgr.DBPartition, name, functions string, step int64, windows []int, filter string) (SeriesSet, error) {
 
-	mint, maxt := partition.GetPartitionRange(q.maxt)
-	if q.mint > mint {
-		mint = q.mint
-	}
+	mint, maxt := partition.GetPartitionRange()
+
 	if q.maxt < maxt {
 		maxt = q.maxt
 	}
 
+	if q.mint > mint {
+		mint = q.mint
+		if step != 0 && step < (maxt-mint) {
+			// temporary Aggregation fix: if mint is not aligned with steps, move it to the next step tick
+			mint += (maxt - mint) % step
+		}
+	}
+
 	newSet := &V3ioSeriesSet{mint: mint, maxt: maxt, partition: partition, logger: q.logger}
 
 	// if there are aggregations to be made

Diff for: pkg/querier/seriesset.go

@@ -105,31 +105,30 @@ func (s *V3ioSeriesSet) Next() bool {
 
 		s.nullSeries = false
 
-		if s.aggrSeries.CanAggregate(s.partition.AggrType()) && s.maxt-s.mint > s.interval {
+		if s.aggrSeries.CanAggregate(s.partition.AggrType()) && s.maxt-s.mint >= s.interval {
 
 			// create series from aggregation arrays (in DB) if the partition stored the desired aggregates
 			maxtUpdate := s.maxt
 			maxTime := s.iter.GetField("_maxtime")
 			if maxTime != nil && int64(maxTime.(int)) < s.maxt {
 				maxtUpdate = int64(maxTime.(int))
 			}
-			mint := s.partition.CyclicMinTime(s.mint, maxtUpdate)
 
-			start := s.partition.Time2Bucket(mint)
+			start := s.partition.Time2Bucket(s.mint)
 			end := s.partition.Time2Bucket(s.maxt+s.interval) + 1
 
-			// len of the returned array, cropped at the end in case of cyclic overlap
-			length := int((maxtUpdate-mint)/s.interval) + 2
+			// len of the returned array, time-range / interval + 2
+			length := int((maxtUpdate-s.mint)/s.interval) + 2
 
 			if s.overlapWin != nil {
-				s.baseTime = s.maxt //- int64(s.overlapWin[0]) * s.interval
+				s.baseTime = s.maxt
 			} else {
-				s.baseTime = mint
+				s.baseTime = s.mint
 			}
 
 			if length > 0 {
 				attrs := s.iter.GetFields()
-				aggrSet, err := s.aggrSeries.NewSetFromAttrs(length, start, end, mint, s.maxt, &attrs)
+				aggrSet, err := s.aggrSeries.NewSetFromAttrs(length, start, end, s.mint, s.maxt, &attrs)
 				if err != nil {
 					s.err = err
 					return false
@@ -145,12 +144,8 @@ func (s *V3ioSeriesSet) Next() bool {
 			// create series from raw chunks
 			s.currSeries = NewSeries(s)
 
-			// the number of cells is equal to divisor of (maxt-mint) and interval. if there's a
-			// remainder or if there are no cells (e.g. diff is smaller than interval), add a cell
-			numCells := (s.maxt - s.mint) / s.interval
-			if ((s.maxt-s.mint)%s.interval) != 0 || numCells == 0 {
-				numCells++
-			}
+			// the number of cells is equal to divisor of (maxt-mint) and interval.
+			numCells := (s.maxt-s.mint)/s.interval + 1
 
 			s.aggrSet = s.aggrSeries.NewSetFromChunks(int(numCells))
 			if s.overlapWin != nil {
@@ -171,12 +166,8 @@ func (s *V3ioSeriesSet) chunks2IntervalAggregates() {
 
 	iter := s.currSeries.Iterator()
 	if iter.Next() {
-		t0, _ := iter.At()
 
-		// the base time should be the first sample we receive. initially, baseTime is zero
-		if s.baseTime == 0 {
-			s.baseTime = t0
-		}
+		s.baseTime = s.mint
 
 		for {
 			t, v := iter.At()

Diff for: pkg/tsdb/v3iotsdb_integration_test.go

@@ -199,6 +199,17 @@ func TestQueryData(t *testing.T) {
 			aggregators: "sum",
 			expected:    map[string][]tsdbtest.DataPoint{"sum": {{Time: 1532940510, Value: 701.0}}}},
 
+		{desc: "Should ingest and query an aggregator EXTRA", metricName: "cpu",
+			labels: utils.FromStrings("os", "linux", "iguaz", "yesplease"),
+			data: []tsdbtest.DataPoint{{Time: 1532940510, Value: 300.3},
+				{Time: 1532940510 + 60, Value: 300.3},
+				{Time: 1532940510 + 2*60, Value: 100.4},
+				{Time: 1532940510 + 2*60, Value: 200.0}},
+			from:        1532940510,
+			to:          1532940510 + 6*60,
+			aggregators: "sum",
+			expected:    map[string][]tsdbtest.DataPoint{"sum": {{Time: 1532940510, Value: 901.0}}}},
+
 		{desc: "Should ingest and query multiple aggregators", metricName: "cpu",
 			labels: utils.FromStrings("os", "linux", "iguaz", "yesplease"),
 			data: []tsdbtest.DataPoint{{Time: 1532940510, Value: 300.3},
@@ -252,7 +263,8 @@ func testQueryDataCase(test *testing.T, v3ioConfig *config.V3ioConfig,
 		}
 	}
 
-	set, err := qry.Select(metricsName, aggregator, 1000, filter)
+	step := int64(5 * 60 * 1000) // 5 minutes
+	set, err := qry.Select(metricsName, aggregator, step, filter)
 	if err != nil {
 		test.Fatalf("Failed to run Select. reason: %v", err)
 	}

Diff for: pkg/tsdbctl/create.go

@@ -27,13 +27,15 @@ import (
 	"github.com/v3io/v3io-tsdb/pkg/aggregate"
 	"github.com/v3io/v3io-tsdb/pkg/config"
 	"github.com/v3io/v3io-tsdb/pkg/tsdb"
+	"github.com/v3io/v3io-tsdb/pkg/utils"
 	"strconv"
+	"time"
 )
 
 const (
 	schemaVersion               = 0
 	defaultStorageClass         = "local"
-	defaultIngestionRate        = "1/s"
+	defaultIngestionRate        = ""
 	defaultRollupInterval       = "1h"
 	defaultShardingBuckets      = 8
 	defaultSampleRetentionHours = 0
@@ -86,7 +88,7 @@ func (cc *createCommandeer) create() error {
 		return err
 	}
 
-	if err := cc.validateFormat(cc.rollupInterval); err != nil {
+	if err := cc.validateRollupInterval(); err != nil {
 		return errors.Wrap(err, "failed to parse rollup interval")
 	}
 
@@ -95,6 +97,9 @@ func (cc *createCommandeer) create() error {
 		return errors.Wrap(err, "failed to parse default rollups")
 	}
 
+	if cc.sampleRate == "" {
+		return errors.New(`sample rate not provided! Please provide sample rate with --rate flag in the format of [0-9]+/[hms]. Example: 12/m`)
+	}
 	rateInHours, err := rateToHours(cc.sampleRate)
 	if err != nil {
 		return errors.Wrap(err, "failed to parse sample rate")
@@ -147,14 +152,15 @@ func (cc *createCommandeer) create() error {
 	return tsdb.CreateTSDB(cc.rootCommandeer.v3iocfg, &schema)
 }
 
-func (cc *createCommandeer) validateFormat(format string) error {
-	interval := format[0 : len(format)-1]
-	if _, err := strconv.Atoi(interval); err != nil {
-		return errors.New("format is incorrect, not a number")
+func (cc *createCommandeer) validateRollupInterval() error {
+	dayMillis := 24 * int64(time.Hour/time.Millisecond)
+	duration, err := utils.Str2duration(cc.rollupInterval)
+	if err != nil {
+		return err
 	}
-	unit := string(format[len(format)-1])
-	if !(unit == "m" || unit == "d" || unit == "h") {
-		return errors.New("format is incorrect, not part of m,d,h")
+
+	if dayMillis%duration != 0 && duration%dayMillis != 0 {
+		return errors.New("rollup interval should be a divisor or a dividend of 1 day. Example: 10m, 30m, 2h, etc.")
 	}
 	return nil
 }
@@ -176,13 +182,21 @@ func (cc *createCommandeer) calculatePartitionAndChunkInterval(rateInHours int)
 	}
 
 	actualCapacityOfChunk := chunkInterval * rateInHours * cc.rootCommandeer.v3iocfg.MaximumSampleSize
-	numberOfChunksInPartition := cc.rootCommandeer.v3iocfg.MaximumPartitionSize / actualCapacityOfChunk
+	numberOfChunksInPartition := 0
+
+	for (numberOfChunksInPartition+24)*actualCapacityOfChunk < cc.rootCommandeer.v3iocfg.MaximumPartitionSize {
+		numberOfChunksInPartition += 24
+	}
+	if numberOfChunksInPartition == 0 {
+		return "", "", errors.Errorf("given rate is too high, can not fit a partition in a day interval with the calculated chunk size %vh", chunkInterval)
+	}
+
 	partitionInterval := numberOfChunksInPartition * chunkInterval
 	return strconv.Itoa(chunkInterval) + "h", strconv.Itoa(partitionInterval) + "h", nil
 }
 
 func rateToHours(sampleRate string) (int, error) {
-	parsingError := errors.New(`not a valid rate. Accepted pattern: [0-9]+/[hms]. Examples: 12/m`)
+	parsingError := errors.New(`not a valid rate. Accepted pattern: [0-9]+/[hms]. Example: 12/m`)
 
 	if len(sampleRate) < 3 {
 		return 0, parsingError

Diff for: pkg/tsdbctl/create_test.go

@@ -16,6 +16,8 @@ func TestRateToHour(t *testing.T) {
 		{input: "1m", shouldFail: true},
 		{input: "1/t", shouldFail: true},
 		{input: "-431/t", shouldFail: true},
+		{input: "-1", shouldFail: true},
+		{input: "", shouldFail: true},
 	}
 
 	for _, testCase := range cases {