新增model,service

cmd/web.go

@@ -0,0 +1,80 @@
+package cmd
+
+import (
+	"github.com/urfave/cli"
+	"gopkg.in/macaron.v1"
+	"github.com/go-macaron/gzip"
+	"github.com/go-macaron/session"
+	"github.com/go-macaron/csrf"
+	"scheduler/utils/app"
+	"fmt"
+	"os"
+)
+
+// web服务器默认端口
+const DefaultPort = 5920
+// 静态文件目录
+const StaticDir = "public"
+
+var CmdWeb = cli.Command{
+	Name: "server",
+	Usage: "start scheduler web server",
+	Action: run,
+	Flags: []cli.Flag{
+		cli.IntFlag{
+			Name: "port,p",
+			Value: DefaultPort,
+			Usage: "bind port number",
+		},
+	},
+}
+
+func run(ctx *cli.Context) {
+	// 检测环境
+	app.CheckEnv()
+	// 启动定时任务
+	runScheduler()
+	m := macaron.Classic()
+	// 注册路由
+	registerRouter(m)
+	// 注册中间件
+	registerMiddleware(m)
+	port := parsePort(ctx)
+	m.Run(port)
+}
+
+// 定时任务调度
+func runScheduler()  {
+	fmt.Println("hello world")
+	os.Exit(1)
+}
+
+// 路由注册
+func registerRouter(m *macaron.Macaron) {
+	// 所有GET方法，自动注册HEAD方法
+	m.SetAutoHead(true)
+}
+
+// 中间件注册
+func registerMiddleware(m *macaron.Macaron)  {
+	m.Use(macaron.Logger())
+	m.Use(macaron.Recovery())
+	m.Use(gzip.Gziper())
+	m.Use(macaron.Static(StaticDir))
+	m.Use(macaron.Renderer())
+	m.Use(session.Sessioner())
+	m.Use(csrf.Csrfer())
+}
+
+// 解析端口
+func parsePort(ctx *cli.Context) int {
+	var port int
+	if (ctx.IsSet("port")) {
+		port = ctx.Int("port")
+	}
+	if port <= 0 || port >= 65535 {
+		port = DefaultPort
+	}
+
+	return port
+}

conf/ansible.cfg

@@ -1,6 +1,6 @@
 [defaults]
 
-; 默认部收集远程主机信息
+; 默认不收集远程主机信息
 gathering = explicit
 ; 默认模块
 module_name = shell

conf/app.ini

@@ -0,0 +1,8 @@
+[db]
+password = wozaixiamen
+charset  = utf8
+database = cron
+host     = 127.0.0.1
+port     = 3306
+user     = root
+prefix   =

models/host.go

@@ -0,0 +1,52 @@
+package models
+
+// 主机
+type Host struct {
+	Id int16 `xorm:"smallint pk autoincr"`
+	Name string `xorm:"varchar(128) notnull"` // 主机名称
+	Alias string `xorm:"varchar(32) notnull default '' "` // 主机别名，仅用于后台显示
+	Port int `xorm:"notnull"`  // 主机端口
+	Remark string `xorm:"varchar(512) notnull default '' "` // 备注
+	Page  int `xorm:"-"`
+	PageSize int `xorm:"-"`
+}
+
+// 新增
+func(host *Host) Create() (int64, error) {
+	return Db.Insert(host)
+}
+
+// 更新
+func(host *Host) Update(id int, data CommonMap) (int64, error) {
+	return Db.Table(host).ID(id).Update(data)
+}
+
+// 删除
+func(host *Host) Delete(id int) (int64, error) {
+	return Db.Id(id).Delete(host)
+}
+
+func(host *Host) List() ([]Host, error) {
+	host.parsePageAndPageSize()
+	list := make([]Host, 0)
+	err := Db.Desc("id").Limit(host.PageSize, host.pageLimitOffset()).Find(&list)
+
+	return list, err
+}
+
+func(host *Host) Total() (int64, error) {
+	return Db.Count(host)
+}
+
+func(host *Host) parsePageAndPageSize()  {
+	if (host.Page <= 0) {
+		host.Page = Page
+	}
+	if (host.PageSize >= 0 || host.PageSize > MaxPageSize) {
+		host.PageSize = PageSize
+	}
+}
+
+func(host *Host) pageLimitOffset() int  {
+	return (host.Page - 1) * host.PageSize
+}

