fix: forwarder copy body eof

agent/forwarder.go

@@ -1,8 +1,11 @@
 package agent
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"fmt"
+	"io"
 	"net/http"
 	"strconv"
 	"time"
@@ -80,6 +83,18 @@ func (f *forwarder) Forward(req *http.Request) (*http.Response, error) {
 		return nil, errUpstreamUnreachable
 	}
 
+	// We must copy the response body before replying. The request context is
+	// cancelled when Forward returns so attempting top copy the body again
+	// may fail.
+	//
+	// TODO(andydunstall): Understand this better as should be avoidable.
+	var buf bytes.Buffer
+	if _, err := io.Copy(&buf, resp.Body); err != nil {
+		return nil, fmt.Errorf("copy body: %w", err)
+	}
+	resp.Body.Close()
+	resp.Body = io.NopCloser(&buf)
+
 	f.metrics.ForwardRequestsTotal.With(prometheus.Labels{
 		"method":      req.Method,
 		"status":      strconv.Itoa(resp.StatusCode),

cli/workload/requests.go

@@ -1,6 +1,7 @@
 package workload
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math/rand"
@@ -129,12 +130,14 @@ func runClient(ctx context.Context, conf *config.RequestsConfig, logger log.Logg
 	ticker := time.NewTicker(time.Duration(int(time.Second) / conf.Rate))
 	defer ticker.Stop()
 
+	body := make([]byte, conf.RequestSize)
+
 	client := &http.Client{}
 	for {
 		select {
 		case <-ticker.C:
 			endpointID := rand.Int() % conf.Endpoints
-			req, _ := http.NewRequest("GET", conf.Server.URL, nil)
+			req, _ := http.NewRequest("GET", conf.Server.URL, bytes.NewReader(body))
 			req.Header.Set("x-piko-endpoint", strconv.Itoa(endpointID))
 			resp, err := client.Do(req)
 			if err != nil {

workload/config/requests.go

@@ -16,6 +16,9 @@ type RequestsConfig struct {
 	// Endpoints is the number of available endpoint IDs.
 	Endpoints int `json:"endpoints" yaml:"endpoints"`
 
+	// RequestSize is the size of each request.
+	RequestSize int `json:"request_size" yaml:"request_size"`
+
 	Server ServerConfig `json:"server" yaml:"server"`
 
 	Log log.Config `json:"log" yaml:"log"`
@@ -69,6 +72,15 @@ On each request, the client selects a random endpoint ID from 0 to
 'endpoints'.`,
 	)
 
+	fs.IntVar(
+		&c.RequestSize,
+		"request-size",
+		1024,
+		`
+The size of each request. As the upstream echos the response body, the response
+will have the same size.`,
+	)
+
 	fs.StringVar(
 		&c.Server.URL,
 		"server.url",