diff --git a/apps/loadgen/go.mod b/apps/loadgen/go.mod new file mode 100644 index 0000000..079758d --- /dev/null +++ b/apps/loadgen/go.mod @@ -0,0 +1,3 @@ +module git.fiwlabs.dev/fiwdev/nereus/apps/loadgen + +go 1.26 diff --git a/apps/loadgen/main.go b/apps/loadgen/main.go new file mode 100644 index 0000000..c2958b3 --- /dev/null +++ b/apps/loadgen/main.go @@ -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 +} diff --git a/apps/loadgen/main_test.go b/apps/loadgen/main_test.go new file mode 100644 index 0000000..6978b21 --- /dev/null +++ b/apps/loadgen/main_test.go @@ -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) + } +}