Merge pull request #69 from v3io/development

Merge: development -> master

Author: Gal Topper

.travis.yml

@@ -4,6 +4,7 @@ go:
   - "1.10.x"
 
 script:
+  - make get
   - make test
   - make lint
 

Makefile

@@ -18,15 +18,15 @@ get:
 	go get -v -t -tags "unit integration" $(TOPLEVEL_DIRS)
 
 .PHONY: test
-test: get
+test:
 	go test -race -tags unit -count 1 $(TOPLEVEL_DIRS)
 
 .PHONY: integration
-integration: get
+integration:
 	go test -race -tags integration -p 1 -count 1 $(TOPLEVEL_DIRS) # p=1 to force Go to run pkg tests serially.
 
 .PHONY: build
-build: get
+build:
 	go build -v -o "$(GOPATH)/bin/$(TSDBCTL_BIN_NAME)" ./cmd/tsdbctl
 
 .PHONY: lint

pkg/appender/store.go

@@ -274,7 +274,7 @@ func (cs *chunkStore) WriteChunks(mc *MetricsCache, metric *MetricState) (bool,
 	var pendingSamplesCount int
 
 	// loop over pending samples, add to chunks & aggregates (create required update expressions)
-	for {
+	for pendingSampleIndex < len(cs.pending) && pendingSamplesCount < mc.cfg.BatchSize && partition.InRange(cs.pending[pendingSampleIndex].t) {
 		sampleTime := cs.pending[pendingSampleIndex].t
 
 		if sampleTime <= cs.initMaxTime && !mc.cfg.OverrideOld {
@@ -308,7 +308,6 @@ func (cs *chunkStore) WriteChunks(mc *MetricsCache, metric *MetricState) (bool,
 		if (pendingSampleIndex == len(cs.pending)-1) || pendingSamplesCount == mc.cfg.BatchSize-1 || !partition.InRange(cs.pending[pendingSampleIndex+1].t) {
 			expr = expr + cs.aggrList.SetOrUpdateExpr("v", bucket, isNewBucket)
 			expr = expr + cs.appendExpression(activeChunk)
-			cs.aggrList.Clear()
 			pendingSampleIndex++
 			pendingSamplesCount++
 			break
@@ -334,6 +333,7 @@ func (cs *chunkStore) WriteChunks(mc *MetricsCache, metric *MetricState) (bool,
 		pendingSamplesCount++
 	}
 
+	cs.aggrList.Clear()
 	if pendingSampleIndex == len(cs.pending) {
 		cs.pending = cs.pending[:0]
 	} else {

pkg/partmgr/partmgr.go

@@ -124,12 +124,12 @@ func (p *PartitionManager) Init() error {
 func (p *PartitionManager) TimeToPart(t int64) (*DBPartition, error) {
 	if p.headPartition == nil {
 		// Rounding t to the nearest PartitionInterval multiple
-		_, err := p.createNewPartition(p.currentPartitionInterval * (t / p.currentPartitionInterval))
+		_, err := p.createAndUpdatePartition(p.currentPartitionInterval * (t / p.currentPartitionInterval))
 		return p.headPartition, err
 	} else {
 		if t >= p.headPartition.startTime {
-			if (t - p.headPartition.startTime) > p.currentPartitionInterval {
-				_, err := p.createNewPartition(p.headPartition.startTime + p.currentPartitionInterval)
+			if (t - p.headPartition.startTime) >= p.currentPartitionInterval {
+				_, err := p.createAndUpdatePartition(p.headPartition.startTime + p.currentPartitionInterval)
 				if err != nil {
 					return nil, err
 				}
@@ -144,21 +144,35 @@ func (p *PartitionManager) TimeToPart(t int64) (*DBPartition, error) {
 					return p.partitions[i], nil
 				}
 			}
+			head := p.headPartition
+			part, _ := p.createAndUpdatePartition(p.currentPartitionInterval * (t / p.currentPartitionInterval))
+			p.headPartition = head
+			return part, nil
 		}
 	}
-	return p.headPartition, nil
 }
 
-func (p *PartitionManager) createNewPartition(t int64) (*DBPartition, error) {
+func (p *PartitionManager) createAndUpdatePartition(t int64) (*DBPartition, error) {
 	time := t & 0x7FFFFFFFFFFFFFF0
 	partPath := path.Join(p.path, strconv.FormatInt(time/1000, 10)) + "/"
 	partition, err := NewDBPartition(p, time, partPath)
 	if err != nil {
 		return nil, err
 	}
 	p.currentPartitionInterval = partition.partitionInterval
-	p.headPartition = partition
-	p.partitions = append(p.partitions, partition)
+	if p.headPartition == nil || time > p.headPartition.startTime {
+		p.headPartition = partition
+		p.partitions = append(p.partitions, partition)
+	} else {
+		for i, part := range p.partitions {
+			if part.startTime > time {
+				p.partitions = append(p.partitions, nil)
+				copy(p.partitions[i+1:], p.partitions[i:])
+				p.partitions[i] = partition
+				break
+			}
+		}
+	}
 	err = p.updatePartitionInSchema(partition)
 	return partition, err
 }
@@ -169,7 +183,9 @@ func (p *PartitionManager) updatePartitionInSchema(partition *DBPartition) error
 	if err != nil {
 		return errors.Wrap(err, "Failed to update new partition in schema file")
 	}
-	err = p.container.Sync.PutObject(&v3io.PutObjectInput{Path: path.Join(p.path, config.SCHEMA_CONFIG), Body: data})
+	if p.container != nil { //tests use case only
+		err = p.container.Sync.PutObject(&v3io.PutObjectInput{Path: path.Join(p.path, config.SCHEMA_CONFIG), Body: data})
+	}
 	return err
 }
 

pkg/partmgr/partmgr_test.go

@@ -0,0 +1,53 @@
+// +build unit
+
+/*
+Copyright 2018 Iguazio Systems Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License") with
+an addition restriction as set forth herein. You may not use this
+file except in compliance with the License. You may obtain a copy of
+the License at http://www.apache.org/licenses/LICENSE-2.0.
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+implied. See the License for the specific language governing
+permissions and limitations under the License.
+
+In addition, you may not use the software for any purposes that are
+illegal under applicable law, and the grant of the foregoing license
+under the Apache 2.0 license is conditioned upon your compliance with
+such restriction.
+*/
+
+package partmgr
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/v3io/v3io-tsdb/pkg/tsdb/tsdbtest/testutils"
+	"github.com/v3io/v3io-tsdb/pkg/utils"
+	"testing"
+)
+
+func TestCreateNewPartition(tst *testing.T) {
+	schema := testutils.CreateSchema(tst, "*")
+	interval, _ := utils.Str2duration(schema.PartitionSchemaInfo.PartitionerInterval)
+	manager, _ := NewPartitionMngr(&schema, "/", nil)
+	startTime := interval + 1
+	//first partition
+	part, _ := manager.TimeToPart(startTime + interval)
+	assert.Equal(tst, 1, len(manager.partitions))
+	assert.Equal(tst, manager.headPartition, part)
+	//new head
+	part, _ = manager.TimeToPart(startTime + (interval * 3))
+	assert.Equal(tst, 3, len(manager.partitions))
+	assert.Equal(tst, manager.headPartition, part)
+	//add in the middle
+	part, _ = manager.TimeToPart(startTime + (interval * 2))
+	assert.Equal(tst, 3, len(manager.partitions))
+	assert.Equal(tst, manager.partitions[1], part)
+	//add first
+	part, _ = manager.TimeToPart(startTime)
+	assert.Equal(tst, 4, len(manager.partitions))
+	assert.Equal(tst, manager.partitions[0], part)
+}

pkg/tsdb/tsdbtest/testutils/schema.go

@@ -0,0 +1,53 @@
+package testutils
+
+import (
+	"github.com/v3io/v3io-tsdb/pkg/aggregate"
+	"github.com/v3io/v3io-tsdb/pkg/config"
+	"testing"
+)
+
+func CreateSchema(t testing.TB, agg string) config.Schema {
+	rollups, err := aggregate.AggregatorsToStringList(agg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defaultRollup := config.Rollup{
+		Aggregators:            rollups,
+		AggregatorsGranularity: "1h",
+		StorageClass:           "local",
+		SampleRetention:        0,
+		LayerRetentionTime:     "1y",
+	}
+
+	tableSchema := config.TableSchema{
+		Version:             0,
+		RollupLayers:        []config.Rollup{defaultRollup},
+		ShardingBuckets:     8,
+		PartitionerInterval: "2d",
+		ChunckerInterval:    "1h",
+	}
+
+	fields, err := aggregate.SchemaFieldFromString(rollups, "v")
+	if err != nil {
+		t.Fatal("Failed to create aggregators list", err)
+	}
+	fields = append(fields, config.SchemaField{Name: "_name", Type: "string", Nullable: false, Items: ""})
+
+	partitionSchema := config.PartitionSchema{
+		Version:                tableSchema.Version,
+		Aggregators:            rollups,
+		AggregatorsGranularity: "1h",
+		StorageClass:           "local",
+		SampleRetention:        0,
+		ChunckerInterval:       tableSchema.ChunckerInterval,
+		PartitionerInterval:    tableSchema.PartitionerInterval,
+	}
+
+	schema := config.Schema{
+		TableSchemaInfo:     tableSchema,
+		PartitionSchemaInfo: partitionSchema,
+		Partitions:          []config.Partition{},
+		Fields:              fields,
+	}
+	return schema
+}

pkg/tsdb/tsdbtest/tsdbtest.go

@@ -2,9 +2,9 @@ package tsdbtest
 
 import (
 	"fmt"
-	"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/tsdb/tsdbtest/testutils"
 	"github.com/v3io/v3io-tsdb/pkg/utils"
 	"regexp"
 	"strings"
@@ -42,7 +42,7 @@ func DeleteTSDB(t testing.TB, v3ioConfig *config.V3ioConfig) {
 }
 
 func CreateTestTSDB(t testing.TB, v3ioConfig *config.V3ioConfig) {
-	schema := CreateSchema(t, "*")
+	schema := testutils.CreateSchema(t, "*")
 	if err := CreateTSDB(v3ioConfig, &schema); err != nil {
 		t.Fatalf("Failed to create TSDB. reason: %s", err)
 	}
@@ -153,49 +153,3 @@ func NormalizePath(path string) string {
 	re := regexp.MustCompile("[" + r + "]+")
 	return re.ReplaceAllString(path, "_")
 }
-
-func CreateSchema(t testing.TB, agg string) config.Schema {
-	rollups, err := aggregate.AggregatorsToStringList(agg)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defaultRollup := config.Rollup{
-		Aggregators:            rollups,
-		AggregatorsGranularity: "1h",
-		StorageClass:           "local",
-		SampleRetention:        0,
-		LayerRetentionTime:     "1y",
-	}
-
-	tableSchema := config.TableSchema{
-		Version:             0,
-		RollupLayers:        []config.Rollup{defaultRollup},
-		ShardingBuckets:     8,
-		PartitionerInterval: "2d",
-		ChunckerInterval:    "1h",
-	}
-
-	fields, err := aggregate.SchemaFieldFromString(rollups, "v")
-	if err != nil {
-		t.Fatal("Failed to create aggregators list", err)
-	}
-	fields = append(fields, config.SchemaField{Name: "_name", Type: "string", Nullable: false, Items: ""})
-
-	partitionSchema := config.PartitionSchema{
-		Version:                tableSchema.Version,
-		Aggregators:            rollups,
-		AggregatorsGranularity: "1h",
-		StorageClass:           "local",
-		SampleRetention:        0,
-		ChunckerInterval:       tableSchema.ChunckerInterval,
-		PartitionerInterval:    tableSchema.PartitionerInterval,
-	}
-
-	schema := config.Schema{
-		TableSchemaInfo:     tableSchema,
-		PartitionSchemaInfo: partitionSchema,
-		Partitions:          []config.Partition{},
-		Fields:              fields,
-	}
-	return schema
-}

pkg/tsdbctl/add.go

@@ -107,7 +107,7 @@ func (ac *addCommandeer) add() error {
 		return err
 	}
 
-	append, err := ac.rootCommandeer.adapter.Appender()
+	appender, err := ac.rootCommandeer.adapter.Appender()
 	if err != nil {
 		return errors.Wrap(err, "failed to create Appender")
 	}
@@ -127,22 +127,22 @@ func (ac *addCommandeer) add() error {
 			return err
 		}
 
-		_, err = ac.appendMetric(append, lset, tarray, varray, true)
+		_, err = ac.appendMetric(appender, lset, tarray, varray, true)
 		if err != nil {
 			return err
 		}
 
-		_, err = append.WaitForCompletion(0)
+		_, err = appender.WaitForCompletion(0)
 		return err
 	}
 
-	err = ac.appendMetrics(append, lset)
+	err = ac.appendMetrics(appender, lset)
 	if err != nil {
 		return err
 	}
 
 	// make sure all writes are committed
-	_, err = append.WaitForCompletion(0)
+	_, err = appender.WaitForCompletion(0)
 	if err != nil {
 		return errors.Wrap(err, "operation timed out")
 	}
@@ -239,6 +239,10 @@ func (ac *addCommandeer) appendMetric(
 
 func strToLabels(name, lbls string) (utils.Labels, error) {
 
+	if err := utils.IsValidMetricName(name); err != nil {
+		return nil, errors.Wrap(err, fmt.Sprintf("illegal metric name: '%s'", name))
+	}
+
 	lset := utils.Labels{utils.Label{Name: "__name__", Value: name}}
 
 	if lbls != "" {
@@ -248,6 +252,10 @@ func strToLabels(name, lbls string) (utils.Labels, error) {
 			if len(splitLbl) != 2 {
 				return nil, errors.New("labels must be in the form: key1=label1,key2=label2,...")
 			}
+
+			if err := utils.IsValidLabelName(splitLbl[0]); err != nil {
+				return nil, errors.Wrap(err, fmt.Sprintf("illegal label name: '%s'", splitLbl[0]))
+			}
 			lset = append(lset, utils.Label{Name: splitLbl[0], Value: splitLbl[1]})
 		}
 	}

pkg/tsdbctl/add_test.go

@@ -69,6 +69,30 @@ func (suite *testAddSuite) TestStrToTVInvalidTime() {
 	suite.Require().Error(err)
 }
 
+func (suite *testAddSuite) TestStrToLabelsWithEmptyName() {
+
+	labels, err := strToLabels("", "a=A,b=B")
+
+	suite.Require().Nil(labels)
+	suite.Require().Error(err)
+}
+
+func (suite *testAddSuite) TestStrToLabelsWithIllegalName() {
+
+	labels, err := strToLabels("illegal-name", "a=A,b=B")
+
+	suite.Require().Nil(labels)
+	suite.Require().Error(err)
+}
+
+func (suite *testAddSuite) TestStrToLabelsWithIllegalLabel() {
+
+	labels, err := strToLabels("valid_name", "a=A,b-b=B")
+
+	suite.Require().Nil(labels)
+	suite.Require().Error(err)
+}
+
 func TestAddSuite(t *testing.T) {
 	suite.Run(t, new(testAddSuite))
 }