feat: 换到 Gin 主流栈并补齐运维能力

HTTP 用 Gin,配置用 Viper,日志用 zap,数据用 sqlx;保留原有接口与会商行为。
顺带补上就绪探活、请求号、滚动日志、systemd 与 Dockerfile。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
gozerosandCursor committed 2026-11-04 09:51:41 +08:00
1 parent 0c45b86d67
commit bd427b57a1
26 files changed
+1213 -302

No files matched your search

+91 -21
View File
@@ -1,18 +1,27 @@
// 会商后台进程入口。
//
// 启动顺序不能颠倒:先日志(后面的失败才看得到)→ 校验配置 → 连库建表 → 挂 Gin → 按模式起一个 Worker → 听信号停机。
// 边界:live 与 demo 只起一个 Worker。两进程不要同时消费同一库的 queued。
package main
import (
"database/sql"
"log"
"context"
"net/http"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
"github.com/gin-gonic/gin"
_ "github.com/go-sql-driver/mysql"
"github.com/jmoiron/sqlx"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/api"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/config"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/deerflow"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/logging"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/repo"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/service"
"git.gozeros.com/wsk/eightbf-analysis-go/internal/worker"
@@ -20,59 +29,120 @@ import (
func main() {
cfg := config.Load()
log.Printf("eightbf-analysis-go starting port=%d mode=%s auth=%v demo=%v",
cfg.Port, cfg.AnalysisMode, cfg.AuthEnabled, cfg.DemoEnabled)
closeLog := logging.Setup(logging.Options{
Level: cfg.LogLevel,
File: cfg.LogFile,
MaxMB: cfg.LogMaxMB,
MaxAge: cfg.LogMaxAgeDays,
MaxBak: cfg.LogMaxBackups,
})
defer closeLog()
db, err := sql.Open("mysql", cfg.DBDSN)
warns, err := cfg.Validate()
if err != nil {
log.Fatalf("open db: %v", err)
logging.L.Errorw("配置校验失败", "err", err)
os.Exit(1)
}
for _, w := range warns {
logging.L.Warnw("配置告警", "msg", w)
}
logging.L.Infow("eightbf-analysis-go 启动",
"version", cfg.Version,
"port", cfg.Port,
"mode", cfg.AnalysisMode,
"auth", cfg.AuthEnabled,
"demo", cfg.DemoEnabled,
"logFile", cfg.LogFile,
)
db, err := sqlx.Open("mysql", cfg.DBDSN)
if err != nil {
logging.L.Errorw("打开数据库失败", "err", err)
os.Exit(1)
}
db.SetMaxOpenConns(20)
db.SetMaxIdleConns(5)
db.SetConnMaxLifetime(time.Hour)
if err := db.Ping(); err != nil {
log.Fatalf("ping db: %v(请检查 EBA_DB_DSN / EBA_DB_*)", err)
logging.L.Errorw("数据库不可达", "err", err)
os.Exit(1)
}
defer db.Close()
jobRepo := repo.NewJobRepo(db)
if schema, err := os.ReadFile("schema.sql"); err != nil {
log.Printf("未找到 schema.sql,跳过自动建表: %v", err)
if schema, err := os.ReadFile(cfg.SchemaPath); err != nil {
logging.L.Warnw("未找到 schema,跳过自动建表", "path", cfg.SchemaPath, "err", err)
} else {
jobRepo.ApplySchema(string(schema))
}
access := &service.AccessService{AuthEnabled: cfg.AuthEnabled}
jobs := &service.JobService{Repo: jobRepo, Access: access}
h := &api.Handler{Jobs: jobs}
h := &api.Handler{
Jobs: jobs,
DB: db,
Version: cfg.Version,
Started: time.Now(),
}
mux := http.NewServeMux()
h.Routes(mux)
// Release 模式关掉 Gin 自带的逐条访问日志。作业页会轮询 GET,那些行对排障没有帮助。
if strings.EqualFold(cfg.LogLevel, "debug") {
gin.SetMode(gin.DebugMode)
} else {
gin.SetMode(gin.ReleaseMode)
}
engine := gin.New()
engine.Use(api.RequestID(), api.CORS(), api.Recovery())
h.Register(engine)
if cfg.AnalysisMode == "live" && !cfg.DemoEnabled {
auth := deerflow.NewAuth(cfg)
df := deerflow.NewClient(cfg, auth)
runner := &worker.LiveChainRunner{DF: df, Repo: jobRepo, Coord: cfg.DeerFlowCoordinator}
lw := &worker.LiveWorker{
Repo: jobRepo,
Runner: runner,
PollEvery: time.Duration(cfg.LivePollMS) * time.Millisecond,
Repo: jobRepo,
Runner: runner,
PollEvery: time.Duration(cfg.LivePollMS) * time.Millisecond,
MaxInflight: int32(cfg.LiveMaxInflight),
}
lw.Start()
log.Printf("LiveWorker 已启动 poll=%dms", cfg.LivePollMS)
logging.L.Infow("LiveWorker 已启动", "pollMs", cfg.LivePollMS, "maxInflight", cfg.LiveMaxInflight)
} else if cfg.DemoEnabled || cfg.AnalysisMode == "demo" {
dw := &service.DemoWorker{
Repo: jobRepo,
Every: time.Duration(cfg.DemoTickMS) * time.Millisecond,
}
dw.Start()
log.Printf("DemoWorker 已启动 tick=%dms", cfg.DemoTickMS)
logging.L.Infow("DemoWorker 已启动", "tickMs", cfg.DemoTickMS)
} else {
log.Printf("警告:未启动任何 Worker(mode=%s demo=%v)", cfg.AnalysisMode, cfg.DemoEnabled)
logging.L.Warnw("未启动任何 Worker", "mode", cfg.AnalysisMode, "demo", cfg.DemoEnabled)
}
addr := ":" + strconv.Itoa(cfg.Port)
log.Printf("监听 %s (健康检查 GET /api/v1/analysis/health)", addr)
if err := http.ListenAndServe(addr, api.WithCORS(mux)); err != nil {
log.Fatal(err)
srv := &http.Server{
Addr: addr,
Handler: engine,
ReadHeaderTimeout: 10 * time.Second,
}
go func() {
logging.L.Infow("HTTP 监听中", "addr", addr)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logging.L.Errorw("HTTP 服务异常退出", "err", err)
os.Exit(1)
}
}()
// 优雅停机:停接收新连接,给进行中请求一点时间(Runner 仍可能在后台跑完当前席)。
stop := make(chan os.Signal, 1)
signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
sig := <-stop
logging.L.Infow("收到退出信号,开始优雅停机", "signal", sig.String())
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
logging.L.Warnw("优雅停机超时或失败", "err", err)
}
logging.L.Infow("服务已退出")
}