Merge pull request #78 from v3io/development

dev->master

Author: Gal Topper

Diff for: pkg/aggregate/iterator.go

@@ -45,11 +45,11 @@ func NewAggregateSeries(functions, col string, buckets int, interval, rollupTime
 
 	split := strings.Split(functions, ",")
 	var aggrMask AggrType
-	aggrList := []AggrType{}
+	var aggrList []AggrType
 	for _, s := range split {
 		aggr, ok := aggrTypeString[s]
 		if !ok {
-			return nil, fmt.Errorf("Invalid aggragator type %s", s)
+			return nil, fmt.Errorf("invalid aggragator type %s", s)
 		}
 		aggrMask = aggrMask | aggr
 		aggrList = append(aggrList, aggr)
@@ -87,7 +87,7 @@ func (as *AggregateSeries) toAttrName(aggr AggrType) string {
 }
 
 func (as *AggregateSeries) GetAttrNames() []string {
-	names := []string{}
+	var names []string
 
 	for _, aggr := range rawAggregators {
 		if aggr&as.aggrMask != 0 {
@@ -115,7 +115,7 @@ func (as *AggregateSeries) NewSetFromAttrs(
 		if aggr&as.aggrMask != 0 {
 			attrBlob, ok := (*attrs)[as.toAttrName(aggr)]
 			if !ok {
-				return nil, fmt.Errorf("Aggregation Attribute %s was not found", as.toAttrName(aggr))
+				return nil, fmt.Errorf("aggregation attribute %s was not found", as.toAttrName(aggr))
 			}
 			aggrArrays[aggr] = utils.AsInt64Array(attrBlob.([]byte))
 			dataArrays[aggr] = make([]float64, length, length)
@@ -124,7 +124,6 @@ func (as *AggregateSeries) NewSetFromAttrs(
 
 	aggrSet := AggregateSet{length: length, interval: as.interval, overlapWin: as.overlapWindows}
 	aggrSet.dataArrays = dataArrays
-	aggrSet.validCells = make([]bool, length, length)
 
 	arrayIndex := start
 	i := 0
@@ -138,7 +137,6 @@ func (as *AggregateSeries) NewSetFromAttrs(
 			for aggr, array := range aggrArrays {
 				aggrSet.mergeArrayCell(aggr, cellIndex, array[arrayIndex])
 			}
-			aggrSet.setValid(i)
 		} else {
 
 			// overlapping time windows (last 1hr, 6hr, ..)
@@ -149,7 +147,6 @@ func (as *AggregateSeries) NewSetFromAttrs(
 						for aggr, array := range aggrArrays {
 							aggrSet.mergeArrayCell(aggr, i, array[arrayIndex])
 						}
-						aggrSet.setValid(i)
 					}
 				}
 			}
@@ -180,13 +177,11 @@ func (as *AggregateSeries) NewSetFromChunks(length int) *AggregateSet {
 	}
 
 	newAggregateSet.dataArrays = dataArrays
-	newAggregateSet.validCells = make([]bool, length, length)
 	return &newAggregateSet
 }
 
 type AggregateSet struct {
 	dataArrays map[AggrType][]float64
-	validCells []bool
 	length     int
 	maxCell    int
 	baseTime   int64
@@ -198,27 +193,20 @@ func (as *AggregateSet) GetMaxCell() int {
 	return as.maxCell
 }
 
-func (as *AggregateSet) setValid(cell int) {
-	if cell < as.length {
-		as.validCells[cell] = true
-	}
-}
-
 // append the value to a cell in all relevant aggregation arrays
 func (as *AggregateSet) AppendAllCells(cell int, val float64) {
 
-	if cell >= as.length {
+	if !isValidCell(cell, as) {
 		return
 	}
 
 	if cell > as.maxCell {
 		as.maxCell = cell
 	}
 
-	for aggr, _ := range as.dataArrays {
+	for aggr := range as.dataArrays {
 		as.updateCell(aggr, cell, val)
 	}
-	as.validCells[cell] = true
 }
 
 // append/merge (v3io) aggregation values into aggregation per requested interval/step
@@ -241,9 +229,22 @@ func (as *AggregateSet) mergeArrayCell(aggr AggrType, cell int, val uint64) {
 	}
 }
 
+func isValidValue(v float64) bool {
+	return !(math.IsNaN(v) || math.IsInf(v, 1) || math.IsInf(v, -1))
+}
+
+func isValidCell(cellIndex int, aSet *AggregateSet) bool {
+	return cellIndex >= 0 &&
+		cellIndex < aSet.length
+}
+
 // function specific aggregation
 func (as *AggregateSet) updateCell(aggr AggrType, cell int, val float64) {
 
+	if !isValidCell(cell, as) {
+		return
+	}
+
 	switch aggr {
 	case aggrTypeCount:
 		as.dataArrays[aggr][cell] += 1
@@ -252,11 +253,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] || !as.validCells[cell] {
+		if val < as.dataArrays[aggr][cell] {
 			as.dataArrays[aggr][cell] = val
 		}
 	case aggrTypeMax:
-		if val > as.dataArrays[aggr][cell] || !as.validCells[cell] {
+		if val > as.dataArrays[aggr][cell] {
 			as.dataArrays[aggr][cell] = val
 		}
 	case aggrTypeLast:
@@ -267,13 +268,24 @@ func (as *AggregateSet) updateCell(aggr AggrType, cell int, val float64) {
 // return the value per aggregate or complex function
 func (as *AggregateSet) GetCellValue(aggr AggrType, cell int) (float64, bool) {
 
-	if cell > as.maxCell || cell >= as.length || !as.validCells[cell] { // TODO: should >Len return NaN or Zero ?
+	if !isValidCell(cell, as) {
+		return math.NaN(), false
+	}
+
+	dependsOnSumAndCount := aggr == aggrTypeStddev || aggr == aggrTypeStdvar || aggr == aggrTypeAvg
+	dependsOnSqr := aggr == aggrTypeStddev || aggr == aggrTypeStdvar
+	dependsOnLast := aggr == aggrTypeLast || aggr == aggrTypeRate
+
+	// return undefined result one dependant fields is missing
+	if (dependsOnSumAndCount && !(isValidValue(as.dataArrays[aggrTypeSum][cell]) && isValidValue(as.dataArrays[aggrTypeCount][cell]))) ||
+		(dependsOnSqr && !isValidValue(as.dataArrays[aggrTypeSqr][cell])) ||
+		(dependsOnLast && !isValidValue(as.dataArrays[aggrTypeLast][cell])) {
 		return math.NaN(), false
 	}
 
 	// if no samples in this bucket the result is undefined
 	var cnt float64
-	if aggr == aggrTypeAvg || aggr == aggrTypeStddev || aggr == aggrTypeStdvar {
+	if dependsOnSumAndCount {
 		cnt = as.dataArrays[aggrTypeCount][cell]
 		if cnt == 0 {
 			return math.NaN(), false
@@ -295,13 +307,13 @@ func (as *AggregateSet) GetCellValue(aggr AggrType, cell int) (float64, bool) {
 		if cell == 0 {
 			return math.NaN(), false
 		}
+		// TODO: need to clarify the meaning of this type of aggregation. IMHO, rate has meaning for monotonic counters only
 		last := as.dataArrays[aggrTypeLast][cell-1]
 		this := as.dataArrays[aggrTypeLast][cell]
-		return (this - last) / float64(as.interval/1000), true // clac rate per sec
+		return (this - last) / float64(as.interval/1000), true // rate per sec
 	default:
 		return as.dataArrays[aggr][cell], true
 	}
-
 }
 
 // get the time per aggregate cell
@@ -317,7 +329,7 @@ func (as *AggregateSet) GetCellTime(base int64, index int) int64 {
 
 func (as *AggregateSet) Clear() {
 	as.maxCell = 0
-	for aggr, _ := range as.dataArrays {
+	for aggr := range as.dataArrays {
 		as.dataArrays[aggr] = as.dataArrays[aggr][:0]
 	}
 }

Diff for: pkg/appender/ingest.go

@@ -23,6 +23,7 @@ package appender
 import (
 	"github.com/pkg/errors"
 	"github.com/v3io/v3io-go-http"
+	"net/http"
 	"reflect"
 	"time"
 )
@@ -274,7 +275,7 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 		} else {
 			// Metrics with too many update errors go into Error state
 			metric.retryCount++
-			if metric.retryCount == maxRetriesOnWrite {
+			if e, hasStatusCode := resp.Error.(v3io.ErrorWithStatusCode); metric.retryCount == maxRetriesOnWrite || hasStatusCode && e.StatusCode() != http.StatusServiceUnavailable {
 				mc.logger.ErrorWith("Metric error, exceeded retry", "metric", metric.Lset)
 				metric.setError(errors.Wrap(resp.Error, "chunk update failed after few retries"))
 				return false

Diff for: pkg/config/config.go

@@ -53,6 +53,14 @@ type V3ioConfig struct {
 	DefaultTimeout int `json:"timeout,omitempty"`
 	// The size of batch to use during ingestion
 	BatchSize int `json:"batchSize,omitempty"`
+	// Size of sample in bytes for worst an best compression scenarios
+	MinimumSampleSize int `json:"minimumSampleSize,omitempty"`
+	MaximumSampleSize int `json:"maximumSampleSize,omitempty"`
+	// Max size of a partition object
+	MaximumPartitionSize int `json:"maximumPartitionSize,omitempty"`
+	// Size of chunk in bytes for worst an best compression scenarios
+	MinimumChunkSize int `json:"minimumChunkSize,omitempty"`
+	MaximumChunkSize int `json:"maximumChunkSize,omitempty"`
 }
 
 type Rollup struct {

Diff for: pkg/partmgr/partmgr.go

@@ -230,7 +230,7 @@ func (p *DBPartition) GetStartTime() int64 {
 }
 
 func (p *DBPartition) GetEndTime() int64 {
-	return p.startTime + p.partitionInterval
+	return p.startTime + p.partitionInterval - 1
 }
 
 // return path to metrics table
@@ -272,7 +272,10 @@ func (p *DBPartition) Time2Bucket(t int64) int {
 	if p.rollupTime == 0 {
 		return 0
 	}
-	return int((t-p.startTime)/p.rollupTime) % p.rollupBuckets
+	if t > p.GetEndTime() {
+		return p.rollupBuckets - 1
+	}
+	return int((t - p.startTime) / p.rollupTime)
 }
 
 // get nearest chunk start

Diff for: pkg/tsdb/tsdbtest/testutils/schema.go

@@ -23,8 +23,8 @@ func CreateSchema(t testing.TB, agg string) config.Schema {
 		Version:             0,
 		RollupLayers:        []config.Rollup{defaultRollup},
 		ShardingBuckets:     8,
-		PartitionerInterval: "2d",
-		ChunckerInterval:    "1h",
+		PartitionerInterval: "340h",
+		ChunckerInterval:    "10h",
 	}
 
 	fields, err := aggregate.SchemaFieldFromString(rollups, "v")

Diff for: pkg/tsdb/v3iotsdb_integration_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/v3io/v3io-tsdb/pkg/config"
 	. "github.com/v3io/v3io-tsdb/pkg/tsdb"
 	"github.com/v3io/v3io-tsdb/pkg/tsdb/tsdbtest"
+	"github.com/v3io/v3io-tsdb/pkg/tsdb/tsdbtest/testutils"
 	"github.com/v3io/v3io-tsdb/pkg/utils"
 	"path/filepath"
 	"sort"
@@ -408,9 +409,9 @@ func TestCreateTSDB(t *testing.T) {
 		conf         config.Schema
 		ignoreReason string
 	}{
-		{desc: "Should create TSDB with standard configuration", conf: tsdbtest.CreateSchema(t, "sum,count")},
+		{desc: "Should create TSDB with standard configuration", conf: testutils.CreateSchema(t, "sum,count")},
 
-		{desc: "Should create TSDB with wildcard aggregations", conf: tsdbtest.CreateSchema(t, "*")},
+		{desc: "Should create TSDB with wildcard aggregations", conf: testutils.CreateSchema(t, "*")},
 	}
 
 	for _, test := range testCases {
@@ -442,7 +443,7 @@ func TestDeleteTSDB(t *testing.T) {
 		t.Fatalf("Failed to load test configuration. reason: %s", err)
 	}
 
-	schema := tsdbtest.CreateSchema(t, "count,sum")
+	schema := testutils.CreateSchema(t, "count,sum")
 	v3ioConfig.Path = t.Name()
 	if err := CreateTSDB(v3ioConfig, &schema); err != nil {
 		t.Fatalf("Failed to create TSDB. reason: %s", err)