247 lines
6.9 KiB
Go
247 lines
6.9 KiB
Go
package main
|
||
|
||
import (
|
||
"bufio"
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"os"
|
||
"os/exec"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
// scriptTimeout 脚本执行超时上限（6 小时）
|
||
const scriptTimeout = 6 * time.Hour
|
||
|
||
// buildTask 单次构建任务的快照,handler 入队时构造。
|
||
type buildTask struct {
|
||
scriptPath string
|
||
repoName string
|
||
ref string
|
||
commitID string
|
||
}
|
||
|
||
// debounceEntry 单 key 的冷却去重状态。
|
||
type debounceEntry struct {
|
||
mu sync.Mutex
|
||
lastExec time.Time
|
||
}
|
||
|
||
var (
|
||
// buildCh 单槽队列;handler enqueue 始终非阻塞,worker 唯一消费者
|
||
buildCh = make(chan buildTask, 1)
|
||
// lastExecMap 冷却去重状态,key: "repoName:ref", value: *debounceEntry
|
||
lastExecMap sync.Map
|
||
// 防抖冷却时间限制:3 分钟
|
||
debounceDuration = 3 * time.Minute
|
||
)
|
||
|
||
// enqueueBuildTask 入队;队列已满时丢老取新(drop-oldest)
|
||
func enqueueBuildTask(t buildTask) {
|
||
select {
|
||
case buildCh <- t:
|
||
// 入队成功后再登记到 dashboard，保证 channel 与列表一致
|
||
if dashboard != nil {
|
||
dashboard.AddPending(t)
|
||
}
|
||
return
|
||
default:
|
||
// 队列已满,丢弃旧 task
|
||
select {
|
||
case <-buildCh:
|
||
if dashboard != nil {
|
||
// 同步移除最旧一条，避免 pending 列表与 channel 错位
|
||
dashboard.RemovePendingAndPeek(t)
|
||
}
|
||
default:
|
||
}
|
||
buildCh <- t // 阻塞不会发生,因为我们是唯一发送方
|
||
if dashboard != nil {
|
||
dashboard.AddPending(t)
|
||
}
|
||
}
|
||
}
|
||
|
||
// startBuildWorker 启动单 worker goroutine 持续消费 buildCh
|
||
func startBuildWorker() {
|
||
go consumeBuild()
|
||
}
|
||
|
||
// consumeBuild worker 主循环;取出 task,按冷却规则执行或跳过
|
||
func consumeBuild() {
|
||
for task := range buildCh {
|
||
if dashboard != nil {
|
||
dashboard.RemovePendingAndPeek(task)
|
||
}
|
||
|
||
lockKey := fmt.Sprintf("%s:%s", task.repoName, task.ref)
|
||
entryVal, _ := lastExecMap.LoadOrStore(lockKey, &debounceEntry{})
|
||
entry := entryVal.(*debounceEntry)
|
||
|
||
entry.mu.Lock()
|
||
now := time.Now()
|
||
if !entry.lastExec.IsZero() && now.Sub(entry.lastExec) < debounceDuration {
|
||
remainingSeconds := int((debounceDuration - now.Sub(entry.lastExec)).Seconds())
|
||
if remainingSeconds < 0 {
|
||
remainingSeconds = 0
|
||
}
|
||
entry.mu.Unlock()
|
||
log.Printf("[SKIP] 3 分钟冷却中,跳过构建 [%s] commit %s 剩余 %d 秒\n",
|
||
lockKey, shortCommit(task.commitID), remainingSeconds)
|
||
// 冷却跳过也写入历史，方便追踪
|
||
if dashboard != nil {
|
||
dashboard.StartExecution(task)
|
||
dashboard.FinishExecution(task, statusSkipped,
|
||
fmt.Sprintf("冷却中,剩余 %d 秒", remainingSeconds),
|
||
time.Duration(0))
|
||
}
|
||
continue
|
||
}
|
||
entry.lastExec = now
|
||
entry.mu.Unlock()
|
||
|
||
log.Printf("[INFO] 开始执行脚本: %s (仓库: %s, 分支: %s, commit: %s)\n",
|
||
task.scriptPath, task.repoName, task.ref, shortCommit(task.commitID))
|
||
go runScript(task.scriptPath, task.repoName, task.ref, task.commitID)
|
||
}
|
||
}
|
||
|
||
// shortCommit 安全地取 commit 短 ID，避免空值/短值切片越界 panic
|
||
func shortCommit(commitID string) string {
|
||
if len(commitID) > 7 {
|
||
return commitID[:7]
|
||
}
|
||
if commitID == "" {
|
||
return "unknown"
|
||
}
|
||
return commitID
|
||
}
|
||
|
||
// streamLog 逐行读取并实时打印脚本输出，直到 r 关闭
|
||
// onLine 在每行非空时回调一次（携带换行符）
|
||
func streamLog(r io.Reader, tag string, onLine func(line string)) {
|
||
br := bufio.NewReader(r)
|
||
for {
|
||
line, err := br.ReadString('\n')
|
||
if line != "" {
|
||
log.Printf("[SCRIPT %s] %s", tag, strings.TrimRight(line, "\r\n"))
|
||
if onLine != nil {
|
||
onLine(line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// notifyResult 发送执行结果通知
|
||
func notifyResult(ok bool, repoName, ref, commitID string, cost time.Duration, reason string) {
|
||
state := "成功"
|
||
if !ok {
|
||
state = "失败"
|
||
}
|
||
title := fmt.Sprintf("%s更新%s", repoName, state)
|
||
body := fmt.Sprintf("分支: %s\nCommit: %s\n耗时: %s\n原因: %s\n服务器已接收到 Push，更新%s",
|
||
ref, shortCommit(commitID), cost.Round(time.Second), reason, state)
|
||
|
||
if err := SendNotification(pushURL, globalConfig.DeviceKeys, title, body); err != nil {
|
||
log.Printf("[WARNING] 推送通知失败: %v", err)
|
||
return
|
||
}
|
||
log.Println("[INFO] 推送通知发送成功")
|
||
}
|
||
|
||
// runScript 异步执行 Shell 脚本，输出实时流式打印，超时上限 scriptTimeout
|
||
func runScript(scriptPath, repoName, ref, commitID string) {
|
||
task := buildTask{
|
||
scriptPath: scriptPath,
|
||
repoName: repoName,
|
||
ref: ref,
|
||
commitID: commitID,
|
||
}
|
||
|
||
start := time.Now()
|
||
tag := fmt.Sprintf("%s:%s", repoName, ref)
|
||
log.Printf("[INFO] 开始执行脚本: %s (仓库: %s, 分支: %s) 超时上限 %s",
|
||
scriptPath, repoName, ref, scriptTimeout)
|
||
|
||
if dashboard != nil {
|
||
dashboard.StartExecution(task)
|
||
}
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), scriptTimeout)
|
||
defer cancel()
|
||
|
||
cmd := exec.CommandContext(ctx, "bash", scriptPath)
|
||
// 超时/取消后最多再给 10 秒让进程优雅退出；超时则强制 SIGKILL
|
||
cmd.WaitDelay = 10 * time.Second
|
||
|
||
// 传入环境变量，脚本内可读取
|
||
cmd.Env = append(os.Environ(),
|
||
fmt.Sprintf("GIT_REPO_NAME=%s", repoName),
|
||
fmt.Sprintf("GIT_REF=%s", ref),
|
||
fmt.Sprintf("GIT_COMMIT_ID=%s", commitID),
|
||
)
|
||
|
||
// stdout / stderr 合并到同一个管道，保持原有交错顺序
|
||
pr, pw := io.Pipe()
|
||
cmd.Stdout = pw
|
||
cmd.Stderr = pw
|
||
|
||
if err := cmd.Start(); err != nil {
|
||
pw.Close()
|
||
pr.Close()
|
||
log.Printf("[ERROR] 脚本启动失败 [%s]: %v", scriptPath, err)
|
||
notifyResult(false, repoName, ref, commitID, time.Since(start), fmt.Sprintf("启动失败: %v", err))
|
||
if dashboard != nil {
|
||
dashboard.FinishExecution(task, statusFailed,
|
||
fmt.Sprintf("启动失败: %v", err), time.Since(start))
|
||
}
|
||
return
|
||
}
|
||
|
||
var wg sync.WaitGroup
|
||
wg.Go(func() {
|
||
streamLog(pr, tag, func(line string) {
|
||
if dashboard != nil {
|
||
dashboard.AppendLog(line)
|
||
}
|
||
})
|
||
})
|
||
|
||
err := cmd.Wait() // 进程退出 + 内部 copy goroutine 结束
|
||
pw.Close() // 关闭写端，读端收到 EOF
|
||
wg.Wait() // 等日志全部刷完再往下走
|
||
|
||
cost := time.Since(start)
|
||
|
||
// 先判断超时（ctx.Err() 在超时/取消时非 nil）
|
||
switch {
|
||
case errors.Is(ctx.Err(), context.DeadlineExceeded):
|
||
log.Printf("[TIMEOUT] 脚本执行超时 [%s] 已运行 %s 上限 %s", scriptPath, cost.Round(time.Second), scriptTimeout)
|
||
notifyResult(false, repoName, ref, commitID, cost, "执行超时，已强制终止")
|
||
if dashboard != nil {
|
||
dashboard.FinishExecution(task, statusTimeout, "执行超时，已强制终止", cost)
|
||
}
|
||
return
|
||
case err != nil:
|
||
log.Printf("[ERROR] 脚本执行失败 [%s] 耗时 %s: %v", scriptPath, cost.Round(time.Millisecond), err)
|
||
notifyResult(false, repoName, ref, commitID, cost, err.Error())
|
||
if dashboard != nil {
|
||
dashboard.FinishExecution(task, statusFailed, err.Error(), cost)
|
||
}
|
||
return
|
||
}
|
||
|
||
log.Printf("[SUCCESS] 脚本执行成功 [%s] 耗时 %s", scriptPath, cost.Round(time.Millisecond))
|
||
notifyResult(true, repoName, ref, commitID, cost, "执行成功")
|
||
if dashboard != nil {
|
||
dashboard.FinishExecution(task, statusSuccess, "执行成功", cost)
|
||
}
|
||
}
|