sup_task/supplierTaskAgent.go

package main

import (
	"bytes"
	"crypto/md5"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"os"
	"os/exec"
	"time"

	"github.com/mitchellh/go-homedir"
	"golang.org/x/crypto/ssh"
	gossh "golang.org/x/crypto/ssh"
	"gopkg.in/yaml.v2"
)

//声明变量
var (
	err        error
	f          []byte
	logFile    *os.File
	Config_dir string
	Conf       Config
	Loger      *log.Logger
	TaskCache  []Task
	resp       *http.Response
	content    []byte
)

type Config struct {
	Url              string
	User             string
	Password         string
	Logdir           string
	Cmdb_url         string
	DingDing_webHook string
	SshKeyPath       string
	Type_cmd         struct {
		N1 []string
		N2 []string
		N8 []string
	}
	Cmdb_Token string
}
type Hostinfo struct {
	Id               int     //id
	Ip               string  //ip地址
	Status           int     //节点状态
	Hostname         string  //主机名
	Node_type        float64 //节点类型
	Server_room_name string  //机房
	ServerId         int     //服务器id
	Idc_id           int     //idc id
	Sup_id           int     //供应商id
	Node_id          float64 //node类型id

}

type Task struct {
	Id         int
	Hostnumber int
	Hostgroup  string

	Describe   string
	Status     string
	Name       string
	ExecResult string
	Starttime  int64
	Stoptime   int64
	Createtime int64
}

type TaskResult struct {
	Ip               string  //ip地址
	Status           int     //节点状态
	Hostname         string  //主机名
	Node_type        float64 //节点类型
	Server_room_name string  //机房
	Result           struct {
		Cmdb_api_result string //cmdb结果
		Cmd_result      string //命令结果
	}
}

//声明变量 end

//钉钉发消息

func SendDingMsg(msg string) {
	//请求地址模板
	webHook := Conf.DingDing_webHook
	// content := `{'msgtype': 'text',
	// 	'text': {'content': '` + msg + `'}
	// }`
	content := make(map[string]interface{})
	text := make(map[string]string)
	text["content"] = msg
	content["msgtype"] = "text"
	content["text"] = text

	fmt.Println(content)
	ddd, err := json.Marshal(content)
	fmt.Println("钉钉发消息", string(ddd), err)

	//创建一个请求
	req, err := http.NewRequest("POST", webHook, bytes.NewReader(ddd))
	if err != nil {
		Loger.Println("钉钉发消息失败", err)
	}

	client := &http.Client{}
	//设置请求头
	req.Header.Set("Content-Type", "application/json; charset=utf-8")
	//发送请求
	resp, err := client.Do(req)

	//关闭请求
	defer resp.Body.Close()

	if err != nil {
		Loger.Println("钉钉发消息失败", err)
	}
}

//不实时执行linux命令
func Cmd(command string) string {
	cmd := exec.Command("/bin/bash", "-c", command)
	bytes, err := cmd.Output()
	if err != nil {

		return fmt.Sprintln("执行命令:", command+"\n", err)
	}
	resp := string(bytes)
	return "执行命令:" + command + "\n" + resp
}

//检测任务并执行

func TimeCheck() {
	fmt.Println("1")
	t := TaskCache
	d := time.Now().Unix()
	for _, v := range t {
		if v.Status == "false" {
			// 开始任务
			if d == v.Starttime {
				r, err := TaskStart(v.Hostgroup, v.Describe, v.Name, v.Starttime, v.Stoptime, v.Hostnumber)
				fmt.Println("开始任务", err)

				SendDingMsg(r)
			}
			//结束任务
			if d == v.Stoptime {
				r, err := TaskStop(v.Hostgroup)
				fmt.Println("结束任务", err)

				SendDingMsg(r)

			}

		}

	}
}

//更新任务缓存
func TaskUpcache() {

	for {
		content, s := CmdbApi("get_task", nil)
		if s == true {
			if err = json.Unmarshal(content, &TaskCache); err != nil {
				fmt.Println(err, string(content))
			}
		}
		time.Sleep(1 * time.Second)
	}
}

type Cli struct {
	user       string
	pwd        string
	addr       string
	client     *gossh.Client
	session    *gossh.Session
	LastResult string
	sshKeyPath string
}

