Merge pull request #98 from v3io/development

dev -> master

Author: Gal Topper

Diff for: pkg/aggregate/iterator.go

@@ -39,13 +39,10 @@ type AggregateSeries struct {
 
 func NewAggregateSeries(functions, col string, buckets int, interval, rollupTime int64, windows []int) (*AggregateSeries, error) {
 
-	if functions == "" || interval == 0 {
-		return nil, nil
-	}
-
 	split := strings.Split(functions, ",")
 	var aggrMask AggrType
 	var aggrList []AggrType
+
 	for _, s := range split {
 		aggr, ok := aggrTypeString[s]
 		if !ok {
@@ -56,8 +53,14 @@ func NewAggregateSeries(functions, col string, buckets int, interval, rollupTime
 	}
 
 	newAggregateSeries := AggregateSeries{
-		aggrMask: aggrMask, functions: aggrList, colName: col, buckets: buckets, rollupTime: rollupTime,
-		interval: interval, overlapWindows: windows}
+		aggrMask:       aggrMask,
+		functions:      aggrList,
+		colName:        col,
+		buckets:        buckets,
+		rollupTime:     rollupTime,
+		interval:       interval,
+		overlapWindows: windows,
+	}
 
 	return &newAggregateSeries, nil
 }
@@ -154,7 +157,7 @@ func (as *AggregateSeries) NewSetFromAttrs(
 		}
 
 		i++
-		arrayIndex = (arrayIndex + 1) % as.buckets
+		arrayIndex = (arrayIndex + 1) % (as.buckets + 1)
 	}
 
 	return &aggrSet, nil
@@ -321,9 +324,11 @@ func (as *AggregateSet) GetCellTime(base int64, index int) int64 {
 	if as.overlapWin == nil {
 		return base + int64(index)*as.interval
 	}
+
 	if index >= len(as.overlapWin) {
 		return base
 	}
+
 	return base - int64(as.overlapWin[index])*as.interval
 }
 

Diff for: pkg/appender/appender.go

@@ -251,11 +251,10 @@ func (mc *MetricsCache) WaitForCompletion(timeout time.Duration) (int, error) {
 	} else if timeout > 0 {
 		maxWaitTime = timeout
 	} else {
-		maxWaitTime = time.Duration(mc.cfg.DefaultTimeout) * time.Second
+		// if negative - use default value from configuration
+		maxWaitTime = time.Duration(mc.cfg.DefaultTimeoutInSeconds) * time.Second
 	}
 
-	//fmt.Printf("\nmaxWaitTime=%d\n", maxWaitTime)
-
 	select {
 	case res := <-waitChan:
 		return res, nil

Diff for: pkg/config/config.go

@@ -26,9 +26,16 @@ import (
 	"os"
 )
 
-const V3ioConfigEnvironmentVariable = "V3IO_CONF"
-const DefaultConfigurationFileName = "v3io.yaml"
-const SCHEMA_CONFIG = ".schema"
+const (
+	V3ioConfigEnvironmentVariable = "V3IO_CONF"
+	DefaultConfigurationFileName  = "v3io.yaml"
+	SCHEMA_CONFIG                 = ".schema"
+
+	defaultNumberOfIngestWorkers = 1
+	defaultNumberOfQueryWorkers  = 8
+	defaultBatchSize             = 64
+	defaultTimeoutInSeconds      = 24 * 60 * 60 // 24 hours
+)
 
 type V3ioConfig struct {
 	// V3IO Connection details: Url, Data container, relative path for this dataset, credentials
@@ -51,7 +58,7 @@ type V3ioConfig struct {
 	// Override last chunk (by default on restart it will append from the last point if possible)
 	OverrideOld bool `json:"overrideOld"`
 	// Default timeout duration in Seconds (if not set, 1 Hour timeout will be used )
-	DefaultTimeout int `json:"timeout,omitempty"`
+	DefaultTimeoutInSeconds 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
@@ -153,20 +160,24 @@ func LoadFromData(data []byte) (*V3ioConfig, error) {
 func InitDefaults(cfg *V3ioConfig) {
 	// Initialize default number of workers
 	if cfg.Workers == 0 {
-		cfg.Workers = 8
+		cfg.Workers = defaultNumberOfIngestWorkers
 	}
 
 	// init default number Query workers if not set to Min(8,Workers)
 	if cfg.QryWorkers == 0 {
-		if cfg.Workers < 8 {
+		if cfg.Workers < defaultNumberOfQueryWorkers {
 			cfg.QryWorkers = cfg.Workers
 		} else {
-			cfg.QryWorkers = 8
+			cfg.QryWorkers = defaultNumberOfQueryWorkers
 		}
 	}
 
 	// init default batch size
 	if cfg.BatchSize <= 0 {
-		cfg.BatchSize = 64
+		cfg.BatchSize = defaultBatchSize
+	}
+
+	if cfg.DefaultTimeoutInSeconds == 0 {
+		cfg.DefaultTimeoutInSeconds = int(defaultTimeoutInSeconds)
 	}
 }

Diff for: pkg/querier/querier.go

@@ -116,24 +116,34 @@ func (q *V3ioQuerier) queryNumericPartition(
 
 	newSet := &V3ioSeriesSet{mint: mint, maxt: maxt, partition: partition, logger: q.logger}
 
-	if functions != "" && step == 0 && partition.RollupTime() != 0 {
-		step = partition.RollupTime()
-	}
+	// if there are aggregations to be made
+	if functions != "" {
 
-	newAggrSeries, err := aggregate.NewAggregateSeries(
-		functions, "v", partition.AggrBuckets(), step, partition.RollupTime(), windows)
-	if err != nil {
-		return nil, err
-	}
+		// if step isn't passed (e.g. when using the console) - the step is the difference between max
+		// and min times (e.g. 5 minutes)
+		if step == 0 {
+			step = maxt - mint
+		}
+
+		newAggrSeries, err := aggregate.NewAggregateSeries(functions,
+			"v",
+			partition.AggrBuckets(),
+			step,
+			partition.RollupTime(),
+			windows)
+
+		if err != nil {
+			return nil, err
+		}
 
-	if newAggrSeries != nil && step != 0 {
 		newSet.aggrSeries = newAggrSeries
 		newSet.interval = step
 		newSet.aggrIdx = newAggrSeries.NumFunctions() - 1
 		newSet.overlapWin = windows
 	}
 
-	err = newSet.getItems(partition, name, filter, q.container, q.cfg.QryWorkers)
+	err := newSet.getItems(partition, name, filter, q.container, q.cfg.QryWorkers)
+
 	return newSet, err
 }
 

Diff for: pkg/querier/series.go

@@ -213,16 +213,28 @@ func NewAggrSeries(set *V3ioSeriesSet, aggr aggregate.AggrType) *V3ioSeries {
 	newSeries := V3ioSeries{set: set}
 	lset := append(initLabels(set), utils.Label{Name: "Aggregator", Value: aggr.String()})
 	newSeries.lset = lset
+
 	if set.nullSeries {
 		newSeries.iter = &nullSeriesIterator{}
 	} else {
-		newSeries.iter = &aggrSeriesIterator{set: set, aggrType: aggr, index: -1}
+
+		// `set`, the thing this iterator "iterates" over is stateful - it holds a "current" set and aggrSet.
+		// this means we need to copy all the stateful things we need into the iterator (e.g. aggrSet) so that
+		// when it's evaluated, it'll hold the proper pointer
+		newSeries.iter = &aggrSeriesIterator{
+			set:      set,
+			aggrSet:  set.aggrSet,
+			aggrType: aggr,
+			index:    -1,
+		}
 	}
+
 	return &newSeries
 }
 
 type aggrSeriesIterator struct {
 	set      *V3ioSeriesSet
+	aggrSet  *aggregate.AggregateSet
 	aggrType aggregate.AggrType
 	index    int
 	err      error
@@ -235,7 +247,7 @@ func (s *aggrSeriesIterator) Seek(t int64) bool {
 		return true
 	}
 
-	if t > s.set.baseTime+int64(s.set.aggrSet.GetMaxCell())*s.set.interval {
+	if t > s.set.baseTime+int64(s.aggrSet.GetMaxCell())*s.set.interval {
 		return false
 	}
 
@@ -245,7 +257,7 @@ func (s *aggrSeriesIterator) Seek(t int64) bool {
 
 // advance to the next time interval/bucket
 func (s *aggrSeriesIterator) Next() bool {
-	if s.index >= s.set.aggrSet.GetMaxCell() {
+	if s.index >= s.aggrSet.GetMaxCell() {
 		return false
 	}
 
@@ -255,8 +267,8 @@ func (s *aggrSeriesIterator) Next() bool {
 
 // return the time & value at the current bucket
 func (s *aggrSeriesIterator) At() (t int64, v float64) {
-	val, _ := s.set.aggrSet.GetCellValue(s.aggrType, s.index)
-	return s.set.aggrSet.GetCellTime(s.set.baseTime, s.index), val
+	val, _ := s.aggrSet.GetCellValue(s.aggrType, s.index)
+	return s.aggrSet.GetCellTime(s.set.baseTime, s.index), val
 }
 
 func (s *aggrSeriesIterator) Err() error { return s.err }

Diff for: pkg/querier/seriesset.go

@@ -116,7 +116,7 @@ func (s *V3ioSeriesSet) Next() bool {
 			mint := s.partition.CyclicMinTime(s.mint, maxtUpdate)
 
 			start := s.partition.Time2Bucket(mint)
-			end := s.partition.Time2Bucket(s.maxt + s.interval)
+			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
@@ -144,7 +144,15 @@ func (s *V3ioSeriesSet) Next() bool {
 
 			// create series from raw chunks
 			s.currSeries = NewSeries(s)
-			s.aggrSet = s.aggrSeries.NewSetFromChunks(int((s.maxt-s.mint)/s.interval) + 1)
+
+			// 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++
+			}
+
+			s.aggrSet = s.aggrSeries.NewSetFromChunks(int(numCells))
 			if s.overlapWin != nil {
 				s.chunks2WindowedAggregates()
 			} else {
@@ -164,7 +172,11 @@ func (s *V3ioSeriesSet) chunks2IntervalAggregates() {
 	iter := s.currSeries.Iterator()
 	if iter.Next() {
 		t0, _ := iter.At()
-		s.baseTime = (t0 / s.interval) * s.interval
+
+		// the base time should be the first sample we receive. initially, baseTime is zero
+		if s.baseTime == 0 {
+			s.baseTime = t0
+		}
 
 		for {
 			t, v := iter.At()

Diff for: pkg/tsdb/testdata/v3io.yaml.template

@@ -15,7 +15,5 @@ verbose: "warn"
 # Default timeout 10 seconds
 timeout: 10
 
-workers: 8
-
 username: <user name>
 password: <password>

Diff for: pkg/tsdb/v3iotsdb_integration_test.go

@@ -196,7 +196,7 @@ func TestQueryData(t *testing.T) {
 				{Time: 1532940510 + 10, Value: 100.4}},
 			from: 1532940510, to: 1532940510 + 11,
 			aggregators: "sum",
-			expected:    map[string][]tsdbtest.DataPoint{"sum": {{Time: 1532940000, Value: 701.0}}}},
+			expected:    map[string][]tsdbtest.DataPoint{"sum": {{Time: 1532940510, Value: 701.0}}}},
 
 		{desc: "Should ingest and query multiple aggregators", metricName: "cpu",
 			labels: utils.FromStrings("os", "linux", "iguaz", "yesplease"),
@@ -205,8 +205,8 @@ func TestQueryData(t *testing.T) {
 				{Time: 1532940510 + 10, Value: 100.4}},
 			from: 1532940510, to: 1532940510 + 11,
 			aggregators: "sum,count",
-			expected: map[string][]tsdbtest.DataPoint{"sum": {{Time: 1532940000, Value: 701.0}},
-				"count": {{Time: 1532940000, Value: 3}}}},
+			expected: map[string][]tsdbtest.DataPoint{"sum": {{Time: 1532940510, Value: 701.0}},
+				"count": {{Time: 1532940510, Value: 3}}}},
 
 		{desc: "Should ingest and query with illegal time (switch from and to)", metricName: "cpu",
 			labels: utils.FromStrings("os", "linux", "iguaz", "yesplease"),

Diff for: pkg/tsdbctl/check.go

@@ -33,8 +33,7 @@ import (
 type checkCommandeer struct {
 	cmd            *cobra.Command
 	rootCommandeer *RootCommandeer
-	name           string
-	lset           string
+	objPath        string
 	attrs          []string
 }
 
@@ -50,14 +49,10 @@ func newCheckCommandeer(rootCommandeer *RootCommandeer) *checkCommandeer {
 		RunE: func(cmd *cobra.Command, args []string) error {
 
 			if len(args) == 0 {
-				return errors.New("add require metric name and/or labels")
+				return errors.New("check requires an object path")
 			}
 
-			commandeer.name = args[0]
-
-			if len(args) > 1 {
-				commandeer.lset = args[1]
-			}
+			commandeer.objPath = args[0]
 
 			// initialize params
 			return commandeer.check()
@@ -85,15 +80,11 @@ func (cc *checkCommandeer) check() error {
 		return err
 	}
 
-	if lset, err = strToLabels(cc.name, cc.lset); err != nil {
-		return err
-	}
-
 	// get metric data and metadata
-	allAtters := append(cc.attrs, "__name", "_name", "_lset", "_maxtime")
-	container, path := cc.rootCommandeer.adapter.GetContainer()
-	objPath := fmt.Sprintf("%s/0/%s.%016x", path, cc.name, lset.Hash())
-	input := v3io.GetItemInput{Path: objPath, AttributeNames: allAtters}
+	allAttrs := append(cc.attrs, "__name", "_name", "_lset", "_maxtime")
+	container, tablePath := cc.rootCommandeer.adapter.GetContainer()
+	objPath := fmt.Sprintf("%s/%s", tablePath, cc.objPath)
+	input := v3io.GetItemInput{Path: objPath, AttributeNames: allAttrs}
 	resp, err := container.Sync.GetItem(&input)
 	if err != nil {
 		return errors.Wrap(err, "failed to GetItem")
@@ -108,10 +99,10 @@ func (cc *checkCommandeer) check() error {
 	fmt.Printf("Object: %s,  %s {%s}  maxtime: %d\n", objName, metricName, lsetString, maxtime)
 
 	// decompress and print metrics
-	for k, attr := range cc.attrs {
+	for _, attr := range cc.attrs {
 
 		values := item.GetField(attr)
-		fmt.Println("Attr:", k)
+		fmt.Println("Attr:", attr)
 
 		if values != nil {
 			bytes := values.([]byte)

Diff for: pkg/tsdbctl/create.go

@@ -30,8 +30,15 @@ import (
 	"strconv"
 )
 
-const schemaVersion = 0
-const defaultStorageClass = "local"
+const (
+	schemaVersion               = 0
+	defaultStorageClass         = "local"
+	defaultIngestionRate        = "1/s"
+	defaultRollupInterval       = "1h"
+	defaultShardingBuckets      = 8
+	defaultSampleRetentionHours = 0
+	defaultLayerRetentionTime   = "1y"
+)
 
 type createCommandeer struct {
 	cmd             *cobra.Command
@@ -62,10 +69,10 @@ func newCreateCommandeer(rootCommandeer *RootCommandeer) *createCommandeer {
 
 	cmd.Flags().StringVarP(&commandeer.defaultRollups, "rollups", "r", "",
 		"Default aggregation rollups, comma seperated: count,avg,sum,min,max,stddev")
-	cmd.Flags().StringVarP(&commandeer.rollupInterval, "rollup-interval", "i", "1h", "aggregation interval")
-	cmd.Flags().IntVarP(&commandeer.shardingBuckets, "sharding-buckets", "b", 8, "number of buckets to split key")
-	cmd.Flags().IntVarP(&commandeer.sampleRetention, "sample-retention", "a", 0, "sample retention in hours")
-	cmd.Flags().StringVar(&commandeer.sampleRate, "rate", "12/m", "sample rate")
+	cmd.Flags().StringVarP(&commandeer.rollupInterval, "rollup-interval", "i", defaultRollupInterval, "aggregation interval")
+	cmd.Flags().IntVarP(&commandeer.shardingBuckets, "sharding-buckets", "b", defaultShardingBuckets, "number of buckets to split key")
+	cmd.Flags().IntVarP(&commandeer.sampleRetention, "sample-retention", "a", defaultSampleRetentionHours, "sample retention in hours")
+	cmd.Flags().StringVar(&commandeer.sampleRate, "rate", defaultIngestionRate, "sample rate")
 
 	commandeer.cmd = cmd
 
@@ -103,7 +110,7 @@ func (cc *createCommandeer) create() error {
 		AggregatorsGranularity: cc.rollupInterval,
 		StorageClass:           defaultStorageClass,
 		SampleRetention:        cc.sampleRetention,
-		LayerRetentionTime:     "1y", //TODO
+		LayerRetentionTime:     defaultLayerRetentionTime, //TODO: make configurable
 	}
 
 	tableSchema := config.TableSchema{