Merge pull request #177 from v3io/development

Development --> Master (0.0.13)

Author: gshatz

Diff for: .travis.yml

@@ -6,6 +6,7 @@ go:
 script:
   - make test
   - make lint
+  - V3IO_TSDB_CONFIG="$TRAVIS_BUILD_DIR/travis_v3io.yaml" TSDB_TEST_TABLE_PATH="TSDB_INTEGRATION_TESTS/$TRAVIS_BUILD_NUMBER" make integration
 
 before_deploy:
   - GOOS=linux GOARCH=amd64 make build

Diff for: README.md

@@ -69,7 +69,7 @@ The code is separated to Prometheus compliant adapter in [/promtsdb](promtsdb) a
 [v3iotsdb_test.go](/pkg/tsdb/v3iotsdb_test.go), both have similar semantics.
 
 For Prometheus you would need to use the fork found in `https://github.com/v3io/prometheus`, it already loads this
-library, you would need to place a `v3io.yaml` file with relevant configuration in the same folder as the Prometheus
+library, you would need to place a `v3io-tsdb-config.yaml` file with relevant configuration in the same folder as the Prometheus
 executable (see details on configurations below).
 
 A developer using this library should first create a TSDB, this can be done using the CLI or an API call (`CreateTSDB`) 
@@ -102,20 +102,20 @@ For use with nuclio function you can see function example under [\nuclio](exampl
 ### Creating and Configuring a TSDB Adapter 
 
 The first step is to create a TSDB, this is done only once per TSDB and generates the required metadata and configuration
-such as partitioning strategy, retention, aggregators, etc. this can be done via the CLI or a function call.
+such as partitioning strategy, retention, aggregates, etc. this can be done via the CLI or a function call.
 
 ```go
 	// Load v3io connection/path details (see YAML below)
-	v3iocfg, err := config.GetOrLoadFromFile("v3io.yaml")
+	v3iocfg, err := config.GetOrLoadFromFile("v3io-tsdb-config.yaml")
 	if err != nil {
 		// TODO: handle error
 	}
 
 	// Specify the default DB configuration (can be modified per partition)
-	sampleRate = "1/s"
-	aggregatorGranularity = "1h"
+	samplesIngestionRate = "1/s"
+	aggregationGranularity = "1h"
 	aggregatesList = "scount,avg,min,max"
-	schema, err := schema.NewSchema(v3iocfg, sampleRate, aggregatorGranularity, aggregatesList)
+	schema, err := schema.NewSchema(v3iocfg, samplesIngestionRate, aggregationGranularity, aggregatesList)
 	if err != nil {
 		// TODO: handle error
 	}
@@ -132,22 +132,24 @@ parameters: the configuration structure, v3io data container object and logger o
 you already have container and logger (when using nuclio data bindings).
 
 Configuration is specified in a YAML or JSON format, and can be read from a file using `config.GetOrLoadFromFile(path string)` 
-or can be loaded from a local buffer using `config.GetOrLoadFromData(data []byte)`. You can see details on the configuration
-options in [config](internal/pkg/config/config.go), a minimal configuration looks like: 
+or can be loaded from a local buffer using `config.GetOrLoadFromData(data []byte)`.
+You can see details on the configuration options in the V3IO TSDB [**config.go**](pkg/config/config.go) source file.
+A template configuration file is found at **examples/v3io-tsdb-config.yaml.template**.
+You can use it as a reference for creating your own TSDB configuration file.
+For example:
 
 ```yaml
-v3ioUrl: "v3io address:port"
+webApiEndpoint: "192.168.1.100:8081"
 container: "tsdb"
-path: performance
-username: "<username>"
-password: "<password>"
+username: "johnd"
+password: "OpenSesame"
 ```
 
-example of creating an adapter:
+Following is an example of code for creating an adapter:
 
 ```go
 	// create configuration object from file
-	cfg, err := config.GetOrLoadFromFile("v3io.yaml")
+	cfg, err := config.GetOrLoadFromFile("v3io-tsdb-config.yaml")
 	if err != nil {
 		// TODO: handle error
 	}
@@ -175,8 +177,8 @@ Example:
 
 	// create metrics labels, `__name__` label specify the metric type (e.g. cpu, temperature, ..)
 	// the other labels can be used in searches (filtering or grouping) or aggregations
-	// use utils.FromStrings(s ...string) for string list input or utils.FromMap(m map[string]string) for map input
-	lset := utils.FromStrings("__name__","http_req", "method", "post")
+	// use utils.LabelsFromStrings(s ...string) for string list input or utils.LabelsFromMap(m map[string]string) for map input
+	lset := utils.LabelsFromStrings("__name__","http_req", "method", "post")
 
 	// Add a sample with current time (in milisec) and the value of 7.9
 	ref, err := appender.Add(lset, time.Now().Unix * 1000, 7.9)
@@ -197,8 +199,8 @@ The `Querier` interface is used to query the database and return one or more met
 and specify the query window (min and max times), once we did we can use `Select()` or `SelectOverlap()` commands which will 
 return a list of series (as an iterator object).
 
-Every returned series have two interfaces, `Labels()` which returns the series or aggregator labels, and `Iterator()`
-which returns an iterator over the series or aggregator values.
+Every returned series have two interfaces, `Labels()` which returns the series or aggregate labels, and `Iterator()`
+which returns an iterator over the series or aggregate values.
 
 The `Select()` call accepts 4 parameters:
 * name (string) - optional, metric type (e.g. cpu, memory, ..), specifying it accelerate performance (use range queries)   
@@ -208,7 +210,7 @@ The `Select()` call accepts 4 parameters:
 
 using `functions` and `step` is optional, use it only when you are interested in pre-aggregation and the step is >> than 
 the sampling interval (and preferably equal or greater than the partition RollupMin interval). when using aggregates it will
-return one series per aggregate function, the `Aggregator` label will be added to that series with the function name.
+return one series per aggregate function, the `Aggregate` label will be added to that series with the function name.
 
 In some cases we would like to retrieve overlapping aggregates instead of fixed interval ones, e.g. stats for last 1hr, 6hr, 24hr
 the `SelectOverlap()` call adds the `win` integer array ([]int) which allow specifying the requested windows. the windows are 

Diff for: examples/nuclio/ingest/ingest_example.go

@@ -12,13 +12,13 @@ import (
 )
 
 // Configuration
-// Note: the TSDB (path) must be first created using the CLI or API
-// the user must also define the v3io data binding in the nuclio function with path, username, password and name it db0
+// Note: the TSDB instance (`path`) must first be created using the CLI or API.
+// The user must also define the V3IO data binding in the Nuclio function - including path, username, and password - and name it "db0".
 var tsdbConfig = `
 path: "pmetric"
 `
 
-// example event
+// Example event
 const pushEvent = `
 {
   "Lset": { "__name__":"cpu", "os" : "win", "node" : "xyz123"},
@@ -38,12 +38,15 @@ func Handler(context *nuclio.Context, event nuclio.Event) (interface{}, error) {
 	}
 	app := context.UserData.(tsdb.Appender)
 
-	// if time is not specified assume "now"
+	// If time isn't specified, assume "now" (default)
 	if sample.Time == "" {
 		sample.Time = "now"
 	}
 
-	// convert time string to time int, string can be: now, now-2h, int (unix milisec time), or RFC3339 date string
+	// Convert a time string to a Unix timestamp in milliseconds integer.
+	// The input time string can be of the format "now", "now-[0-9]+[mdh]"
+	// (for example, "now-2h"), "<Unix timestamp in milliseconds>", or
+	// "<RFC3339 time>" (for example, "2018-09-26T14:10:20Z").
 	t, err := utils.Str2unixTime(sample.Time)
 	if err != nil {
 		return "", err
@@ -54,15 +57,15 @@ func Handler(context *nuclio.Context, event nuclio.Event) (interface{}, error) {
 	return "", err
 }
 
-// InitContext runs only once when the function runtime starts
+// InitContext runs only once, when the function runtime starts
 func InitContext(context *nuclio.Context) error {
 
 	var err error
 	defer adapterMtx.Unlock()
 	adapterMtx.Lock()
 
 	if adapter == nil {
-		// create adapter once for all contexts
+		// Create an adapter once for all contexts
 		cfg, _ := config.GetOrLoadFromData([]byte(tsdbConfig))
 		data := context.DataBinding["db0"].(*v3io.Container)
 		adapter, err = tsdb.NewV3ioAdapter(cfg, data, context.Logger)

Diff for: examples/nuclio/ingest/ingest_example_test.go renamed to examples/nuclio/ingest/ingest_example_integration_test.go

@@ -19,24 +19,25 @@ path: "pmetric"
 `
 
 type tsdbQuery struct {
-	Name        string
-	Aggregators []string
-	Step        string
-	Filter      string
-	From        string
-	To          string
-	Last        string
+	Name       string
+	Aggregates []string
+	Step       string
+	Filter     string
+	From       string
+	To         string
+	Last       string
 }
 
-// example query event
+// Example query event
 const queryEvent = `
 {
   "Name": "cpu", 
   "Last":"2h"
 }
 `
 
-// Note: the user must define the v3io data binding in the nuclio function with path, username, password and name it db0
+// Note: The user must define a "db0" V3IO data binding in the nuclio function
+// with path, username, and password.
 
 func Handler(context *nuclio.Context, event nuclio.Event) (interface{}, error) {
 
@@ -46,30 +47,33 @@ func Handler(context *nuclio.Context, event nuclio.Event) (interface{}, error) {
 		return nil, err
 	}
 
-	// convert string times (unix or RFC3339 or relative like now-2h) to unix milisec times
+	// Convert time strings to Unix timestamp in milliseconds integers.
+	// The input time string can be of the format "now", "now-[0-9]+[mdh]"
+	// (for example, "now-2h"), "<Unix timestamp in milliseconds>", or
+	// "<RFC3339 time>" (for example, "2018-09-26T14:10:20Z").
 	from, to, step, err := utils.GetTimeFromRange(query.From, query.To, query.Last, query.Step)
 	if err != nil {
-		return nil, errors.Wrap(err, "Error parsing query time range")
+		return nil, errors.Wrap(err, "Error parsing query time range.")
 	}
 
-	// Create TSDB Querier
+	// Create a TSDB Querier
 	context.Logger.DebugWith("Query", "params", query)
 	adapter := context.UserData.(*tsdb.V3ioAdapter)
 	qry, err := adapter.Querier(nil, from, to)
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to initialize Querier")
 	}
 
-	// Select Query to get back a series set iterator
-	set, err := qry.Select(query.Name, strings.Join(query.Aggregators, ","), step, query.Filter)
+	// Select a query to get back a series-set iterator
+	set, err := qry.Select(query.Name, strings.Join(query.Aggregates, ","), step, query.Filter)
 	if err != nil {
 		return nil, errors.Wrap(err, "Select Failed")
 	}
 
-	// convert SeriesSet to Json (Grafana simpleJson format)
+	// Convert a SeriesSet to a JSON object (Grafana simpleJson format)
 	f, err := formatter.NewFormatter("json", nil)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to start json formatter")
+		return nil, errors.Wrap(err, "Failed to start the JSON formatter.")
 	}
 
 	var b bytes.Buffer
