Merge pull request #434 from v3io/development

Development -> master

Author: dinal

Diff for: Makefile

@@ -32,7 +32,15 @@ BUILD_OPTS := -ldflags " \
   -X $(CONFIG_PKG).branch=$(GIT_BRANCH)" \
  -v -o "$(GOPATH)/bin/$(TSDBCTL_BIN_NAME)"
 
-TSDB_BUILD_COMMAND ?= CGO_ENABLED=0 go build $(BUILD_OPTS) ./cmd/tsdbctl
+TSDB_BUILD_COMMAND ?= GO111MODULE="on" CGO_ENABLED=0 go build $(BUILD_OPTS) ./cmd/tsdbctl
+
+.PHONY: fmt
+fmt:
+	gofmt -l -s -w .
+
+.PHONY: get
+get:
+	GO111MODULE="on" go mod tidy
 
 .PHONY: test
 test:

Diff for: go.mod

@@ -7,10 +7,12 @@ require (
 	github.com/cpuguy83/go-md2man v1.0.10 // indirect
 	github.com/ghodss/yaml v1.0.0
 	github.com/imdario/mergo v0.3.7
+	github.com/kr/pretty v0.2.0 // indirect
 	github.com/nuclio/logger v0.0.1
 	github.com/nuclio/nuclio-sdk-go v0.0.0-20190205170814-3b507fbd0324
 	github.com/nuclio/nuclio-test-go v0.0.0-20180704132150-0ce6587f8e37
 	github.com/nuclio/zap v0.0.2
+	github.com/pavius/impi v0.0.0-20180302134524-c1cbdcb8df2b // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a
 	github.com/spf13/cobra v0.0.3
@@ -21,6 +23,8 @@ require (
 	github.com/v3io/v3io-go v0.0.5-0.20191205125653-9003ae83f0b6
 	github.com/v3io/v3io-go-http v0.0.0-20190415143924-cc2fbcde6663 // indirect
 	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
+	google.golang.org/genproto v0.0.0-20181026194446-8b5d7a19e2d9 // indirect
+	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
 )
 
 replace (

Diff for: go.sum

@@ -34,6 +34,8 @@ github.com/klauspost/cpuid v0.0.0-20180405133222-e7e905edc00e h1:+lIPJOWl+jSiJOc
 github.com/klauspost/cpuid v0.0.0-20180405133222-e7e905edc00e/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
+github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
@@ -61,6 +63,8 @@ github.com/nuclio/nuclio-test-go v0.0.0-20180704132150-0ce6587f8e37/go.mod h1:aO
 github.com/nuclio/zap v0.0.0-20180228181516-4a2bd2f9ef28/go.mod h1:SUxPsgePvlyjx6c5MtGdB50pf0IQThtlyLwISLboeuc=
 github.com/nuclio/zap v0.0.2 h1:rY5PkMOl8CTkqRqIPuxziBiKK6Mq/8oEurfgRnNtqf0=
 github.com/nuclio/zap v0.0.2/go.mod h1:SUxPsgePvlyjx6c5MtGdB50pf0IQThtlyLwISLboeuc=
+github.com/pavius/impi v0.0.0-20180302134524-c1cbdcb8df2b h1:yS0+/i6mwRZCdssUd+MkFJkCn/Evh1PlUKCYe3aCtQw=
+github.com/pavius/impi v0.0.0-20180302134524-c1cbdcb8df2b/go.mod h1:x/hU0bfdWIhuOT1SKwiJg++yvkk6EuOtJk8WtDZqgr8=
 github.com/pavius/zap v0.0.0-20180228181622-8d52692529b8 h1:1N/m7VjDY1Pd30Uwv6bLttZVFQm3n8RUK9Ylf2J+4a4=
 github.com/pavius/zap v0.0.0-20180228181622-8d52692529b8/go.mod h1:6FWOCx06uh50GClv8S2cfk3asqTJs3qq3ZNRtLZE77I=
 github.com/pavius/zap v1.4.2-0.20180228181622-8d52692529b8 h1:WqLgmr/wj9TO5Sc6oYPQRAJBxuHE0NTeuVeFnT+FZVo=
@@ -76,9 +80,9 @@ github.com/prometheus/prometheus v2.5.0+incompatible h1:7QPitgO2kOFG8ecuRn9O/4L9
 github.com/prometheus/prometheus v2.5.0+incompatible/go.mod h1:oAIUtOny2rjMX0OWN5vPR5/q/twIROJvdqnQKDdil/s=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
-github.com/russross/blackfriday v1.5.2+incompatible h1:/YIL6L1Deczl4O/cQ7ZVdrdKwuB6y7EWpw9LkD8xofE=
 github.com/russross/blackfriday v1.5.2+incompatible/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
+github.com/russross/blackfriday v1.5.2+incompatible h1:/YIL6L1Deczl4O/cQ7ZVdrdKwuB6y7EWpw9LkD8xofE=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=

Diff for: internal/pkg/performance/metrics.go

@@ -190,8 +190,5 @@ func (mr *MetricReporter) isEnabled() bool {
 }
 
 func (mr *MetricReporter) isRunning() bool {
-	mr.lock.Lock()
-	defer mr.lock.Unlock()
-
-	return mr.running
+	return false
 }

Diff for: pkg/appender/appender.go

@@ -122,14 +122,18 @@ type MetricsCache struct {
 	updatesComplete chan int
 	newUpdates      chan int
 
-	lastMetric     uint64
+	lastMetric uint64
+
+	// TODO: consider switching to synch.Map (https://golang.org/pkg/sync/#Map)
 	cacheMetricMap map[cacheKey]*MetricState // TODO: maybe use hash as key & combine w ref
 	cacheRefMap    map[uint64]*MetricState   // TODO: maybe turn to list + free list, periodically delete old matrics
 
 	NameLabelMap map[string]bool // temp store all lable names
 
 	lastError           error
 	performanceReporter *performance.MetricReporter
+
+	stopChan chan int
 }
 
 func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config.V3ioConfig,
@@ -146,6 +150,7 @@ func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config
 	newCache.metricQueue = NewElasticQueue()
 	newCache.updatesComplete = make(chan int, 100)
 	newCache.newUpdates = make(chan int, 1000)
+	newCache.stopChan = make(chan int, 3)
 
 	newCache.NameLabelMap = map[string]bool{}
 	newCache.performanceReporter = performance.ReporterInstanceFromConfig(cfg)
@@ -217,11 +222,18 @@ func (mc *MetricsCache) Add(lset utils.LabelsIfc, t int64, v interface{}) (uint6
 
 	isValueVariantType := false
 	// If the value is not of Float type assume it's variant type.
-	if _, ok := v.(float64); !ok {
+	switch v.(type) {
+	case int, int64, float64, float32:
+		isValueVariantType = false
+	default:
 		isValueVariantType = true
 	}
 
 	name, key, hash := lset.GetKey()
+	err = utils.IsValidMetricName(name)
+	if err != nil {
+		return 0, err
+	}
 	metric, ok := mc.getMetric(name, hash)
 
 	var aggrMetrics []*MetricState
@@ -301,6 +313,12 @@ func verifyTimeValid(t int64) error {
 	}
 	return nil
 }
+func (mc *MetricsCache) Close() {
+	//for 3 go funcs
+	mc.stopChan <- 0
+	mc.stopChan <- 0
+	mc.stopChan <- 0
+}
 
 func (mc *MetricsCache) WaitForCompletion(timeout time.Duration) (int, error) {
 	waitChan := make(chan int, 2)

Diff for: pkg/appender/ingest.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/v3io/v3io-go/pkg/dataplane"
 	"github.com/v3io/v3io-go/pkg/errors"
+	"github.com/v3io/v3io-tsdb/pkg/utils"
 )
 
 // Start event loops for handling metric updates (appends and Get/Update DB responses)
@@ -53,6 +54,8 @@ func (mc *MetricsCache) metricFeed(index int) {
 
 		for {
 			select {
+			case _ = <-mc.stopChan:
+				return
 			case inFlight = <-mc.updatesComplete:
 				// Handle completion notifications from the update loop
 				length := mc.metricQueue.Length()
@@ -149,6 +152,8 @@ func (mc *MetricsCache) metricsUpdateLoop(index int) {
 		counter := 0
 		for {
 			select {
+			case _ = <-mc.stopChan:
+				return
 			case _ = <-mc.newUpdates:
 				// Handle new metric notifications (from metricFeed)
 				for mc.updatesInFlight < mc.cfg.Workers*2 { //&& newMetrics > 0{
@@ -266,7 +271,7 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 
 	if resp.Error != nil && metric.getState() != storeStateGet {
 		req := reqInput.(*v3io.UpdateItemInput)
-		mc.logger.ErrorWith("I/O failure", "id", resp.ID, "err", resp.Error, "key", metric.key,
+		mc.logger.WarnWith("I/O failure", "id", resp.ID, "err", resp.Error, "key", metric.key,
 			"in-flight", mc.updatesInFlight, "mqueue", mc.metricQueue.Length(),
 			"numsamples", metric.store.samplesQueueLength(), "path", req.Path, "update expression", req.Expression)
 	} else {
@@ -300,8 +305,17 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 			// Metrics with too many update errors go into Error state
 			metric.retryCount++
 			if e, hasStatusCode := resp.Error.(v3ioerrors.ErrorWithStatusCode); hasStatusCode && e.StatusCode() != http.StatusServiceUnavailable {
-				mc.logger.ErrorWith(fmt.Sprintf("Chunk update failed with status code %d.", e.StatusCode()))
-				setError(mc, metric, errors.Wrap(resp.Error, fmt.Sprintf("Chunk update failed due to status code %d.", e.StatusCode())))
+				// If condition was evaluated as false log this and report this error upstream.
+				if utils.IsFalseConditionError(resp.Error) {
+					req := reqInput.(*v3io.UpdateItemInput)
+					// This might happen on attempt to add metric value of wrong type, i.e. float <-> string
+					errMsg := fmt.Sprintf("trying to ingest values of incompatible data type. Metric %q has not been updated.", req.Path)
+					mc.logger.ErrorWith(errMsg)
+					setError(mc, metric, errors.Wrap(resp.Error, errMsg))
+				} else {
+					mc.logger.ErrorWith(fmt.Sprintf("Chunk update failed with status code %d.", e.StatusCode()))
+					setError(mc, metric, errors.Wrap(resp.Error, fmt.Sprintf("Chunk update failed due to status code %d.", e.StatusCode())))
+				}
 				clear()
 				return false
 			} else if metric.retryCount == maxRetriesOnWrite {
@@ -347,24 +361,27 @@ func (mc *MetricsCache) nameUpdateRespLoop() {
 
 	go func() {
 		for {
-			resp := <-mc.nameUpdateChan
-			// Handle V3IO PutItem in names table
-
-			metric, ok := resp.Context.(*MetricState)
-			if ok {
-				metric.Lock()
-				if resp.Error != nil {
-					// Count errors
-					mc.performanceReporter.IncrementCounter("UpdateNameError", 1)
-
-					mc.logger.ErrorWith("Update-name process failed", "id", resp.ID, "name", metric.name)
-				} else {
-					mc.logger.DebugWith("Update-name process response", "id", resp.ID, "name", metric.name)
+			select {
+			case _ = <-mc.stopChan:
+				return
+			case resp := <-mc.nameUpdateChan:
+				// Handle V3IO PutItem in names table
+				metric, ok := resp.Context.(*MetricState)
+				if ok {
+					metric.Lock()
+					if resp.Error != nil {
+						// Count errors
+						mc.performanceReporter.IncrementCounter("UpdateNameError", 1)
+
+						mc.logger.ErrorWith("Update-name process failed", "id", resp.ID, "name", metric.name)
+					} else {
+						mc.logger.DebugWith("Update-name process response", "id", resp.ID, "name", metric.name)
+					}
+					metric.Unlock()
 				}
-				metric.Unlock()
-			}
 
-			resp.Release()
+				resp.Release()
+			}
 		}
 	}()
 }

Diff for: pkg/appender/store.go

@@ -232,7 +232,7 @@ func (cs *chunkStore) Append(t int64, v interface{}) {
 
 	cs.pending = append(cs.pending, pendingData{t: t, v: v})
 	// If the new time is older than previous times, sort the list
-	if len(cs.pending) > 1 && cs.pending[len(cs.pending)-2].t < t {
+	if len(cs.pending) > 1 && cs.pending[len(cs.pending)-2].t > t {
 		sort.Sort(cs.pending)
 	}
 }
@@ -418,17 +418,20 @@ func (cs *chunkStore) writeChunks(mc *MetricsCache, metric *MetricState) (hasPen
 
 			var encodingExpr string
 			if !cs.isAggr() {
-				encodingExpr = fmt.Sprintf("%v='%d'; ", config.EncodingAttrName, activeChunk.appender.Encoding())
+				encodingExpr = fmt.Sprintf("%s='%d'; ", config.EncodingAttrName, activeChunk.appender.Encoding())
 			}
-			lsetExpr := fmt.Sprintf("%v='%s'; ", config.LabelSetAttrName, metric.key)
+			lsetExpr := fmt.Sprintf("%s='%s'; ", config.LabelSetAttrName, metric.key)
 			expr = lblexpr + encodingExpr + lsetExpr + expr
 		}
 
 		// Call the V3IO async UpdateItem method
+		conditionExpr := fmt.Sprintf("NOT exists(%s) OR (exists(%s) AND %s == '%d')",
+			config.EncodingAttrName, config.EncodingAttrName,
+			config.EncodingAttrName, activeChunk.appender.Encoding())
 		expr += fmt.Sprintf("%v=%d;", config.MaxTimeAttrName, cs.maxTime) // TODO: use max() expr
 		path := partition.GetMetricPath(metric.name, metric.hash, cs.labelNames, cs.isAggr())
 		request, err := mc.container.UpdateItem(
-			&v3io.UpdateItemInput{Path: path, Expression: &expr}, metric, mc.responseChan)
+			&v3io.UpdateItemInput{Path: path, Expression: &expr, Condition: conditionExpr}, metric, mc.responseChan)
 		if err != nil {
 			mc.logger.ErrorWith("UpdateItem failed", "err", err)
 			hasPendingUpdates = false

Diff for: pkg/chunkenc/vartype.go

@@ -108,36 +108,18 @@ func (a *varAppender) Chunk() Chunk {
 }
 
 func (a *varAppender) Append(t int64, v interface{}) {
-
 	if v == nil {
 		a.appendNoValue(t, varTypeNil, varValueNone)
 		return
 	}
 
-	switch vType := v.(type) {
-	case float64:
-		val := v.(float64)
-		if val == 0 {
-			a.appendNoValue(t, varTypeFloat64, varValueZero)
-			return
-
-		}
-
-		if math.IsNaN(val) {
-			a.appendNoValue(t, varTypeFloat64, varValueNone)
-			return
-		}
-
-		a.appendWithUint(t, varTypeFloat64, math.Float64bits(val))
-
+	switch val := v.(type) {
 	case string:
-		val := []byte(v.(string))
-		a.appendWithValue(t, varTypeString, val)
+		a.appendWithValue(t, varTypeString, []byte(val))
 
 	default:
-		a.logger.Error("unsupported type %v of value %v\n", vType, v)
+		a.logger.Error("unsupported type %T of value %v\n", v, v)
 	}
-
 }
 
 func (a *varAppender) appendNoValue(t int64, varType, varVal byte) {

Diff for: pkg/chunkenc/xor.go

@@ -173,7 +173,18 @@ func (a *xorAppender) Chunk() Chunk {
 func (a *xorAppender) Append(t int64, vvar interface{}) {
 	var tDelta uint64
 	num := *a.samples
-	v := vvar.(float64)
+
+	var v float64
+	switch typedValue := vvar.(type) {
+	case int:
+		v = float64(typedValue)
+	case float64:
+		v = typedValue
+	default:
+		a.logger.Warn("Discarding sample {time: %d, value: %v}, as it's value is of incompatible data type. "+
+			"Reason: expected 'float' actual '%T'.", t, vvar, vvar)
+		return
+	}
 
 	// Do not append if sample is too old.
 	if t < a.t {

Diff for: pkg/config/config.go

@@ -170,7 +170,7 @@ type V3ioConfig struct {
 	// 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"`
-	LoadPartitionsFromSchemaFile  bool `json:"loadPartitionsFromSchemaFile,omitempty"`
+	LoadPartitionsFromSchemaAttr  bool `json:"loadPartitionsFromSchemaAttr,omitempty"`
 }
 
 type MetricsReporterConfig struct {