func (c *Cli) Connect() (*Cli, error) {
	config := &gossh.ClientConfig{}
	config.SetDefaults()
	config.User = c.user
	// config.Auth = []gossh.AuthMethod{gossh.Password(c.pwd)}
	config.Auth = []ssh.AuthMethod{publicKeyAuthFunc(c.sshKeyPath)}
	config.HostKeyCallback = func(hostname string, remote net.Addr, key gossh.PublicKey) error { return nil }
	client, err := gossh.Dial("tcp", c.addr, config)
	if nil != err {
		return c, err
	}
	c.client = client
	return c, nil
}

func (c Cli) Run(shell string) (string, error) {
	if c.client == nil {
		if _, err := c.Connect(); err != nil {
			return "", err
		}
	}
	session, err := c.client.NewSession()
	if err != nil {
		return "", err
	}
	defer session.Close()
	buf, err := session.CombinedOutput(shell)

	c.LastResult = string(buf)
	return c.LastResult, err
}

//cmdb api
func CmdbApi(action string, data interface{}) ([]byte, bool) {

	now := time.Now().Unix()
	u := make(map[string]string)
	m := md5.Sum([]byte(Conf.User + Conf.Password + fmt.Sprintln(now)))
	client := http.Client{Timeout: 5 * time.Second}

	u["action"] = "get_task"
	u["user"] = Conf.User
	u["time"] = fmt.Sprintln(now)

	u["sign"] = fmt.Sprintf("%x", m)
	d, err1 := json.Marshal(data)
	if err1 != nil {
		fmt.Println(err1)

	}

	u["data"] = string(d)

	_u, err2 := json.Marshal(u)
	if err2 != nil {
		fmt.Println(err2)

	}
	resp, err3 := client.Post(Conf.Url, "application/json;charset=utf-8", bytes.NewBuffer(_u))

	if err2 != nil {
		fmt.Println("1", err3)

	}

	defer resp.Body.Close()

	content, err4 := ioutil.ReadAll(resp.Body)

	if err4 != nil {
		fmt.Println("Read failed:", err4)
		return []byte(""), false
	}

	return content, true

}

//http请求
func Httprequest(url, method, data string) ([]byte, error) {
	var (
		req  *http.Request
		resp *http.Response
		body []byte
		err  error
	)
	d := []byte(data)

	if data != "" {
		if req, err = http.NewRequest(method, url, bytes.NewReader(d)); err != nil {
			return body, err
		}

	} else {
		if req, err = http.NewRequest(method, url, nil); err != nil {
			return body, err

		}
	}

	req.Header.Set("Content-Type", "application/json;charset=UTF-8")
	req.Header.Set("Authorization", "Token 	"+Conf.Cmdb_Token)
	if resp, err = http.DefaultClient.Do(req); err != nil {
		return body, err
	}
	defer resp.Body.Close()

	if body, err = ioutil.ReadAll(resp.Body); err != nil {
		return body, err
	}
	return body, err
}

func publicKeyAuthFunc(kPath string) ssh.AuthMethod {
	//交叉编译打开文件

	keyPath, err := homedir.Expand(kPath)
	if err != nil {
		log.Fatal("find key's home dir failed", err)
	}

	//读文件
	key, err := ioutil.ReadFile(keyPath)
	if err != nil {
		log.Fatal("ssh key file read failed", err)
	}
	//解析为一个对象
	// Create the Signer for this private key.
	signer, err := ssh.ParsePrivateKey(key)
	if err != nil {
		log.Fatal("ssh key signer failed", err)
	}
	//放入key对象  生成一个指针

	return ssh.PublicKeys(signer)
}

