bench: Add benchmark functionality for single node Tarantool topology

Author: Kirill-Churkin

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 
+- Tarantool benchmark tool.
 - Ability to reverse search in ``cartridge enter`` and ``cartridge connect`` commands.
 - Added support for functionality from Golang 1.17.
 

cli/bench/bench.go

@@ -0,0 +1,140 @@
+package bench
+
+import (
+	bctx "context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/common"
+	"github.com/tarantool/cartridge-cli/cli/context"
+)
+
+// Output benchmark foramatted results.
+func printResults(results Results) {
+	fmt.Printf("\nResults:\n")
+	fmt.Printf("\tSuccess operations: %d\n", results.successResultCount)
+	fmt.Printf("\tFailed  operations: %d\n", results.failedResultCount)
+	fmt.Printf("\tRequest count: %d\n", results.handledRequestsCount)
+	fmt.Printf("\tTime (seconds): %f\n", results.duration)
+	fmt.Printf("\tRequests per second: %d\n\n", results.requestsPerSecond)
+}
+
+// Delete and create specified space for benchmark.
+func spacePreset(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection) error {
+	dropSpace(ctx, tarantoolConnection)
+	return createSpace(tarantoolConnection)
+}
+
+// Increases the counter of successful/failed requests depending on the presence of an error.
+func incrementRequest(err error, results *Results) {
+	if err == nil {
+		results.successResultCount++
+	} else {
+		results.failedResultCount++
+	}
+	results.handledRequestsCount++
+}
+
+// Сontinuously executes the specified query until the benchmark time runs out.
+func requestsLoop(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection, results *Results, backgroundCtx bctx.Context) {
+	for {
+		select {
+		case <-backgroundCtx.Done():
+			return
+		default:
+			_, err := tarantoolConnection.Exec(
+				tarantool.Insert(
+					benchSpaceName,
+					[]interface{}{common.RandomString(ctx.KeySize), common.RandomString(ctx.DataSize)}))
+			incrementRequest(err, results)
+		}
+	}
+}
+
+// Represents a separate connection with a certain number of parallel requests.
+func connectionLoop(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection, results *Results, backgroundCtx bctx.Context) {
+	var connectionWait sync.WaitGroup
+	for i := 0; i < ctx.SimultaneousRequests; i++ {
+		connectionWait.Add(1)
+		go func() {
+			defer connectionWait.Done()
+			requestsLoop(ctx, tarantoolConnection, results, backgroundCtx)
+		}()
+	}
+
+	connectionWait.Wait()
+}
+
+// Main benchmark function.
+func Run(ctx context.BenchCtx) error {
+	rand.Seed(time.Now().UnixNano())
+
+	// Connect to tarantool and preset space for benchmark.
+	tarantoolConnection, err := tarantool.Connect(ctx.URL, tarantool.Opts{
+		User:     ctx.User,
+		Password: ctx.Password,
+	})
+	if err != nil {
+		return fmt.Errorf(
+			"Couldn't connect to Tarantool %s. Did you forget to enter tarantool -e 'box.cfg ({listen = 3301})'?",
+			ctx.URL)
+	}
+	defer tarantoolConnection.Close()
+
+	printConfig(ctx, tarantoolConnection)
+
+	if err := spacePreset(ctx, tarantoolConnection); err != nil {
+		return err
+	}
+
+	// Create connections before detaching.
+	connectionPool := make([]*tarantool.Connection, ctx.Connections)
+	for i := 0; i < ctx.Connections; i++ {
+		connectionPool[i], err = tarantool.Connect(ctx.URL, tarantool.Opts{
+			User:     ctx.User,
+			Password: ctx.Password,
+		})
+		if err != nil {
+			return err
+		}
+		defer connectionPool[i].Close()
+	}
+
+	fmt.Println("Benchmark start")
+	fmt.Println("...")
+
+	backgroundCtx, cancel := bctx.WithCancel(bctx.Background())
+	var waitGroup sync.WaitGroup
+	results := Results{}
+	connectionId := 0
+
+	startTime := time.Now()
+	timer := time.NewTimer(time.Duration(ctx.Duration * int(time.Second)))
+
+	// Start detached connections.
+	for i := 0; i < ctx.Connections; i++ {
+		waitGroup.Add(1)
+		connection := connectionPool[connectionId]
+		go func() {
+			defer waitGroup.Done()
+			connectionLoop(ctx, connection, &results, backgroundCtx)
+		}()
+		connectionId++
+	}
+	// Waiting for the time runs out.
+	<-timer.C
+	cancel()
+	waitGroup.Wait()
+
+	results.duration = time.Since(startTime).Seconds()
+	results.requestsPerSecond = int(float64(results.handledRequestsCount) / results.duration)
+
+	dropSpace(ctx, tarantoolConnection)
+	fmt.Println("Benchmark stop")
+
+	printResults(results)
+	return nil
+}

