schedule_converter/main.go

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)
	}
}