Merge pull request #529 from v3io/development

Development

Author: dinal

pkg/aggregate/aggregate.go

@@ -310,7 +310,7 @@ func MaskToString(mask AggrType) string {
 	return output.String()
 }
 
-func ToAttrName(aggr AggrType) string {
+func ToAttrName(aggr fmt.Stringer) string {
 	return config.AggregateAttrPrefix + aggr.String()
 }
 

pkg/aggregate/iterator.go

@@ -134,7 +134,7 @@ func (as *Series) NewSetFromAttrs(
 			}
 			aggrArrays[aggr] = utils.AsInt64Array(attrBlob.([]byte))
 
-			dataArrays[aggr] = make([]float64, length, length)
+			dataArrays[aggr] = make([]float64, length)
 			copy(dataArrays[aggr], getOrCreateInitDataArray(aggr, length))
 		}
 	}
@@ -189,7 +189,7 @@ func (as *Series) NewSetFromChunks(length int) *Set {
 
 	for _, aggr := range rawAggregates {
 		if aggr&as.aggrMask != 0 {
-			dataArrays[aggr] = make([]float64, length, length) // TODO: len/capacity & reuse (pool)
+			dataArrays[aggr] = make([]float64, length) // TODO: len/capacity & reuse (pool)
 			initArray := getOrCreateInitDataArray(aggr, length)
 			copy(dataArrays[aggr], initArray)
 		}
@@ -373,7 +373,7 @@ func getOrCreateInitDataArray(aggrType AggrType, length int) []float64 {
 
 func createInitDataArray(aggrType AggrType, length int) []float64 {
 	// Prepare "clean" array for fastest reset of "uninitialized" data arrays
-	resultArray := make([]float64, length, length)
+	resultArray := make([]float64, length)
 
 	var initWith float64
 	switch aggrType {

pkg/appender/appender.go

@@ -248,9 +248,9 @@ func (mc *MetricsCache) Add(lset utils.LabelsIfc, t int64, v interface{}) (uint6
 		for _, preAggr := range mc.partitionMngr.GetConfig().TableSchemaInfo.PreAggregates {
 			subLset := lset.Filter(preAggr.Labels)
 			name, key, hash := subLset.GetKey()
-			aggrMetric, ok := mc.getMetric(name, hash)
+			_, ok := mc.getMetric(name, hash)
 			if !ok {
-				aggrMetric = &MetricState{Lset: subLset, key: key, name: name, hash: hash}
+				aggrMetric := &MetricState{Lset: subLset, key: key, name: name, hash: hash}
 				aggrMetric.store = newChunkStore(mc.logger, subLset.LabelNames(), true)
 				mc.addMetric(hash, name, aggrMetric)
 				aggrMetrics = append(aggrMetrics, aggrMetric)

pkg/appender/ingest.go

@@ -53,7 +53,7 @@ func (mc *MetricsCache) metricFeed(index int) {
 
 		for {
 			select {
-			case _ = <-mc.stopChan:
+			case <-mc.stopChan:
 				return
 			case app := <-mc.asyncAppendChan:
 				newMetrics := 0
@@ -150,9 +150,9 @@ func (mc *MetricsCache) metricsUpdateLoop(index int) {
 		counter := 0
 		for {
 			select {
-			case _ = <-mc.stopChan:
+			case <-mc.stopChan:
 				return
-			case _ = <-mc.newUpdates:
+			case <-mc.newUpdates:
 				// Handle new metric notifications (from metricFeed)
 				for mc.updatesInFlight < mc.cfg.Workers*2 {
 					freeSlots := mc.cfg.Workers*2 - mc.updatesInFlight
@@ -390,7 +390,7 @@ func (mc *MetricsCache) nameUpdateRespLoop() {
 	go func() {
 		for {
 			select {
-			case _ = <-mc.stopChan:
+			case <-mc.stopChan:
 				return
 			case resp := <-mc.nameUpdateChan:
 				// Handle V3IO PutItem in names table

pkg/appender/store.go

@@ -475,7 +475,6 @@ func (cs *chunkStore) writeChunks(mc *MetricsCache, metric *MetricState) (hasPen
 
 		hasPendingUpdates = true
 		cs.performanceReporter.UpdateHistogram("WriteChunksSizeHistogram", int64(pendingSamplesCount))
-		return
 	})
 
 	return

pkg/config/config.go

@@ -171,9 +171,8 @@ type V3ioConfig struct {
 	UsePreciseAggregations bool `json:"usePreciseAggregations,omitempty"`
 	// Coefficient to decide whether or not to use server aggregates optimization
 	// use server aggregations if ` <requested step> / <rollup interval>  >  UseServerAggregateCoefficient`
-	UseServerAggregateCoefficient int  `json:"useServerAggregateCoefficient,omitempty"`
-	LoadPartitionsFromSchemaAttr  bool `json:"loadPartitionsFromSchemaAttr,omitempty"`
-	RequestChanLength             int  `json:"RequestChanLength,omitempty"`
+	UseServerAggregateCoefficient int `json:"useServerAggregateCoefficient,omitempty"`
+	RequestChanLength             int `json:"RequestChanLength,omitempty"`
 }
 
 type MetricsReporterConfig struct {
@@ -225,8 +224,7 @@ type PartitionSchema struct {
 }
 
 type Partition struct {
-	StartTime  int64           `json:"startTime"`
-	SchemaInfo PartitionSchema `json:"schemaInfo"`
+	StartTime int64 `json:"startTime"`
 }
 
 type SchemaField struct {
@@ -260,7 +258,6 @@ func GetOrDefaultConfig() (*V3ioConfig, error) {
 func GetOrLoadFromFile(path string) (*V3ioConfig, error) {
 	once.Do(func() {
 		instance, failure = loadConfig(path)
-		return
 	})
 
 	return instance, failure
@@ -269,7 +266,6 @@ func GetOrLoadFromFile(path string) (*V3ioConfig, error) {
 func GetOrLoadFromData(data []byte) (*V3ioConfig, error) {
 	once.Do(func() {
 		instance, failure = loadFromData(data)
-		return
 	})
 
 	return instance, failure
@@ -280,7 +276,6 @@ func GetOrLoadFromStruct(cfg *V3ioConfig) (*V3ioConfig, error) {
 	once.Do(func() {
 		initDefaults(cfg)
 		instance = cfg
-		return
 	})
 
 	return instance, nil

pkg/partmgr/partmgr.go

@@ -25,7 +25,6 @@ import (
 	"fmt"
 	"math"
 	"path"
-	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -50,7 +49,7 @@ func NewPartitionMngr(schemaConfig *config.Schema, cont v3io.Container, v3ioConf
 		return nil, err
 	}
 	newMngr := &PartitionManager{schemaConfig: schemaConfig, cyclic: false, container: cont, currentPartitionInterval: currentPartitionInterval, v3ioConfig: v3ioConfig}
-	err = newMngr.updatePartitionsFromSchema(schemaConfig, nil)
+	err = newMngr.updatePartitionsFromSchema(schemaConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -141,7 +140,7 @@ func (p *PartitionManager) createAndUpdatePartition(t int64) (*DBPartition, erro
 	}
 	p.currentPartitionInterval = partition.partitionInterval
 
-	schemaPartition := &config.Partition{StartTime: partition.startTime, SchemaInfo: p.schemaConfig.PartitionSchemaInfo}
+	schemaPartition := &config.Partition{StartTime: partition.startTime}
 	if p.headPartition == nil || time > p.headPartition.startTime {
 		p.headPartition = partition
 		p.partitions = append(p.partitions, partition)
@@ -182,24 +181,6 @@ func (p *PartitionManager) updateSchema() error {
 			err = p.container.PutObjectSync(&v3io.PutObjectInput{Path: schemaFilePath, Body: data})
 			if err != nil {
 				outerError = err
-				return
-			}
-			attributes := make(map[string]interface{}, len(p.partitions))
-			for _, part := range p.partitions {
-				marshalledPartition, err := json.Marshal(part.ToMap())
-				if err != nil {
-					outerError = err
-					return
-				}
-				attributes[part.GetPartitionAttributeName()] = marshalledPartition
-			}
-
-			input := &v3io.PutItemInput{Path: schemaFilePath, Attributes: attributes}
-			_, err := p.container.PutItemSync(input)
-
-			if err != nil {
-				outerError = errors.Wrap(err, "failed to update partitions table.")
-				return
 			}
 		}
 	})
@@ -256,7 +237,7 @@ func (p *PartitionManager) ReadAndUpdateSchema() (err error) {
 		err = errors.Wrap(err, "Failed to create timer ReadAndUpdateSchemaTimer.")
 		return
 	}
-	schemaInfoResp, err := p.container.GetItemSync(&v3io.GetItemInput{Path: schemaFilePath, AttributeNames: []string{"**"}})
+	schemaInfoResp, err := p.container.GetItemSync(&v3io.GetItemInput{Path: schemaFilePath, AttributeNames: []string{"__mtime_secs", "__mtime_nsecs"}})
 	if err != nil {
 		err = errors.Wrapf(err, "Failed to read schema at path '%s'.", schemaFilePath)
 		return
@@ -280,29 +261,13 @@ func (p *PartitionManager) ReadAndUpdateSchema() (err error) {
 		p.schemaMtimeNanosecs = mtimeNsecs
 
 		metricReporter.WithTimer("ReadAndUpdateSchemaTimer", func() {
-			err = p.updatePartitionsFromSchema(nil, schemaGetItemResponse)
-			return
+			err = p.updatePartitionsFromSchema(nil)
 		})
 	}
 	return
 }
 
-func (p *PartitionManager) updatePartitionsFromSchema(schemaConfig *config.Schema, schemaGetItemResponse *v3io.GetItemOutput) error {
-	var currentSchemaVersion int
-	if schemaConfig == nil {
-		currentSchemaVersion = p.schemaConfig.TableSchemaInfo.Version
-	} else {
-		currentSchemaVersion = schemaConfig.TableSchemaInfo.Version
-	}
-
-	if currentSchemaVersion == 4 && p.v3ioConfig.LoadPartitionsFromSchemaAttr {
-		return p.newLoadPartitions(schemaGetItemResponse)
-	}
-
-	return p.oldLoadPartitions(schemaConfig)
-}
-
-func (p *PartitionManager) oldLoadPartitions(schema *config.Schema) error {
+func (p *PartitionManager) updatePartitionsFromSchema(schema *config.Schema) error {
 	if schema == nil {
 		schemaFilePath := p.GetSchemaFilePath()
 		resp, innerError := p.container.GetObjectSync(&v3io.GetObjectInput{Path: schemaFilePath})
@@ -334,58 +299,6 @@ func (p *PartitionManager) oldLoadPartitions(schema *config.Schema) error {
 	return nil
 }
 
-func (p *PartitionManager) newLoadPartitions(schemaAttributesResponse *v3io.GetItemOutput) error {
-	if p.container == nil { // Tests use case only
-		return nil
-	}
-
-	if schemaAttributesResponse == nil {
-		schemaFilePath := p.GetSchemaFilePath()
-		schemaInfoResp, err := p.container.GetItemSync(&v3io.GetItemInput{Path: schemaFilePath, AttributeNames: []string{"*"}})
-		if err != nil {
-			return errors.Wrapf(err, "Failed to read schema at path '%s'.", schemaFilePath)
-		}
-
-		schemaAttributesResponse = schemaInfoResp.Output.(*v3io.GetItemOutput)
-	}
-
-	p.partitions = []*DBPartition{}
-	for partitionStartTime, partitionAttrBlob := range schemaAttributesResponse.Item {
-		// Only process "partition" attributes
-		if !strings.HasPrefix(partitionStartTime, partitionAttributePrefix) {
-			continue
-		}
-		intStartTime, err := strconv.ParseInt(partitionStartTime[1:], 10, 64)
-		if err != nil {
-			return errors.Wrapf(err, "invalid partition name '%v'", partitionStartTime)
-		}
-
-		partPath := path.Join(p.Path(), strconv.FormatInt(intStartTime/1000, 10)) + "/"
-
-		partitionAttr := make(map[string]interface{}, 5)
-		err = json.Unmarshal(partitionAttrBlob.([]byte), &partitionAttr)
-		if err != nil {
-			return err
-		}
-		newPart, err := NewDBPartitionFromMap(p, intStartTime, partPath, partitionAttr)
-		if err != nil {
-			return err
-		}
-		p.partitions = append(p.partitions, newPart)
-		if p.headPartition == nil {
-			p.headPartition = newPart
-		} else if p.headPartition.startTime < newPart.startTime {
-			p.headPartition = newPart
-		}
-	}
-
-	sort.SliceStable(p.partitions, func(i, j int) bool {
-		return p.partitions[i].startTime < p.partitions[j].startTime
-	})
-
-	return nil
-}
-
 //if inclusive is true than partial partitions (not fully in range) will be retireved as well
 func (p *PartitionManager) PartsForRange(mint, maxt int64, inclusive bool) []*DBPartition {
 	var parts []*DBPartition

pkg/pquerier/collector.go

@@ -66,7 +66,7 @@ func mainCollector(ctx *selectQueryContext, responseChannel chan *qryResults) {
 
 	for {
 		select {
-		case _ = <-ctx.stopChan:
+		case <-ctx.stopChan:
 			return
 		case res, ok := <-responseChannel:
 			if !ok {

pkg/pquerier/pqueriertest/raw_query_integration_test.go

@@ -3,7 +3,6 @@
 package pqueriertest
 
 import (
-	"errors"
 	"fmt"
 	"math"
 	"testing"
@@ -718,75 +717,3 @@ func (suite *testRawQuerySuite) TestQueryMultiMetricsInconsistentLabels() {
 		suite.NotNil(iter.At(), "Iterator yielded a nil series")
 	}
 }
-
-func (suite *testRawQuerySuite) TestLoadPartitionsFromAttributes() {
-	suite.v3ioConfig.LoadPartitionsFromSchemaAttr = true
-	defer func() { suite.v3ioConfig.LoadPartitionsFromSchemaAttr = false }()
-
-	adapter, err := tsdb.NewV3ioAdapter(suite.v3ioConfig, nil, nil)
-	if err != nil {
-		suite.T().Fatalf("failed to create v3io adapter. reason: %s", err)
-	}
-
-	labels1 := utils.LabelsFromStringList("os", "linux")
-	labels2 := utils.LabelsFromStringList("os", "mac")
-	numberOfEvents := 5
-	eventsInterval := int64(tsdbtest.MinuteInMillis)
-	expectedData := []tsdbtest.DataPoint{{suite.basicQueryTime - 7*tsdbtest.DaysInMillis, 10},
-		{int64(suite.basicQueryTime + tsdbtest.MinuteInMillis), 20},
-		{suite.basicQueryTime + 2*eventsInterval, 30},
-		{suite.basicQueryTime + 3*eventsInterval, 40}}
-
-	testParams := tsdbtest.NewTestParams(suite.T(),
-		tsdbtest.TestOption{
-			Key: tsdbtest.OptTimeSeries,
-			Value: tsdbtest.TimeSeries{tsdbtest.Metric{
-				Name:   "cpu",
-				Labels: labels1,
-				Data:   expectedData},
-				tsdbtest.Metric{
-					Name:   "cpu",
-					Labels: labels2,
-					Data:   expectedData},
-			}})
-
-	tsdbtest.InsertData(suite.T(), testParams)
-
-	querierV2, err := adapter.QuerierV2()
-	if err != nil {
-		suite.T().Fatalf("Failed to create querier v2, err: %v", err)
-	}
-
-	params := &pquerier.SelectParams{Name: "cpu", From: suite.basicQueryTime - 8*tsdbtest.DaysInMillis, To: suite.basicQueryTime + int64(numberOfEvents)*eventsInterval}
-	set, err := querierV2.Select(params)
-	if err != nil {
-		suite.T().Fatalf("Failed to exeute query, err: %v", err)
-	}
-
-	var seriesCount int
-	for set.Next() {
-		seriesCount++
-		iter := set.At().Iterator()
-		data, err := tsdbtest.IteratorToSlice(iter)
-		if err != nil {
-			suite.T().Fatal(err)
-		}
-
-		for i := 0; i < len(expectedData); i++ {
-			assert.Equal(suite.T(), expectedData[i].Time, data[i].Time)
-			currentExpected := expectedData[i].Value
-			switch val := currentExpected.(type) {
-			case float64:
-				assert.Equal(suite.T(), val, data[i].Value)
-			case int:
-				assert.Equal(suite.T(), float64(val), data[i].Value)
-			case string:
-				assert.Equal(suite.T(), val, data[i].Value)
-			default:
-				assert.Error(suite.T(), errors.New("unsupported data type"))
-			}
-		}
-	}
-
-	assert.Equal(suite.T(), 2, seriesCount, "series count didn't match expected")
-}

pkg/pquerier/querier.go

@@ -187,7 +187,6 @@ func (q *V3ioQuerier) baseSelectQry(params *SelectParams, showAggregateLabel boo
 		}
 
 		iter, err = selectContext.start(parts, params)
-		return
 	})
 
 	return
@@ -227,7 +226,7 @@ func (q *V3ioQuerier) getMetricNames() ([]string, error) {
 		metricNames = append(metricNames, iter.GetField(config.ObjectNameAttrName).(string))
 	}
 
-	sort.Sort(sort.StringSlice(metricNames))
+	sort.Strings(metricNames)
 
 	if iter.Err() != nil {
 		return nil, fmt.Errorf("failed to read metric names; err = %v", iter.Err().Error())