Ripley performance improvements

Dockerfile

@@ -1,6 +1,21 @@
+FROM golang:1.20-alpine as build
+
+ADD . /src/
+
+RUN cd /src && \
+    go mod download && \
+    go build -o /ripley . && \
+    go build -o /dummyweb etc/dummyweb.go && \
+    go build -o /linegen etc/linegen.go
+
+##################################
 # Start fresh from a smaller image
-FROM alpine:3.15.0
+##################################
+FROM alpine:latest
 RUN apk add ca-certificates
 
-COPY ripley /usr/bin/ripley
-ENTRYPOINT ["/usr/bin/ripley"]
+COPY --from=build /ripley /usr/bin/ripley
+COPY --from=build /linegen /usr/bin/linegen
+COPY --from=build /dummyweb /usr/bin/dummyweb
+
+ENTRYPOINT ["/usr/bin/ripley"]

README.md

@@ -129,3 +129,11 @@ cat etc/requests.jsonl | ./ripley -pace "30s@1" -dry-run
 ```bash
 go test pkg/*.go
 ```
+
+## Memory profiling
+
+```
+./ripley -metricsServerEnable -printStat -memprofile dist/mem.pprof -pace 1m@10 -workers 100  < etc/requests.jsonl
+
+go tool pprof --alloc_objects ripley dist/mem.pprof
+```bash

etc/dummyweb.go

@@ -1,23 +1,100 @@
 package main
 
 import (
+	"encoding/json"
+	"flag"
+	"fmt"
 	"log"
-	"net/http"
-	"net/http/httputil"
+	"os"
+	"os/signal"
+	"syscall"
 	"time"
+
+	"github.com/valyala/fasthttp"
+	"github.com/valyala/fasthttp/reuseport"
 )
 
