新增ansible、任务管理

utils/ansible/ansible.go

@@ -0,0 +1,100 @@
+package ansible
+
+// ansible ad-hoc playbook命令封装
+
+import (
+	"os"
+	"scheduler/utils"
+	"errors"
+	"gopkg.in/yaml.v2"
+	"io/ioutil"
+	"scheduler/utils/app"
+)
+
+func init()  {
+		// ansible配置文件目录
+		os.Setenv("ANSIBLE_CONFIG", app.ConfDir)
+}
+
+type Handler map[string]interface{}
+
+type Playbook struct {
+	Name string
+	Hosts string
+	Tasks []Handler
+	Handlers []Handler
+}
+
+func(playbook *Playbook) SetHosts(hosts string) {
+	playbook.Hosts = hosts
+}
+
+func(playbook *Playbook) SetName(name string) {
+	playbook.Name = name
+}
+
+func(playbook *Playbook) AddTask(handler Handler) {
+	playbook.Tasks = append(playbook.Tasks, handler)
+}
+
+
+func(playbook *Playbook) AddHandler(handler Handler) {
+	playbook.Handlers = append(playbook.Handlers, handler)
+}
+
+
+/**
+ * 执行ad-hoc
+ * hosts  主机文件路径
+ * module 调用模块
+ * args   传递给模块的参数
+*/
+func ExecCommand(hostPath string, module string, args... string) (output string, err error) {
+	if hostPath == "" || module == ""  {
+		err = errors.New("参数不完整")
+		return
+	}
+	commandArgs := []string{"-i", , hostPath, "-m", module}
+	if len(args) != 0 {
+		commandArgs = append(commandArgs, "-a", args...)
+	}
+	output, err = utils.ExecShell("ansible", commandArgs...)
+
+	return
+}
+
+// 执行playbook
+func ExecPlaybook(hostPath string, playbook Playbook) (result string, err error)  {
+	data, err := yaml.Marshal([]Playbook{playbook})
+	if err != nil {
+		return
+	}
+
+	tmpFile, err := ioutil.TempFile(getTmpDir(), "playbook")
+	if err != nil {
+		return
+	}
+	defer func() {
+		tmpFile.Close()
+		os.Remove(tmpFile.Name())
+	}()
+	_, err = tmpFile.Write(data)
+	if err != nil {
+		return
+	}
+	commandArgs := []string{"-i", hostPath, tmpFile.Name()}
+	result, err = utils.ExecShell("ansible-playbook", commandArgs...)
+
+	return
+}
+
+// 判断 获取临时目录，默认/dev/shm
+func getTmpDir() string {
+	dir := "/dev/shm"
+	_, err := os.Stat(dir)
+	if os.IsPermission(err) {
+		return ""
+	}
+
+	return dir
+}

utils/app/app.go

@@ -0,0 +1,69 @@
+package app
+
+import (
+	"os"
+	"scheduler/utils"
+	"runtime"
+)
+
+var  (
+	AppDir string    // 应用根目录
+	ConfDir string   // 配置目录
+	LogDir string    // 日志目录
+	DataDir string   // 数据目录，存放session文件等
+	AppConfig string // 应用配置文件
+	Installed bool = isInstalled()  // 应用是否安装过
+)
+
+func init() {
+	wd, err := os.Getwd()
+	if err != nil {
+		panic(err)
+	}
+	AppDir = wd
+	ConfDir = AppDir + "/conf"
+	LogDir  = AppDir + "/log"
+	DataDir = AppDir + "/data"
+	AppConfig = AppDir + "/app.ini"
+	checkDirExists(ConfDir, LogDir, DataDir)
+}
+
+
+// 检测环境
+func CheckEnv()  {
+	// ansible不支持安装在windows上, windows只能作为被控机
+	if runtime.GOOS == "windows" {
+		panic("不支持在windows上运行")
+	}
+	_, err := utils.ExecShell("ansible", "--version")
+	if err != nil {
+		panic(err)
+	}
+	_, err = utils.ExecShell("ansible-playbook", "--version")
+	if err != nil {
+		panic("ansible-playbook not found")
+	}
+}
+
+// 判断应用是否安装过
+func isInstalled() bool {
+	_, err := os.Stat(ConfDir + "/install.lock")
+	if os.IsExist(err) {
+		return true
+	}
+
+	return false
+}
+
+// 检测目录是否存在
+func checkDirExists(path... string)  {
+	for _, value := range(path) {
+		_, err := os.Stat(value)
+		if os.IsNotExist(err) {
+			panic(value + "目录不存在")
+		}
+		if os.IsPermission(err) {
+			panic(value + "目录无权限操作")
+		}
+	}
+}