//开始任务
func TaskStart(hostgroup string, desc string, name string, starttime int64, stoptime int64, hostnumber int) (string, error) {
	//关节点
	var (
		h          []Hostinfo
		v          []byte
		taskResult []TaskResult
	)

	if err = json.Unmarshal([]byte(hostgroup), &h); err != nil {
		return "", err

	}

	for index := range h {
		switch h[index].Node_type {
		case 16:
			t := TaskResult{}
			if v, err = Httprequest(Conf.Cmdb_url+"nodes/"+fmt.Sprintln(h[index].Node_id), "PUT", `{"status":1,"node_type":`+fmt.Sprintln(h[index].Node_type)+`}`); err != nil {
				continue
			}
			c := ""
			for cmd_index := range Conf.Type_cmd.N8 {
				c += Cmd(fmt.Sprintln("ssh root@127.0.0.1 '%s' -p 7721 ", Conf.Type_cmd.N8[cmd_index]))
			}
			t.Server_room_name = h[index].Server_room_name
			t.Node_type = h[index].Node_type
			t.Ip = h[index].Ip
			t.Hostname = h[index].Hostname
			t.Status = h[index].Status
			t.Result.Cmd_result = c
			t.Result.Cmdb_api_result = string(v)
			taskResult = append(taskResult, t)
		case 32:
			t := TaskResult{}
			if v, err = Httprequest(Conf.Cmdb_url+"nodes/"+fmt.Sprintln(h[index].Node_id), "PUT", `{"status":1,"node_type":`+fmt.Sprintln(h[index].Node_type)+`}`); err != nil {
				continue
			}
			c := ""
			for cmd_index := range Conf.Type_cmd.N8 {
				c += Cmd(Conf.Type_cmd.N8[cmd_index])
			}
			t.Server_room_name = h[index].Server_room_name
			t.Node_type = h[index].Node_type
			t.Ip = h[index].Ip
			t.Hostname = h[index].Hostname
			t.Status = h[index].Status
			t.Result.Cmd_result = c
			t.Result.Cmdb_api_result = string(v)
			taskResult = append(taskResult, t)
		case 64:
			t := TaskResult{}
			if v, err = Httprequest(Conf.Cmdb_url+"nodes/"+fmt.Sprintln(h[index].Node_id), "PUT", `{"status":1,"node_type":`+fmt.Sprintln(h[index].Node_type)+`}`); err != nil {
				continue
			}
			c := ""
			for cmd_index := range Conf.Type_cmd.N8 {
				c += Cmd(Conf.Type_cmd.N8[cmd_index])
			}
			t.Server_room_name = h[index].Server_room_name
			t.Node_type = h[index].Node_type
			t.Ip = h[index].Ip
			t.Hostname = h[index].Hostname
			t.Status = h[index].Status
			t.Result.Cmd_result = c
			t.Result.Cmdb_api_result = string(v)
			taskResult = append(taskResult, t)
		case 8:
			t := TaskResult{}
			if v, err = Httprequest(Conf.Cmdb_url+"nodes/"+fmt.Sprintln(h[index].Node_id), "PUT", `{"status":1,"node_type":`+fmt.Sprintln(h[index].Node_type)+`}`); err != nil {
				continue
			}
			c := ""
			for cmd_index := range Conf.Type_cmd.N8 {
				cli := Cli{
					user:       "root",
					pwd:        "",
					addr:       h[index].Ip + ":7721",
					sshKeyPath: Conf.SshKeyPath,
				}

				fmt.Println(cli)
				output, err := cli.Run(Conf.Type_cmd.N8[cmd_index])
				if err != nil {
					c += fmt.Sprintln(err)
				}
				c += fmt.Sprintln(output)

			}

			t.Server_room_name = h[index].Server_room_name
			t.Node_type = h[index].Node_type
			t.Ip = h[index].Ip
			t.Hostname = h[index].Hostname
			t.Status = h[index].Status
			t.Result.Cmd_result = c
			t.Result.Cmdb_api_result = string(v)
			taskResult = append(taskResult, t)
		default:
			t := TaskResult{}

			t.Server_room_name = h[index].Server_room_name
			t.Node_type = h[index].Node_type
			t.Ip = h[index].Ip
			t.Hostname = h[index].Hostname
			t.Status = h[index].Status
			taskResult = append(taskResult, t)

		}

	}
	m, err := UpVersion()

	t := struct {
		Title       string
		Hostlength  int
		TaskResult  []TaskResult
		VersionInfo string
		Describe    string
		Name        string
		Startime    string
		Stoptime    string
	}{
		Title:       "开始任务",
		Hostlength:  hostnumber,
		TaskResult:  taskResult,
		VersionInfo: m,
		Describe:    desc,
		Name:        name,
		Startime:    time.Unix(starttime, 0).Format("2006-01-02 15:04:05"),
		Stoptime:    time.Unix(stoptime, 0).Format("2006-01-02 15:04:05"),
	}

	d, err := yaml.Marshal(t)
	return string(d), err
}

//结束任务