cli/bench/config.go

@@ -0,0 +1,34 @@
+package bench
+
+import (
+	"fmt"
+	"os"
+	"text/tabwriter"
+
+	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/context"
+)
+
+var (
+	benchSpaceName             = "__benchmark_space__"
+	benchSpacePrimaryIndexName = "__bench_primary_key__"
+)
+
+// Output formatted config parameters.
+func printConfig(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection) {
+	fmt.Printf("%s\n", tarantoolConnection.Greeting().Version)
+	fmt.Printf("Parameters:\n")
+	fmt.Printf("\tURL: %s\n", ctx.URL)
+	fmt.Printf("\tuser: %s\n", ctx.User)
+	fmt.Printf("\tconnections: %d\n", ctx.Connections)
+	fmt.Printf("\tsimultaneous requests: %d\n", ctx.SimultaneousRequests)
+	fmt.Printf("\tduration: %d seconds\n", ctx.Duration)
+	fmt.Printf("\tkey size: %d bytes\n", ctx.KeySize)
+	fmt.Printf("\tdata size: %d bytes\n", ctx.DataSize)
+	fmt.Printf("Data schema\n")
+	w := tabwriter.NewWriter(os.Stdout, 1, 1, 1, ' ', 0)
+	fmt.Fprintf(w, "|\tkey\t|\tvalue\n")
+	fmt.Fprintf(w, "|\t------------------------------\t|\t------------------------------\n")
+	fmt.Fprintf(w, "|\trandom(%d)\t|\trandom(%d)\n", ctx.KeySize, ctx.DataSize)
+	w.Flush()
+}

cli/bench/space.go

@@ -0,0 +1,57 @@
+package bench
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/context"
+)
+
+// Create specified benchmark space with formatting and primary index.
+func createSpace(tarantoolConnection *tarantool.Connection) error {
+	// Creationg space.
+	createCommand := "return box.schema.space.create(...).name"
+	_, err := tarantoolConnection.Exec(tarantool.Eval(createCommand, []interface{}{benchSpaceName, map[string]bool{"if_not_exists": true}}))
+	if err != nil {
+		return err
+	}
+
+	// Formatting space.
+	formatCommand := fmt.Sprintf("box.space.%s:format", benchSpaceName)
+	_, err = tarantoolConnection.Exec(tarantool.Call(formatCommand, [][]map[string]string{
+		{
+			{"name": "key", "type": "string"},
+			{"name": "value", "type": "string"},
+		},
+	}))
+	if err != nil {
+		return err
+	}
+
+	// Create primary index.
+	createIndexCommand := fmt.Sprintf("box.space.%s:create_index", benchSpaceName)
+	_, err = tarantoolConnection.Exec(tarantool.Call(createIndexCommand, []interface{}{
+		benchSpacePrimaryIndexName,
+		map[string]interface{}{
+			"parts":         []string{"key"},
+			"if_not_exists": true,
+		},
+	}))
+	return err
+}
+
+// Delete specified benchmark space.
+func dropSpace(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection) error {
+	checkCommand := fmt.Sprintf("return box.space.%s.index[0].name", benchSpaceName)
+	indexName, err := tarantoolConnection.Exec(tarantool.Eval(checkCommand, []interface{}{}))
+	if err != nil {
+		return err
+	}
+	if reflect.ValueOf(indexName.Data).Index(0).Elem().String() == benchSpacePrimaryIndexName {
+		dropCommand := fmt.Sprintf("box.space.%s:drop", benchSpaceName)
+		_, err := tarantoolConnection.Exec(tarantool.Call(dropCommand, []interface{}{}))
+		return err
+	}
+	return nil
+}

cli/bench/types.go

@@ -0,0 +1,10 @@
+package bench
+
+// Describes set of benchmark results.
+type Results struct {
+	handledRequestsCount int     // count of all executed requests.
+	successResultCount   int     // count of successful request in all connections.
+	failedResultCount    int     // count of failed request in all connections.
+	duration             float64 // benchmark duration.
+	requestsPerSecond    int     // number of requests per second - the main measured value.
+}

cli/commands/bench.go

