Merge pull request #235 from bojand/pace

Load and concurrency control

Author: bojand

.github/workflows/build.yaml

@@ -31,7 +31,7 @@ jobs:
     runs-on: ubuntu-latest
     env:
       GO111MODULE: on
-      GOLANGCI_LINT_VERSION: v1.27.0
+      GOLANGCI_LINT_VERSION: v1.31.0
 
     steps:
     - name: Set up Go

.golangci.yml

@@ -24,5 +24,17 @@ issues:
     - path: cmd/ghz/main.go
       text: "Error return value of `logger.Sync` is not checked"
 
+    # sync.once copy in pacer test
+    - path: load/pacer_test.go
+      text: "copylocks"
+
+    # TODO fix protobuf deprecated
+    - path: runner/
+      text: "SA1019: package github.com/golang/protobuf"
+  
+    # TODO fix protobuf deprecated
+    - path: protodesc/
+      text: "SA1019: package github.com/golang/protobuf"
+
     - path: runner/requester.go
       text: "SA1019: grpc.WithBalancerName"

Makefile

@@ -36,10 +36,6 @@ else
 	OPEN_COVERAGE_HTML :=
 endif
 
-#################################################
-##### Everything below should not be edited #####
-#################################################
-
 # UNAME_OS stores the value of uname -s.
 UNAME_OS := $(shell uname -s)
 # UNAME_ARCH stores the value of uname -m.
@@ -48,25 +44,16 @@ UNAME_ARCH := $(shell uname -m)
 # TMP_BASE is the base directory used for TMP.
 # Use TMP and not TMP_BASE as the temporary directory.
 TMP_BASE := .tmp
-# TMP is the temporary directory used.
-# This is based on UNAME_OS and UNAME_ARCH to make sure there are no issues
-# switching between platform builds.
-TMP := $(TMP_BASE)/$(UNAME_OS)/$(UNAME_ARCH)
-# TMP_BIN is where we install binaries for usage during development.
-TMP_BIN = $(TMP)/bin
 # TMP_COVERAGE is where we store code coverage files.
 TMP_COVERAGE := $(TMP_BASE)/coverage
 
 # Run all by default when "make" is invoked.
 .DEFAULT_GOAL := all
 
-# Install all the build and lint dependencies
-setup:
-	if [ ! -f $(GOPATH)/bin/tparse ]; then go get github.com/mfridman/tparse; fi;
-	if [ ! -f $(GOPATH)/bin/goimports ]; then go get golang.org/x/tools/cmd/goimports; fi;
-	if [ ! -f $(GOPATH)/bin/golangci-lint ]; then curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(GOPATH)/bin v1.23.8; fi;
-	go mod download
-.PHONY: setup
+# Tools
+.PHONY: tools
+tools:
+	go generate -tags tools tools/tools.go
 
 # All runs the default lint, test, and code coverage targets.
 .PHONY: all
@@ -88,11 +75,12 @@ lint:
 test:
 	go test $(GO_TEST_FLAGS) $(GO_PKGS)
 
-# gofmt and goimports all go files
+# Formats using gofmt and goimports all go files
 .PHONY: fmt
 fmt:
 	find . -name '*.go' | while read -r file; do gofmt -w -s "$$file"; goimports -w "$$file"; done
 
+# Build
 .PHONY: build
 build:
 	CGO_ENABLED=0 go build --ldflags="-s -w" -o $(DIST_DIR)/ghz ./cmd/ghz/...

README.md

@@ -70,16 +70,31 @@ Flags:
       --key=                   File containing client private key, to present to the server. Must also provide -cert option.
       --cname=                 Server name override when validating TLS certificate - useful for self signed certs.
       --skipTLS                Skip TLS client verification of the server's certificate chain and host name.
-      --skipFirst=0            Skip the first X requests from the timing calculations (useful for initial warmup)
       --insecure               Use plaintext and insecure connection.
       --authority=             Value to be used as the :authority pseudo-header. Only works if -insecure is used.
