Enable persisting ingestion/query benchmark results in a common format across targets (timescale#147)

resultset: removing intermedia non-work chars from JSON label
Co-authored-by: Ante Kresic <[email protected]>

Author: filipecosta90

load/loader.go

@@ -1,8 +1,10 @@
 package load
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/timescale/tsbs/pkg/targets"
+	"io/ioutil"
 	"log"
 	"math/rand"
 	"sync"
@@ -29,21 +31,22 @@ var (
 
 // BenchmarkRunnerConfig contains all the configuration information required for running BenchmarkRunner.
 type BenchmarkRunnerConfig struct {
-	DBName          string        `yaml:"db-name" mapstructure:"db-name"`
-	BatchSize       uint          `yaml:"batch-size" mapstructure:"batch-size"`
-	Workers         uint          `yaml:"workers" mapstructure:"workers"`
-	Limit           uint64        `yaml:"limit" mapstructure:"limit"`
-	DoLoad          bool          `yaml:"do-load" mapstructure:"do-load"`
-	DoCreateDB      bool          `yaml:"do-create-db" mapstructure:"do-create-db"`
-	DoAbortOnExist  bool          `yaml:"do-abort-on-exist" mapstructure:"do-abort-on-exist"`
-	ReportingPeriod time.Duration `yaml:"reporting-period" mapstructure:"reporting-period"`
-	HashWorkers     bool          `yaml:"hash-workers" mapstructure:"hash-workers"`
-	NoFlowControl   bool          `yaml:"no-flow-control" mapstructure:"no-flow-control"`
-	ChannelCapacity uint          `yaml:"channel-capacity" mapstructure:"channel-capacity"`
-	InsertIntervals string        `yaml:"insert-intervals" mapstructure:"insert-intervals"`
+	DBName          string        `yaml:"db-name" mapstructure:"db-name" json:"db-name"`
+	BatchSize       uint          `yaml:"batch-size" mapstructure:"batch-size" json:"batch-size"`
+	Workers         uint          `yaml:"workers" mapstructure:"workers" json:"workers"`
+	Limit           uint64        `yaml:"limit" mapstructure:"limit" json:"limit"`
+	DoLoad          bool          `yaml:"do-load" mapstructure:"do-load" json:"do-load"`
+	DoCreateDB      bool          `yaml:"do-create-db" mapstructure:"do-create-db" json:"do-create-db"`
+	DoAbortOnExist  bool          `yaml:"do-abort-on-exist" mapstructure:"do-abort-on-exist" json:"do-abort-on-exist"`
+	ReportingPeriod time.Duration `yaml:"reporting-period" mapstructure:"reporting-period" json:"reporting-period"`
+	HashWorkers     bool          `yaml:"hash-workers" mapstructure:"hash-workers" json:"hash-workers"`
+	NoFlowControl   bool          `yaml:"no-flow-control" mapstructure:"no-flow-control" json:"no-flow-control"`
+	ChannelCapacity uint          `yaml:"channel-capacity" mapstructure:"channel-capacity" json:"channel-capacity"`
+	InsertIntervals string        `yaml:"insert-intervals" mapstructure:"insert-intervals" json:"insert-intervals"`
+	ResultsFile     string        `yaml:"results-file" mapstructure:"results-file" json:"results-file"`
 	// deprecated, should not be used in other places other than tsbs_load_xx commands
-	FileName string `yaml:"file" mapstructure:"file"`
-	Seed     int64  `yaml:"seed" mapstructure:"seed"`
+	FileName string `yaml:"file" mapstructure:"file" json:"file"`
+	Seed     int64  `yaml:"seed" mapstructure:"seed" json:"seed"`
 }
 
 // AddToFlagSet adds command line flags needed by the BenchmarkRunnerConfig to the flag set.
@@ -60,6 +63,7 @@ func (c BenchmarkRunnerConfig) AddToFlagSet(fs *pflag.FlagSet) {
 	fs.Int64("seed", 0, "PRNG seed (default: 0, which uses the current timestamp)")
 	fs.String("insert-intervals", "", "Time to wait between each insert, default '' => all workers insert ASAP. '1,2' = worker 1 waits 1s between inserts, worker 2 and others wait 2s")
 	fs.Bool("hash-workers", false, "Whether to consistently hash insert data to the same workers (i.e., the data for a particular host always goes to the same worker)")
+	fs.String("results-file", "", "Write the test results summary json to this file")
 }
 
 type BenchmarkRunner interface {
@@ -138,7 +142,41 @@ func (l *CommonBenchmarkRunner) postRun(wg *sync.WaitGroup, start *time.Time) {
 	// Wait for all workers to finish
 	wg.Wait()
 	end := time.Now()
-	l.summary(end.Sub(*start))
+	took := end.Sub(*start)
+	l.summary(took)
+	if l.BenchmarkRunnerConfig.ResultsFile != "" {
+		metricRate := float64(l.metricCnt) / took.Seconds()
+		rowRate := float64(l.rowCnt) / took.Seconds()
+		l.saveTestResult(took, *start, end, metricRate, rowRate)
+	}
+}
+
+func (l *CommonBenchmarkRunner) saveTestResult(took time.Duration, start time.Time, end time.Time, metricRate, rowRate float64) {
+	totals := make(map[string]interface{})
+	totals["metricRate"] = metricRate
+	if l.rowCnt > 0 {
+		totals["rowRate"] = rowRate
+	}
+
+	testResult := LoaderTestResult{
+		ResultFormatVersion: LoaderTestResultVersion,
+		RunnerConfig:        l.BenchmarkRunnerConfig,
+		StartTime:           start.Unix(),
+		EndTime:             end.Unix(),
+		DurationMillis:      took.Milliseconds(),
+		Totals:              totals,
+	}
+
+	_, _ = fmt.Printf("Saving results json file to %s\n", l.BenchmarkRunnerConfig.ResultsFile)
+	file, err := json.MarshalIndent(testResult, "", " ")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = ioutil.WriteFile(l.BenchmarkRunnerConfig.ResultsFile, file, 0644)
+	if err != nil {
+		log.Fatal(err)
+	}
 }
 
 // RunBenchmark takes in a Benchmark b and uses it to run the load benchmark

load/loader_test_result.go

@@ -0,0 +1,20 @@
+package load
+
+const LoaderTestResultVersion = "0.1"
+
+// LoaderTestResult aggregates the results of an insert or load benchmark in a common format across targets
+type LoaderTestResult struct {
+	// Format Configs
+	ResultFormatVersion string `json:"ResultFormatVersion"`
+
+	// RunnerConfig Configs
+	RunnerConfig BenchmarkRunnerConfig `json:"RunnerConfig"`
+
+	// Run info
+	StartTime      int64 `json:"StartTime`
+	EndTime        int64 `json:"EndTime"`
+	DurationMillis int64 `json:"DurationMillis"`
+
+	// Totals
+	Totals map[string]interface{} `json:"Totals"`
+}

pkg/query/benchmark_result.go

@@ -0,0 +1,20 @@
+package query
+
+const BenchmarkTestResultVersion = "0.1"
+
+// LoaderTestResult aggregates the results of an query benchmark in a common format across targets
+type LoaderTestResult struct {
+	// Format Configs
+	ResultFormatVersion string `json:"ResultFormatVersion"`
+
+	// RunnerConfig Configs
+	RunnerConfig BenchmarkRunnerConfig `json:"RunnerConfig"`
+
+	// Run info
+	StartTime      int64 `json:"StartTime`
+	EndTime        int64 `json:"EndTime"`
+	DurationMillis int64 `json:"DurationMillis"`
+
+	// Totals
+	Totals map[string]interface{} `json:"Totals"`
+}

pkg/query/benchmarker.go

@@ -2,7 +2,9 @@ package query
 
 import (
 	"bufio"
+	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"log"
 	"os"
 	"runtime/pprof"
@@ -35,6 +37,7 @@ type BenchmarkRunnerConfig struct {
 	BurnIn           uint64 `mapstructure:"burn-in"`
 	PrintInterval    uint64 `mapstructure:"print-interval"`
 	PrewarmQueries   bool   `mapstructure:"prewarm-queries"`
+	ResultsFile      string `mapstructure:"results-file"`
 }
 
 // AddToFlagSet adds command line flags needed by the BenchmarkRunnerConfig to the flag set.
@@ -51,6 +54,7 @@ func (c BenchmarkRunnerConfig) AddToFlagSet(fs *pflag.FlagSet) {
 	fs.Bool("print-responses", false, "Pretty print response bodies for correctness checking (default false).")
 	fs.Int("debug", 0, "Whether to print debug messages.")
 	fs.String("file", "", "File name to read queries from")
+	fs.String("results-file", "", "Write the test results summary json to this file")
 }
 
 // BenchmarkRunner contains the common components for running a query benchmarking
@@ -183,6 +187,33 @@ func (b *BenchmarkRunner) Run(queryPool *sync.Pool, processorCreateFn ProcessorC
 		pprof.WriteHeapProfile(f)
 		f.Close()
 	}
+
+	// (Optional) save the results file:
+	if len(b.BenchmarkRunnerConfig.ResultsFile) > 0 {
+		b.saveTestResult(wallTook, wallStart, wallEnd)
+	}
+}
+
+func (b *BenchmarkRunner) saveTestResult(took time.Duration, start time.Time, end time.Time) {
+	testResult := LoaderTestResult{
+		ResultFormatVersion: BenchmarkTestResultVersion,
+		RunnerConfig:        b.BenchmarkRunnerConfig,
+		StartTime:           start.UTC().Unix() * 1000,
+		EndTime:             end.UTC().Unix() * 1000,
+		DurationMillis:      took.Milliseconds(),
+		Totals:              b.sp.GetTotalsMap(),
+	}
+
+	_, _ = fmt.Printf("Saving results json file to %s\n", b.BenchmarkRunnerConfig.ResultsFile)
+	file, err := json.MarshalIndent(testResult, "", " ")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = ioutil.WriteFile(b.BenchmarkRunnerConfig.ResultsFile, file, 0644)
+	if err != nil {
+		log.Fatal(err)
+	}
 }
 
 func (b *BenchmarkRunner) processorHandler(wg *sync.WaitGroup, rateLimiter *rate.Limiter, queryPool *sync.Pool, processor Processor, workerNum int) {

pkg/query/benchmarker_test.go

@@ -323,6 +323,10 @@ func (m *mockStatProcessor) CloseAndWait() {
 	m.closed = true
 	m.wg.Done()
 }
+func (m *mockStatProcessor) GetTotalsMap() map[string]interface{} {
+	totals := make(map[string]interface{})
+	return totals
+}
 
 type mockProcessor struct {
 	processRes []*Stat