feat(loadgen): sustain rollback traffic
This commit is contained in:
parent
6c1329c2a9
commit
e1f728ff9b
3 changed files with 308 additions and 0 deletions
3
apps/loadgen/go.mod
Normal file
3
apps/loadgen/go.mod
Normal file
|
|
@ -0,0 +1,3 @@
|
||||||
|
module git.fiwlabs.dev/fiwdev/nereus/apps/loadgen
|
||||||
|
|
||||||
|
go 1.26
|
||||||
199
apps/loadgen/main.go
Normal file
199
apps/loadgen/main.go
Normal file
|
|
@ -0,0 +1,199 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
|
"math"
|
||||||
|
"math/rand/v2"
|
||||||
|
"net/http"
|
||||||
|
"os"
|
||||||
|
"os/signal"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"syscall"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
type generator struct {
|
||||||
|
client *http.Client
|
||||||
|
target string
|
||||||
|
buoys []string
|
||||||
|
mu sync.Mutex
|
||||||
|
counts map[string]int
|
||||||
|
}
|
||||||
|
|
||||||
|
type buoy struct {
|
||||||
|
ID string `json:"id"`
|
||||||
|
Name string `json:"name,omitempty"`
|
||||||
|
Latitude float64 `json:"latitude,omitempty"`
|
||||||
|
Longitude float64 `json:"longitude,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
if err := run(); err != nil {
|
||||||
|
slog.Error("load generator stopped", "error", err)
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func run() error {
|
||||||
|
target := strings.TrimRight(os.Getenv("TARGET_URL"), "/")
|
||||||
|
if target == "" {
|
||||||
|
return errors.New("TARGET_URL is required")
|
||||||
|
}
|
||||||
|
rps, err := strconv.Atoi(envOr("RPS", "5"))
|
||||||
|
if err != nil || rps < 1 || rps > 100 {
|
||||||
|
return errors.New("RPS must be between 1 and 100")
|
||||||
|
}
|
||||||
|
logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
|
||||||
|
slog.SetDefault(logger)
|
||||||
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||||
|
defer stop()
|
||||||
|
g := &generator{client: &http.Client{Timeout: 5 * time.Second}, target: target, counts: make(map[string]int)}
|
||||||
|
if err := g.seed(ctx); err != nil {
|
||||||
|
logger.Warn("initial buoy seed failed; traffic will continue", "error", err)
|
||||||
|
}
|
||||||
|
go g.seedUntilReady(ctx, logger)
|
||||||
|
go g.summarize(ctx, logger)
|
||||||
|
ticker := time.NewTicker(time.Second / time.Duration(rps))
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil
|
||||||
|
case <-ticker.C:
|
||||||
|
go g.send(ctx)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *generator) seed(ctx context.Context) error {
|
||||||
|
var existing []buoy
|
||||||
|
if err := g.request(ctx, http.MethodGet, "/api/v1/buoys", nil, &existing); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
g.mu.Lock()
|
||||||
|
for _, item := range existing {
|
||||||
|
g.buoys = append(g.buoys, item.ID)
|
||||||
|
}
|
||||||
|
ready := len(g.buoys) > 0
|
||||||
|
g.mu.Unlock()
|
||||||
|
if ready {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
positions := [][2]float64{{44.5, -9}, {44.2, -8}, {44.3, -7}, {44.4, -6}, {44.5, -5}, {44.6, -4}, {44.7, -3}, {44.5, -2}, {43.8, -2}, {43.5, -9.8}, {42.5, -10.2}, {41.5, -10.5}, {40.5, -10.8}, {39.5, -11}, {38.5, -11}, {37.5, -10.8}, {36.5, -10}, {35.9, -9}, {35.8, -8}, {35.7, -7}, {35.2, -7.5}, {35, -8.5}, {35.8, -4.5}, {36, -3.5}, {36.2, -2.5}, {36.5, -1.5}, {36.8, -.5}, {37.2, .5}, {37.7, 1.2}, {38.1, 2}, {38, 3}, {38.5, 4}, {39.5, 1}, {40, 1.5}, {40.5, 2}, {41, 2.7}, {41.5, 3.3}, {42, 3.8}, {42.5, 4.5}, {43, -9.8}, {41, -10.5}, {39, -10.8}}
|
||||||
|
for i, position := range positions {
|
||||||
|
input := buoy{Name: fmt.Sprintf("Synthetic %d", i+1), Latitude: position[0], Longitude: position[1]}
|
||||||
|
var created buoy
|
||||||
|
if err := g.request(ctx, http.MethodPost, "/api/v1/buoys", input, &created); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
g.mu.Lock()
|
||||||
|
g.buoys = append(g.buoys, created.ID)
|
||||||
|
g.mu.Unlock()
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *generator) send(ctx context.Context) { g.sendRoll(ctx, rand.IntN(100)) }
|
||||||
|
func (g *generator) sendRoll(ctx context.Context, roll int) {
|
||||||
|
path, method, body := "/api/v1/readings", http.MethodGet, any(nil)
|
||||||
|
if roll >= 60 && roll < 85 {
|
||||||
|
g.mu.Lock()
|
||||||
|
ids := append([]string(nil), g.buoys...)
|
||||||
|
g.mu.Unlock()
|
||||||
|
if len(ids) == 0 {
|
||||||
|
_ = g.request(ctx, method, path, nil, nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
method = http.MethodPost
|
||||||
|
phase := float64(time.Now().UnixMilli()) / 60000
|
||||||
|
body = map[string]any{"buoy_id": ids[rand.IntN(len(ids))], "water_temp": 19 + 2*math.Sin(phase) + rand.Float64()*.4, "wave_height": 1.8 + .8*math.Sin(phase*.7+1) + rand.Float64()*.25, "salinity": 35.5 + .5*math.Sin(phase*.3) + rand.Float64()*.15}
|
||||||
|
} else if roll >= 85 {
|
||||||
|
path = "/api/v1/readings/aggregate?window=1h"
|
||||||
|
}
|
||||||
|
_ = g.request(ctx, method, path, body, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *generator) seedUntilReady(ctx context.Context, logger *slog.Logger) {
|
||||||
|
ticker := time.NewTicker(10 * time.Second)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
g.mu.Lock()
|
||||||
|
ready := len(g.buoys) > 0
|
||||||
|
g.mu.Unlock()
|
||||||
|
if ready {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := g.seed(ctx); err != nil {
|
||||||
|
logger.Warn("buoy seed failed; traffic continues", "error", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *generator) request(ctx context.Context, method, path string, body, output any) error {
|
||||||
|
var payload *bytes.Reader
|
||||||
|
if body == nil {
|
||||||
|
payload = bytes.NewReader(nil)
|
||||||
|
} else {
|
||||||
|
encoded, err := json.Marshal(body)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("encode request: %w", err)
|
||||||
|
}
|
||||||
|
payload = bytes.NewReader(encoded)
|
||||||
|
}
|
||||||
|
req, err := http.NewRequestWithContext(ctx, method, g.target+path, payload)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("create request: %w", err)
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
resp, err := g.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
g.record("network_error")
|
||||||
|
return fmt.Errorf("send request: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = resp.Body.Close() }()
|
||||||
|
g.record(strconv.Itoa(resp.StatusCode))
|
||||||
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||||
|
return fmt.Errorf("API returned %s", resp.Status)
|
||||||
|
}
|
||||||
|
if output != nil && json.NewDecoder(resp.Body).Decode(output) != nil {
|
||||||
|
return errors.New("decode response")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *generator) record(status string) { g.mu.Lock(); defer g.mu.Unlock(); g.counts[status]++ }
|
||||||
|
func (g *generator) summarize(ctx context.Context, logger *slog.Logger) {
|
||||||
|
ticker := time.NewTicker(10 * time.Second)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
g.mu.Lock()
|
||||||
|
counts := g.counts
|
||||||
|
g.counts = make(map[string]int)
|
||||||
|
g.mu.Unlock()
|
||||||
|
logger.Info("traffic summary", "status_counts", counts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func envOr(name, fallback string) string {
|
||||||
|
if value := os.Getenv(name); value != "" {
|
||||||
|
return value
|
||||||
|
}
|
||||||
|
return fallback
|
||||||
|
}
|
||||||
106
apps/loadgen/main_test.go
Normal file
106
apps/loadgen/main_test.go
Normal file
|
|
@ -0,0 +1,106 @@
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRequestWeighting(t *testing.T) {
|
||||||
|
var mu sync.Mutex
|
||||||
|
counts := make(map[string]int)
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
mu.Lock()
|
||||||
|
counts[r.Method+" "+r.URL.RequestURI()]++
|
||||||
|
mu.Unlock()
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
g := &generator{client: server.Client(), target: server.URL, buoys: []string{"buoy-1"}, counts: make(map[string]int)}
|
||||||
|
for roll := range 100 {
|
||||||
|
g.sendRoll(context.Background(), roll)
|
||||||
|
}
|
||||||
|
|
||||||
|
want := map[string]int{
|
||||||
|
"GET /api/v1/readings": 60,
|
||||||
|
"POST /api/v1/readings": 25,
|
||||||
|
"GET /api/v1/readings/aggregate?window=1h": 15,
|
||||||
|
}
|
||||||
|
if fmt.Sprint(counts) != fmt.Sprint(want) {
|
||||||
|
t.Fatalf("request weighting = %v, want %v", counts, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSeedEmptyAPI(t *testing.T) {
|
||||||
|
var mu sync.Mutex
|
||||||
|
created := 0
|
||||||
|
var posted []buoy
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
switch {
|
||||||
|
case r.Method == http.MethodGet && r.URL.Path == "/api/v1/buoys":
|
||||||
|
_ = json.NewEncoder(w).Encode([]buoy{})
|
||||||
|
case r.Method == http.MethodPost && r.URL.Path == "/api/v1/buoys":
|
||||||
|
var input buoy
|
||||||
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
||||||
|
http.Error(w, "invalid buoy", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
mu.Lock()
|
||||||
|
created++
|
||||||
|
posted = append(posted, input)
|
||||||
|
id := fmt.Sprintf("buoy-%d", created)
|
||||||
|
mu.Unlock()
|
||||||
|
_ = json.NewEncoder(w).Encode(buoy{ID: id})
|
||||||
|
default:
|
||||||
|
http.NotFound(w, r)
|
||||||
|
}
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
g := &generator{client: server.Client(), target: server.URL, counts: make(map[string]int)}
|
||||||
|
if err := g.seed(context.Background()); err != nil {
|
||||||
|
t.Fatalf("seed: %v", err)
|
||||||
|
}
|
||||||
|
if created != 42 || len(g.buoys) != 42 {
|
||||||
|
t.Fatalf("created %d buoys and retained %d IDs, want 42 and 42", created, len(g.buoys))
|
||||||
|
}
|
||||||
|
want := [][2]float64{{44.5, -9}, {44.2, -8}, {44.3, -7}, {44.4, -6}, {44.5, -5}, {44.6, -4}, {44.7, -3}, {44.5, -2}, {43.8, -2}, {43.5, -9.8}, {42.5, -10.2}, {41.5, -10.5}, {40.5, -10.8}, {39.5, -11}, {38.5, -11}, {37.5, -10.8}, {36.5, -10}, {35.9, -9}, {35.8, -8}, {35.7, -7}, {35.2, -7.5}, {35, -8.5}, {35.8, -4.5}, {36, -3.5}, {36.2, -2.5}, {36.5, -1.5}, {36.8, -.5}, {37.2, .5}, {37.7, 1.2}, {38.1, 2}, {38, 3}, {38.5, 4}, {39.5, 1}, {40, 1.5}, {40.5, 2}, {41, 2.7}, {41.5, 3.3}, {42, 3.8}, {42.5, 4.5}, {43, -9.8}, {41, -10.5}, {39, -10.8}}
|
||||||
|
for i, position := range want {
|
||||||
|
if posted[i].Latitude != position[0] || posted[i].Longitude != position[1] {
|
||||||
|
t.Errorf("buoy %d position = (%v, %v), want (%v, %v)", i, posted[i].Latitude, posted[i].Longitude, position[0], position[1])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTrafficContinuesThroughHTTP500(t *testing.T) {
|
||||||
|
const requests = 30
|
||||||
|
var mu sync.Mutex
|
||||||
|
received := 0
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
mu.Lock()
|
||||||
|
received++
|
||||||
|
mu.Unlock()
|
||||||
|
http.Error(w, "unavailable", http.StatusInternalServerError)
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
g := &generator{client: server.Client(), target: server.URL, buoys: []string{"buoy-1"}, counts: make(map[string]int)}
|
||||||
|
for roll := range requests {
|
||||||
|
g.sendRoll(context.Background(), roll)
|
||||||
|
}
|
||||||
|
|
||||||
|
mu.Lock()
|
||||||
|
got := received
|
||||||
|
mu.Unlock()
|
||||||
|
if got != requests {
|
||||||
|
t.Fatalf("received %d requests, want %d", got, requests)
|
||||||
|
}
|
||||||
|
if g.counts["500"] != requests {
|
||||||
|
t.Fatalf("recorded %d HTTP 500 responses, want %d", g.counts["500"], requests)
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Add table
Reference in a new issue