259 lines
10 KiB
Go
259 lines
10 KiB
Go
package loadgen
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestRunExecutesFixedBoundedHTTPWorkload(t *testing.T) {
|
|
var requests atomic.Int64
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
if request.Method != http.MethodPost || request.Header.Get("X-Scenario") != "fixed" {
|
|
t.Errorf("request = %s %q", request.Method, request.Header.Get("X-Scenario"))
|
|
}
|
|
body, err := io.ReadAll(request.Body)
|
|
if err != nil || string(body) != `{"count":1}` {
|
|
t.Errorf("request body = %q, error = %v", body, err)
|
|
}
|
|
requests.Add(1)
|
|
response.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
TargetURL: server.URL, Method: http.MethodPost, Headers: http.Header{"X-Scenario": {"fixed"}}, RequestBody: []byte(`{"count":1}`),
|
|
Requests: 12, Concurrency: 3, RequestTimeout: time.Second,
|
|
})
|
|
if err != nil || report.Requests != 12 || report.Completed != 12 || report.Succeeded != 12 || report.Failed != 0 ||
|
|
requests.Load() != 12 || report.Latency.P50UpperBound <= 0 || report.Runtime.NumCPU <= 0 {
|
|
t.Fatalf("Run() = (%+v, %v); handler requests=%d", report, err, requests.Load())
|
|
}
|
|
}
|
|
|
|
func TestRunClassifiesHTTPFailure(t *testing.T) {
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) {
|
|
response.WriteHeader(http.StatusServiceUnavailable)
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
|
|
})
|
|
if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.Status5xx != 1 || report.Completed != 1 {
|
|
t.Fatalf("Run() = (%+v, %v)", report, err)
|
|
}
|
|
}
|
|
|
|
func TestRunTimeBoxedRateIsBounded(t *testing.T) {
|
|
var requests atomic.Int64
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) {
|
|
requests.Add(1)
|
|
response.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
TargetURL: server.URL, Duration: 100 * time.Millisecond, Rate: 100, Concurrency: 2, RequestTimeout: time.Second,
|
|
})
|
|
if err != nil || report.Requests == 0 || report.Requests > 20 || report.Requests != report.Completed ||
|
|
report.Succeeded != report.Completed || requests.Load() != int64(report.Requests) {
|
|
t.Fatalf("Run() = (%+v, %v); handler requests=%d", report, err, requests.Load())
|
|
}
|
|
}
|
|
|
|
func TestRunReportsRateStartsDroppedByConcurrency(t *testing.T) {
|
|
started := make(chan struct{}, 1)
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
select {
|
|
case started <- struct{}{}:
|
|
default:
|
|
}
|
|
<-request.Context().Done()
|
|
response.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
TargetURL: server.URL, Duration: 80 * time.Millisecond, Rate: 100_000, Concurrency: 1, RequestTimeout: time.Second,
|
|
})
|
|
select {
|
|
case <-started:
|
|
default:
|
|
t.Fatal("rate-limited workload did not start a request")
|
|
}
|
|
if err != nil || report.RateStartsGenerated <= report.Requests || report.RateStartsDropped == 0 ||
|
|
report.RateStartsDropped > report.RateStartsGenerated {
|
|
t.Fatalf("Run() = (%+v, %v)", report, err)
|
|
}
|
|
}
|
|
|
|
func TestRunExecutesBoundedCONNECTTunnelWorkload(t *testing.T) {
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatalf("Listen(): %v", err)
|
|
}
|
|
defer listener.Close()
|
|
|
|
var tunnels atomic.Int64
|
|
serveDone := make(chan struct{})
|
|
go func() {
|
|
defer close(serveDone)
|
|
for range 3 {
|
|
connection, acceptErr := listener.Accept()
|
|
if acceptErr != nil {
|
|
return
|
|
}
|
|
go func(connection net.Conn) {
|
|
defer connection.Close()
|
|
request, readErr := http.ReadRequest(bufio.NewReader(connection))
|
|
if readErr != nil || request.Method != http.MethodConnect || request.Host != "target.example:443" {
|
|
t.Errorf("CONNECT request = %+v, error = %v", request, readErr)
|
|
return
|
|
}
|
|
tunnels.Add(1)
|
|
_, _ = fmt.Fprint(connection, "HTTP/1.1 200 Connection Established\r\n\r\n")
|
|
_, _ = io.Copy(io.Discard, connection)
|
|
}(connection)
|
|
}
|
|
}()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioConnect, TargetURL: "https://target.example/health", ProxyURL: "http://" + listener.Addr().String(),
|
|
Requests: 3, Concurrency: 2, RequestTimeout: time.Second, TunnelHold: 10 * time.Millisecond,
|
|
})
|
|
if err != nil || report.Requests != 3 || report.Completed != 3 || report.Succeeded != 3 || report.Failed != 0 ||
|
|
report.TunnelsEstablished != 3 || tunnels.Load() != 3 || report.Latency.P50UpperBound <= 0 {
|
|
t.Fatalf("Run() = (%+v, %v); CONNECT requests=%d", report, err, tunnels.Load())
|
|
}
|
|
<-serveDone
|
|
}
|
|
|
|
func TestRunClassifiesCONNECTProxyRejection(t *testing.T) {
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatalf("Listen(): %v", err)
|
|
}
|
|
defer listener.Close()
|
|
go func() {
|
|
connection, acceptErr := listener.Accept()
|
|
if acceptErr != nil {
|
|
return
|
|
}
|
|
defer connection.Close()
|
|
_, _ = http.ReadRequest(bufio.NewReader(connection))
|
|
_, _ = fmt.Fprint(connection, "HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n")
|
|
}()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioConnect, TargetURL: "https://target.example/health", ProxyURL: "http://" + listener.Addr().String(),
|
|
Requests: 1, Concurrency: 1, RequestTimeout: time.Second, TunnelHold: time.Millisecond,
|
|
})
|
|
if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.Status4xx != 1 || report.TunnelsEstablished != 0 {
|
|
t.Fatalf("Run() = (%+v, %v)", report, err)
|
|
}
|
|
}
|
|
|
|
func TestRunExecutesAndValidatesExtractWorkload(t *testing.T) {
|
|
var requests atomic.Int64
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
if request.Method != http.MethodPost || request.Header.Get("Content-Type") != "application/json" ||
|
|
request.Header.Get("X-Request-ID") == "" || request.Header.Get("Idempotency-Key") == "" {
|
|
t.Errorf("extract request = method=%s content-type=%q request-id=%q idempotency=%q",
|
|
request.Method, request.Header.Get("Content-Type"), request.Header.Get("X-Request-ID"), request.Header.Get("Idempotency-Key"))
|
|
}
|
|
var body struct {
|
|
Count int `json:"count"`
|
|
Fulfillment string `json:"fulfillment"`
|
|
}
|
|
if err := json.NewDecoder(request.Body).Decode(&body); err != nil || body.Count != 2 || body.Fulfillment != "partial" {
|
|
t.Errorf("extract body = %+v, error = %v", body, err)
|
|
}
|
|
requests.Add(1)
|
|
response.Header().Set("Content-Type", "application/json")
|
|
_, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"},{"id":"proxy-b"}]}`,
|
|
request.Header.Get("X-Request-ID"))
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 2, Concurrency: 1, RequestTimeout: time.Second,
|
|
ExtractCount: 2, ExtractFulfillment: "partial",
|
|
})
|
|
if err != nil || report.Requests != 2 || report.Completed != 2 || report.Succeeded != 2 || report.Failed != 0 ||
|
|
report.ExtractResponsesValidated != 2 || report.ExtractReturned != 4 || report.ExtractValidationFailures != 0 || requests.Load() != 2 {
|
|
t.Fatalf("Run() = (%+v, %v); extract requests=%d", report, err, requests.Load())
|
|
}
|
|
}
|
|
|
|
func TestRunRejectsInvalidExtractResponse(t *testing.T) {
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
response.Header().Set("Content-Type", "application/json")
|
|
_, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"}]}`,
|
|
request.Header.Get("X-Request-ID"))
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
|
|
ExtractCount: 2, ExtractFulfillment: "partial",
|
|
})
|
|
if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.ExtractResponsesValidated != 0 ||
|
|
report.ExtractValidationFailures != 1 || report.ExtractReturned != 0 {
|
|
t.Fatalf("Run() = (%+v, %v)", report, err)
|
|
}
|
|
}
|
|
|
|
func TestRunRejectsDuplicateProxyIDsInExtractResponse(t *testing.T) {
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
response.Header().Set("Content-Type", "application/json")
|
|
_, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"},{"id":"proxy-a"}]}`,
|
|
request.Header.Get("X-Request-ID"))
|
|
}))
|
|
defer server.Close()
|
|
|
|
report, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
|
|
ExtractCount: 2, ExtractFulfillment: "partial",
|
|
})
|
|
if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.ExtractResponsesValidated != 0 ||
|
|
report.ExtractValidationFailures != 1 {
|
|
t.Fatalf("Run() = (%+v, %v)", report, err)
|
|
}
|
|
}
|
|
|
|
func TestRunRejectsCONNECTWithoutProxyOrHold(t *testing.T) {
|
|
_, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioConnect, TargetURL: "https://target.example/health", Requests: 1, Concurrency: 1, RequestTimeout: time.Second,
|
|
})
|
|
if !errors.Is(err, ErrInvalidOptions) {
|
|
t.Fatalf("Run(connect without proxy or hold) error = %v, want ErrInvalidOptions", err)
|
|
}
|
|
}
|
|
|
|
func TestRunRejectsInvalidExtractOptions(t *testing.T) {
|
|
_, err := Run(context.Background(), Options{
|
|
Scenario: ScenarioExtract, TargetURL: "http://127.0.0.1:8081/api/v1/proxies/extract", Requests: 1, Concurrency: 1,
|
|
RequestTimeout: time.Second, ExtractCount: 0,
|
|
})
|
|
if !errors.Is(err, ErrInvalidOptions) {
|
|
t.Fatalf("Run(invalid extract options) error = %v, want ErrInvalidOptions", err)
|
|
}
|
|
}
|
|
|
|
func TestRunRejectsUnboundedWorkload(t *testing.T) {
|
|
_, err := Run(context.Background(), Options{TargetURL: "http://127.0.0.1:8080", Concurrency: 1})
|
|
if !errors.Is(err, ErrInvalidOptions) {
|
|
t.Fatalf("Run(unbounded) error = %v, want ErrInvalidOptions", err)
|
|
}
|
|
}
|