Merge pull request #152 from v3io/development

Development --> Master (0.0.12)

Author: gshatz

Diff for: Makefile

@@ -27,7 +27,7 @@ integration: get
 
 .PHONY: bench
 bench: get
-	go test -run=XXX -bench='^BenchmarkIngest$$' -benchtime 10s -timeout 1m ./test/benchmark/...
+	go test -run=XXX -bench='^BenchmarkIngest$$' -benchtime 10s -timeout 5m ./test/benchmark/...
 
 .PHONY: build
 build: get

Diff for: README.md

@@ -81,17 +81,17 @@ A user can run the CLI to add (append) or query the DB, to use the CLI, build th
 it has built-in help, see the following add/query examples:
 
 ```
-	# create a DB with some aggregates (at 30 min interval) 
-	tsdbctl create -p <path> -r count,sum,max -i 30
+	# create a DB with expected ingestion rate of one sample per second and some aggregates (at 30 min interval) 
+	tsdbctl create -t <table> --ingestion-rate 1/s -a count,sum,max -i 30m
 
 	# display DB info with metric names (types) 
-	tsdbctl info -n
+	tsdbctl info -t <table> -n
 	
 	# append a sample (73.2) to the specified metric type (cpu) + labels at the current time
-	tsdbctl add cpu os=win,node=xyz123 -d 73.2
+	tsdbctl add -t <table> cpu os=win,node=xyz123 -d 73.2
 	
 	# display all the CPU metrics for win servers from the last hours, in CSV format 
-	tsdbctl query cpu -f "os=='win'" -l 1h -o csv
+	tsdbctl query -t <table> cpu -f "os=='win'" -l 1h -o csv
 	
 ```
 
@@ -106,17 +106,21 @@ such as partitioning strategy, retention, aggregators, etc. this can be done via
 
 ```go
 	// Load v3io connection/path details (see YAML below)
-	v3iocfg, _ := cfg, err = config.LoadConfig("v3io.yaml")
+	v3iocfg, err := config.GetOrLoadFromFile("v3io.yaml")
+	if err != nil {
+		// TODO: handle error
+	}
 
 	// Specify the default DB configuration (can be modified per partition)
-	dbcfg := config.DBPartConfig{
-		DaysPerObj:     1,
-		HrInChunk:      1,
-		DefaultRollups: "count,avg,sum,stddev",
-		RollupMin:      30,
+	sampleRate = "1/s"
+	aggregatorGranularity = "1h"
+	aggregatesList = "scount,avg,min,max"
+	schema, err := schema.NewSchema(v3iocfg, sampleRate, aggregatorGranularity, aggregatesList)
+	if err != nil {
+		// TODO: handle error
 	}
-
-	return tsdb.CreateTSDB(v3iocfg, &dbcfg)
+	
+	return tsdb.CreateTSDB(v3iocfg, schema)
 ```
 
 > If you plan on using pre-aggregation to speed aggregate queries you should specify the `Rollups` (function list) and 
@@ -127,14 +131,14 @@ In order to use the TSDB we need to create an adapter, the `NewV3ioAdapter` func
 parameters: the configuration structure, v3io data container object and logger object. The last 2 are optional, in case
 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.LoadConfig(path string)` 
-or can be loaded from a local buffer using `config.LoadFromData(data []byte)`. You can see details on the configuration
+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: 
 
 ```yaml
 v3ioUrl: "v3io address:port"
 container: "tsdb"
-path: "metrics"
+path: performance
 username: "<username>"
 password: "<password>"
 ```
@@ -143,15 +147,15 @@ example of creating an adapter:
 
 ```go
 	// create configuration object from file