@@ -78,7 +82,7 @@ func Handler(context *nuclio.Context, event nuclio.Event) (interface{}, error) {
 	return b.String(), err
 }
 
-// InitContext runs only once when the function runtime starts
+// InitContext runs only once, when the function runtime starts
 func InitContext(context *nuclio.Context) error {
 	cfg, _ := config.GetOrLoadFromData([]byte(tsdbConfig))
 	data := context.DataBinding["db0"].(*v3io.Container)
@@ -87,7 +91,7 @@ func InitContext(context *nuclio.Context) error {
 		return err
 	}
 
-	// Store adapter in user cache
+	// Store the adapter in the user-data cache
 	context.UserData = adapter
 	return nil
 }

Diff for: examples/nuclio/query/query_example.go

@@ -1,4 +1,4 @@
-# File:         v3io.yaml.template
+# File:         v3io-tsdb-config.yaml
 # Description:  Template of a V3IO TSDB Configuration File
 
 # TODO: In your configuration file, delete the configuration keys that you
@@ -14,13 +14,11 @@ webApiEndpoint: "<IP address/host name>:8081"
 # Example: "bigdata"
 container: "<container name>"
 
-# Logging verbosity level
+# Log level
 # Valid values: "debug" | "info" | "warn" | "error"
-# Default value: "debug"
-verbose: "warn"
+logLevel: "warn"
 
 # Authentication credentials for the web-API service
