427 lines
11 KiB
Go
427 lines
11 KiB
Go
package scheduler
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"os/exec"
|
||
"runtime"
|
||
"strings"
|
||
"time"
|
||
|
||
"ppgo_job/dao"
|
||
"ppgo_job/model/entity"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
)
|
||
|
||
// TaskResult 任务执行结果
|
||
type TaskResult struct {
|
||
Output string
|
||
Error string
|
||
IsOk bool
|
||
IsTimeout bool
|
||
}
|
||
|
||
// executeTask 执行任务主逻辑
|
||
func (s *schedulerService) executeTask(ctx context.Context, task *entity.Task) {
|
||
startTime := time.Now()
|
||
g.Log().Infof(ctx, "开始执行任务 [%d] %s", task.Id, task.TaskName)
|
||
|
||
var finalResult *TaskResult
|
||
serverName := "本地服务器"
|
||
var serverId int
|
||
|
||
if task.TaskType == "http" {
|
||
// HTTP 任务：直接从调度器所在机器发起请求
|
||
result := s.execHTTP(ctx, task)
|
||
finalResult = result
|
||
} else {
|
||
// Shell 任务：按原有逻辑执行（本地/远程）
|
||
serverIds := strings.Split(task.ServerIds, ",")
|
||
for _, sid := range serverIds {
|
||
sid = strings.TrimSpace(sid)
|
||
if sid == "" {
|
||
continue
|
||
}
|
||
|
||
var result *TaskResult
|
||
if sid == "0" {
|
||
serverName = "本地服务器"
|
||
serverId = 0
|
||
result = s.execLocal(ctx, task.Command, task.Timeout)
|
||
} else {
|
||
if srv, _ := dao.TaskServer.GetById(ctx, parseInt(sid)); srv != nil {
|
||
serverName = srv.ServerName
|
||
serverId = srv.Id
|
||
}
|
||
result = s.execRemote(ctx, task.Command, task.Timeout, parseInt(sid))
|
||
}
|
||
|
||
if finalResult == nil || !result.IsOk {
|
||
finalResult = result
|
||
}
|
||
}
|
||
}
|
||
|
||
if finalResult == nil {
|
||
finalResult = &TaskResult{IsOk: true}
|
||
}
|
||
|
||
elapsed := time.Since(startTime).Milliseconds()
|
||
|
||
// 记录日志
|
||
log := &entity.TaskLog{
|
||
TaskId: task.Id,
|
||
ServerId: serverId,
|
||
ServerName: serverName,
|
||
Output: finalResult.Output,
|
||
Error: finalResult.Error,
|
||
ProcessTime: int(elapsed),
|
||
CreateTime: time.Now().Unix(),
|
||
}
|
||
|
||
if finalResult.IsTimeout {
|
||
log.Status = -2
|
||
} else if !finalResult.IsOk {
|
||
log.Status = -1
|
||
} else {
|
||
log.Status = 0
|
||
}
|
||
|
||
_, _ = dao.TaskLog.Insert(ctx, log)
|
||
|
||
// 更新任务统计
|
||
_ = dao.Task.UpdateFields(ctx, task.Id, g.Map{"prev_time": startTime.Unix()})
|
||
_, _ = g.DB().Exec(ctx, "UPDATE pp_task SET execute_times = COALESCE(execute_times, 0) + 1 WHERE id = ?", task.Id)
|
||
|
||
// 失败时通知
|
||
if log.Status < 0 && task.IsNotify == 1 {
|
||
s.sendNotify(ctx, task, log)
|
||
}
|
||
|
||
g.Log().Infof(ctx, "任务 [%d] %s 执行完成, 耗时 %dms", task.Id, task.TaskName, elapsed)
|
||
}
|
||
|
||
// execLocal 本地执行命令
|
||
func (s *schedulerService) execLocal(ctx context.Context, command string, timeoutSec int) *TaskResult {
|
||
var cmd *exec.Cmd
|
||
if runtime.GOOS == "windows" {
|
||
cmd = exec.Command("CMD", "/C", command)
|
||
} else {
|
||
cmd = exec.Command("sh", "-c", command)
|
||
}
|
||
|
||
result := &TaskResult{}
|
||
done := make(chan error, 1)
|
||
|
||
go func() {
|
||
out, err := cmd.CombinedOutput()
|
||
result.Output = string(out)
|
||
done <- err
|
||
}()
|
||
|
||
timeout := time.Duration(timeoutSec) * time.Second
|
||
if timeoutSec <= 0 {
|
||
timeout = 24 * time.Hour
|
||
}
|
||
|
||
select {
|
||
case err := <-done:
|
||
result.IsOk = (err == nil)
|
||
if err != nil {
|
||
result.Error = err.Error()
|
||
}
|
||
case <-time.After(timeout):
|
||
_ = cmd.Process.Kill()
|
||
result.IsTimeout = true
|
||
result.Error = fmt.Sprintf("任务执行超过 %d 秒，已强制终止", timeoutSec)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
// execRemote 远程执行（SSH）
|
||
func (s *schedulerService) execRemote(ctx context.Context, command string, timeoutSec int, serverId int) *TaskResult {
|
||
if serverId <= 0 {
|
||
return s.execLocal(ctx, command, timeoutSec)
|
||
}
|
||
|
||
server, err := dao.TaskServer.GetById(ctx, serverId)
|
||
if err != nil || server == nil {
|
||
return &TaskResult{IsOk: false, Error: fmt.Sprintf("服务器不存在: %d", serverId)}
|
||
}
|
||
|
||
g.Log().Infof(ctx, "远程执行 -> %s@%s:%d", server.ServerAccount, server.ServerIp, server.Port)
|
||
|
||
sshClient := &SSHClient{}
|
||
connectErr := sshClient.Connect(server.ServerIp, server.Port, server.ServerAccount, server.Password, server.PrivateKeySrc, server.Type)
|
||
if connectErr != nil {
|
||
return &TaskResult{IsOk: false, Error: "SSH连接失败: " + connectErr.Error()}
|
||
}
|
||
defer sshClient.Close()
|
||
|
||
result := &TaskResult{}
|
||
stdout, stderr, runErr := sshClient.Exec(command, timeoutSec)
|
||
result.Output = stdout
|
||
result.Error = stderr
|
||
|
||
if runErr != nil {
|
||
if strings.Contains(runErr.Error(), "timeout") {
|
||
result.IsTimeout = true
|
||
} else {
|
||
result.IsOk = false
|
||
}
|
||
if stderr == "" {
|
||
result.Error = runErr.Error()
|
||
}
|
||
} else {
|
||
result.IsOk = true
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
// execHTTP 执行 HTTP 请求任务
|
||
func (s *schedulerService) execHTTP(ctx context.Context, task *entity.Task) *TaskResult {
|
||
result := &TaskResult{}
|
||
client := &http.Client{Timeout: time.Duration(task.Timeout) * time.Second}
|
||
if task.Timeout <= 0 {
|
||
client.Timeout = 30 * time.Second
|
||
}
|
||
|
||
method := strings.ToUpper(task.Method)
|
||
if method == "" {
|
||
method = "GET"
|
||
}
|
||
|
||
var bodyReader io.Reader
|
||
if task.Body != "" && (method == "POST" || method == "PUT" || method == "PATCH") {
|
||
bodyReader = bytes.NewBufferString(task.Body)
|
||
}
|
||
|
||
req, err := http.NewRequest(method, task.Url, bodyReader)
|
||
if err != nil {
|
||
result.Error = "创建请求失败: " + err.Error()
|
||
return result
|
||
}
|
||
|
||
// 解析并设置自定义 Header
|
||
if task.Headers != "" {
|
||
var headers map[string]string
|
||
if err := json.Unmarshal([]byte(task.Headers), &headers); err == nil {
|
||
for k, v := range headers {
|
||
req.Header.Set(k, v)
|
||
}
|
||
} else {
|
||
result.Error = "Headers 格式错误（需为 JSON 对象）"
|
||
return result
|
||
}
|
||
}
|
||
|
||
// 默认 Content-Type
|
||
if req.Header.Get("Content-Type") == "" && bodyReader != nil {
|
||
req.Header.Set("Content-Type", "application/json")
|
||
}
|
||
|
||
g.Log().Infof(ctx, "HTTP请求 -> %s %s (超时=%ds)", method, task.Url, task.Timeout)
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
if isTimeoutError(err) {
|
||
result.IsTimeout = true
|
||
result.Error = fmt.Sprintf("请求超时（%d秒）: %s", task.Timeout, task.Url)
|
||
} else {
|
||
result.Error = "请求失败: " + err.Error()
|
||
}
|
||
return result
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
bodyBytes, _ := io.ReadAll(resp.Body)
|
||
bodyStr := string(bodyBytes)
|
||
|
||
// 截断过长的输出
|
||
if len(bodyStr) > 10000 {
|
||
bodyStr = bodyStr[:10000] + "\n... (truncated)"
|
||
}
|
||
|
||
statusLine := fmt.Sprintf("[HTTP %d]", resp.StatusCode)
|
||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||
result.Output = statusLine + " " + bodyStr
|
||
|
||
// 尝试解析 JSON 响应体，检测应用层错误
|
||
if appErr := detectJSONError(bodyBytes); appErr != "" {
|
||
result.IsOk = false
|
||
result.Error = fmt.Sprintf("%s %s", statusLine, appErr)
|
||
} else {
|
||
result.IsOk = true
|
||
}
|
||
} else {
|
||
result.IsOk = false
|
||
result.Error = statusLine + " " + bodyStr
|
||
}
|
||
|
||
g.Log().Infof(ctx, "HTTP响应 -> %s %s -> %s (ok=%v)", method, task.Url, statusLine, result.IsOk)
|
||
return result
|
||
}
|
||
|
||
// detectJSONError 解析 JSON 响应体，递归检测应用层错误。
|
||
// 支持常见 API 格式：
|
||
//
|
||
// {"code":1,...} — code ≠ 0 表示错误
|
||
// {"success":false,...} — success=false 表示错误
|
||
// {"status":"error",...} — status="error"/"fail" 表示错误
|
||
// {"errno":-1,...} — errno ≠ 0 表示错误
|
||
// {"error":"..."} — error 非空字符串表示错误
|
||
// {"data":{"success":false}} — 嵌套在 data 字段中的错误也会检测
|
||
//
|
||
// 返回空字符串表示无错误或无法解析。
|
||
func detectJSONError(body []byte) string {
|
||
body = bytes.TrimSpace(body)
|
||
if len(body) == 0 || body[0] != '{' {
|
||
return ""
|
||
}
|
||
var data map[string]interface{}
|
||
if err := json.Unmarshal(body, &data); err != nil {
|
||
return ""
|
||
}
|
||
return detectJSONErrorMap(data)
|
||
}
|
||
|
||
// detectJSONErrorMap 递归检测 map 中的应用层错误。
|
||
// 优先级：error 字段 > success 字段 > code 字段 > status 字段 > errno 字段 > 嵌套 data。
|
||
func detectJSONErrorMap(data map[string]interface{}) string {
|
||
// 1. 检查 error 字段（字符串且非空）
|
||
if errVal, ok := data["error"]; ok {
|
||
switch v := errVal.(type) {
|
||
case string:
|
||
if v != "" {
|
||
return fmt.Sprintf("业务错误: %s", v)
|
||
}
|
||
case map[string]interface{}:
|
||
// error 可能是一个对象，如 {"error": {"message": "..."}}
|
||
if msg, hasMsg := v["message"]; hasMsg {
|
||
return fmt.Sprintf("业务错误: %v", msg)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 2. 检查 success 字段（明确 false 表示错误，true 表示成功）
|
||
if success, ok := data["success"]; ok {
|
||
if b, ok := success.(bool); ok {
|
||
if !b {
|
||
msg := extractMsg(data, "message")
|
||
if msg != "" {
|
||
return fmt.Sprintf("业务错误(success=false): %s", msg)
|
||
}
|
||
return "业务错误(success=false)"
|
||
}
|
||
// success=true 明确表示成功，无需继续检查
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// 3. 检查 code 字段（数字类型，非 0 表示错误 — 国内 API 最常见格式）
|
||
if code, ok := data["code"]; ok {
|
||
switch v := code.(type) {
|
||
case float64:
|
||
if v != 0 {
|
||
msg := extractMsg(data, "message")
|
||
if msg != "" {
|
||
return fmt.Sprintf("业务错误(code=%.0f): %s", v, msg)
|
||
}
|
||
return fmt.Sprintf("业务错误(code=%.0f)", v)
|
||
}
|
||
// code=0 可能是 GoFrame 等框架的全局成功码，
|
||
// 不返回 success，继续检查嵌套 data
|
||
}
|
||
}
|
||
|
||
// 4. 检查 status 字段
|
||
if status, ok := data["status"]; ok {
|
||
if s, ok := status.(string); ok {
|
||
switch s {
|
||
case "error", "fail", "failed":
|
||
msg := extractMsg(data, "message")
|
||
if msg != "" {
|
||
return fmt.Sprintf("业务错误(status=%s): %s", s, msg)
|
||
}
|
||
return fmt.Sprintf("业务错误(status=%s)", s)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 5. 检查 errno 字段（部分 API 用 errno/errmsg 表示错误）
|
||
if errno, ok := data["errno"]; ok {
|
||
switch v := errno.(type) {
|
||
case float64:
|
||
if v != 0 {
|
||
msg := extractMsg(data, "errmsg", "message")
|
||
if msg != "" {
|
||
return fmt.Sprintf("业务错误(errno=%.0f): %s", v, msg)
|
||
}
|
||
return fmt.Sprintf("业务错误(errno=%.0f)", v)
|
||
}
|
||
// errno=0 明确表示成功
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// 6. 递归检查嵌套的 data 字段
|
||
if nested, ok := data["data"].(map[string]interface{}); ok {
|
||
if errMsg := detectJSONErrorMap(nested); errMsg != "" {
|
||
return errMsg
|
||
}
|
||
}
|
||
|
||
return ""
|
||
}
|
||
|
||
// extractMsg 从 JSON 对象中按优先级提取消息字段的值。
|
||
func extractMsg(data map[string]interface{}, keys ...string) string {
|
||
for _, key := range keys {
|
||
if v, ok := data[key]; ok {
|
||
if s, ok := v.(string); ok && s != "" {
|
||
return s
|
||
}
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// isTimeoutError 判断是否为超时错误
|
||
func isTimeoutError(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
// Go http.Client 超时返回 context.DeadlineExceeded
|
||
if errors.Is(err, context.DeadlineExceeded) {
|
||
return true
|
||
}
|
||
// 也检查 net.Error.Timeout()
|
||
var netErr interface{ Timeout() bool }
|
||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||
return true
|
||
}
|
||
// 兜底：关键字匹配
|
||
msg := err.Error()
|
||
return strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline exceeded")
|
||
}
|
||
|
||
// parseInt 简单字符串转整数
|
||
func parseInt(s string) int {
|
||
id := 0
|
||
for _, c := range s {
|
||
if c >= '0' && c <= '9' {
|
||
id = id*10 + int(c-'0')
|
||
} else {
|
||
break
|
||
}
|
||
}
|
||
return id
|
||
}
|