-  -c, --concurrency=50         Number of requests to run concurrently. Total number of requests cannot be smaller than the concurrency level. Default is 50.
+      --async                  Make requests asynchronous as soon as possible. Does not wait for request to finish before sending next one.
+  -r, --rps=0                  Requests per second (RPS) rate limit for constant load schedule. Default is no rate limit.
+      --load-schedule="const"  Specifies the load schedule. Options are const, step, or line. Default is const.
+      --load-start=0           Specifies the RPS load start value for step or line schedules.
+      --load-step=0            Specifies the load step value or slope value.
+      --load-end=0             Specifies the load end value for step or line load schedules.
+      --load-step-duration=0   Specifies the load step duration value for step load schedule.
+      --load-max-duration=0    Specifies the max load duration value for step or line load schedule.
+  -c, --concurrency=50         Number of request workers to run concurrently for const concurrency schedule. Default is 50.
+      --concurrency-schedule="const"
+                               Concurrency change schedule. Options are const, step, or line. Default is const.
+      --concurrency-start=0    Concurrency start value for step and line concurrency schedules.
+      --concurrency-end=0      Concurrency end value for step and line concurrency schedules.
+      --concurrency-step=1     Concurrency step / slope value for step and line concurrency schedules.
+      --concurrency-step-duration=0
+                               Specifies the concurrency step duration value for step concurrency schedule.
+      --concurrency-max-duration=0
+                               Specifies the max concurrency adjustment duration value for step or line concurrency schedule.
   -n, --total=200              Number of requests to run. Default is 200.
-  -q, --qps=0                  Rate limit, in queries per second (QPS). Default is no rate limit.
   -t, --timeout=20s            Timeout for each request. Default is 20s, use 0 for infinite.
   -z, --duration=0             Duration of application to send requests. When duration is reached, application stops and exits. If duration is specified, n is ignored. Examples: -z 10s -z 3m.
   -x, --max-duration=0         Maximum duration of application to send requests with n setting respected. If duration is reached before n requests are completed, application stops and exits. Examples: -x 10s -x 3m.
-      --duration-stop="close"  Specifies how duration stop is reported. Options are close, wait or ignore.
+      --duration-stop="close"  Specifies how duration stop is reported. Options are close, wait or ignore. Default is close.
   -d, --data=                  The call data as stringified JSON. If the value is '@' then the request contents are read from stdin.
   -D, --data-file=             File path for call data JSON file. Examples: /home/user/file.json or ./file.json.
   -b, --binary                 The call data comes as serialized binary message or multiple count-prefixed messages read from stdin.
@@ -90,6 +105,7 @@ Flags:
       --reflect-metadata=      Reflect metadata as stringified JSON used only for reflection request.
   -o, --output=                Output path. If none provided stdout is used.
   -O, --format=                Output format. One of: summary, csv, json, pretty, html, influx-summary, influx-details. Default is summary.
+      --skipFirst=0            Skip the first X requests when doing the results tally.
       --connections=1          Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.
       --connect-timeout=10s    Connection timeout for the initial connection dial. Default is 10s.
       --keepalive=0            Keepalive time duration. Only used if present and above 0.

cmd/ghz/main.go