-	cfg, err := config.LoadConfig("v3io.yaml")
+	cfg, err := config.GetOrLoadFromFile("v3io.yaml")
 	if err != nil {
-		panic(err)
+		// TODO: handle error
 	}
 
 	// create and start a new TSDB adapter 
 	adapter, err := tsdb.NewV3ioAdapter(cfg, nil, nil)
 	if err != nil {
-		panic(err)
+		// TODO: handle error
 	}
 ```
 

Diff for: cmd/tsdbctl/tsdbctl.go

@@ -7,13 +7,19 @@ import (
 
 func main() {
 	if err := Run(); err != nil {
-
 		os.Exit(1)
 	}
-
 	os.Exit(0)
 }
 
 func Run() error {
-	return tsdbctl.NewRootCommandeer().Execute()
+	rootCmd := tsdbctl.NewRootCommandeer()
+	defer tearDown(rootCmd)
+	return rootCmd.Execute()
+}
+
+func tearDown(cmd *tsdbctl.RootCommandeer) {
+	if cmd.Reporter != nil { // could be nil if has failed on initialisation
+		cmd.Reporter.Stop()
+	}
 }

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

@@ -63,7 +63,7 @@ func InitContext(context *nuclio.Context) error {
 
 	if adapter == nil {
 		// create adapter once for all contexts
-		cfg, _ := config.LoadFromData([]byte(tsdbConfig))
+		cfg, _ := config.GetOrLoadFromData([]byte(tsdbConfig))
 		data := context.DataBinding["db0"].(*v3io.Container)
 		adapter, err = tsdb.NewV3ioAdapter(cfg, data, context.Logger)
 		if err != nil {

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

@@ -12,12 +12,15 @@ import (
 
 func TestIngestIntegration(t *testing.T) {
 	v3ioConfig, err := tsdbtest.LoadV3ioConfig()
+	if err != nil {
+		t.Fatalf("unable to load configuration. Error: %v", err)
+	}
 	defer tsdbtest.SetUp(t, v3ioConfig)()
-	tsdbConfig = fmt.Sprintf(`path: "%v"`, v3ioConfig.Path)
+	tsdbConfig = fmt.Sprintf(`path: "%v"`, v3ioConfig.TablePath)
 
 	url := os.Getenv("V3IO_SERVICE_URL")
 	if url == "" {
-		url = v3ioConfig.V3ioUrl
+		url = v3ioConfig.WebApiEndpoint
 	}
 
 	data := nutest.DataBind{

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

@@ -80,7 +80,7 @@ func Handler(context *nuclio.Context, event nuclio.Event) (interface{}, error) {
 
 // InitContext runs only once when the function runtime starts
 func InitContext(context *nuclio.Context) error {
-	cfg, _ := config.LoadFromData([]byte(tsdbConfig))
+	cfg, _ := config.GetOrLoadFromData([]byte(tsdbConfig))
 	data := context.DataBinding["db0"].(*v3io.Container)
 	adapter, err := tsdb.NewV3ioAdapter(cfg, data, context.Logger)
 	if err != nil {

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

@@ -12,12 +12,15 @@ import (
 
 func TestQueryIntegration(t *testing.T) {
 	v3ioConfig, err := tsdbtest.LoadV3ioConfig()
+	if err != nil {
+		t.Fatalf("unable to load configuration. Error: %v", err)
+	}
 	defer tsdbtest.SetUp(t, v3ioConfig)()
-	tsdbConfig = fmt.Sprintf(`path: "%v"`, v3ioConfig.Path)
+	tsdbConfig = fmt.Sprintf(`path: "%v"`, v3ioConfig.TablePath)
 
 	url := os.Getenv("V3IO_SERVICE_URL")
 	if url == "" {
-		url = v3ioConfig.V3ioUrl
+		url = v3ioConfig.WebApiEndpoint
 	}
 
 	data := nutest.DataBind{

Diff for: examples/v3io.yaml.template

@@ -0,0 +1,26 @@
+# File:         v3io.yaml.template
+# Description:  Template of a V3IO TSDB Configuration File
+
+# TODO: In your configuration file, delete the configuration keys that you
+#       don't need and replace the "<...>" placeholders.
+
+# Endpoint of an Iguazio Continuous Data Platform web-gateway (web-API) service,
+# consisting of an IP address or resolvable host domain name, and a port number
+# (currently, always port 8081)
+# Example: "192.168.1.100:8081"
+webApiEndpoint: "<IP address/host name>:8081"
+
+# Name of an Iguazio Continuous Data Platform container for storing the TSDB table
+# Example: "bigdata"
+container: "<container name>"
+
+# Logging verbosity level
+# Valid values: "debug" | "info" | "warn" | "error"
+# Default value: "debug"
+verbose: "warn"
+
+# Authentication credentials for the web-API service
+# Username
+username: "<username>"
+# Password
+password: "<password>"

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

@@ -0,0 +1,184 @@
+package performance
+
+import (
+	"fmt"
+	"github.com/pkg/errors"
+	"github.com/rcrowley/go-metrics"
+	"github.com/v3io/v3io-tsdb/pkg/config"
+	"io"
+	"log"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+)
+
+var instance *MetricReporter
+var once sync.Once
+
+const (
+	STDOUT = "stdout"
+	STDERR = "stderr"
+)
+
+type MetricReporter struct {
+	lock                  sync.Mutex
+	running               bool
+	registry              metrics.Registry
+	logWriter             io.Writer
+	reportPeriodically    bool
+	reportIntervalSeconds int
+	reportOnShutdown      bool
+}
+
+func DefaultReporterInstance() (reporter *MetricReporter, err error) {
+	cfg, err := config.GetOrDefaultConfig()
+
+	if err != nil {
+		// DO NOT return the error to prevent failures of unit tests
+		fmt.Fprintf(os.Stderr, "unable to load configuration. Reason: %v\n"+
+			"Will use default reporter configuration instead.", err)
+		reporter = ReporterInstance(STDOUT, true, 60, true)
+	} else {
+		reporter = ReporterInstanceFromConfig(cfg)
+	}
+
+	return reporter, nil
+}
+
+func ReporterInstance(writeTo string, reportPeriodically bool, reportIntervalSeconds int, reportOnShutdown bool) *MetricReporter {
+	once.Do(func() {
+		var writer io.Writer
+		switch writeTo {
+		case STDOUT:
+			writer = os.Stdout
+		case STDERR:
+			writer = os.Stderr
+		default:
+			writer = os.Stdout
+		}
+
+		instance = newMetricReporter(writer, reportPeriodically, reportIntervalSeconds, reportOnShutdown)
+	})
+	return instance
+}
+
+func ReporterInstanceFromConfig(config *config.V3ioConfig) *MetricReporter {
+	return ReporterInstance(
+		config.MetricsReporter.Output,
+		config.MetricsReporter.ReportPeriodically,
+		config.MetricsReporter.RepotInterval,
+		config.MetricsReporter.ReportOnShutdown)
+}
+
+func (mr *MetricReporter) Start() error {
+	mr.lock.Lock()
+	defer mr.lock.Unlock()
+
+	if !mr.running {
+		mr.running = true
+	} else {
+		return errors.Errorf("metric reporter is already running.")
+	}
+
+	return nil
+}
+
+func (mr *MetricReporter) Stop() error {
+	mr.lock.Lock()
+	defer mr.lock.Unlock()
+
+	if mr.running {
+		mr.running = false
+		if mr.reportOnShutdown {
+			time.Sleep(300 * time.Millisecond) // postpone performance report on shutdown to avoid mixing with other log messages
+			metrics.WriteOnce(mr.registry, mr.logWriter)
+		}
+		mr.registry.UnregisterAll()
+	} else {
+		return errors.Errorf("can't stop metric reporter since it's not running.")
+	}
+
+	return nil
+}
+
+func (mr *MetricReporter) GetTimer(name string) (metrics.Timer, error) {
+	if mr.running {
+		return metrics.GetOrRegisterTimer(name, mr.registry), nil
+	} else {
+		return nil, errors.Errorf("failed to create timer '%s'. Reason: metric reporter in not running", name)
+	}
+}
+
+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) 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) 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)
+	}
+}
+
+// Listen to the SIGINT and SIGTERM
+// SIGINT will listen to CTRL-C.
+// SIGTERM will be caught if kill command executed.
+func (mr *MetricReporter) registerShutdownHook() {
+	var gracefulStop = make(chan os.Signal)
+	// Register for specific signals
+	signal.Notify(gracefulStop, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		sig := <-gracefulStop
+		mr.logWriter.Write([]byte(fmt.Sprintf("\n**************************\ncaught sig: %+v\n**************************\n", sig)))
+		metrics.WriteOnce(mr.registry, mr.logWriter)
+	}()
+}
+
+func newMetricReporter(outputWriter io.Writer, reportPeriodically bool, reportIntervalSeconds int, reportOnShutdown bool) *MetricReporter {
+	var writer io.Writer
+
+	if outputWriter != nil {
+		writer = outputWriter
+	} else {
+		writer = os.Stderr
+	}
+
+	reporter := MetricReporter{
+		registry:              metrics.NewPrefixedRegistry("v3io-tsdb -> "),
+		logWriter:             writer,
+		running:               true,
+		reportPeriodically:    reportPeriodically,
+		reportIntervalSeconds: reportIntervalSeconds,
+		reportOnShutdown:      reportOnShutdown,
+	}
+
+	if reportPeriodically && reportIntervalSeconds > 0 {
+		// Log periodically
+		go metrics.Log(reporter.registry,
+			time.Duration(reportIntervalSeconds)*time.Second,
+			log.New(reporter.logWriter, "metrics: ", log.Lmicroseconds))
+	}
+
+	if reportOnShutdown {
+		reporter.registerShutdownHook()
+	}
+
+	return &reporter
+}