105 lines
3.3 KiB
Go
105 lines
3.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
|
|
_ "github.com/go-sql-driver/mysql"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/loadsim"
|
|
)
|
|
|
|
func main() {
|
|
log.SetFlags(log.LstdFlags | log.Lmicroseconds)
|
|
flagCfg := loadsim.RegisterFlags(flag.CommandLine)
|
|
cleanupRegistration := flag.Bool("cleanup-registration", false, "delete this run's loopback JT808 registration rows after the simulation")
|
|
cleanupOnly := flag.Bool("cleanup-only", false, "skip the simulation and only clean the configured JT808 phone range")
|
|
mysqlDSN := flag.String("mysql-dsn", strings.TrimSpace(os.Getenv("MYSQL_DSN")), "MySQL DSN used only by JT808 registration cleanup")
|
|
if err := flag.CommandLine.Parse(os.Args[1:]); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
cfg, err := flagCfg.Build()
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
if *cleanupOnly {
|
|
deleted, err := cleanupJT808Registrations(ctx, cfg, *mysqlDSN)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
log.Printf("jt808 synthetic registration cleanup completed rows_deleted=%d", deleted)
|
|
return
|
|
}
|
|
|
|
log.Printf("load simulation started protocol=%s addr=%s connections=%d connect_rate=%d send_interval=%s duration=%s template=%s",
|
|
cfg.Protocol, cfg.Addr, cfg.Connections, cfg.ConnectRatePerSecond, cfg.SendInterval, cfg.Duration, cfg.Template)
|
|
stats, err := (loadsim.Runner{}).Run(ctx, cfg)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
log.Print(formatStats(stats))
|
|
if *cleanupRegistration {
|
|
deleted, err := cleanupJT808Registrations(ctx, cfg, *mysqlDSN)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
log.Printf("jt808 synthetic registration cleanup completed rows_deleted=%d", deleted)
|
|
}
|
|
}
|
|
|
|
func cleanupJT808Registrations(ctx context.Context, cfg loadsim.Config, dsn string) (int64, error) {
|
|
if cfg.Protocol != loadsim.ProtocolJT808 {
|
|
return 0, fmt.Errorf("registration cleanup only supports jt808")
|
|
}
|
|
if strings.TrimSpace(dsn) == "" {
|
|
return 0, fmt.Errorf("mysql-dsn or MYSQL_DSN is required for registration cleanup")
|
|
}
|
|
db, err := sql.Open("mysql", dsn)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("open mysql for registration cleanup: %w", err)
|
|
}
|
|
defer db.Close()
|
|
return cleanupJT808RegistrationsWithDB(ctx, db, cfg)
|
|
}
|
|
|
|
type cleanupExecer interface {
|
|
ExecContext(context.Context, string, ...any) (sql.Result, error)
|
|
}
|
|
|
|
func cleanupJT808RegistrationsWithDB(ctx context.Context, exec cleanupExecer, cfg loadsim.Config) (int64, error) {
|
|
firstPhone := fmt.Sprintf("%012d", cfg.JT808PhoneBase)
|
|
lastPhone := fmt.Sprintf("%012d", cfg.JT808PhoneBase+int64(cfg.Connections)-1)
|
|
result, err := exec.ExecContext(ctx, `DELETE FROM jt808_registration
|
|
WHERE phone BETWEEN ? AND ?
|
|
AND source_ip IN ('127.0.0.1', '::1')`, firstPhone, lastPhone)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("delete loopback jt808 registrations: %w", err)
|
|
}
|
|
deleted, err := result.RowsAffected()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("read registration cleanup result: %w", err)
|
|
}
|
|
return deleted, nil
|
|
}
|
|
|
|
func formatStats(stats loadsim.Stats) string {
|
|
return fmt.Sprintf("connections_opened=%d connections_failed=%d frames_written=%d write_errors=%d response_bytes=%d read_errors=%d",
|
|
stats.ConnectionsOpened,
|
|
stats.ConnectionsFailed,
|
|
stats.FramesWritten,
|
|
stats.WriteErrors,
|
|
stats.ResponseBytes,
|
|
stats.ReadErrors,
|
|
)
|
|
}
|