@@ -64,10 +64,6 @@ var (
 	skipVerify = kingpin.Flag("skipTLS", "Skip TLS client verification of the server's certificate chain and host name.").
 			Default("false").IsSetByUser(&isSkipSet).Bool()
 
-	isSkipFirstSet = false
-	skipFirst      = kingpin.Flag("skipFirst", "Skip the first X requests when doing the results tally.").
-			Default("0").IsSetByUser(&isSkipFirstSet).Uint()
-
 	isInsecSet = false
 	insecure   = kingpin.Flag("insecure", "Use plaintext and insecure connection.").
 			Default("false").IsSetByUser(&isInsecSet).Bool()
@@ -77,18 +73,72 @@ var (
 			PlaceHolder(" ").IsSetByUser(&isAuthSet).String()
 
 	// Run
+	isAsyncSet = false
+	async      = kingpin.Flag("async", "Make requests asynchronous as soon as possible. Does not wait for request to finish before sending next one.").
+			Default("false").IsSetByUser(&isAsyncSet).Bool()
+
+	isRPSSet = false
+	rps      = kingpin.Flag("rps", "Requests per second (RPS) rate limit for constant load schedule. Default is no rate limit.").
+			Default("0").Short('r').IsSetByUser(&isRPSSet).Uint()
+
+	isScheduleSet = false
+	schedule      = kingpin.Flag("load-schedule", "Specifies the load schedule. Options are const, step, or line. Default is const.").
+			Default("const").IsSetByUser(&isScheduleSet).String()
+
+	isLoadStartSet = false
+	loadStart      = kingpin.Flag("load-start", "Specifies the RPS load start value for step or line schedules.").
+			Default("0").IsSetByUser(&isLoadStartSet).Uint()
+
+	isLoadStepSet = false
+	loadStep      = kingpin.Flag("load-step", "Specifies the load step value or slope value.").
+			Default("0").IsSetByUser(&isLoadStepSet).Int()
+
+	isLoadEndSet = false
+	loadEnd      = kingpin.Flag("load-end", "Specifies the load end value for step or line load schedules.").
+			Default("0").IsSetByUser(&isLoadEndSet).Uint()
+
+	isLoadStepDurSet = false
+	loadStepDuration = kingpin.Flag("load-step-duration", "Specifies the load step duration value for step load schedule.").
+				Default("0").IsSetByUser(&isLoadStepDurSet).Duration()
+
+	isLoadMaxDurSet = false
+	loadMaxDuration = kingpin.Flag("load-max-duration", "Specifies the max load duration value for step or line load schedule.").
+			Default("0").IsSetByUser(&isLoadMaxDurSet).Duration()
+
+	// Concurrency
 	isCSet = false
-	c      = kingpin.Flag("concurrency", "Number of requests to run concurrently. Total number of requests cannot be smaller than the concurrency level. Default is 50.").
+	c      = kingpin.Flag("concurrency", "Number of request workers to run concurrently for const concurrency schedule. Default is 50.").
 		Short('c').Default("50").IsSetByUser(&isCSet).Uint()
 
+	isCScheduleSet = false
+	cschdule       = kingpin.Flag("concurrency-schedule", "Concurrency change schedule. Options are const, step, or line. Default is const.").
+			Default("const").IsSetByUser(&isCScheduleSet).String()
+
+	isCStartSet = false
+	cStart      = kingpin.Flag("concurrency-start", "Concurrency start value for step and line concurrency schedules.").
+			Default("0").IsSetByUser(&isCStartSet).Uint()
+
+	isCEndSet = false
+	cEnd      = kingpin.Flag("concurrency-end", "Concurrency end value for step and line concurrency schedules.").
+			Default("0").IsSetByUser(&isCEndSet).Uint()
+
+	isCStepSet = false
+	cstep      = kingpin.Flag("concurrency-step", "Concurrency step / slope value for step and line concurrency schedules.").
+			Default("1").IsSetByUser(&isCStepSet).Int()
+
+	isCStepDurSet = false
+	cStepDuration = kingpin.Flag("concurrency-step-duration", "Specifies the concurrency step duration value for step concurrency schedule.").
+			Default("0").IsSetByUser(&isCStepDurSet).Duration()
+
+	isCMaxDurSet = false
+	cMaxDuration = kingpin.Flag("concurrency-max-duration", "Specifies the max concurrency adjustment duration value for step or line concurrency schedule.").
+			Default("0").IsSetByUser(&isCMaxDurSet).Duration()
+
+	// Other
 	isNSet = false
 	n      = kingpin.Flag("total", "Number of requests to run. Default is 200.").
 		Short('n').Default("200").IsSetByUser(&isNSet).Uint()
 
-	isQSet = false
-	q      = kingpin.Flag("qps", "Rate limit, in queries per second (QPS). Default is no rate limit.").
-		Default("0").Short('q').IsSetByUser(&isQSet).Uint()
-
 	isTSet = false
 	t      = kingpin.Flag("timeout", "Timeout for each request. Default is 20s, use 0 for infinite.").
 		Default("20s").Short('t').IsSetByUser(&isTSet).Duration()
@@ -102,7 +152,7 @@ var (
 		Short('x').Default("0").IsSetByUser(&isXSet).Duration()
 
 	isZStopSet = false
-	zstop      = kingpin.Flag("duration-stop", "Specifies how duration stop is reported. Options are close, wait or ignore.").
+	zstop      = kingpin.Flag("duration-stop", "Specifies how duration stop is reported. Options are close, wait or ignore. Default is close.").
 			Default("close").IsSetByUser(&isZStopSet).String()
 
 	// Data
@@ -147,6 +197,10 @@ var (
 	format      = kingpin.Flag("format", "Output format. One of: summary, csv, json, pretty, html, influx-summary, influx-details. Default is summary.").
 			Short('O').Default("summary").PlaceHolder(" ").IsSetByUser(&isFormatSet).Enum("summary", "csv", "json", "pretty", "html", "influx-summary", "influx-details")
 
+	isSkipFirstSet = false
+	skipFirst      = kingpin.Flag("skipFirst", "Skip the first X requests when doing the results tally.").
+			Default("0").IsSetByUser(&isSkipFirstSet).Uint()
+
 	// Connection
 	isConnSet = false
 	conns     = kingpin.Flag("connections", "Number of connections to use. Concurrency is distributed evenly among all the connections. Default is 1.").
@@ -360,7 +414,7 @@ func createConfigFromArgs(cfg *runner.Config) error {
 	cfg.CName = *cname
 	cfg.N = *n
 	cfg.C = *c
-	cfg.QPS = *q
+	cfg.RPS = *rps
 	cfg.Z = runner.Duration(*z)
 	cfg.X = runner.Duration(*x)
 	cfg.Timeout = runner.Duration(*t)
@@ -384,6 +438,19 @@ func createConfigFromArgs(cfg *runner.Config) error {
 	cfg.ReflectMetadata = rmdMap
 	cfg.Debug = *debug
 	cfg.EnableCompression = *enableCompression
+	cfg.LoadSchedule = *schedule
+	cfg.LoadStart = *loadStart
+	cfg.LoadStep = *loadStep
+	cfg.LoadEnd = *loadEnd
+	cfg.LoadStepDuration = runner.Duration(*loadStepDuration)
+	cfg.LoadMaxDuration = runner.Duration(*loadMaxDuration)
+	cfg.Async = *async
+	cfg.CSchedule = *cschdule
+	cfg.CStart = *cStart
+	cfg.CStep = *cstep
+	cfg.CEnd = *cEnd
+	cfg.CStepDuration = runner.Duration(*cStepDuration)
+	cfg.CMaxDuration = runner.Duration(*cMaxDuration)
 
 	return nil
 }
@@ -393,6 +460,8 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		return errors.New("config cannot be nil")
 	}
 
+	// proto
+
 	if isProtoSet {
 		dest.Proto = src.Proto
 	}
@@ -405,6 +474,8 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.Call = src.Call
 	}
 
+	// security
+
 	if isCACertSet {
 		dest.RootCert = src.RootCert
 	}
@@ -437,18 +508,12 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.CName = src.CName
 	}
 
+	// run
+
 	if isNSet {
 		dest.N = src.N
 	}
 
-	if isCSet {
-		dest.C = src.C
-	}
-
-	if isQSet {
-		dest.QPS = src.QPS
-	}
-
 	if isZSet {
 		dest.Z = src.Z
 	}
@@ -465,6 +530,8 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.ZStop = src.ZStop
 	}
 
+	// data
+
 	if isDataSet {
 		dest.Data = src.Data
 	}
@@ -489,6 +556,8 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.MetadataPath = src.MetadataPath
 	}
 
