feat($task): 支持任务同时在多个节点上运行

Closes #7

README.md

@@ -2,19 +2,21 @@
 # gocron - 定时任务管理系统
 
 # 项目简介
-使用Go语言开发的定时任务集中调度和管理系统, 用于替代Linux-crontab [查看文档](https://github.com/ouqiang/gocron/wiki)
+使用Go语言开发的定时任务集中调度和管理系统, 用于替代Linux-crontab [查看文档](https://github.com/ouqiang/gocron/wiki)  
+
+原有的延时任务拆分为独立项目[延迟队列](https://github.com/ouqiang/delay-queue)  
 
 ## 功能特性
-* Web界面管理定时任务, 支持动态添加、删除、编辑任务
+* Web界面管理定时任务, 支持动态添加、删除任务
 * crontab时间表达式，精确到秒
 * 任务执行失败重试设置
 * 任务超时设置
 * 任务依赖配置
 * 任务类型
     * shell任务
-    > 在远程服务器上执行shell命令
+    > 在任务节点上执行shell命令, 支持任务同时在多个节点上运行
     * HTTP任务
-    > 访问指定的URL地址
+    > 访问指定的URL地址, 由调度器直接执行, 不依赖任务节点
 * 查看任务执行日志
 * 任务执行结果通知, 支持邮件、Slack
 
@@ -31,36 +33,29 @@
 
 
 ## 下载
-* 调度器(管理后台)
-    * [Linux-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-linux-amd64.tar.gz)
-    * [Mac OS-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-darwin-amd64.tar.gz)
-    * [Windows-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-windows-amd64.zip)
-* 任务执行器(安装在远程主机上, 执行shell命令需安装)
-     * [Linux-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-node-linux-amd64.tar.gz)
-     * [Mac OS-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-node-darwin-amd64.tar.gz)
-     * [Windows-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-node-windows-amd64.zip)
+[v1.1](https://github.com/ouqiang/gocron/releases/tag/v1.1)
 
 
 ## 安装
 
 ###  二进制安装
-> Windows平台默认后台运行)
 1. 解压压缩包   
 2. `cd 解压目录`   
 3. 启动        
 * 调度器启动        
   * Windows: `gocron.exe web`   
   * Linux、Mac OS:  `./gocron web`
-* 任务执行器启动
-  * Windows:  `gocron-node.exe ip:port (默认0.0.0.0:5921)`            
-  * Linux、Mac OS:  `./gocron-node ip:port (默认0.0.0.0:5921)`   
+* 任务节点启动, 默认监听0.0.0.0:5921
+  * Windows:  `gocron-node.exe`
+  * Linux、Mac OS:  `./gocron-node`
 4. 浏览器访问 http://localhost:5920
+
 ### 源码安装
 1. `go`语言版本1.7+
 2. `go get -d github.com/ouqiang/gocron`
 3. 编译 
     * 调度器 `go build -tags gocron -o gocron`
-    * 任务执行器 `go build -tags node -o gocron-node`
+    * 任务节点 `go build -tags node -o gocron-node`
 4. 启动、访问方式同上
 
 ### 命令
@@ -69,14 +64,13 @@
     * --host 默认0.0.0.0
     * -p 端口, 指定端口, 默认5920
     * -e 指定运行环境, dev|test|prod, dev模式下可查看更多日志信息, 默认prod
-    * -d 后台运行
     * -h 查看帮助
-* gocron serv 
-    * -s stop|status stop:停止gocron status:查看运行状态
-* gocron-node ip:port, 默认0.0.0.0:5921 
+* gocron-node
+    *  -allow-root *nix平台允许以root用户运行
+    * -s ip:port 监听地址
 
 ## 程序使用的组件
-* web框架 [Macaron](http://go-macaron.com/)
+* Web框架 [Macaron](http://go-macaron.com/)
 * 定时任务调度 [Cron](https://github.com/robfig/cron)
 * ORM [Xorm](https://github.com/go-xorm/xorm)
 * UI框架 [Semantic UI](https://semantic-ui.com/)
@@ -85,3 +79,14 @@
 
 ## 反馈
 提交[issue](https://github.com/ouqiang/gocron/issues/new)
+
+## ChangeLog
+
+v1.1
+--------
+
+* 任务可同时在多个节点上运行
+* *nix平台默认禁止以root用户运行任务节点
+* 子任务命令中增加预定义占位符, 子任务可根据主任务运行结果执行相应操作
+* 删除守护进程模块
+* Web访问日志输出到终端

build.sh

@@ -54,7 +54,7 @@ fi
 
 echo '开始编译调度器'
 if [[ $OS = 'windows' ]];then
-    GOOS=$OS GOARCH=$ARCH go build -tags gocron -ldflags '-w -H windowsgui'
+    GOOS=$OS GOARCH=$ARCH go build -tags gocron -ldflags '-w'
 else
     GOOS=$OS GOARCH=$ARCH go build -tags gocron -ldflags '-w'
 fi

build_node.sh

@@ -60,7 +60,7 @@ fi
 
 echo '开始编译任务节点'
 if [[ $OS = 'windows' ]];then
-    GOOS=$OS GOARCH=$ARCH go build -tags node -ldflags '-w -H windowsgui' -o $EXEC_NAME
+    GOOS=$OS GOARCH=$ARCH go build -tags node -ldflags '-w' -o $EXEC_NAME
 else
     GOOS=$OS GOARCH=$ARCH go build -tags node -ldflags '-w' -o $EXEC_NAME
 fi

cmd/serv.go

@@ -1,67 +0,0 @@
-package cmd
-
-import (
-    "github.com/urfave/cli"
-    "fmt"
-    "github.com/ouqiang/gocron/modules/utils"
-    "github.com/ouqiang/gocron/modules/app"
-    "os"
-    "syscall"
-)
-
-var CmdServ = cli.Command{
-    Name:   "serv",
-    Usage:  "manage gocron, ./gocron serv -s stop|status",
-    Action: runServ,
-    Flags: []cli.Flag{
-        cli.StringFlag{
-            Name: "s",
-            Value:"",
-            Usage: "stop|status",
-        },
-    },
-}
-
-func runServ(ctx *cli.Context)  {
-    if utils.IsWindows() {
-        fmt.Println("not support on windows")
-        return
-    }
-    option := ctx.String("s")
-    if !utils.InStringSlice([]string{"stop", "status"}, option) {
-        fmt.Println("invalid option")
-        return
-    }
-    app.InitEnv()
-    pid := app.GetPid()
-    if pid <= 0 {
-        fmt.Println("not running")
-        return
-    }
-    process ,err := os.FindProcess(pid)
-    if err != nil {
-        fmt.Println("not running", err)
-        return
-    }
-    switch option {
-        case "stop":
-            stop(process)
-        case "status":
-            status(process)
-    }
-}
-
-func stop(process *os.Process)  {
-    fmt.Println("stopping gocron......")
-    err := process.Signal(syscall.SIGTERM)
-    if err != nil {
-        fmt.Println("failed to kill process", err)
-    } else {
-        fmt.Println("stopped")
-    }
-}
-
-
-func status(process *os.Process)  {
-    fmt.Printf("running, pid-[%d]\n", process.Pid)
-}

cmd/web.go

@@ -13,18 +13,12 @@ import (
     "github.com/ouqiang/gocron/models"
     "github.com/ouqiang/gocron/modules/setting"
     "time"
-    "io"
-    "fmt"
-    "path/filepath"
-    "os/exec"
-    "github.com/ouqiang/gocron/modules/utils"
     "github.com/ouqiang/gocron/modules/rpc/grpcpool"
 )
 
 // web服务器默认端口
 const DefaultPort = 5920
 
-const InitProcess = 1
 
 var CmdWeb = cli.Command{
     Name:   "web",
@@ -46,26 +40,19 @@ var CmdWeb = cli.Command{
             Value: "prod",
             Usage: "runtime environment, dev|test|prod",
         },
-        cli.BoolFlag{
-            Name: "d",
-            Usage: "-d=true, run as daemon process",
-        },
     },
 }
 
 func runWeb(ctx *cli.Context) {
-    // 设置守护进程
-    becomeDaemon(ctx)
     // 设置运行环境
     setEnvironment(ctx)
     // 初始化应用
     app.InitEnv()
-    app.WritePid()
     // 初始化模块 DB、定时任务等
     initModule()
     // 捕捉信号,配置热更新等
     go catchSignal()
-    m := macaron.NewWithLogger(getWebLogWriter())
+    m := macaron.Classic()
 
     // 注册路由
     routers.Register(m)
@@ -73,38 +60,9 @@ func runWeb(ctx *cli.Context) {
     routers.RegisterMiddleware(m)
     host := parseHost(ctx)
     port := parsePort(ctx)
-    fmt.Println("server start")
     m.Run(host, port)
 }
 
-func becomeDaemon(ctx *cli.Context) {
-    // 不支持windows
-    if utils.IsWindows() {
-        return
-    }
-    if !ctx.IsSet("d") {
-        return
-    }
-
-    if os.Getppid() == InitProcess {
-        // 子进程不再处理
-        return
-    }
-
-    filePath, _:= filepath.Abs(os.Args[0])
-    cmd := exec.Command(filePath, os.Args[1:]...)
-    cmd.Stdin = os.Stdin
-    cmd.Stdout = os.Stdout
-    cmd.Stderr = os.Stderr
-    err := cmd.Start()
-    if err != nil {
-        logger.Fatal("创建守护进程失败", err)
-    }
-
-    // 父进程退出, 子进程由init-1号进程收养
-    os.Exit(0)
-}
-
 func initModule()  {
     if !app.Installed {
         return
@@ -177,25 +135,10 @@ func catchSignal()  {
     }
 }
 
-func getWebLogWriter() io.Writer  {
-    if macaron.Env == macaron.DEV {
-        return os.Stdout
-    }
-    logFile := app.LogDir + "/access.log"
-    var err error
-    w, err := os.OpenFile(logFile, os.O_RDWR|os.O_CREATE|os.O_APPEND ,0666)
-    if err != nil {
-        fmt.Printf("日志文件[%s]打开失败", logFile)
-        panic(err)
-    }
-
-    return w
-}
 
 // 应用退出
 func shutdown()  {
     defer func() {
-        app.RemovePid()
         logger.Info("已退出")
         os.Exit(0)
     }()
@@ -208,8 +151,6 @@ func shutdown()  {
     // 停止所有任务调度
     logger.Info("停止定时任务调度")
     serviceTask.StopAll()
-    // 释放gRPC连接池
-    grpcpool.Pool.ReleaseAll()
 
     taskNumInRunning := service.TaskNum.Num()
     logger.Infof("正在运行的任务有%d个", taskNumInRunning)
@@ -223,4 +164,7 @@ func shutdown()  {
         time.Sleep(3 * time.Second)
         taskNumInRunning = service.TaskNum.Num()
     }
+
+    // 释放gRPC连接池
+    grpcpool.Pool.ReleaseAll()
 }

gocron-node.go

@@ -5,17 +5,33 @@ package main
 
 import (
 	"github.com/ouqiang/gocron/modules/rpc/server"
-	"os"
+    "flag"
+    "runtime"
+    "os"
+    "fmt"
 )
 
+const AppVersion = "1.1"
+
 func main()  {
-	var addr string
-	if (len(os.Args) < 2) {
-		addr = "0.0.0.0:5921"
-	} else {
-        addr = os.Args[1]
+	var serverAddr string
+    var allowRoot bool
+    var version bool
+    flag.BoolVar(&allowRoot, "allow-root", false, "./gocron-node -allow-root")
+    flag.StringVar(&serverAddr, "s", "0.0.0.0:5921", "./gocron-node -s ip:port")
+    flag.BoolVar(&version, "v", false, "./gocron-node -v")
+    flag.Parse()
+
+    if version {
+        fmt.Println(AppVersion)
+        os.Exit(0)
     }
-	for {
-		server.Start(addr)
-	}
+
+    if runtime.GOOS != "windows" && os.Getuid() == 0 && !allowRoot   {
+        fmt.Println("Do not run gocron-node as root user")
+        os.Exit(1)
+    }
+
+
+	server.Start(serverAddr)
 }

gocron.go

@@ -10,7 +10,7 @@ import (
     "github.com/ouqiang/gocron/cmd"
 )
 
-const AppVersion = "1.0"
+const AppVersion = "1.1"
 
 func main() {
     app := cli.NewApp()
@@ -19,7 +19,6 @@ func main() {
     app.Version = AppVersion
     app.Commands = []cli.Command{
         cmd.CmdWeb,
-        cmd.CmdServ,
     }
     app.Flags = append(app.Flags, []cli.Flag{}...)
     app.Run(os.Args)

models/host.go

@@ -12,6 +12,7 @@ type Host struct {
     Port      int       `xorm:"notnull default 22"`               // 主机端口
     Remark    string    `xorm:"varchar(100) notnull default '' "` // 备注
     BaseModel       `xorm:"-"`
+    Selected bool   `xorm:"-"`
 }
 
 // 新增

models/migration.go

@@ -15,7 +15,7 @@ func (migration *Migration) Exec(dbName string) error {
     setting := new(Setting)
     task := new(Task)
     tables := []interface{}{
-        &User{}, task, &TaskLog{}, &Host{}, setting,&LoginLog{},
+        &User{}, task, &TaskLog{}, &Host{}, setting,&LoginLog{},&TaskHost{},
     }
     for _, table := range tables {
         exist, err:= Db.IsTableExist(table)

models/model.go

@@ -54,8 +54,8 @@ func (model *BaseModel) parsePageAndPageSize(params CommonMap) {
     if model.Page <= 0 {
         model.Page = Page
     }
-    if model.PageSize <= 0 || model.PageSize > MaxPageSize {
-        model.PageSize = PageSize
+    if model.PageSize <= 0  {
+        model.PageSize = MaxPageSize
     }
 }
 

models/task.go

@@ -41,7 +41,6 @@ type Task struct {
     Timeout  int       `xorm:"mediumint notnull default 0"`      // 任务执行超时时间(单位秒),0不限制
     Multi    int8      `xorm:"tinyint notnull default 1"`        // 是否允许多实例运行
     RetryTimes int8    `xorm:"tinyint notnull default 0"`         // 重试次数
-    HostId   int16    `xorm:"smallint notnull index default 0"`        // RPC host id，
     NotifyStatus int8  `xorm:"smallint notnull default 1"`       // 任务执行结束是否通知 0: 不通知 1: 失败通知 2: 执行结束通知
     NotifyType int8 `xorm:"smallint notnull default 0"`  // 通知类型 1: 邮件 2: slack
     NotifyReceiverId string `xorm:"varchar(256) notnull default '' "` // 通知接受者ID, setting表主键ID，多个ID逗号分隔
@@ -50,17 +49,11 @@ type Task struct {
     Created  time.Time `xorm:"datetime notnull created"`         // 创建时间
     Deleted  time.Time `xorm:"datetime deleted"`                 // 删除时间
     BaseModel `xorm:"-"`
+    Hosts []TaskHostDetail `xorm:"-"`
 }
 
-type TaskHost struct {
-    Task `xorm:"extends"`
-    Name string
-    Port int
-    Alias string
-}
-
-func (TaskHost) TableName() string  {
-    return TablePrefix + "task"
+func taskHostTableName() []string {
+    return []string{TablePrefix + "task_host", "th"}
 }
 
 // 新增
@@ -88,7 +81,7 @@ func (task *Task) CreateTestTask() {
 
 func (task *Task) UpdateBean(id int) (int64, error)  {
     return Db.ID(id).
-    Cols("name,spec,protocol,command,timeout,multi,retry_times,host_id,remark,notify_status,notify_type,notify_receiver_id, dependency_task_id, dependency_status").
+    Cols("name,spec,protocol,command,timeout,multi,retry_times,remark,notify_status,notify_type,notify_receiver_id, dependency_task_id, dependency_status").
     Update(task)
 }
 
@@ -113,36 +106,48 @@ func (task *Task) Enable(id int) (int64, error) {
 }
 
 // 获取所有激活任务
-func (task *Task) ActiveList() ([]TaskHost, error) {
-    list := make([]TaskHost, 0)
-    fields := "t.*, host.alias,host.name,host.port"
-    err := Db.Alias("t").
-            Join("LEFT", hostTableName(), "t.host_id=host.id").
-            Where("t.status = ? AND t.level = ?", Enabled, TaskLevelParent).
-            Cols(fields).
+func (task *Task) ActiveList() ([]Task, error) {
+    list := make([]Task, 0)
+    err := Db.Where("status = ? AND level = ?", Enabled, TaskLevelParent).
             Find(&list)
 
-    return list, err
+    if err != nil {
+        return list, err
+    }
+
+    return task.setHostsForTasks(list)
 }
 
 // 获取某个主机下的所有激活任务
-func (task *Task) ActiveListByHostId(hostId int16) ([]TaskHost, error) {
-    list := make([]TaskHost, 0)
-    fields := "t.*, host.alias,host.name,host.port"
-    err := Db.Alias("t").
-        Join("LEFT", hostTableName(), "t.host_id=host.id").
-        Where("t.status = ? AND t.host_id = ? AND t.level = ?", Enabled, hostId, TaskLevelParent).
-        Cols(fields).
+func (task *Task) ActiveListByHostId(hostId int16) ([]Task, error) {
+    taskHostModel := new(TaskHost)
+    taskIds, err := taskHostModel.GetTaskIdsByHostId(hostId)
+    if err != nil {
+        return nil, err
+    }
+    list := make([]Task, 0)
+    err = Db.Where("status = ?  AND level = ?", Enabled, TaskLevelParent).
+        In("id", taskIds...).
         Find(&list)
+    if err != nil {
+        return list, err
+    }
 
-    return list, err
+    return task.setHostsForTasks(list)
 }
 
-// 判断主机id是否有引用
-func (task *Task) HostIdExist(hostId int16) (bool, error) {
-    count, err := Db.Where("host_id = ?", hostId).Count(task);
+func (task *Task) setHostsForTasks(tasks []Task) ([]Task, error) {
+    taskHostModel := new(TaskHost)
+    var err error
+    for i, value := range tasks {
+        taskHostDetails, err := taskHostModel.GetHostIdsByTaskId(value.Id)
+        if err != nil {
+            return nil, err
+        }
+        tasks[i].Hosts = taskHostDetails
+    }
 
-    return count > 0, err
+    return tasks, err
 }
 
 // 判断任务名称是否存在
@@ -168,28 +173,37 @@ func (task *Task) GetStatus(id int) (Status, error) {
     return task.Status, nil
 }
 
-func(task *Task) Detail(id int) (TaskHost, error)  {
-    taskHost := TaskHost{}
-    fields := "t.*, host.alias,host.name,host.port"
-    _, err := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id").Where("t.id=?", id).Cols(fields).Get(&taskHost)
+func(task *Task) Detail(id int) (Task, error)  {
+    t := Task{}
+    _, err := Db.Where("id=?", id).Get(&t)
 
-    return taskHost, err
+    if err != nil {
+        return t, err
+    }
+
+    taskHostModel := new(TaskHost)
+    t.Hosts, err = taskHostModel.GetHostIdsByTaskId(id)
+
+    return t, err
 }
 
-func (task *Task) List(params CommonMap) ([]TaskHost, error) {
+func (task *Task) List(params CommonMap) ([]Task, error) {
     task.parsePageAndPageSize(params)
-    list := make([]TaskHost, 0)
-    fields := "t.*, host.alias,host.name"
-    session := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id")
+    list := make([]Task, 0)
+    session := Db.Alias("t").Join("LEFT", taskHostTableName(), "t.id = th.task_id")
     task.parseWhere(session, params)
-    err := session.Cols(fields).Desc("t.id").Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
+    err := session.GroupBy("t.id").Desc("t.id").Cols("t.*").Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
 
-    return list, err
+    if err != nil {
+        return nil, err
+    }
+
+    return task.setHostsForTasks(list)
 }
 
 // 获取依赖任务列表
-func (task *Task) GetDependencyTaskList(ids string) ([]TaskHost, error) {
-    list := make([]TaskHost, 0)
+func (task *Task) GetDependencyTaskList(ids string) ([]Task, error) {
+    list := make([]Task, 0)
     if ids == "" {
         return list, nil
     }
@@ -198,21 +212,24 @@ func (task *Task) GetDependencyTaskList(ids string) ([]TaskHost, error) {
     for i,  v := range idList {
         taskIds[i] = v
     }
-    fields := "t.*, host.alias,host.name,host.port"
+    fields := "t.*"
     err := Db.Alias("t").
-            Join("LEFT", hostTableName(), "t.host_id=host.id").
             Where("t.level = ?", TaskLevelChild).
             In("t.id", taskIds).
             Cols(fields).
             Find(&list)
 
-    return list, err
+    if err != nil {
+        return list, err
+    }
+
+    return task.setHostsForTasks(list)
 }
 
 func (task *Task) Total(params CommonMap) (int64, error) {
-    session := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id")
+    session := Db.Alias("t").Join("LEFT", taskHostTableName(), "t.id = th.task_id")
     task.parseWhere(session, params)
-    return session.Count(task)
+    return session.GroupBy("t.id").Count(task)
 }
 
 // 解析where
@@ -226,7 +243,7 @@ func (task *Task) parseWhere(session *xorm.Session, params CommonMap)  {
     }
     hostId, ok := params["HostId"]
     if ok && hostId.(int) > 0 {
-        session.And("host_id = ?", hostId)
+        session.And("th.host_id = ?", hostId)
     }
     name, ok := params["Name"]
     if ok && name.(string) != "" {
@@ -242,6 +259,3 @@ func (task *Task) parseWhere(session *xorm.Session, params CommonMap)  {
     }
 }
 
-func hostTableName() []string {
-    return []string{TablePrefix + "host", "host"}
-}

models/task_host.go

@@ -0,0 +1,81 @@
+package models
+
+
+type TaskHost struct {
+    Id        int     `xorm:"int pk autoincr"`
+    TaskId    int     `xorm:"int not null index"`
+    HostId    int16   `xorm:"smallint not null index"`
+}
+
+type TaskHostDetail struct {
+    TaskHost `xorm:"extends"`
+    Name string
+    Port int
+    Alias string
+}
+
+func (TaskHostDetail) TableName() string  {
+    return TablePrefix + "task_host"
+}
+
+func hostTableName() []string {
+    return []string{TablePrefix + "host", "h"}
+}
+
+func (th *TaskHost) Remove(taskId int) error {
+    _, err := Db.Where("task_id = ?", taskId).Delete(new(TaskHost))
+
+    return err
+}
+
+func (th *TaskHost) Add(taskId int, hostIds []int) error {
+
+    err := th.Remove(taskId)
+    if err != nil {
+        return err
+    }
+
+    taskHosts := make([]TaskHost, len(hostIds))
+    for i, value := range hostIds {
+        taskHosts[i].TaskId = taskId
+        taskHosts[i].HostId = int16(value)
+    }
+
+    _, err = Db.Insert(&taskHosts)
+
+    return err
+}
+
+func (th *TaskHost) GetHostIdsByTaskId(taskId int) ([]TaskHostDetail, error) {
+    list := make([]TaskHostDetail, 0)
+    fields := "th.id,th.host_id,h.alias,h.name,h.port"
+    err := Db.Alias("th").
+        Join("LEFT", hostTableName(), "th.host_id=h.id").
+        Where("th.task_id = ?", taskId).
+        Cols(fields).
+        Find(&list)
+
+    return list, err
+}
+
+func (th *TaskHost) GetTaskIdsByHostId(hostId int16) ([]interface{}, error)  {
+    list := make([]TaskHost, 0)
+    err := Db.Where("host_id = ?", hostId).Cols("task_id").Find(&list)
+    if err != nil {
+        return nil, err
+    }
+
+    taskIds := make([]interface{}, len(list))
+    for i, value := range list {
+        taskIds[i] = value.TaskId
+    }
+
+    return taskIds, err
+}
+
+// 判断主机id是否有引用
+func (th *TaskHost) HostIdExist(hostId int16) (bool, error) {
+    count, err := Db.Where("host_id = ?", hostId).Count(th);
+
+    return count > 0, err
+}

modules/app/app.go

@@ -4,11 +4,8 @@ import (
     "os"
 
     "github.com/ouqiang/gocron/modules/logger"
-    "runtime"
     "github.com/ouqiang/gocron/modules/utils"
     "gopkg.in/ini.v1"
-    "io/ioutil"
-    "strconv"
 )
 
 var (
@@ -19,12 +16,10 @@ var (
     AppConfig    string // 应用配置文件
     Installed    bool   // 应用是否安装过
     Setting      *ini.Section // 应用配置
-    PidFile      string
 )
 
 
 func InitEnv() {
-    runtime.GOMAXPROCS(runtime.NumCPU())
     logger.InitLogger()
     wd, err := os.Getwd()
     if err != nil {
@@ -35,7 +30,6 @@ func InitEnv() {
     LogDir = AppDir + "/log"
     DataDir = AppDir + "/data"
     AppConfig = ConfDir + "/app.ini"
-    PidFile = LogDir + "/gocron.pid"
     checkDirExists(ConfDir, LogDir, DataDir)
     Installed = IsInstalled()
 }
@@ -50,33 +44,6 @@ func IsInstalled() bool {
     return true
 }
 
-func WritePid()  {
-    pid := os.Getpid()
-    pidStr := strconv.Itoa(pid)
-    err := ioutil.WriteFile(PidFile, []byte(pidStr), 0644)
-    if err != nil {
-        logger.Fatal("写入pid文件失败", err)
-    }
-}
-
-func GetPid() int {
-    bytes, err := ioutil.ReadFile(PidFile)
-    if err != nil {
-        return 0
-    }
-    pidStr := string(bytes)
-    pid, err := strconv.Atoi(pidStr)
-    if err != nil {
-        return 0
-    }
-
-    return pid
-}
-
-func RemovePid()  {
-    os.Remove(PidFile)
-}
-
 // 创建安装锁文件
 func CreateInstallLock() error {
     _, err := os.Create(ConfDir + "/install.lock")

modules/rpc/server/server.go

@@ -41,7 +41,7 @@ func Start(addr string)  {
     }
     s := grpc.NewServer()
     pb.RegisterTaskServer(s, Server{})
-    grpclog.Println("listen address ", addr)
+    grpclog.Println("listen ", addr)
     err = s.Serve(l)
     if err != nil {
         grpclog.Fatal(err)

routers/host/host.go

@@ -117,7 +117,7 @@ func Store(ctx *macaron.Context, form HostForm) string  {
             grpcpool.Pool.Release(oldAddr)
         }
 
-        taskModel := new(models.TaskHost)
+        taskModel := new(models.Task)
         tasks, err := taskModel.ActiveListByHostId(id)
         if  err != nil {
             return json.CommonFailure("刷新任务主机信息失败", err)
@@ -135,8 +135,8 @@ func Remove(ctx *macaron.Context) string  {
     if err != nil {
         return json.CommonFailure("参数错误", err)
     }
-    taskModel := new(models.Task)
-    exist,err := taskModel.HostIdExist(int16(id))
+    taskHostModel := new(models.TaskHost)
+    exist,err := taskHostModel.HostIdExist(int16(id))
     if err != nil {
         return json.CommonFailure("操作失败", err)
     }

routers/routers.go

@@ -19,6 +19,7 @@ import (
     "github.com/ouqiang/gocron/routers/loginlog"
     "time"
     "strconv"
+    "html/template"
 )
 
 // 静态文件目录
@@ -138,6 +139,14 @@ func RegisterMiddleware(m *macaron.Macaron) {
         IndentJSON: true,
         // 渲染具有缩进格式的 XML，默认为不缩进
         IndentXML: true,
+        Funcs: []template.FuncMap{map[string]interface{} {
+            "HostFormat": func(index int) bool {
+                return (index + 1) % 3 == 0
+            },
+            "unescape": func(str string) template.HTML {
+                return template.HTML(str)
+            },
+        }},
     }))
     m.Use(session.Sessioner(session.Options{
         Provider:       "file",

routers/task/task.go

@@ -28,7 +28,7 @@ type TaskForm struct {
     Timeout int `binding:"Range(0,86400)"`
     Multi  int8 `binding:"In(1,2)"`
     RetryTimes int8
-    HostId int16
+    HostId string
     Remark string
     NotifyStatus int8 `binding:"In(1,2,3)"`
     NotifyType int8 `binding:"In(1,2,3)"`
@@ -91,9 +91,22 @@ func Edit(ctx *macaron.Context)  {
         logger.Errorf("编辑任务#获取任务详情失败#任务ID-%d#%s", id, err.Error())
         ctx.Redirect("/task")
     }
+    hostModel := new(models.Host)
+    hostModel.PageSize = -1
+    hosts, err := hostModel.List(models.CommonMap{})
+    if err != nil {
+        logger.Error(err)
+    } else {
+        for i, host := range(hosts) {
+            if inHosts(task.Hosts, host.Id) {
+                hosts[i].Selected = true
+            }
+        }
+    }
+
     ctx.Data["Task"]  = task
+    ctx.Data["Hosts"] = hosts
     ctx.Data["Title"] = "编辑"
-    setHostsToTemplate(ctx)
     ctx.HTML(200, "task/task_form")
 }
 
@@ -110,15 +123,10 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
         return json.CommonFailure("任务名称已存在")
     }
 
-    if form.Protocol == models.TaskRPC && form.HostId <= 0 {
+    if form.Protocol == models.TaskRPC && form.HostId == "" {
         return json.CommonFailure("请选择主机名")
     }
 
-    if form.Protocol == models.TaskRPC {
-        taskModel.HostId = form.HostId
-    } else {
-        taskModel.HostId = 0
-    }
     taskModel.Name = form.Name
     taskModel.Protocol = form.Protocol
     taskModel.Command = form.Command
@@ -187,6 +195,18 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
         return json.CommonFailure("保存失败", err)
     }
 
+    taskHostModel := new(models.TaskHost)
+    if form.Protocol == models.TaskRPC {
+        hostIdStrList := strings.Split(form.HostId, ",")
+        hostIds := make([]int, len(hostIdStrList))
+        for i, hostIdStr := range hostIdStrList {
+            hostIds[i], _ = strconv.Atoi(hostIdStr)
+        }
+        taskHostModel.Add(id, hostIds)
+    } else {
+        taskHostModel.Remove(id)
+    }
+
     status, err := taskModel.GetStatus(id)
     if status == models.Enabled && taskModel.Level == models.TaskLevelParent {
         addTaskToTimer(id)
@@ -205,6 +225,9 @@ func Remove(ctx *macaron.Context) string {
         return json.CommonFailure(utils.FailureContent, err)
     }
 
+    taskHostModel := new(models.TaskHost)
+    taskHostModel.Remove(id)
+
     service.Cron.RemoveJob(strconv.Itoa(id))
 
     return json.Success(utils.SuccessContent, nil)
@@ -290,9 +313,20 @@ func parseQueryParams(ctx *macaron.Context) (models.CommonMap) {
 
 func setHostsToTemplate(ctx *macaron.Context)  {
     hostModel := new(models.Host)
+    hostModel.PageSize = -1
     hosts, err := hostModel.List(models.CommonMap{})
     if err != nil {
         logger.Error(err)
     }
     ctx.Data["Hosts"] = hosts
+}
+
+func inHosts(slice []models.TaskHostDetail, element int16) bool {
+    for _, v := range slice {
+        if v.HostId == element {
+            return true
+        }
+    }
+
+    return false
 }

service/task.go

@@ -14,6 +14,8 @@ import (
     rpcClient "github.com/ouqiang/gocron/modules/rpc/client"
     pb "github.com/ouqiang/gocron/modules/rpc/proto"
     "strings"
+    "text/template"
+    "bytes"
 )
 
 // 定时任务调度管理器
@@ -107,14 +109,14 @@ func (task *Task) Initialize() {
 }
 
 // 批量添加任务
-func (task *Task) BatchAdd(tasks []models.TaskHost)  {
+func (task *Task) BatchAdd(tasks []models.Task)  {
     for _, item := range tasks {
         task.Add(item)
     }
 }
 
 // 添加任务
-func (task *Task) Add(taskModel models.TaskHost) {
+func (task *Task) Add(taskModel models.Task) {
     if taskModel.Level == models.TaskLevelChild {
         logger.Errorf("添加任务失败#不允许添加子任务到调度器#任务Id-%d", taskModel.Id);
         return
@@ -140,12 +142,12 @@ func (task *Task) StopAll()  {
 }
 
 // 直接运行任务
-func (task *Task) Run(taskModel models.TaskHost)  {
+func (task *Task) Run(taskModel models.Task)  {
     go createJob(taskModel)()
 }
 
 type Handler interface {
-    Run(taskModel models.TaskHost) (string, error)
+    Run(taskModel models.Task) (string, error)
 }
 
 
@@ -155,7 +157,7 @@ type HTTPHandler struct{}
 // http任务执行时间不超过300秒
 const HttpExecTimeout = 300
 
-func (h *HTTPHandler) Run(taskModel models.TaskHost) (result string, err error) {
+func (h *HTTPHandler) Run(taskModel models.Task) (result string, err error) {
     if taskModel.Timeout <= 0 || taskModel.Timeout > HttpExecTimeout {
         taskModel.Timeout = HttpExecTimeout
     }
@@ -171,26 +173,54 @@ func (h *HTTPHandler) Run(taskModel models.TaskHost) (result string, err error)
 // RPC调用执行任务
 type RPCHandler struct {}
 
-func (h *RPCHandler) Run(taskModel models.TaskHost) (result string, err error)  {
+func (h *RPCHandler) Run(taskModel models.Task) (result string, err error)  {
     taskRequest := new(pb.TaskRequest)
     taskRequest.Timeout = int32(taskModel.Timeout)
     taskRequest.Command = taskModel.Command
+    var resultChan chan TaskResult = make(chan TaskResult, len(taskModel.Hosts))
+    for _, taskHost := range taskModel.Hosts {
+        go func(th models.TaskHostDetail) {
+            output, err := rpcClient.ExecWithRetry(th.Name, th.Port, taskRequest)
+            var errorMessage string = ""
+            if err != nil {
+                errorMessage = err.Error()
+            }
+            outputMessage := fmt.Sprintf("主机: [%s-%s]\n%s\n%s\n\n",
+                th.Alias, th.Name, errorMessage, output,
+            )
+            resultChan <- TaskResult{Err:err, Result: outputMessage}
+        }(taskHost)
+    }
 
-    return rpcClient.ExecWithRetry(taskModel.Name, taskModel.Port, taskRequest)
+    var aggregationErr error = nil
+    var aggregationResult string = ""
+    for i := 0; i < len(taskModel.Hosts); i++ {
+        taskResult := <- resultChan
+        aggregationResult += taskResult.Result
+        if taskResult.Err != nil {
+            aggregationErr = taskResult.Err
+        }
+    }
+
+    return aggregationResult, aggregationErr
 }
 
 
 // 创建任务日志
-func createTaskLog(taskModel models.TaskHost, status models.Status) (int64, error) {
+func createTaskLog(taskModel models.Task, status models.Status) (int64, error) {
     taskLogModel := new(models.TaskLog)
     taskLogModel.TaskId = taskModel.Id
-    taskLogModel.Name = taskModel.Task.Name
+    taskLogModel.Name = taskModel.Name
     taskLogModel.Spec = taskModel.Spec
     taskLogModel.Protocol = taskModel.Protocol
     taskLogModel.Command = taskModel.Command
     taskLogModel.Timeout = taskModel.Timeout
     if taskModel.Protocol == models.TaskRPC {
-        taskLogModel.Hostname = taskModel.Alias + "-" + taskModel.Name
+        var aggregationHost string = ""
+        for _, host := range taskModel.Hosts {
+            aggregationHost += fmt.Sprintf("%s-%s<br>", host.Alias, host.Name)
+        }
+        taskLogModel.Hostname = aggregationHost
     }
     taskLogModel.StartTime = time.Now()
     taskLogModel.Status = status
@@ -217,7 +247,7 @@ func updateTaskLog(taskLogId int64, taskResult TaskResult) (int64, error) {
 
 }
 
-func createJob(taskModel models.TaskHost) cron.FuncJob {
+func createJob(taskModel models.Task) cron.FuncJob {
     var handler Handler = createHandler(taskModel)
     if handler == nil {
         return nil
@@ -229,16 +259,16 @@ func createJob(taskModel models.TaskHost) cron.FuncJob {
         if taskLogId <= 0 {
             return
         }
-        logger.Infof("开始执行任务#%s#命令-%s", taskModel.Task.Name, taskModel.Command)
+        logger.Infof("开始执行任务#%s#命令-%s", taskModel.Name, taskModel.Command)
         taskResult := execJob(handler, taskModel)
-        logger.Infof("任务完成#%s#命令-%s", taskModel.Task.Name, taskModel.Command)
+        logger.Infof("任务完成#%s#命令-%s", taskModel.Name, taskModel.Command)
         afterExecJob(taskModel, taskResult, taskLogId)
     }
 
     return taskFunc
 }
 
-func createHandler(taskModel models.TaskHost) Handler  {
+func createHandler(taskModel models.Task) Handler  {
     var handler Handler = nil
     switch taskModel.Protocol {
         case models.TaskHTTP:
@@ -252,7 +282,7 @@ func createHandler(taskModel models.TaskHost) Handler  {
 }
 
 // 任务前置操作
-func beforeExecJob(taskModel models.TaskHost) (taskLogId int64)  {
+func beforeExecJob(taskModel models.Task) (taskLogId int64)  {
     if taskModel.Multi == 0 && runInstance.has(taskModel.Id) {
         createTaskLog(taskModel, models.Cancel)
         return
@@ -272,7 +302,7 @@ func beforeExecJob(taskModel models.TaskHost) (taskLogId int64)  {
 }
 
 // 任务执行后置操作
-func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId int64)  {
+func afterExecJob(taskModel models.Task, taskResult TaskResult, taskLogId int64)  {
     if taskResult.Err != nil {
         taskResult.Result = taskResult.Err.Error() + "\n" + taskResult.Result
     }
@@ -288,7 +318,7 @@ func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId in
 }
 
 // 执行依赖任务, 多个任务并发执行
-func execDependencyTask(taskModel models.TaskHost, taskResult TaskResult)  {
+func execDependencyTask(taskModel models.Task, taskResult TaskResult)  {
     // 父任务才能执行子任务
     if taskModel.Level != models.TaskLevelParent {
         return
@@ -316,16 +346,43 @@ func execDependencyTask(taskModel models.TaskHost, taskResult TaskResult)  {
     if len(tasks) == 0 {
         logger.Errorf("依赖任务列表为空#主任务ID-%d", taskModel.Id)
     }
-
     serviceTask := new(Task)
     for _, task := range tasks {
+        task.Command = appendResultToCommand(task.Command, taskResult)
         task.Spec = fmt.Sprintf("依赖任务(主任务ID-%d)", taskModel.Id)
         serviceTask.Run(task)
     }
 }
 
+/**
+ * 添加主任务执行结果到子任务命令中, 占位符{{.Code}} {{.Message}}
+ */
+func appendResultToCommand(command string, taskResult TaskResult) string {
+    var code int8 = 0
+    if taskResult.Err != nil {
+        code = 1
+    }
+    data := map[string]interface{} {
+        "Code": code,
+        "Message": taskResult.Result,
+    }
+    var buf *bytes.Buffer = new(bytes.Buffer)
+    tmpl, err := template.New("command").Parse(command)
+    if err != nil {
+        logger.Errorf("替换子任务命令占位符失败#%s", err.Error())
+        return command
+    }
+    err = tmpl.Execute(buf, data)
+    if err != nil {
+        logger.Errorf("替换子任务命令占位符失败#%s", err.Error())
+        return command
+    }
+
+    return buf.String()
+}
+
 // 发送任务结果通知
-func SendNotification(taskModel models.TaskHost, taskResult TaskResult)  {
+func SendNotification(taskModel models.Task, taskResult TaskResult)  {
     var statusName string
     // 未开启通知
     if taskModel.NotifyStatus == 0 {
@@ -347,7 +404,7 @@ func SendNotification(taskModel models.TaskHost, taskResult TaskResult)  {
     msg := notify.Message{
         "task_type": taskModel.NotifyType,
         "task_receiver_id": taskModel.NotifyReceiverId,
-        "name": taskModel.Task.Name,
+        "name": taskModel.Name,
         "output": taskResult.Result,
         "status": statusName,
         "taskId": taskModel.Id,
@@ -356,7 +413,7 @@ func SendNotification(taskModel models.TaskHost, taskResult TaskResult)  {
 }
 
 // 执行具体任务
-func execJob(handler Handler, taskModel models.TaskHost) TaskResult  {
+func execJob(handler Handler, taskModel models.Task) TaskResult  {
     defer func() {
        if err := recover(); err != nil {
            logger.Error("panic#service/task.go:execJob#", err)

templates/common/header.html

@@ -61,12 +61,11 @@
         <div class="bigcontainer">
             <div class="right menu">
                 <a class="item {{{if or (eq .Controller "task") (eq .Controller "delaytask")}}}active{{{end}}}" href="/task"><i class="tasks icon"></i>任务</a>
-                <a class="item {{{if eq .Controller "host"}}}active{{{end}}}" href="/host"><i class="linux icon"></i>主机</a>
+                <a class="item {{{if eq .Controller "host"}}}active{{{end}}}" href="/host"><i class="linux icon"></i>任务节点</a>
                 <!-- <a class="item {{{if eq .Controller "user"}}}active{{{end}}}" href="/user"><i class="user icon"></i>账户</a> -->
                 {{{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>

templates/host/host_form.html

@@ -29,7 +29,7 @@
                     </div>
                 </div>
                 <div class="field">
-                    <label>主机别名 (方便记忆和引用)</label>
+                    <label>节点名称</label>
                     <div class="ui small  input">
                         <input type="text" name="alias" value="{{{.Host.Alias}}}">
                     </div>

templates/host/index.html

@@ -10,7 +10,7 @@
                     <a href="/host/create">
                         <i class="large add icon"></i>
                         <div class="content">
-                            添加主机
+                            添加节点
                         </div>
                     </a>
                 </h3>

templates/host/menu.html

@@ -2,7 +2,7 @@
     <div class="verticalMenu">
         <div class="ui vertical pointing menu fluid">
             <a class="{{{if eq .URI "/host"}}}active teal{{{end}}}  item" href="/host">
-                <i class="linux icon"></i> 主机列表
+                <i class="linux icon"></i> 节点列表
             </a>
         </div>
     </div>

templates/task/index.html

@@ -24,7 +24,7 @@
                 </div>
                 <div class="field">
                     <select name="host_id" id="hostId">
-                        <option value="">选择主机</option>
+                        <option value="">选择节点</option>
                         {{{range $i, $v := .Hosts}}}
                         <option value="{{{.Id}}}" {{{if eq $.Params.HostId .Id }}} selected {{{end}}} >{{{.Alias}}}-{{{.Name}}}</option>
                         {{{end}}}
@@ -60,7 +60,7 @@
                 <th>超时时间</th>
                 <th>重试次数</th>
                 <th>单实例运行</th>
-                <th>主机</th>
+                <th>任务节点</th>
                 <th>状态</th>
                 <th>操作</th>
             </tr>
@@ -69,14 +69,18 @@
                 {{{range $i, $v := .Tasks}}}
                     <tr>
                         <td>{{{.Id}}}</td>
-                        <td>{{{.Task.Name}}}</td>
+                        <td>{{{.Name}}}</td>
                         <td>{{{if eq .Level 1}}}主任务{{{else}}}子任务{{{end}}}</td>
                         <td>{{{.Spec}}}</td>
                         <td>{{{if eq .Protocol 1}}} HTTP {{{else if eq .Protocol 2}}} SHELL {{{end}}}</td>
                         <td>{{{if eq .Timeout -1}}}后台运行{{{else if gt .Timeout 0}}}{{{.Timeout}}}秒{{{else}}}不限制{{{end}}}</td>
                         <td>{{{.RetryTimes}}}</td>
                         <td>{{{if gt .Multi 0}}}否{{{else}}}是{{{end}}}</td>
-                        <td>{{{.Alias}}}-{{{.Name}}}</td>
+                        <td>
+                            {{{range $k, $h := .Hosts}}}
+                                {{{$h.Alias}}}<br>
+                            {{{end}}}
+                        </td>
                         <td>
                             {{{if eq .Level 1}}}
                                 {{{if eq .Status 1}}}<span style="color: green;">激活</span>{{{else}}}<span style="color: red;">停止<span>{{{end}}}

templates/task/log.html

@@ -54,9 +54,9 @@
                 <th>任务ID</th>
                 <th>任务名称</th>
                 <th>cron表达式</th>
-                <th>协议</th>
+                <th>执行方式</th>
                 <th>重试次数</th>
-                <th>主机</th>
+                <th>任务节点</th>
                 <th>执行时长</th>
                 <th>状态</th>
                 <th>执行结果</th>
@@ -70,7 +70,7 @@
                 <td>{{{.Spec}}}</td>
                 <td>{{{if eq .Protocol 1}}} HTTP {{{else if eq .Protocol 2}}} SHELL {{{end}}}</td>
                 <td>{{{.RetryTimes}}}</td>
-                <td>{{{.Hostname}}}</td>
+                <td>{{{unescape .Hostname}}}</td>
                 <td>
                     {{{if and (ne .Status 3) (ne .Status 4)}}}
                     {{{if gt .TotalTime 0}}}{{{.TotalTime}}}秒{{{else}}}1秒{{{end}}}<br>

templates/task/task_form.html

@@ -19,7 +19,7 @@
                     <div class="content">任务名称</div>
                 </label>
                 <div class="ui small input">
-                    <input type="text" name="name" placeholder="订单量统计" value="{{{.Task.Task.Name}}}">
+                    <input type="text" name="name" placeholder="订单量统计" value="{{{.Task.Name}}}">
                 </div>
             </div>
         </div>
@@ -83,20 +83,23 @@
             <div class="field">
                 <label>执行方式</label>
                 <select name="protocol" id="protocol">
-                    <option value="2" {{{if .Task}}} {{{if eq .Task.Protocol 2}}}selected{{{end}}} {{{end}}} data-match="host_id" data-validate-type="selectProtocol">SHELL</option>
+                    <option value="2" {{{if .Task}}} {{{if eq .Task.Protocol 2}}}selected{{{end}}} {{{end}}}
+                            data-validate-type="selectProtocol">SHELL</option>
                     <option value="1" {{{if .Task}}} {{{if eq .Task.Protocol 1}}}selected{{{end}}} {{{end}}}>HTTP</option>
                 </select>
             </div>
         </div>
-        <div class="three fields" id="hostField">
+        <div class="fields" id="hostField">
             <div class="field">
-                <label>任务执行器</label>
-                <select name="host_id" id="hostId">
-                    <option value="">选择主机</option>
+                <label>选择任务节点</label>
+                <div id="hostId">
                     {{{range $i, $v := .Hosts}}}
-                     <option value="{{{.Id}}}" {{{if $.Task}}}{{{if eq $.Task.HostId .Id }}} selected {{{end}}} {{{end}}}>{{{.Alias}}}-{{{.Name}}}</option>
+                        <label>
+                            <input type="checkbox" value="{{{.Id}}}" {{{if $.Task}}}{{{if $v.Selected}}} checked {{{end}}}{{{end}}}  style="width:25px;height: 25px;">{{{.Alias}}}-{{{.Name}}}
+                            {{{if  (HostFormat $i) }}}<br>{{{end}}}
+                        </label>
                     {{{end}}}
-                 </select> &nbsp; <a class="ui blue button" href="/host/create" target="_blank">添加主机</a>
+                 </div> &nbsp; <br> <a class="ui blue button" href="/host/create" target="_blank">添加节点</a>
 
             </div>
         </div>
@@ -109,11 +112,11 @@
         <div class="three fields">
             <div class="field">
                 <label>任务超时时间(秒, 0-86400)</label>
-                <input type="text"  name="timeout" placeholder="默认0, 不限制" value="{{{if .Task}}} {{{.Task.Timeout}}} {{{else}}} 0 {{{end}}}">
+                <input type="text"  name="timeout" placeholder="默认0, 不限制" value="{{{if .Task}}} {{{.Task.Timeout}}} {{{else}}}0{{{end}}}">
             </div>
             <div class="field">
                 <label>任务失败重试次数 (0-10)</label>
-                <input type="text"  name="retry_times" placeholder="默认0, 不重试" value="{{{if .Task}}} {{{.Task.RetryTimes}}} {{{else}}} 0 {{{end}}}">
+                <input type="text"  name="retry_times" placeholder="默认0, 不重试" value="{{{if .Task}}} {{{.Task.RetryTimes}}} {{{else}}}0{{{end}}}">
             </div>
             <div class="field">
                 <label>允许多实例同时运行</label>
@@ -313,6 +316,15 @@
         return receivers.join(",");
     }
 
+    function parseHostId() {
+        var hostIds = [];
+        $('#hostId input:checked').each(function () {
+           hostIds.push($(this).val());
+        });
+
+        return hostIds.join(",");
+    }
+
     function changeLevel() {
         var selected = $('#level').val();
         if (selected == 1) {
@@ -336,6 +348,12 @@
                         return false;
                     }
                     fields.notify_receiver_id = parseNotifyReceiver();
+                    fields.host_id = parseHostId();
+                    if (fields.protocol == 2 && fields.host_id == "") {
+                        swal('错误提示', '请选择任务节点');
+                        return false;
+                    }
+
                     util.post('/task/store', fields, function(code, message) {
                         location.href = "/task"
                     });