func TaskStop(hostgroup string) (string, error) {
	//开节点
	var (
		title string
		msg   string
		h     []Hostinfo
		v     []byte
	)
	if err = json.Unmarshal([]byte(hostgroup), &h); err != nil {
		return "", err

	}

	msg += "结束任务" + "\n"
	msg += fmt.Sprintln("执行主机长度", len(h))

	for index := range h {
		switch h[index].Node_type {
		case 16:
			msg += fmt.Sprintln(h[index].Ip, "执行汇集节点命令")
		case 32:
			msg += fmt.Sprintln(h[index].Ip, "执行普通几点")

		case 64:
			msg += fmt.Sprintln(h[index].Ip, "执行普通64")
		case 8:

			if v, err = Httprequest(Conf.Cmdb_url+"nodes/"+fmt.Sprintln(h[index].Node_id), "PUT", `{"status":1,"node_type":`+fmt.Sprintln(h[index].Node_type)+`}`); err != nil {
				return msg, err
			}
			c := ""
			for cmd_index := range Conf.Type_cmd.N8 {
				c += Cmd(Conf.Type_cmd.N8[cmd_index])

			}

			title = fmt.Sprintln("主机", h[index].Ip, "类型", h[index].Node_type, "状态", h[index].Status, "机房", h[index].Server_room_name, "---------主机执行")

			msg += fmt.Sprintln(title, "执行cmdb接口"+string(v)+"\n", "------执行shells\n", c, "\n------执行shell结束")
			msg += fmt.Sprintln("主机", h[index].Ip, "类型", h[index].Node_type, "状态", h[index].Status, "机房", h[index].Server_room_name, "---------主机执行结束")
		case 2:
			//关闭节点
			msg += fmt.Sprintln(h[index].Ip, "执行普通64")

		default:
			msg += fmt.Sprintln(index, "无节点类型", h[index], "类型", h[index].Node_type, "节点状态", h[index].Status)

		}

	}
	m, err := UpVersion()
	msg += m
	return msg, err
}

//更新版本号+1
func UpVersion() (string, error) {
	var (
		err error
		msg string
		v   []byte
	)

	version := make(map[string]int)
	if v, err = Httprequest(Conf.Cmdb_url+"node-version/1", "GET", ""); err != nil {
		return "", err

	}

	if err = json.Unmarshal(v, &version); err != nil {
		return "", err

	}
	msg += fmt.Sprintln("version执行前版本", version["version"])

	version["version"] = version["version"] + 1
	if v, err = json.Marshal(version); err != nil {
		return "", err

	}

	if v, err = Httprequest(Conf.Cmdb_url+"node-version/1", "PUT", string(v)); err != nil {
		return "", err
	}
	msg += fmt.Sprintln("当前版本号", string(v))

	return msg, err
}

//初始化
func init() {
	flag.StringVar(&Config_dir, "taskConfig", "./taskConfig.json", "配置文件默认地址")
	flag.Parse()
	currentTime := time.Now()

	//加载配置文件
	if f, err = ioutil.ReadFile(Config_dir); err != nil {
		fmt.Println("读取错误", err)
		os.Exit(255)
	}
	if err = json.Unmarshal(f, &Conf); err != nil {
		fmt.Println("配置文件加载错误", err)
		os.Exit(255)
	}
	// dddd, _ := json.MarshalIndent(Conf, "", "    ")
	// fmt.Println("初始化配置文件内容如下\n", string(dddd))

	if logFile, err = os.OpenFile(currentTime.Format(Conf.Logdir+"/2006-01-02.log"), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0766); err != nil {
		if err != nil {
			fmt.Println("创建日志失败", err)
			os.Exit(255)

		}
	}
	Loger = log.New(logFile, "[LOG]", log.Ldate|log.Lshortfile|log.Ltime)

}

func main() {

	go TaskUpcache()

	content, s := CmdbApi("get_task", nil)

	if s == true {
		if err = json.Unmarshal(content, &TaskCache); err != nil {
			fmt.Println(err, string(content))
		}
	}
	r, err := TaskStart(TaskCache[0].Hostgroup, TaskCache[0].Describe, TaskCache[0].Name, TaskCache[0].Starttime, TaskCache[0].Stoptime, TaskCache[0].Hostnumber)
	fmt.Println("开始任务", err)
	fmt.Println(r)
	SendDingMsg(r)
	// for {
	// 	if len(TaskCache) == 0 {
	// 		fmt.Println(111)
	// 		return
	// 	}
	// 	TimeCheck()

	// 	time.Sleep(1 * time.Second)
	// }

}