+	// other
+
 	if isSISet {
 		dest.SI = src.SI
 	}
@@ -541,6 +610,70 @@ func mergeConfig(dest *runner.Config, src *runner.Config) error {
 		dest.Host = src.Host
 	}
 
+	// load
+
+	if isAsyncSet {
+		dest.Async = src.Async
+	}
+
+	if isRPSSet {
+		dest.RPS = src.RPS
+	}
+
+	if isScheduleSet {
+		dest.LoadSchedule = src.LoadSchedule
+	}
+
+	if isLoadStartSet {
+		dest.LoadStart = src.LoadStart
+	}
+
+	if isLoadStepSet {
+		dest.LoadStep = src.LoadStep
+	}
+
+	if isLoadEndSet {
+		dest.LoadEnd = src.LoadEnd
+	}
+
+	if isLoadStepDurSet {
+		dest.LoadStepDuration = src.LoadStepDuration
+	}
+
+	if isLoadMaxDurSet {
+		dest.LoadMaxDuration = src.LoadMaxDuration
+	}
+
+	// concurrency
+
+	if isCSet {
+		dest.C = src.C
+	}
+
+	if isCScheduleSet {
+		dest.CSchedule = src.CSchedule
+	}
+
+	if isCStartSet {
+		dest.CStart = src.CStart
+	}
+
+	if isCStepSet {
+		dest.CStep = src.CStep
+	}
+
+	if isCEndSet {
+		dest.CEnd = src.CEnd
+	}
+
+	if isCStepDurSet {
+		dest.CStepDuration = src.CStepDuration
+	}
+
+	if isCMaxDurSet {
+		dest.CMaxDuration = src.CMaxDuration
+	}
+
 	return nil
 }