utils/cron_task.go

@@ -0,0 +1,77 @@
+package utils
+
+import (
+	"github.com/robfig/cron"
+	"errors"
+)
+
+// todo map并发访问加锁
+
+var DefaultCronTask = &CronTask{
+	make(map[string]*cron.Cron),
+}
+
+type CronTask struct {
+	tasks map[string]*cron.Cron
+}
+
+// 新增定时任务,如果name存在，则添加失败
+func(cronTask *CronTask) Add(name string, spec string, cmd func() ) error {
+	if name == "" || spec == "" || cmd == nil {
+		return errors.New("参数不完整")
+	}
+	if cronTask.IsExist(name) {
+		return errors.New("任务已存在")
+	}
+
+	cronTask.tasks[name] = cron.New()
+	err := cronTask.tasks[name].AddFunc(spec, cmd)
+
+	return err
+}
+
+// 任务不存在则新增，任务已存在则替换任务
+func(cronTask *CronTask) addOrReplace(name string, spec string, cmd func() ) error {
+	if cronTask.IsExist(name) {
+		cronTask.Delete(name)
+	}
+
+	return cronTask.Add(name, spec, cmd)
+}
+
+
+// 判断任务是否存在
+func(cronTask *CronTask) IsExist(name string) bool {
+	_, ok := cronTask.tasks[name]
+
+	return ok
+}
+
+// 启动任务
+func(cronTask *CronTask) Start(name string) {
+	if cronTask.IsExist(name) {
+		cronTask.tasks[name].Start()
+	}
+}
+
+// 停止任务
+func(cronTask *CronTask) Stop(name string) {
+	if cronTask.IsExist(name) {
+		cronTask.tasks[name].Stop()
+	}
+}
+
+// 删除任务
+func(cronTask *CronTask) Delete(name string) {
+	cronTask.Stop(name)
+	cronTask.tasks[name] = nil
+	delete(cronTask.tasks, name)
+}
+
+// 运行所有任务
+func(cronTask *CronTask) run() {
+	for _, cron := range cronTask.tasks {
+		// cron内部有开启goroutine,此处不用新建
+		cron.Start()
+	}
+}

utils/setting/setting.go

@@ -0,0 +1,45 @@
+package setting
+
+import (
+	"gopkg.in/ini.v1"
+	"errors"
+	"scheduler/utils/app"
+)
+
+// 读取配置
+func Read() (config *ini.File, err error) {
+	config, err = ini.Load(app.AppConfig)
+	if err != nil {
+		return
+	}
+
+	return
+}
+
+
+// 写入配置
+func Write(config map[string]map[string]string) (error) {
+	if len(config) == 0 {
+		return errors.New("参数不能为空")
+	}
+
+	file := ini.Empty()
+	for sectionName, items := range(config) {
+		if sectionName == "" {
+			return errors.New("节名称不能为空")
+		}
+		section, err := file.NewSection(sectionName)
+		if err != nil {
+			return err
+		}
+		for key, value := range(items) {
+			_, err = section.NewKey(key, value)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	err := file.SaveTo(app.AppConfig)
+
+	return err
+}

utils/utils.go

@@ -0,0 +1,52 @@
+package utils
+
+import (
+	"os/exec"
+	"math/rand"
+	"time"
+	"crypto/md5"
+	"encoding/hex"
+	"log"
+)
+
+// 执行shell命令
+func ExecShell(command string, args... string) (string, error) {
+	result, err := exec.Command(command, args...).CombinedOutput()
+
+	return string(result), err
+}
+
+// 生成长度为length的随机字符串
+func RandString(length int64) string {
+	sources := []byte("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
+	result := []byte{}
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	sourceLength := len(sources)
+	var i int64 = 0
+	for ; i < length; i++ {
+		result = append(result, sources[r.Intn(sourceLength)])
+	}
+
+	return string(result)
+}
+
+// 生成32位MD5摘要
+func Md5(str string) string {
+	m := md5.New()
+	m.Write([]byte(str))
+
+	return hex.EncodeToString(m.Sum(nil))
+}
+
+// 生成0-max之间随机数
+func RandNumber(max int) int {
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+
+	return r.Intn(max)
+}
+
+// 日志记录
+// todo 保存到哪里 文件,数据库还是elasticsearch?，暂时输出到终端
+func RecordLog(v... interface{})  {
+	log.Println(v)
+}