package main import ( "context" "flag" "fmt" "log/slog" "os" "os/signal" "syscall" "time" "schedule_converter/config" grpcClient "schedule_converter/grpc" "schedule_converter/pkg/logger" ) const ( defaultGRPCAddr = "localhost:50051" runnerIDPrefix = "runner-" defaultVersion = "1.0.0" initialReconnectInterval = 5 * time.Second maxReconnectInterval = 60 * time.Second heartbeatInterval = 30 * time.Second ) func main() { grpcAddr := flag.String("grpc-addr", defaultGRPCAddr, "gRPC 服务器地址") debug := flag.Bool("debug", false, "启用调试模式") flag.Parse() if *debug { logger.SetLevel(slog.LevelDebug) logger.SetTextHandler() } cfg := config.Load() slog.Info("配置加载完成", "base_url", cfg.BaseURL) runGRPCClient(*grpcAddr) } func resolveRunnerID() string { if id := os.Getenv("RUNNER_ID"); id != "" { return id } if hostname, err := os.Hostname(); err == nil && hostname != "" { return runnerIDPrefix + hostname } return fmt.Sprintf("%s%d", runnerIDPrefix, os.Getpid()) } func resolveVersion() string { if v := os.Getenv("RUNNER_VERSION"); v != "" { return v } return defaultVersion } func runGRPCClient(serverAddr string) { runnerID := resolveRunnerID() version := resolveVersion() slog.Info("启动 gRPC 客户端", "runner_id", runnerID, "version", version) ctx, cancel := context.WithCancel(context.Background()) defer cancel() sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) go func() { sig := <-sigChan slog.Info("收到信号,正在关闭...", "signal", sig) cancel() }() retryCount := 0 for { select { case <-ctx.Done(): slog.Info("客户端已停止", "total_retries", retryCount) return default: } retryCount++ currentInterval := initialReconnectInterval if retryCount > 1 { currentInterval = min(initialReconnectInterval*time.Duration(retryCount-1), maxReconnectInterval) } slog.Info("正在连接服务器", "attempt", retryCount, "server", serverAddr) c := grpcClient.NewClient(serverAddr, runnerID, version) handler := grpcClient.NewTaskHandler() c.SetHandler(handler) if err := c.Connect(ctx); err != nil { slog.Warn("连接失败", "error", err, "attempt", retryCount, "retry_after", currentInterval) time.Sleep(currentInterval) continue } slog.Info("连接成功", "attempt", retryCount) retryCount = 0 heartbeatCtx, heartbeatCancel := context.WithCancel(ctx) go c.HeartbeatLoop(heartbeatCtx, heartbeatInterval) err := c.Run(ctx) heartbeatCancel() c.Close() if ctx.Err() != nil { return } if err != nil { slog.Warn("连接断开", "error", err, "retry_after", initialReconnectInterval) } else { slog.Info("连接已关闭") } time.Sleep(initialReconnectInterval) } }