jinql
/
gocron
mirror of https://github.com/ouqiang/gocron
1
0
Fork 0
Code Issues Releases Wiki Activity

支持后台手动停止运行中的shell任务

pull/21/merge
ouqiang 2018-01-27 11:10:08 +08:00
parent fd39434446
commit f081d89fb4
12 changed files with 108 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
build.sh
View File

@ -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`

3
build_node.sh
View File

@ -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

2
cmd/web.go
View File

@ -150,7 +150,7 @@ func shutdown() {
serviceTask := new(service.Task)
// 停止所有任务调度
logger.Info("停止定时任务调度")
serviceTask.Stop()
serviceTask.WaitAndExit()
}
// 判断应用是否需要升级, 当存在版本号文件且版本小于app.VersionId时升级

4
modules/app/app.go
View File

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

24
modules/rpc/client/client.go
View File

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

33
modules/rpc/proto/task.pb.go
View File

@ -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,
}

1
modules/rpc/proto/task.proto
View File

@ -9,6 +9,7 @@ service Task {
message TaskRequest {
string command = 2; // 命令
int32 timeout = 3; // 任务执行超时时间
int64 id = 4; // 执行任务唯一ID
}
message TaskResponse {

1
routers/routers.go
View File

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

26
routers/tasklog/task_log.go
View File

@ -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")

25
service/task.go
View File

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

13
templates/task/log.html
View File

@ -100,6 +100,11 @@
>查看结果
</button>
{{{end}}}
{{{if and (eq .Status 1) (eq .Protocol 2) }}}
<button class="ui small blue button" onclick="stopTask({{{.Id}}}, {{{.TaskId}}})">停止任务</button>
{{{end}}}
</td>
</tr>
{{{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() {

40
upload_package_to_qiniu.sh
View File

@ -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 '打包并上传成功'