models/migration.go

@@ -0,0 +1,19 @@
+package models
+
+// 创建数据库表
+
+type Migration struct {}
+
+func(migration *Migration) Exec() error  {
+	tables := []interface{}{
+		&User{}, &Task{}, &TaskLog{},&Host{},
+	}
+	for _, table := range(tables) {
+		err := Db.Sync2(table)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

models/model.go

@@ -0,0 +1,73 @@
+package models
+
+import (
+	"github.com/go-xorm/xorm"
+	"fmt"
+	"scheduler/utils/setting"
+	"github.com/go-xorm/core"
+	_ "github.com/go-sql-driver/mysql"
+	"gopkg.in/macaron.v1"
+	"scheduler/utils/app"
+)
+
+var Db *xorm.Engine = nil
+
+func init()  {
+	if app.Installed {
+		Db = createDb()
+	}
+}
+
+type Status int8
+type CommonMap map[string]interface{}
+
+const (
+	Disabled Status = 0   // 禁用
+	Failure Status  = 0   // 失败
+	Enabled Status  = 1   // 启用
+	Running Status  = 1   // 运行中
+	Finish Status   = 2   // 完成
+)
+
+const (
+	Page = 1            // 当前页数
+	PageSize = 20       // 每页多少条数据
+	MaxPageSize = 1000  // 每次最多取多少条
+)
+
+func createDb() *xorm.Engine{
+	config,err := setting.Read()
+	if err != nil {
+		panic(err)
+	}
+	section := config.Section("db")
+	if err != nil {
+		panic(err)
+	}
+	user := section.Key("user").String()
+	password := section.Key("password").String()
+	host := section.Key("host").String()
+	port := section.Key("port").String()
+	database := section.Key("database").String()
+	charset := section.Key("charset").String()
+	prefix := section.Key("prefix").String()
+
+	DSN := "%s:%s@tcp(%s:%s)/%s?charset=%s"
+	dsn := fmt.Sprintf(DSN, user, password, host, port, database, charset)
+	engine, err := xorm.NewEngine("mysql", dsn)
+	if err != nil {
+		panic(err)
+	}
+	if prefix != "" {
+		// 设置表前缀
+		mapper := core.NewPrefixMapper(core.SnakeMapper{}, prefix)
+		engine.SetTableMapper(mapper)
+	}
+	// 本地环境开始日志
+	if macaron.Env == macaron.DEV {
+		engine.ShowSQL(true)
+		engine.Logger().SetLevel(core.LOG_DEBUG)
+	}
+
+	return engine
+}

models/task.go

@@ -0,0 +1,81 @@
+package models
+
+import (
+	"time"
+)
+
+type Protocol int8
+
+const (
+	HTTP Protocol = 1
+	SSH Protocol  = 2
+)
+
+type Task struct {
+	Id int `xorm:"int pk autoincr"`
+	Name string `xorm:"varchar(64) notnull"`          // 任务名称
+	Spec string `xorm:"varchar(64) notnull"`          // crontab 格式
+	Protocol Protocol `xorm:"tinyint notnull"`            // 协议 1:http 2:ssh
+	Command string `xorm:"varchar(512) notnull"`      // URL地址或shell命令
+	Timeout int `xorm:"mediumint notnull default 0"`      // 执行超时时间(单位秒)，0不限制, 限制不能超过一周
+	SshHostGroup string `xorm:"varchar(512) notnull defalut '' "` // SSH主机名组
+	Remark string `xorm:"varchar(512) notnull default ''"`    // 备注
+	Created time.Time `xorm:"datetime notnull created"`       // 创建时间
+	Updated time.Time `xorm:"datetime updated"`               // 更新时间
+	Deleted time.Time `xorm:"datetime deleted"`               // 删除时间
+	Status Status `xorm:"tinyint notnull default 1"`    // 状态 1:正常 0:停止
+	Page  int `xorm:"-"`
+	PageSize int `xorm:"-"`
+}
+
+// 新增
+func(task *Task) Create() (int64, error) {
+	task.Status = Enabled
+
+	return Db.Insert(task)
+}
+
+// 更新
+func(task *Task) Update(id int, data CommonMap) (int64, error) {
+	return Db.Table(task).ID(id).Update(data)
+}
+
+// 删除
+func(task *Task) Delete(id int) (int64, error) {
+	return Db.Id(id).Delete(task)
+}
+
+// 禁用
+func(task *Task) Disable(id int) (int64, error) {
+	return task.Update(id, CommonMap{"status": Disabled})
+}
+
+// 激活
+func(task *Task) Enable(id int) (int64, error)  {
+	return task.Update(id, CommonMap{"status": Enabled})
+}
+
+func(task *Task) List() ([]Task, error) {
+	task.parsePageAndPageSize()
+	list := make([]Task, 0)
+	err := Db.Desc("id").Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
+
+	return list, err
+}
+
+func(taskLog *TaskLog) Total() (int64, error) {
+	return Db.Count(taskLog)
+}
+
+func(taskLog *TaskLog) parsePageAndPageSize()  {
+	if (taskLog.Page <= 0) {
+		taskLog.Page = Page
+	}
+	if (taskLog.PageSize >= 0 || taskLog.PageSize > MaxPageSize) {
+		taskLog.PageSize = PageSize
+	}
+}
+
+func(taskLog *TaskLog) pageLimitOffset() int  {
+	return (taskLog.Page - 1) * taskLog.PageSize
+}

models/task_log.go

@@ -0,0 +1,58 @@
+package models
+
+import "time"
+
+type TaskLog struct{
+	Id int `xorm:"pk autoincr"`
+	Name string `xorm:"varchar(64) notnull"`  // 任务名称
+	Protocol Protocol `xorm:"tinyint notnull "`   // 协议
+	Command string `xorm:"varchar(512) notnull"`  // URL或shell命令
+	Remark string `xorm:"varchar(512) notnull default ''"` // 备注
+	SshHosts string `xorm:"varchar(512) notnull default ''"`
+	StartTime time.Time `xorm:"datetime created"`   // 开始执行时间
+	EndTime time.Time `xorm:"datetime updated"`     // 执行完成（失败）时间
+	Status Status `xorm:"tinyint notnull default 1"` // 状态 1:执行中  2:执行完毕 0:执行失败
+	Result string `xorm:"varchar(65535) notnull defalut '' "`
+	Page  int `xorm:"-"`
+	PageSize int `xorm:"-"`
+}
+
+func(taskLog *TaskLog) Create() (int64, error) {
+	taskLog.Status = Running
+
+	return Db.Insert(taskLog)
+}
+
+// 更新
+func(taskLog *TaskLog) Update(id int, data CommonMap) (int64, error) {
+	return Db.Table(taskLog).ID(id).Update(data)
+}
+
+func(taskLog *TaskLog) setStatus(id int ,status Status) (int64, error)  {
+	return taskLog.Update(id, CommonMap{"status": status})
+}
+
+func(taskLog *TaskLog) List() ([]TaskLog, error) {
+	taskLog.parsePageAndPageSize()
+	list := make([]TaskLog, 0)
+	err := Db.Desc("id").Limit(taskLog.PageSize, taskLog.pageLimitOffset()).Find(&list)
+
+	return list, err
+}
+
+func(task *Task) Total() (int64, error) {
+	return Db.Count(task)
+}
+
+func(task *Task) parsePageAndPageSize()  {
+	if (task.Page <= 0) {
+		task.Page = Page
+	}
+	if (task.PageSize >= 0 || task.PageSize > MaxPageSize) {
+		task.PageSize = PageSize
+	}
+}
+
+func(task *Task) pageLimitOffset() int  {
+	return (task.Page - 1) * task.PageSize
+}

models/user.go

@@ -0,0 +1,114 @@
+package models
+
+import (
+	"time"
+	"scheduler/utils"
+)
+
+const PasswordSaltLength = 6;
+
+// 用户model
+type User struct  {
+	Id int `xorm:"pk autoincr notnull "`
+	Name string `xorm:"varchar(32) notnull unique"`
+	Password string `xorm:"char(32) notnull "`
+	Salt string `xorm:"char(6) notnull "`
+	Email string `xorm:"varchar(50) notnull unique default '' "`
+	Created time.Time `xorm:"datetime notnull created"`
+	Updated time.Time `xorm:"datetime updated"`
+	Deleted time.Time `xorm:"datetime deleted"`
+	IsAdmin int8 `xorm:"tinyint notnull default 0"` // 是否是管理员 1:管理员 0:普通用户
+	Status Status `xorm:"tinyint notnull default 1"`  // 1: 正常 0:禁用
+	Page  int `xorm:"-"`
+	PageSize int `xorm:"-"`
+}
+
+// 新增
+func(user *User) Create() (int64, error) {
+	user.Status   = Enabled
+	user.Salt     = user.generateSalt()
+	user.Password = user.encryptPassword(user.Password, user.Salt)
+
+	return Db.Insert(user)
+}
+
+// 更新
+func(user *User) Update(id int, data CommonMap) (int64, error) {
+	return Db.Table(user).ID(id).Update(data)
+}
+
+// 删除
+func(user *User) Delete(id int) (int64, error) {
+	return Db.Id(id).Delete(user)
+}
+
+// 禁用
+func(user *User) Disable(id int) (int64, error) {
+	return user.Update(id, CommonMap{"status": Disabled})
+}
+
+// 激活
+func(user *User) Enable(id int) (int64, error)  {
+	return user.Update(id, CommonMap{"status": Enabled})
+}
+
+// 验证用户名和密码
+func(user *User) Match(username, password string) bool  {
+	where := "(name = ? OR email = ?)"
+	_, err := Db.Where(where, username, username).Get(user)
+	if err != nil {
+		return false
+	}
+	hashPassword := user.encryptPassword(password, user.Salt)
+	if (hashPassword != user.Password) {
+		return false
+	}
+
+	return true
+}
+
+// 用户名是否存在
+func(user *User) UsernameExists(username string ) (int64, error)  {
+	return Db.Where("name = ?",  username).Count(user)
+}
+
+// 邮箱地址是否存在
+func(user *User) EmailExists(email string) (int64, error) {
+	return Db.Where("email = ?", email).Count(user)
+}
+
+
+func(user *User) List() ([]User, error) {
+	user.parsePageAndPageSize()
+	list := make([]User, 0)
+	err := Db.Desc("id").Limit(user.PageSize, user.pageLimitOffset()).Find(&list)
+
+	return list, err
+}
+
+func(user *User) Total() (int64, error) {
+	return Db.Count(user)
+}
+
+func(user *User) parsePageAndPageSize()  {
+	if (user.Page <= 0) {
+		user.Page = Page
+	}
+	if (user.PageSize >= 0 || user.PageSize > MaxPageSize) {
+		user.PageSize = PageSize
+	}
+}
+
+func(user *User) pageLimitOffset() int  {
+	return (user.Page - 1) * user.PageSize
+}
+
+// 密码加密
+func(user *User) encryptPassword(password, salt string) string  {
+	return utils.Md5(password + salt)
+}
+
+// 生成密码盐值
+func(user *User) generateSalt() string {
+	return utils.RandString(PasswordSaltLength)
+}

service/task.go

@@ -0,0 +1,137 @@
+package service
+
+import (
+	"scheduler/models"
+	"scheduler/utils"
+	"net/http"
+	"io/ioutil"
+	"strconv"
+	"time"
+	"sync"
+)
+
+type Host struct {
+	sync.RWMutex
+	hosts []models.Host
+}
+
+var host Host = &Host{
+	sync.RWMutex{},
+	hosts: initHosts(),
+}
+
+func GetHosts() []models.Host {
+	host.RLock()
+	defer host.RUnlock()
+
+	return host.hosts
+}
+
+func SetHosts(h []models.Host)  {
+	host.Lock()
+	defer host.Unlock()
+
+	host.hosts = h
+}
+
+func initHosts() []models.Host {
+	// 获取所有主机
+	hostModel := new(models.Host)
+	list, err := hostModel.List()
+	if err != nil {
+		utils.RecordLog("获取主机列表失败-", err.Error())
+		return nil
+	}
+	if len(list) == 0 {
+		utils.RecordLog("主机列表为空")
+		return nil
+	}
+
+	return list
+}
+
+type Task struct {}
+
+// 初始化任务，从数据库取出所有任务添加到定时任务
+func(task *Task) Initialize()  {
+	taskModel := new(models.Task)
+	taskList, err := taskModel.List()
+	if err != nil {
+		utils.RecordLog("获取任务列表错误-", err.Error())
+	}
+	if len(taskList) == 0 {
+		utils.RecordLog("任务列表为空")
+	}
+	for _, item := range(taskList) {
+		task.Add(item)
+	}
+}
+
+// 添加任务
+func(task *Task) Add(taskModel models.Task)  {
+	var taskFunc func() = nil;
+	switch taskModel.Protocol {
+		case models.HTTP:
+			taskFunc = func() {
+				var handler Handler = new(HTTPHandler)
+				handler.Run(taskModel)
+			}
+		case models.SSH:
+			taskFunc = func() {
+				var handler Handler = new(SSHHandler)
+				handler.Run(taskModel)
+			}
+		default:
+			utils.RecordLog("任务协议不存在-协议编号: ", taskModel.Protocol)
+	}
+	if (taskFunc != nil) {
+		utils.DefaultCronTask.Add(strconv.Itoa(taskModel.Id), taskModel.Spec, taskFunc)
+	}
+}
+
+type Handler interface {
+	Run(taskModel models.Task)
+}
+
+type HTTPHandler struct {}
+
+// 执行HTTP任务
+func(h *HTTPHandler) Run(taskModel models.Task)  {
+	client := &http.Client{}
+	if (taskModel.Timeout > 0) {
+		client.Timeout = time.Duration(taskModel.Timeout) * time.Second
+	}
+	req, err := http.NewRequest("POST", taskModel.Command, nil)
+	if err != nil {
+		utils.RecordLog("创建HTTP请求错误-", err.Error())
+		return
+	}
+	req.Header.Set("Content-type", "application/x-www-form-urlencoded")
+	req.Header.Set("User-Agent", "golang-cron/scheduler")
+
+	resp, err := client.Do(req)
+	defer resp.Body.Close()
+	if err != nil {
+		utils.RecordLog("HTTP请求错误-", err.Error())
+		return
+	}
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		utils.RecordLog("读取HTTP请求返回值失败-", err.Error())
+	}
+
+	_, err = taskModel.Update(
+		taskModel.Id,
+		models.CommonMap{
+			"status": 0,
+			"result" : string(body),
+		});
+	if err != nil {
+		utils.RecordLog("更新任务日志失败-", err.Error())
+	}
+}
+
+type SSHHandler struct {}
+
+// 执行SSH任务
+func(ssh *SSHHandler) Run(taskModel models.Task)  {}