Add the -requestLengthPercentiles command-line flag (#59)

The strest-grpc client currently only sends very small request payloads,
including only the response spec. This change adds an arbitrary payload
to strest-grpc requests (that are ignored by the server) so that strest
clients can generate request payloads.

client/client.go

@@ -264,9 +264,11 @@ func logFinalReport(good, bad, bytes uint, latencies *hdrhistogram.Histogram, ji
 func safeNonStreamingRequest(worker workerID,
 	client pb.ResponderClient,
 	clientTimeout time.Duration,
-	lengthDistribution distribution.Distribution,
+	requestLengthDistribution,
+	lengthDistribution,
 	latencyDistribution distribution.Distribution,
-	errorRate float32, r *rand.Rand,
+	errorRate float32,
+	r *rand.Rand,
 	responses chan<- *MeasuredResponse) {
 	start := time.Now()
 
@@ -279,10 +281,16 @@ func safeNonStreamingRequest(worker workerID,
 		ctx = context.Background()
 	}
 
+	requestDataLength := lengthDistribution.Get(r.Int31() % 1000)
+	data := make([]byte, requestDataLength)
+	r.Read(data)
+
 	spec := pb.ResponseSpec{
 		Length:    int32(lengthDistribution.Get(r.Int31() % 1000)),
 		Latency:   latencyDistribution.Get(r.Int31() % 1000),
-		ErrorRate: errorRate}
+		ErrorRate: errorRate,
+		Data:      data,
+	}
 
 	log.Debugf("%v: req", worker)
 	resp, err := client.Get(ctx, &spec)
@@ -302,7 +310,8 @@ func sendNonStreamingRequests(
 	client pb.ResponderClient,
 	shutdown <-chan struct{},
 	clientTimeout time.Duration,
-	lengthDistribution distribution.Distribution,
+	requestLengthDistribution,
+	lengthDistribution,
 	latencyDistribution distribution.Distribution,
 	errorRate float32,
 	r *rand.Rand,
@@ -314,7 +323,7 @@ func sendNonStreamingRequests(
 		case <-shutdown:
 			return
 		case <-driver:
-			safeNonStreamingRequest(worker, client, clientTimeout, lengthDistribution, latencyDistribution, errorRate, r, responses)
+			safeNonStreamingRequest(worker, client, clientTimeout, requestLengthDistribution, lengthDistribution, latencyDistribution, errorRate, r, responses)
 		}
 	}
 }
@@ -490,7 +499,8 @@ func connect(
 	shutdowns []chan struct{},
 	connOpts []grpc.DialOption,
 	c uint,
-	latencyDistribution distribution.Distribution,
+	requestLengthDistribution,
+	latencyDistribution,
 	lengthDistribution distribution.Distribution,
 ) {
 	defer mainWait.Done()
@@ -515,7 +525,7 @@ func connect(
 			r := rand.New(rand.NewSource(time.Now().UnixNano()))
 			if !cfg.Streaming {
 				sendNonStreamingRequests(worker, client, shutdown, cfg.ClientTimeout,
-					lengthDistribution, latencyDistribution,
+					requestLengthDistribution, lengthDistribution, latencyDistribution,
 					float32(cfg.ErrorRate), r, driver, responses,
 				)
 			} else {
@@ -720,25 +730,26 @@ func loop(
 
 /// Configuration to run a client.
 type Config struct {
-	Address            string
-	UseUnixAddr        bool
-	ClientTimeout      time.Duration
-	Connections        uint
-	Streams            uint
-	TotalRequests      uint
-	TotalTargetRps     uint
-	Interval           time.Duration
-	NumIterations      uint
-	LatencyPercentiles string
-	LengthPercentiles  string
-	ErrorRate          float64
-	NoIntervalReport   bool
-	NoFinalReport      bool
-	Streaming          bool
-	StreamingRatio     string
-	MetricAddr         string
-	LatencyUnit        string
-	TlsTrustChainFile  string
+	Address                  string
+	UseUnixAddr              bool
+	ClientTimeout            time.Duration
+	Connections              uint
+	Streams                  uint
+	TotalRequests            uint
+	TotalTargetRps           uint
+	Interval                 time.Duration
+	NumIterations            uint
+	LatencyPercentiles       string
+	LengthPercentiles        string
+	RequestLengthPercentiles string
+	ErrorRate                float64
+	NoIntervalReport         bool
+	NoFinalReport            bool
+	Streaming                bool
+	StreamingRatio           string
+	MetricAddr               string
+	LatencyUnit              string
+	TlsTrustChainFile        string
 }
 
 // TODO: this would be much less ugly if the configuration was either stored in a struct, or used viper...
@@ -791,6 +802,16 @@ func (cfg Config) Run() {
 		log.Fatalf("unable to create length distribution: %v", err)
 	}
 
+	requestLengthPercentiles, err := percentiles.ParsePercentiles(cfg.RequestLengthPercentiles)
+	if err != nil {
+		log.Fatalf("requestLengthPercentiles was not valid: %v", err)
+	}
+
+	requestLengthDistribution, err := distribution.FromMap(requestLengthPercentiles)
+	if err != nil {
+		log.Fatalf("unable to create request length distribution: %v", err)
+	}
+
 	// By default, allow enough capacity for each worker.
 	capacity := cfg.Connections * cfg.Streams
 
@@ -862,6 +883,7 @@ func (cfg Config) Run() {
 			shutdowns,
 			connOpts,
 			c,
+			requestLengthDistribution,
 			latencyDistribution,
 			lengthDistribution,
 		)

cmd/client.go

@@ -32,6 +32,7 @@ func init() {
 	flags.UintVar(&clientCfg.NumIterations, "iterations", 0, "Number of iterations (0 for infinite)")
 	flags.StringVar(&clientCfg.LatencyPercentiles, "latencyPercentiles", "100=0", "response latency percentile distribution (in ms). (e.g. 50=10,100=100)")
 	flags.StringVar(&clientCfg.LengthPercentiles, "lengthPercentiles", "100=0", "response body length percentile distribution. (e.g. 50=100,100=1000)")
+	flags.StringVar(&clientCfg.RequestLengthPercentiles, "requestLengthPercentiles", "100=0", "request body length percentile distribution. (e.g. 50=100,100=1000)")
 	flags.Float64Var(&clientCfg.ErrorRate, "errorRate", 0.0, "the chance to return an error")
 	flags.BoolVar(&clientCfg.NoFinalReport, "noFinalReport", false, "do not print a final JSON output report")
 	flags.BoolVar(&clientCfg.NoIntervalReport, "noIntervalReport", false, "only print the final report, nothing intermediate")

protos/strest.proto

@@ -9,11 +9,16 @@ service Responder {
 }
 
 message ResponseSpec {
-  // how many bytes we expect in return.
+  // How many bytes we expect in a response?
   int32 length = 1;
-  // how long (in milliseconds) we want the response to wait.
+
+  // How long (in milliseconds) we want the response to wait.
   int64 latency = 2;
+
   float errorRate = 3;
+
+  // Unused, arbitrary data. Allows for arbitrarily large requests.
+  bytes data = 4;
 }
 
 message StreamingResponseSpec {