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 }