diff --git a/build.sh b/build.sh index 8bfc1a0..cb8d66c 100644 --- a/build.sh +++ b/build.sh @@ -2,11 +2,11 @@ # set -x -u # 构建应用, 生成压缩包 gocron.zip或gocron.tar.gz -# ./build.sh -p windows -a amd64 +# ./build.sh -p windows -a amd64 -v 1.4 # 参数含义 # -p 指定平台(windows|linux|darwin) # -a 指定体系架构(amd64|386), 默认amd64 - +# -v 版本号 TEMP_DIR=`date +%s`-temp-`echo $RANDOM` diff --git a/build_node.sh b/build_node.sh index 1f4c3a2..6748610 100644 --- a/build_node.sh +++ b/build_node.sh @@ -2,10 +2,11 @@ # set -x -u # 任务节点打包, 生成压缩包 gocron-node.zip或gocron-node.tar.gz -# ./build-node.sh -p windows -a amd64 +# ./build-node.sh -p windows -a amd64 -v 1.4 # 参数含义 # -p 指定平台(windows|linux|darwin) # -a 指定体系架构(amd64|386), 默认amd64 +# -v 版本号 # 目标平台 windows,linux,darwin diff --git a/cmd/web.go b/cmd/web.go index 1e59ec1..32bccaa 100644 --- a/cmd/web.go +++ b/cmd/web.go @@ -150,7 +150,7 @@ func shutdown() { serviceTask := new(service.Task) // 停止所有任务调度 logger.Info("停止定时任务调度") - serviceTask.Stop() + serviceTask.WaitAndExit() } // 判断应用是否需要升级, 当存在版本号文件且版本小于app.VersionId时升级 diff --git a/modules/app/app.go b/modules/app/app.go index 0a82a73..5551438 100644 --- a/modules/app/app.go +++ b/modules/app/app.go @@ -36,7 +36,7 @@ func InitEnv(versionString string) { DataDir = AppDir + "/data" AppConfig = ConfDir + "/app.ini" VersionFile = ConfDir + "/.version" - createDirIfNeed(ConfDir, LogDir, DataDir) + createDirIfNotExists(ConfDir, LogDir, DataDir) Installed = IsInstalled() VersionId = ToNumberVersion(versionString) } @@ -108,7 +108,7 @@ func ToNumberVersion(versionString string) int { } // 检测目录是否存在 -func createDirIfNeed(path ...string) { +func createDirIfNotExists(path ...string) { for _, value := range path { if !utils.FileExist(value) { err := os.Mkdir(value, 0755) diff --git a/modules/rpc/client/client.go b/modules/rpc/client/client.go index d6d1b51..fce14b9 100644 --- a/modules/rpc/client/client.go +++ b/modules/rpc/client/client.go @@ -10,12 +10,30 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" "time" + "sync" +) + +var ( + taskMap sync.Map ) var ( errUnavailable = errors.New("无法连接远程服务器") ) +func generateTaskUniqueKey(ip string, port int, id int64) string { + return fmt.Sprintf("%s:%d:%d", ip, port, id) +} + +func Stop(ip string, port int , id int64) { + key := generateTaskUniqueKey(ip, port, id) + cancel, ok := taskMap.Load(key) + if !ok { + return + } + cancel.(context.CancelFunc)() +} + func Exec(ip string, port int, taskReq *pb.TaskRequest) (string, error) { defer func() { if err := recover(); err != nil { @@ -39,7 +57,9 @@ func Exec(ip string, port int, taskReq *pb.TaskRequest) (string, error) { } timeout := time.Duration(taskReq.Timeout) * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() + taskUniqueKey := generateTaskUniqueKey(ip, port, taskReq.Id) + taskMap.Store(taskUniqueKey, cancel) + defer taskMap.Delete(taskUniqueKey) resp, err := c.Run(ctx, taskReq) if err != nil { return parseGRPCError(err, conn, &isConnClosed) @@ -60,6 +80,8 @@ func parseGRPCError(err error, conn *grpc.ClientConn, connClosed *bool) (string, return "", errUnavailable case codes.DeadlineExceeded: return "", errors.New("执行超时, 强制结束") + case codes.Canceled: + return "", errors.New("手动停止") } return "", err } diff --git a/modules/rpc/proto/task.pb.go b/modules/rpc/proto/task.pb.go index 9f0a810..cc9282f 100644 --- a/modules/rpc/proto/task.pb.go +++ b/modules/rpc/proto/task.pb.go @@ -36,6 +36,7 @@ const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package type TaskRequest struct { Command string `protobuf:"bytes,2,opt,name=command" json:"command,omitempty"` Timeout int32 `protobuf:"varint,3,opt,name=timeout" json:"timeout,omitempty"` + Id int64 `protobuf:"varint,4,opt,name=id" json:"id,omitempty"` } func (m *TaskRequest) Reset() { *m = TaskRequest{} } @@ -57,6 +58,13 @@ func (m *TaskRequest) GetTimeout() int32 { return 0 } +func (m *TaskRequest) GetId() int64 { + if m != nil { + return m.Id + } + return 0 +} + type TaskResponse struct { Output string `protobuf:"bytes,1,opt,name=output" json:"output,omitempty"` Error string `protobuf:"bytes,2,opt,name=error" json:"error,omitempty"` @@ -161,16 +169,17 @@ var _Task_serviceDesc = grpc.ServiceDesc{ func init() { proto.RegisterFile("task.proto", fileDescriptor0) } var fileDescriptor0 = []byte{ - // 170 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x4c, 0x8f, 0xb1, 0x0e, 0x82, 0x40, - 0x0c, 0x86, 0x45, 0x04, 0x63, 0x75, 0xd0, 0xc6, 0x98, 0x8b, 0x13, 0x61, 0x62, 0x30, 0x0c, 0xea, - 0xe8, 0xe2, 0x2b, 0x5c, 0x7c, 0x01, 0xc4, 0x1b, 0x0c, 0x81, 0x9e, 0xbd, 0xde, 0xfb, 0x1b, 0x38, - 0x48, 0x18, 0xbf, 0x36, 0xfd, 0xfa, 0xff, 0x00, 0x52, 0xb9, 0xa6, 0xb4, 0x4c, 0x42, 0x18, 0xb3, - 0xad, 0xf3, 0x27, 0x6c, 0x5f, 0x95, 0x6b, 0xb4, 0xf9, 0x79, 0xe3, 0x04, 0x15, 0xac, 0x6b, 0x6a, - 0xdb, 0xaa, 0xfb, 0xa8, 0x65, 0x16, 0x15, 0x1b, 0x3d, 0x61, 0xbf, 0x91, 0x6f, 0x6b, 0xc8, 0x8b, - 0x8a, 0xb3, 0xa8, 0x48, 0xf4, 0x84, 0xf9, 0x03, 0x76, 0x41, 0xe1, 0x2c, 0x75, 0xce, 0xe0, 0x09, - 0x52, 0xf2, 0x62, 0xbd, 0xa8, 0x68, 0x50, 0x8c, 0x84, 0x47, 0x48, 0x0c, 0x33, 0xf1, 0x68, 0x0e, - 0x70, 0xbd, 0xc3, 0xaa, 0xbf, 0xc6, 0x0b, 0xc4, 0xda, 0x77, 0xb8, 0x2f, 0xd9, 0xd6, 0xe5, 0x2c, - 0xd2, 0xf9, 0x30, 0x9b, 0x84, 0x0f, 0xf9, 0xe2, 0x9d, 0x0e, 0x15, 0x6e, 0xff, 0x00, 0x00, 0x00, - 0xff, 0xff, 0xef, 0x3c, 0x71, 0x6b, 0xd0, 0x00, 0x00, 0x00, + // 184 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x4c, 0x8f, 0xcf, 0x8a, 0x83, 0x30, + 0x10, 0xc6, 0x37, 0x46, 0x5d, 0x76, 0x76, 0x59, 0xda, 0xa1, 0x94, 0xd0, 0x93, 0x78, 0xf2, 0x50, + 0x3c, 0xb4, 0x3d, 0xf6, 0x25, 0x1a, 0xfa, 0x02, 0x56, 0x73, 0x10, 0xd1, 0x49, 0x93, 0xc9, 0xfb, + 0x17, 0xff, 0x81, 0xc7, 0xdf, 0x0c, 0xf3, 0xfd, 0xe6, 0x03, 0xe0, 0xca, 0x77, 0xa5, 0x75, 0xc4, + 0x84, 0xd2, 0xd9, 0x3a, 0x7f, 0xc0, 0xef, 0xb3, 0xf2, 0x9d, 0x36, 0xef, 0x60, 0x3c, 0xa3, 0x82, + 0xef, 0x9a, 0xfa, 0xbe, 0x1a, 0x1a, 0x15, 0x65, 0xa2, 0xf8, 0xd1, 0x2b, 0x8e, 0x1b, 0x6e, 0x7b, + 0x43, 0x81, 0x95, 0xcc, 0x44, 0x91, 0xe8, 0x15, 0xf1, 0x1f, 0xa2, 0xb6, 0x51, 0x71, 0x26, 0x0a, + 0xa9, 0xa3, 0xb6, 0xc9, 0xef, 0xf0, 0x37, 0x47, 0x7a, 0x4b, 0x83, 0x37, 0x78, 0x84, 0x94, 0x02, + 0xdb, 0xc0, 0x4a, 0x4c, 0x91, 0x0b, 0xe1, 0x01, 0x12, 0xe3, 0x1c, 0xb9, 0xc5, 0x34, 0xc3, 0xe5, + 0x06, 0xf1, 0x78, 0x8d, 0x67, 0x90, 0x3a, 0x0c, 0xb8, 0x2b, 0x9d, 0xad, 0xcb, 0xcd, 0x8b, 0xa7, + 0xfd, 0x66, 0x32, 0x1b, 0xf2, 0xaf, 0x57, 0x3a, 0x55, 0xba, 0x7e, 0x02, 0x00, 0x00, 0xff, 0xff, + 0xd7, 0x7f, 0x8a, 0x9d, 0xe0, 0x00, 0x00, 0x00, } diff --git a/modules/rpc/proto/task.proto b/modules/rpc/proto/task.proto index 235fd27..f03ae46 100644 --- a/modules/rpc/proto/task.proto +++ b/modules/rpc/proto/task.proto @@ -9,6 +9,7 @@ service Task { message TaskRequest { string command = 2; // 命令 int32 timeout = 3; // 任务执行超时时间 + int64 id = 4; // 执行任务唯一ID } message TaskResponse { diff --git a/routers/routers.go b/routers/routers.go index 55cf84d..a6fc443 100644 --- a/routers/routers.go +++ b/routers/routers.go @@ -66,6 +66,7 @@ func Register(m *macaron.Macaron) { m.Get("", task.Index) m.Get("/log", tasklog.Index) m.Post("/log/clear", tasklog.Clear) + m.Post("/log/stop", tasklog.Stop) m.Post("/remove/:id", task.Remove) m.Post("/enable/:id", task.Enable) m.Post("/disable/:id", task.Disable) diff --git a/routers/tasklog/task_log.go b/routers/tasklog/task_log.go index 6fa5d79..1a0a838 100644 --- a/routers/tasklog/task_log.go +++ b/routers/tasklog/task_log.go @@ -11,6 +11,7 @@ import ( "github.com/ouqiang/gocron/routers/base" "gopkg.in/macaron.v1" "html/template" + "github.com/ouqiang/gocron/service" ) func Index(ctx *macaron.Context) { @@ -48,6 +49,31 @@ func Clear(ctx *macaron.Context) string { return json.Success(utils.SuccessContent, nil) } +// 停止运行中的任务 +func Stop(ctx *macaron.Context) string { + id := ctx.QueryInt64("id") + taskId := ctx.QueryInt("task_id") + taskModel := new(models.Task) + task, err := taskModel.Detail(taskId) + json := utils.JsonResponse{} + if err != nil { + return json.CommonFailure("获取任务信息失败#" + err.Error(), err) + } + if task.Protocol != models.TaskRPC { + return json.CommonFailure("仅支持SHELL任务手动停止") + } + if len(task.Hosts) == 0 { + return json.CommonFailure("任务节点列表为空") + } + serviceTask := new(service.Task) + for _, host := range task.Hosts { + serviceTask.Stop(host.Name, host.Port, id) + + } + + return json.Success("已执行停止操作, 请等待任务退出", nil); +} + // 删除N个月前的日志 func Remove(ctx *macaron.Context) string { month := ctx.ParamsInt(":id") diff --git a/service/task.go b/service/task.go index a4a2884..e5b20c7 100644 --- a/service/task.go +++ b/service/task.go @@ -14,6 +14,7 @@ import ( "strings" "sync" "time" + "net/http" ) // 定时任务调度管理器 @@ -127,8 +128,13 @@ func (task *Task) Add(taskModel models.Task) { } } -// 停止所有任务 -func (task *Task) Stop() { +// 停止运行中的任务 +func (task *Task) Stop(ip string, port int, id int64) { + rpcClient.Stop(ip, port, id) +} + +// 等待所有任务结束后退出 +func (task *Task) WaitAndExit() { Cron.Stop() taskCount.Exit() } @@ -139,7 +145,7 @@ func (task *Task) Run(taskModel models.Task) { } type Handler interface { - Run(taskModel models.Task) (string, error) + Run(taskModel models.Task, taskUniqueId int64) (string, error) } // HTTP任务 @@ -148,13 +154,13 @@ type HTTPHandler struct{} // http任务执行时间不超过300秒 const HttpExecTimeout = 300 -func (h *HTTPHandler) Run(taskModel models.Task) (result string, err error) { +func (h *HTTPHandler) Run(taskModel models.Task, taskUniqueId int64) (result string, err error) { if taskModel.Timeout <= 0 || taskModel.Timeout > HttpExecTimeout { taskModel.Timeout = HttpExecTimeout } resp := httpclient.Get(taskModel.Command, taskModel.Timeout) // 返回状态码非200,均为失败 - if resp.StatusCode != 200 { + if resp.StatusCode != http.StatusOK { return resp.Body, errors.New(fmt.Sprintf("HTTP状态码非200-->%d", resp.StatusCode)) } @@ -164,10 +170,11 @@ func (h *HTTPHandler) Run(taskModel models.Task) (result string, err error) { // RPC调用执行任务 type RPCHandler struct{} -func (h *RPCHandler) Run(taskModel models.Task) (result string, err error) { +func (h *RPCHandler) Run(taskModel models.Task, taskUniqueId int64) (result string, err error) { taskRequest := new(pb.TaskRequest) taskRequest.Timeout = int32(taskModel.Timeout) taskRequest.Command = taskModel.Command + taskRequest.Id = taskUniqueId var resultChan chan TaskResult = make(chan TaskResult, len(taskModel.Hosts)) for _, taskHost := range taskModel.Hosts { go func(th models.TaskHostDetail) { @@ -250,7 +257,7 @@ func createJob(taskModel models.Task) cron.FuncJob { return } logger.Infof("开始执行任务#%s#命令-%s", taskModel.Name, taskModel.Command) - taskResult := execJob(handler, taskModel) + taskResult := execJob(handler, taskModel, taskLogId) logger.Infof("任务完成#%s#命令-%s", taskModel.Name, taskModel.Command) afterExecJob(taskModel, taskResult, taskLogId) } @@ -371,7 +378,7 @@ func SendNotification(taskModel models.Task, taskResult TaskResult) { } // 执行具体任务 -func execJob(handler Handler, taskModel models.Task) TaskResult { +func execJob(handler Handler, taskModel models.Task, taskUniqueId int64) TaskResult { defer func() { if err := recover(); err != nil { logger.Error("panic#service/task.go:execJob#", err) @@ -389,7 +396,7 @@ func execJob(handler Handler, taskModel models.Task) TaskResult { var output string var err error for i < execTimes { - output, err = handler.Run(taskModel) + output, err = handler.Run(taskModel, taskUniqueId) if err == nil { return TaskResult{Result: output, Err: err, RetryTimes: i} } diff --git a/templates/task/log.html b/templates/task/log.html index d9f7fbf..6ff61ab 100644 --- a/templates/task/log.html +++ b/templates/task/log.html @@ -100,6 +100,11 @@ >查看结果 {{{end}}} + + + {{{if and (eq .Status 1) (eq .Protocol 2) }}} + + {{{end}}} {{{end}}} @@ -148,6 +153,14 @@ }).modal('refresh').modal('show'); } + function stopTask(id, taskId) { + util.confirm("确定要停止任务吗", function () { + util.post("/task/log/stop/", {id: id, task_id:taskId}, function () { + location.reload(); + }); + }); +} + function clearLog() { util.confirm("确定要删除所有日志吗?", function() { util.post("/task/log/clear",{}, function() { diff --git a/upload_package_to_qiniu.sh b/upload_package_to_qiniu.sh deleted file mode 100644 index 2a6a28c..0000000 --- a/upload_package_to_qiniu.sh +++ /dev/null @@ -1,40 +0,0 @@ -#!/usr/bin/env bash -# set -x -u -# 上传二进制包到七牛 - -if [[ -z $QINIU_ACCESS_KEY || -z $QINIU_SECRET_KEY || -z $QINIU_URL ]];then - echo 'QINIU_ACCESS_KEY | QINIU_SECRET_KEY | QINIU_URL is need' - exit 1 -fi - -# 打包 -for i in linux darwin windows -do - ./build.sh -p $i - if [[ $? != 0 ]];then - break - fi - ./build_node.sh -p $i - if [[ $? != 0 ]];then - break - fi -done - -# 身份认证 -qrsctl login $QINIU_ACCESS_KEY $QINIU_SECRET_KEY - -# 上传 -for i in `ls gocron*.gz gocron*.zip` -do - # 上传文件 qrsctl put bucket key srcFile - KEY=gocron/$i - qrsctl put github $KEY $i - if [[ $? != 0 ]];then - break - fi - echo "刷新七牛CDN-" $QINIU_URL/$KEY - qrsctl cdn/refresh $QINIU_URL/$KEY - rm $i -done - -echo '打包并上传成功' \ No newline at end of file