-func handler(w http.ResponseWriter, r *http.Request) {
-	dump, err := httputil.DumpRequest(r, true)
+var silent bool
+
+func main() {
+	flag.BoolVar(&silent, "silent", false, "whether to silence output")
+	addr := flag.String("addr", "localhost:8080", "server listen address. Default: localhost:8080")
+	flag.Parse()
+
+	// Create a new listener on the given address using port reuse
+	ln, err := reuseport.Listen("tcp4", *addr)
 	if err != nil {
-		panic(err)
+		log.Fatalf("error creating listener: %v", err)
 	}
+	defer ln.Close()
 
-	log.Printf("%v: %s\n", time.Now().Format(time.UnixDate), string(dump))
-	w.Write([]byte("hi\n"))
+	// Create a new fasthttp server
+	server := &fasthttp.Server{
+		TCPKeepalive:    true,
+		LogAllErrors:    true,
+		ReadBufferSize:  1024 * 1024,
+		WriteBufferSize: 1024 * 1024,
+		ReadTimeout:     90 * time.Second,
+		WriteTimeout:    5 * time.Second,
+		Handler:         requestHandler,
+	}
+
+	// Start the server in a goroutine
+	go func() {
+		if err := server.Serve(ln); err != nil {
+			log.Fatalf("error starting server: %v", err)
+		}
+	}()
+
+	// Wait for a signal to stop the server
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+	<-sig
+
+	// Stop the server
+	server.Shutdown()
 }
 
-func main() {
-	http.HandleFunc("/", handler)
-	log.Fatal(http.ListenAndServe(":8080", nil))
+func requestToJSON(req *fasthttp.Request) ([]byte, error) {
+	type requestJSON struct {
+		URI         string            `json:"uri"`
+		Method      string            `json:"method"`
+		Headers     map[string]string `json:"headers"`
+		ContentType string            `json:"content_type"`
+		Body        string            `json:"body"`
+	}
+
+	// Get the request URI, method, headers, content type, and body
+	uri := string(req.URI().FullURI())
+	method := string(req.Header.Method())
+	headers := make(map[string]string)
+	req.Header.VisitAll(func(k, v []byte) {
+		headers[string(k)] = string(v)
+	})
+	contentType := string(req.Header.ContentType())
+	body := string(req.Body())
+
+	// Create a requestJSON struct and marshal it to JSON
+	reqJSON := &requestJSON{
+		URI:         uri,
+		Method:      method,
+		Headers:     headers,
+		ContentType: contentType,
+		Body:        body,
+	}
+	return json.Marshal(reqJSON)
+}
+
+func requestHandler(ctx *fasthttp.RequestCtx) {
+	jsonData, _ := requestToJSON(&ctx.Request)
+
+	if !silent {
+		fmt.Println(string(jsonData))
+	}
+
+	ctx.SetContentType("application/json")
+	ctx.Response.Header.SetContentLength(len(jsonData))
+	// ctx.Response.Header.Set("Connection", "keep-alive")
+	ctx.SetStatusCode(fasthttp.StatusOK)
+	ctx.Write(jsonData)
 }

etc/linegen.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"bufio"
+	"flag"
+	"fmt"
+	"os"
+	"time"
+)
+
+func main() {
+	rate := flag.Int("rate", 10, "lines per second")
+	flag.Parse()
+
+	tickDuration := time.Duration((1000000 / (*rate))) * time.Microsecond
+
+	ticker := time.NewTicker(tickDuration)
+	defer ticker.Stop()
+
+	scanner := bufio.NewScanner(os.Stdin)
+	if !scanner.Scan() {
+		return
+	}
+	line := scanner.Text()
+
+	for range ticker.C {
+		fmt.Println(line)
+	}
+}

etc/requests.jsonl

@@ -1,15 +1,12 @@
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:50.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:51.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:52.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:53.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:54.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:55.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:56.9Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T18:59:57.9Z"}
-{"url": "http://localhost:8080/", "method": "POST", "body": "{\"foo\": \"bar\"}", "headers": {"Accept": "text/plain"}, "timestamp": "2021-11-08T18:59:58.9Z"}
+{"url": "http://localhost:8080/localhost:8080", "method": "GET", "timestamp": "2021-11-08T18:59:50.9Z", "headers": {"Content-Type": "application/json", "Host": "example.net", "Cookies":"cookie=1234567890", "User-Agent":"Mozilla/5.0", "x-custom-header":"x-custom-header-value"}}
+{"url": "http://localhost:8081/localhost:8081", "method": "GET", "timestamp": "2021-11-08T18:59:50.9Z", "headers": {"Content-Type": "application/json", "Host": "example.net", "Cookies":"cookie=1234567890", "User-Agent":"Mozilla/5.0", "x-custom-header":"x-custom-header-value"}}
+{"url": "http://localhost:8080/localhost:8080", "method": "POST", "body": "{\"foo\": \"bar\"}", "headers": {"Accept": "text/plain"}, "timestamp": "2021-11-08T18:59:58.9Z"}
+{"url": "http://localhost:8081/localhost:8081", "method": "POST", "body": "{\"foo\": \"bar\"}", "headers": {"Accept": "text/plain"}, "timestamp": "2021-11-08T18:59:58.9Z"}
+{"url": "http://localhost:8081/", "method": "GET", "headers": {"Accept": "text/plain"}, "timestamp": "2021-11-08T18:59:59.9Z"}
 {"url": "http://localhost:8080/", "method": "GET", "headers": {"Accept": "text/plain"}, "timestamp": "2021-11-08T18:59:59.9Z"}
-{"url": "http://localhost:8080/", "method": "HEAD", "timestamp": "2021-11-08T19:00:00.00Z"}
+{"url": "http://localhost:8080/", "method": "HEAD", "timestamp": "2021-11-08T19:00:00.00Z", "headers": {"Connection": "close" }}
 {"url": "http://localhost:8080/", "method": "OPTIONS", "timestamp": "2021-11-08T19:00:00.01Z"}
 {"url": "http://localhost:8080/", "method": "TRACE", "timestamp": "2021-11-08T19:00:00.02Z"}
+{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T19:00:00.04Z"}
+{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T19:00:00.04Z"}}}
 {"url": "http://localhost:8080/", "method": "PROPFIND", "timestamp": "2021-11-08T19:00:00.03Z"}
-{"url": "http://localhost:8080/", "method": "GET", "timestamp": "2021-11-08T19:00:00.04Z"}

go.mod

@@ -1,3 +1,20 @@
 module github.com/loveholidays/ripley
 
-go 1.17
+go 1.20
+
+require (
+	github.com/VictoriaMetrics/metrics v1.23.1
+	github.com/montanaflynn/stats v0.7.0
+	github.com/valyala/fasthttp v1.45.0
+	github.com/valyala/fastjson v1.6.4
+)
+
+require (
+	github.com/andybalholm/brotli v1.0.5 // indirect
+	github.com/klauspost/compress v1.16.3 // indirect
+	github.com/valyala/bytebufferpool v1.0.0 // indirect
+	github.com/valyala/fastrand v1.1.0 // indirect
+	github.com/valyala/histogram v1.2.0 // indirect
+	github.com/valyala/tcplisten v1.0.0 // indirect
+	golang.org/x/sys v0.6.0 // indirect
+)

go.sum

@@ -0,0 +1,22 @@
+github.com/VictoriaMetrics/metrics v1.23.1 h1:/j8DzeJBxSpL2qSIdqnRFLvQQhbJyJbbEi22yMm7oL0=
+github.com/VictoriaMetrics/metrics v1.23.1/go.mod h1:rAr/llLpEnAdTehiNlUxKgnjcOuROSzpw0GvjpEbvFc=
+github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=
+github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
+github.com/klauspost/compress v1.16.3 h1:XuJt9zzcnaz6a16/OU53ZjWp/v7/42WcR5t2a0PcNQY=
+github.com/klauspost/compress v1.16.3/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/montanaflynn/stats v0.7.0 h1:r3y12KyNxj/Sb/iOE46ws+3mS1+MZca1wlHQFPsY/JU=
+github.com/montanaflynn/stats v0.7.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
+github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
+github.com/valyala/fasthttp v1.45.0 h1:zPkkzpIn8tdHZUrVa6PzYd0i5verqiPSkgTd3bSUcpA=
+github.com/valyala/fasthttp v1.45.0/go.mod h1:k2zXd82h/7UZc3VOdJ2WaUqt1uZ/XpXAfE9i+HBC3lA=
+github.com/valyala/fastjson v1.6.4 h1:uAUNq9Z6ymTgGhcm0UynUAB6tlbakBrz6CQFax3BXVQ=
+github.com/valyala/fastjson v1.6.4/go.mod h1:CLCAqky6SMuOcxStkYQvblddUtoRxhYMGLrsQns1aXY=
+github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8=
+github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
+github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ=
+github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY=
+github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8=
+github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
+golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

main.go

@@ -20,29 +20,49 @@ package main
 
 import (
 	"flag"
+	"fmt"
+	_ "net/http/pprof"
 	"os"
+	"os/signal"
 	"runtime"
 	"runtime/pprof"
+	"syscall"
 
 	ripley "github.com/loveholidays/ripley/pkg"
 )
 
 func main() {
-	paceStr := flag.String("pace", "10s@1", `[duration]@[rate], e.g. "1m@1 [email protected] 1h@2"`)
-	silent := flag.Bool("silent", false, "Suppress output")
-	dryRun := flag.Bool("dry-run", false, "Consume input but do not send HTTP requests to targets")
-	timeout := flag.Int("timeout", 10, "HTTP client timeout in seconds")
-	strict := flag.Bool("strict", false, "Panic on bad input")
-	memprofile := flag.String("memprofile", "", "Write memory profile to `file` before exit")
-	numWorkers := flag.Int("workers", 1000, "Number of client workers to use")
+	var opts ripley.Options
+
+	flag.StringVar(&opts.Pace, "pace", "10s@1", `[duration]@[rate], e.g. "1m@1 [email protected] 1h@2"`)
+	flag.BoolVar(&opts.Silent, "silent", false, "Suppress output")
+	flag.BoolVar(&opts.SilentHttpError, "silentHttpError", false, "Suppress HTTP errors (http codes 5xx) output")
+
+	flag.BoolVar(&opts.DryRun, "dry-run", false, "Consume input but do not send HTTP requests to targets")
+	flag.IntVar(&opts.Timeout, "timeout", 10, "HTTP client request timeout in seconds")
+	flag.IntVar(&opts.TimeoutConnection, "timeoutConnection", 3, "HTTP client connetion timeout in seconds")
+	flag.BoolVar(&opts.Strict, "strict", false, "Panic on bad input")
+	flag.StringVar(&opts.Memprofile, "memprofile", "", "Write memory profile to `file` before exit")
+	flag.IntVar(&opts.NumWorkers, "workers", 10, "Number of client workers to use")
+
+	flag.BoolVar(&opts.PrintStat, "printStat", false, "Print statistics to stdout at the end")
+	flag.BoolVar(&opts.MetricsServerEnable, "metricsServerEnable", false, "Enable metrics server. Server prometheus statistics on /metrics endpoint")
+	flag.StringVar(&opts.MetricsServerAddr, "metricsServerAddr", "0.0.0.0:8081", "Metrics server listen address")
+
+	flag.IntVar(&opts.PrintNSlowest, "printNSlowest", 0, "Print N slowest Requests")
+
+	flag.Usage = func() {
+		fmt.Fprintf(os.Stderr, "Usage: %s -target string\n", os.Args[0])
+		flag.PrintDefaults()
+	}
 
 	flag.Parse()
 
-	exitCode := ripley.Replay(*paceStr, *silent, *dryRun, *timeout, *strict, *numWorkers)
+	exitCode := ripley.Replay(&opts)
 	defer os.Exit(exitCode)
 
-	if *memprofile != "" {
-		f, err := os.Create(*memprofile)
+	if opts.Memprofile != "" {
+		f, err := os.Create(opts.Memprofile)
 
 		if err != nil {
 			panic(err)
@@ -54,5 +74,10 @@ func main() {
 		if err := pprof.WriteHeapProfile(f); err != nil {
 			panic(err)
 		}
+
+		// Wait for a signal to stop the server
+		sig := make(chan os.Signal, 1)
+		signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+		<-sig
 	}
 }