jinql
/
gocron
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

增加异步任务回调接口

pull/21/merge
ouqiang 2017-05-04 10:47:14 +08:00
parent bc56743a63
commit 90322c1230
12 changed files with 144 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
README.md
View File

@ -39,7 +39,8 @@
2. `go get -d https://github.com/ouqiang/gocron`
3. 编译 `go build`
4. 启动、访问方式同上
5. 生成压缩包（Windows: gocron.zip, 其他平台: gocron.tar.gz） ./build.sh -p 平台 -a CPU架构 例 ./build.sh -p darwin -a amd64
5. 生成压缩包（Windows: gocron.zip, 其他平台: gocron.tar.gz）
> ./build.sh -p 平台 -a CPU架构 例 ./build.sh -p darwin -a amd64
### 启动可选参数

1
models/model.go
View File

@ -26,6 +26,7 @@ const (
Running Status = 1 // 运行中
Finish Status = 2 // 完成
Cancel Status = 3 // 取消
Background Status = 4 // 后台运行
)
const (

9
models/task_log.go
View File

@ -21,7 +21,8 @@ type TaskLog struct {
Hostname string `xorm:"varchar(128) notnull defalut '' "` // SSH主机名，逗号分隔
StartTime time.Time `xorm:"datetime created"` // 开始执行时间
EndTime time.Time `xorm:"datetime updated"` // 执行完成（失败）时间
Status Status `xorm:"tinyint notnull default 1"` // 状态 0:执行失败 1:执行中 2:执行完毕 3:任务取消(上次任务未执行完成)
Status Status `xorm:"tinyint notnull default 1"` // 状态 0:执行失败 1:执行中 2:执行完毕 3:任务取消(上次任务未执行完成) 4:异步执行
NotifyId string `xorm:"varchar(32) notnull default '' "` // 回调通知ID
Result string `xorm:"mediumtext notnull defalut '' "` // 执行结果
TotalTime int `xorm:"-"` // 执行总时长
BaseModel `xorm:"-"`
@ -41,6 +42,12 @@ func (taskLog *TaskLog) Update(id int64, data CommonMap) (int64, error) {
return Db.Table(taskLog).ID(id).Update(data)
}
func (taskLog *TaskLog) UpdateStatus(notifyId string, status Status, result string) (int64, error) {
taskLog.Status = status
taskLog.Result = result
return Db.Cols("status,result").Where("notify_id = ?", notifyId).Update(taskLog)
}
func (taskLog *TaskLog) setStatus(id int64, status Status) (int64, error) {
return taskLog.Update(id, CommonMap{"status": status})
}

10
modules/ssh/ssh.go
View File

@ -86,8 +86,14 @@ func Exec(sshConfig SSHConfig, cmd string) (output string, err error) {
return "", err
}
defer session.Close()
if sshConfig.ExecTimeout <= 0 {
// 后台运行
if sshConfig.ExecTimeout < 0 {
go session.CombinedOutput(cmd)
time.Sleep(5 * time.Second)
return "", nil
}
// 不限制超时
if sshConfig.ExecTimeout == 0 {
outputByte, execErr := session.CombinedOutput(cmd)
output = string(outputByte)
err = execErr

7
routers/routers.go
View File

@ -87,6 +87,11 @@ func Register(m *macaron.Macaron) {
m.Get("/login-log", loginlog.Index)
})
// API
m.Group("/api/v1", func() {
m.Get("/tasklog/update-status", tasklog.UpdateStatus)
});
// 404错误
m.NotFound(func(ctx *macaron.Context) {
if isGetRequest(ctx) && !isAjaxRequest(ctx) {
@ -162,7 +167,7 @@ func userAuth(m *macaron.Macaron) {
}
uri := ctx.Req.URL.Path
found := false
excludePaths := []string{"/install", "/user/login"}
excludePaths := []string{"/install", "/user/login", "/"}
for _, path := range excludePaths {
if strings.HasPrefix(uri, path) {
found = true

2
routers/task/task.go
View File

@ -20,7 +20,7 @@ type TaskForm struct {
Spec string `binding:"Required;MaxSize(64)"`
Protocol models.TaskProtocol `binding:"In(1,2,3)"`
Command string `binding:"Required;MaxSize(512)"`
Timeout int `binding:"Range(0,86400)"`
Timeout int `binding:"Range(-1,86400)"`
Multi int8 `binding:"In(1,2)"`
RetryTimes int8
HostId int16

24
routers/tasklog/task_log.go
View File

@ -49,6 +49,30 @@ func Clear(ctx *macaron.Context) string {
return json.Success(utils.SuccessContent, nil)
}
// 更新任务状态
func UpdateStatus(ctx *macaron.Context) string {
id := ctx.QueryTrim("id")
status := ctx.QueryInt("status")
result := ctx.QueryTrim("result")
json := utils.JsonResponse{}
if id == "" {
return json.CommonFailure("任务ID不能为空")
}
if status != 1 && status != 2 {
return json.CommonFailure("status值错误")
}
if status == 1 {
status -= 1
}
taskLogModel := new(models.TaskLog)
affectRows, err := taskLogModel.UpdateStatus(id, models.Status(status), result)
if err != nil || affectRows == 0 {
return json.CommonFailure("更新任务状态失败")
}
return json.Success("success", nil)
}
// 解析查询参数
func parseQueryParams(ctx *macaron.Context) (models.CommonMap) {
var params models.CommonMap = models.CommonMap{}

133
service/task.go
View File

@ -46,6 +46,7 @@ type TaskResult struct {
Result string
Err error
RetryTimes int8
IsAsync bool
}
// 初始化任务, 从数据库取出所有任务, 添加到定时任务并运行
@ -180,7 +181,7 @@ func (h *SSHCommandHandler) Run(taskModel models.TaskHost) (string, error) {
}
// 创建任务日志
func createTaskLog(taskModel models.TaskHost, status models.Status) (int64, error) {
func createTaskLog(taskModel models.TaskHost, status models.Status) (int64, string, error) {
taskLogModel := new(models.TaskLog)
taskLogModel.TaskId = taskModel.Id
taskLogModel.Name = taskModel.Task.Name
@ -193,9 +194,15 @@ func createTaskLog(taskModel models.TaskHost, status models.Status) (int64, erro
}
taskLogModel.StartTime = time.Now()
taskLogModel.Status = status
// SSH执行远程命令，后台运行
var notifyId string = ""
if taskModel.Timeout == -1 && taskModel.Protocol == models.TaskSSH {
notifyId = utils.RandString(32);
taskLogModel.NotifyId = notifyId;
}
insertId, err := taskLogModel.Create()
return insertId, err
return insertId, notifyId, err
}
// 更新任务日志
@ -205,6 +212,8 @@ func updateTaskLog(taskLogId int64, taskResult TaskResult) (int64, error) {
var result string = taskResult.Result
if taskResult.Err != nil {
status = models.Failure
} else if taskResult.IsAsync {
status = models.Background
} else {
status = models.Finish
}
@ -222,55 +231,12 @@ func createJob(taskModel models.TaskHost) cron.FuncJob {
return nil
}
taskFunc := func() {
if taskModel.Multi == 0 && runInstance.has(taskModel.Id) {
createTaskLog(taskModel, models.Cancel)
return
}
if taskModel.Multi == 0 {
runInstance.add(taskModel.Id)
defer runInstance.done(taskModel.Id)
}
taskLogId, err := createTaskLog(taskModel, models.Running)
if err != nil {
logger.Error("任务开始执行#写入任务日志失败-", err)
taskLogId := beforeExecJob(&taskModel)
if taskLogId <= 0 {
return
}
taskResult := execJob(handler, taskModel)
if taskResult.Err != nil {
taskResult.Result = taskResult.Err.Error() + "\n" + taskResult.Result
}
_, err = updateTaskLog(taskLogId, taskResult)
if err != nil {
logger.Error("任务结束#更新任务日志失败-", err)
}
var statusName string
var enableNotify bool = true
// 未开启通知
if taskModel.NotifyStatus == 0 {
enableNotify = false;
} else if taskModel.NotifyStatus == 1 && taskResult.Err == nil {
// 执行失败才发送通知
enableNotify = false
}
if taskResult.Err != nil {
statusName = "失败"
} else {
statusName = "成功"
}
if !enableNotify {
return
}
// 发送通知
msg := notify.Message{
"task_type": taskModel.NotifyType,
"task_receiver_id": taskModel.NotifyReceiverId,
"name": taskModel.Task.Name,
"output": taskResult.Result,
"status": statusName,
"taskId": taskModel.Id,
};
notify.Push(msg)
afterExecJob(taskModel, taskResult, taskLogId)
}
return taskFunc
@ -290,7 +256,78 @@ func createHandler(taskModel models.TaskHost) Handler {
return handler;
}
func beforeExecJob(taskModel *models.TaskHost) (taskLogId int64) {
if taskModel.Multi == 0 && runInstance.has(taskModel.Id) {
createTaskLog(*taskModel, models.Cancel)
return
}
if taskModel.Multi == 0 {
runInstance.add(taskModel.Id)
}
taskLogId, notifyId, err := createTaskLog(*taskModel, models.Running)
if err != nil {
logger.Error("任务开始执行#写入任务日志失败-", err)
return
}
// 设置notifyId到环境变量中
if notifyId != "" {
taskModel.Command = fmt.Sprintf("export GOCRON_TASK_ID=%s;%s", notifyId, taskModel.Command)
}
return taskLogId
}
func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId int64) {
if taskResult.Err != nil {
taskResult.Result = taskResult.Err.Error() + "\n" + taskResult.Result
}
if taskModel.Protocol == models.TaskSSH && taskModel.Timeout == -1 {
taskResult.IsAsync = true
}
_, err := updateTaskLog(taskLogId, taskResult)
if err != nil {
logger.Error("任务结束#更新任务日志失败-", err)
}
if taskResult.IsAsync {
return
}
sendNotification(taskModel, taskResult)
}
// 发送任务结果通知
func sendNotification(taskModel models.TaskHost, taskResult TaskResult) {
var statusName string
// 未开启通知
if taskModel.NotifyStatus == 0 {
return
}
if taskModel.NotifyStatus == 1 && taskResult.Err == nil {
// 执行失败才发送通知
return
}
if taskResult.Err != nil {
statusName = "失败"
} else {
statusName = "成功"
}
// 发送通知
msg := notify.Message{
"task_type": taskModel.NotifyType,
"task_receiver_id": taskModel.NotifyReceiverId,
"name": taskModel.Task.Name,
"output": taskResult.Result,
"status": statusName,
"taskId": taskModel.Id,
};
notify.Push(msg)
}
// 执行具体任务
func execJob(handler Handler, taskModel models.TaskHost) TaskResult {
if taskModel.Multi == 0 {
defer runInstance.done(taskModel.Id)
}
// 默认只运行任务一次
var execTimes int8 = 1
if (taskModel.RetryTimes > 0) {

1
templates/common/header.html
View File

@ -65,6 +65,7 @@
{{{if gt .LoginUid 0}}}
<a class="item {{{if eq .Controller "manage"}}}active{{{end}}}" href="/manage/slack/edit"><i class="settings icon"></i>管理</a>
{{{end}}}
<a class="item" href="https://github.com/ouqiang/gocron/wiki" target="_blank"><i class="file text icon"></i>查看文档</a>
</div>
</div>
</div>

4
templates/task/index.html
View File

@ -32,7 +32,7 @@
</div>
<div class="field">
<select name="protocol" id="protocol">
<option value="0">选择协议</option>
<option value="0">执行方式</option>
<option value="3" {{{if eq .Params.Protocol 3}}}selected{{{end}}}>系统命令</option>
<option value="2" {{{if eq .Params.Protocol 2}}}selected{{{end}}} data-match="host_id" data-validate-type="selectProtocol">SSH</option>
<option value="1" {{{if eq .Params.Protocol 1}}}selected{{{end}}}>HTTP</option>
@ -59,7 +59,7 @@
</h5>
<p>任务ID： <span class="stress">{{{.Id}}}</span></p>
<p>cron表达式： {{{.Spec}}}</p>
<p>协议: {{{if eq .Protocol 1}}} HTTP {{{else if eq .Protocol 2}}} SSH {{{else if eq .Protocol 3}}}本地命令{{{end}}}</p>
<p>执行方式: {{{if eq .Protocol 1}}} HTTP {{{else if eq .Protocol 2}}} SSH {{{else if eq .Protocol 3}}}本地命令{{{end}}}</p>
<p class="sensorStatus">命令：{{{.Command}}}</p>
<p class="sensorStatus">超时时间：{{{if gt .Timeout 0}}}{{{.Timeout}}}秒{{{else}}}不限制{{{end}}}</p>
<p>重试次数: {{{.RetryTimes}}}</p>

5
templates/task/log.html
View File

@ -42,6 +42,7 @@
<option value="2" {{{if eq .Params.Status 1}}}selected{{{end}}}>执行中</option>
<option value="3" {{{if eq .Params.Status 2}}}selected{{{end}}}>成功</option>
<option value="4" {{{if eq .Params.Status 3}}}selected{{{end}}}>取消</option>
<option value="5" {{{if eq .Params.Status 3}}}selected{{{end}}}>后台运行</option>
</select>
</div>
<div class="field">
@ -75,7 +76,7 @@
<td>{{{.RetryTimes}}}</td>
<td>{{{.Hostname}}}</td>
<td>
{{{if ne .Status 3}}}
{{{if and (ne .Status 3) (ne .Status 4)}}}
{{{if gt .TotalTime 0}}}{{{.TotalTime}}}秒{{{else}}}1秒{{{end}}}<br>
开始时间: {{{.StartTime.Format "2006-01-02 15:04:05" }}}<br>
{{{if ne .Status 1}}}
@ -92,6 +93,8 @@
<span style="color:red">失败</span>
{{{else if eq .Status 3}}}
<span style="color:#4499EE">取消</span>
{{{else if eq .Status 4}}}
<span style="color:#43A102">后台运行</span>
{{{end}}}
</td>
<td>

42
templates/task/task_form.html
View File

@ -28,25 +28,6 @@
<label>
<div class="content">
crontab表达式
<div class="ui blue message">
Linux-crontab时间表达式语法, 支持秒级任务定义 <br>
格式: 秒 分 时 天 月 周 <br>
示例：<br>
1 * * * * * 每分钟第一秒运行 <br>
*/20 * * * * * 每隔20秒运行一次 <br>
0 30 21 * * * 每天晚上21:30:00运行一次 <br>
0 0 23 * * 6 每周六晚上23:00:00 运行一次 <br>
快捷语法: <br>
@yearly 每年运行一次 <br>
@monthly 每月运行一次 <br>
@weekly 每周运行一次 <br>
@daily 每天运行一次 <br>
@midnight 每天午夜运行一次<br>
@hourly 每小时运行一次 <br>
@every 30s 每隔30秒运行一次 <br>
@every 1m20s 每隔1分钟20秒运行一次 <br>
@every 3h5m10s 每隔3小时5分钟10秒运行一次 <br>
</div>
</div>
</label>
<div class="ui small input">
@ -56,12 +37,7 @@
</div>
<div class="three fields">
<div class="field">
<label>协议</label>
<div class="ui blue message">
系统命令: 调用本机系统命令 <br>
SSH: 通过SSH执行远程命令 <br>
HTTP: 执行HTTP-GET请求 <br>
</div>
<label>执行方式</label>
<select name="protocol" id="protocol">
<option value="3" {{{if .Task}}} {{{if eq .Task.Protocol 3}}}selected{{{end}}} {{{end}}}>系统命令</option>
<option value="2" {{{if .Task}}} {{{if eq .Task.Protocol 2}}}selected{{{end}}} {{{end}}} data-match="host_id" data-validate-type="selectProtocol">SSH</option>
@ -87,32 +63,18 @@
<div class="two fields">
<div class="field">
<label>命令</label>
<div class="ui blue message">
根据选择的协议输入相应的命令 <br>
系统命令 - ifconfig -a <br>
SSH - netstat -natpu <br>
HTTP - URL地址 例: http://golang.org <br>
</div>
<textarea rows="5" name="command">{{{.Task.Command}}}</textarea>
</div>
</div>
<div class="three fields">
<div class="field">
<label>任务超时时间(秒)</label>
<div class="ui blue message">
默认0，不限制超时
</div>
<input type="text" name="timeout" value="{{{.Task.Timeout}}}">
</div>
</div>
<div class="two fields">
<div class="field">
<label>任务重试次数</label>
<div class="ui blue message">
无法连接远程主机，shell返回值非0, http响应码非200等异常返回，可重复执行任务,
重试时间间隔 重试次数 * 分钟, 按1分钟、2分钟、3分钟.....的间隔进行重试
取值范围1-10, 默认0，不重试
</div>
<label>任务失败重试次数</label>
<input type="text" name="retry_times" value="{{{.Task.RetryTimes}}}">
</div>
</div>