-# Username
 username: "<username>"
-# Password
-password: "<password>"
+password: "<password>"
+

Diff for: examples/nuclio/query/query_example_test.go renamed to examples/nuclio/query/query_example_integration_test.go

@@ -14,8 +14,13 @@ import (
 	"time"
 )
 
+const (
+	reservoirSize = 100
+)
+
 var instance *MetricReporter
 var once sync.Once
+var sampleInstance metrics.Sample
 
 const (
 	STDOUT = "stdout"
@@ -59,6 +64,7 @@ func ReporterInstance(writeTo string, reportPeriodically bool, reportIntervalSec
 			writer = os.Stdout
 		}
 
+		sampleInstance = metrics.NewUniformSample(reservoirSize)
 		instance = newMetricReporter(writer, reportPeriodically, reportIntervalSeconds, reportOnShutdown)
 	})
 	return instance
@@ -76,7 +82,7 @@ func (mr *MetricReporter) Start() error {
 	mr.lock.Lock()
 	defer mr.lock.Unlock()
 
-	if !mr.running {
+	if mr.isEnabled() && !mr.running {
 		mr.running = true
 	} else {
 		return errors.Errorf("metric reporter is already running.")
@@ -103,36 +109,33 @@ func (mr *MetricReporter) Stop() error {
 	return nil
 }
 
-func (mr *MetricReporter) GetTimer(name string) (metrics.Timer, error) {
-	if mr.running {
-		return metrics.GetOrRegisterTimer(name, mr.registry), nil
+func (mr *MetricReporter) WithTimer(name string, body func()) {
+	if mr.isRunning() {
+		timer := metrics.GetOrRegisterTimer(name, mr.registry)
+		timer.Time(body)
 	} else {
-		return nil, errors.Errorf("failed to create timer '%s'. Reason: metric reporter in not running", name)
+		body()
 	}
 }
 
-func (mr *MetricReporter) GetCounter(name string) (metrics.Counter, error) {
-	if mr.running {
-		return metrics.GetOrRegisterCounter(name, mr.registry), nil
-	} else {
-		return nil, errors.Errorf("failed to create counter '%s'. Reason: metric reporter in not running", name)
+func (mr *MetricReporter) IncrementCounter(name string, count int64) {
+	if mr.isRunning() {
+		counter := metrics.GetOrRegisterCounter(name, mr.registry)
+		counter.Inc(count)
 	}
 }
 
-func (mr *MetricReporter) GetMeter(name string) (metrics.Meter, error) {
-	if mr.running {
-		return metrics.GetOrRegisterMeter(name, mr.registry), nil
-	} else {
-		return nil, errors.Errorf("failed to create meter '%s'. Reason: metric reporter in not running", name)
+func (mr *MetricReporter) UpdateMeter(name string, count int64) {
+	if mr.isRunning() {
+		meter := metrics.GetOrRegisterMeter(name, mr.registry)
+		meter.Mark(count)
 	}
 }
 
-func (mr *MetricReporter) GetHistogram(name string, reservoirSize int) (metrics.Histogram, error) {
-	if mr.running {
-		sample := metrics.NewUniformSample(reservoirSize)
-		return metrics.GetOrRegisterHistogram(name, mr.registry, sample), nil
-	} else {
-		return nil, errors.Errorf("failed to create histogram '%s'. Reason: metric reporter in not running", name)
+func (mr *MetricReporter) UpdateHistogram(name string, value int64) {
+	if mr.isRunning() {
+		histogram := metrics.GetOrRegisterHistogram(name, mr.registry, sampleInstance)
+		histogram.Update(value)
 	}
 }
 
@@ -182,3 +185,14 @@ func newMetricReporter(outputWriter io.Writer, reportPeriodically bool, reportIn
 
 	return &reporter
 }
+
+func (mr *MetricReporter) isEnabled() bool {
+	return mr.reportOnShutdown || mr.reportPeriodically
+}
+
+func (mr *MetricReporter) isRunning() bool {
+	mr.lock.Lock()
+	defer mr.lock.Unlock()
+
+	return mr.running
+}