@@ -0,0 +1,33 @@
+package commands
+
+import (
+	"github.com/apex/log"
+	"github.com/spf13/cobra"
+	"github.com/tarantool/cartridge-cli/cli/bench"
+)
+
+func init() {
+	var benchCmd = &cobra.Command{
+		Use:   "bench",
+		Short: "Util for running benchmarks for Tarantool",
+		Long:  "Benchmark utility that simulates running commands done by N clients at the same time sending M simultaneous queries",
+		Run: func(cmd *cobra.Command, args []string) {
+			if err := bench.Run(ctx.Bench); err != nil {
+				log.Fatalf(err.Error())
+			}
+		},
+	}
+	rootCmd.AddCommand(benchCmd)
+
+	configureFlags(benchCmd)
+
+	benchCmd.Flags().StringVar(&ctx.Bench.URL, "url", "127.0.0.1:3301", "Tarantool address")
+	benchCmd.Flags().StringVar(&ctx.Bench.User, "user", "guest", "Tarantool user for connection")
+	benchCmd.Flags().StringVar(&ctx.Bench.Password, "password", "", "Tarantool password for connection")
+
+	benchCmd.Flags().IntVar(&ctx.Bench.Connections, "connections", 10, "Number of concurrent connections")
+	benchCmd.Flags().IntVar(&ctx.Bench.SimultaneousRequests, "requests", 10, "Number of simultaneous requests per connection")
+	benchCmd.Flags().IntVar(&ctx.Bench.Duration, "duration", 10, "Duration of benchmark test (seconds)")
+	benchCmd.Flags().IntVar(&ctx.Bench.KeySize, "keysize", 10, "Size of key part of benchmark data (bytes)")
+	benchCmd.Flags().IntVar(&ctx.Bench.DataSize, "datasize", 20, "Size of value part of benchmark data (bytes)")
+}

cli/context/context.go

@@ -21,6 +21,7 @@ type Ctx struct {
 	Replicasets ReplicasetsCtx
 	Connect     ConnectCtx
 	Failover    FailoverCtx
+	Bench       BenchCtx
 }
 
 type ProjectCtx struct {
@@ -173,3 +174,14 @@ type FailoverCtx struct {
 	ParamsJSON         string
 	ProviderParamsJSON string
 }
+
+type BenchCtx struct {
+	URL                  string
+	User                 string
+	Password             string
+	Connections          int
+	SimultaneousRequests int
+	Duration             int
+	KeySize              int
+	DataSize             int
+}

test/integration/bench/test_bench.py

@@ -0,0 +1,40 @@
+import os
+import signal
+import time
+from socket import AF_INET, SOCK_STREAM, socket
+from subprocess import PIPE, STDOUT, Popen
+from threading import Thread
+
+from utils import consume_lines, run_command_and_get_output
+
+
+def test_bench(cartridge_cmd, request, tmpdir):
+    base_cmd = [cartridge_cmd, 'bench', '--duration=1']
+    tarantool_cmd = [
+        "tarantool",
+        "-e", f"""box.cfg{{listen="127.0.0.1:3301",work_dir=[[{tmpdir}]]}}""",
+        "-e", """box.schema.user.grant("guest","super",nil,nil,{if_not_exists=true})"""
+    ]
+
+    env = os.environ.copy()
+    process = Popen(tarantool_cmd, stdout=PIPE, stderr=STDOUT, env=env)
+    thread = Thread(target=consume_lines, args=["3301", process.stdout])
+    thread.start()
+
+    def kill():
+        process.send_signal(signal.SIGKILL)
+        if thread is not None:
+            thread.join(5)
+    request.addfinalizer(kill)
+
+    tnt_socket = socket(AF_INET, SOCK_STREAM)
+    attempt = 10
+    while attempt > 0:
+        try:
+            time.sleep(1)
+            tnt_socket.connect(('127.0.0.1', 3301))
+        except Exception:
+            attempt -= 1
+
+    rc, output = run_command_and_get_output(base_cmd, cwd=tmpdir)
+    assert rc == 0

test/utils.py

@@ -1,6 +1,7 @@
 import glob
 import gzip
 import json
+import logging
 import os
 import re
 import shutil
@@ -1394,3 +1395,10 @@ def get_response_data(response):
     assert 'errors' not in response_json, response_json
 
     return response_json['data']
+
+
+def consume_lines(port, pipe):
+    logger = logging.getLogger(f'localhost:{port}')
+    with pipe:
+        for line in iter(pipe.readline, b''):
+            logger.warning(line.rstrip().decode('utf-8'))