Merge pull request #86 from shunfei/feature/cmd

csctl: add sync command for node to sync node info to mongodb

.gitignore

@@ -5,6 +5,9 @@ dist
 bin/*/*server
 .DS_Store
 web/ui/node_modules
+web/ui/package-lock.json
+web/ui/semantic.json
+web/ui/semantic
 web/ui/dist
 .vscode
 *npm-debug.log

bin/csctl/cmd.go

@@ -25,8 +25,8 @@ var rootCmd = &cobra.Command{
 }
 
 func init() {
-	rootCmd.PersistentFlags().StringVarP(&confFile, "conf", "c", "", "base.json file path.")
-	rootCmd.AddCommand(subcmd.BackupCmd, subcmd.RestoreCmd, subcmd.UpgradeCmd)
+	rootCmd.PersistentFlags().StringVarP(&confFile, "conf", "c", "conf/files/base.json", "base.json file path.")
+	rootCmd.AddCommand(subcmd.BackupCmd, subcmd.RestoreCmd, subcmd.UpgradeCmd, subcmd.NodeCmd)
 }
 
 func main() {

bin/csctl/cmd/backup.go

@@ -78,6 +78,7 @@ func init() {
 
 type ExitAction struct {
 	Defer func()
+	After func()
 }
 
 func NewExitAction() *ExitAction {
@@ -97,6 +98,10 @@ func (ea *ExitAction) Exit(format string, v ...interface{}) {
 	}
 
 	fmt.Printf(format+"\n", v...)
+
+	if ea.After != nil {
+		ea.After()
+	}
 	os.Exit(1)
 }
 

bin/csctl/cmd/node.go

@@ -0,0 +1,65 @@
+package cmd
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/spf13/cobra"
+
+	"github.com/shunfei/cronsun"
+)
+
+var (
+	nodeCmd     string
+	nodeInclude string
+	nodeExclude string
+
+	spliter = ","
+)
+
+func init() {
+	NodeCmd.Flags().StringVar(&nodeCmd, "cmd", "", "the command send to node")
+	NodeCmd.Flags().StringVar(&nodeInclude, "include", "", "the node ids that needs to execute the command, split by ',', e.g: '--include=aa,bb,cc', empty means all nodes")
+	NodeCmd.Flags().StringVar(&nodeExclude, "exclude", "", "the node ids that doesn't need to execute the command, split by ',', e.g: '--exclude=aa,bb,cc', empty means none")
+}
+
+var NodeCmd = &cobra.Command{
+	Use:   "node",
+	Short: "Send some commands to nodes",
+	Long: `Send a command to nodes and execute it.
+
+Available Commands:
+  rmold:	remove old version(< 0.3.0) node info from mongodb and etcd
+  sync:		sync node info to mongodb
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		ea := NewExitAction()
+		ea.After = func() {
+			fmt.Println()
+			cmd.Help()
+		}
+		nc, err := cronsun.ToNodeCmd(nodeCmd)
+		if err != nil {
+			ea.Exit(err.Error() + ": " + nodeCmd)
+		}
+
+		var include, exclude []string
+		if len(nodeInclude) > 0 {
+			include = strings.Split(nodeInclude, spliter)
+		}
+		if len(nodeExclude) > 0 {
+			exclude = strings.Split(nodeExclude, spliter)
+		}
+
+		err = cronsun.PutCsctl(&cronsun.CsctlCmd{
+			Cmd:     nc,
+			Include: include,
+			Exclude: exclude,
+		})
+		if err != nil {
+			ea.ExitOnErr(err)
+		}
+
+		fmt.Printf("command[%s] send success\n", nodeCmd)
+	},
+}

bin/csctl/cmd/upgrade.go

@@ -31,7 +31,7 @@ var UpgradeCmd = &cobra.Command{
 			ea.Exit("invalid version number")
 		}
 
-		nodesById := getIPMapper(ea)
+		nodesById := getIPMapper(ea, prever)
 		if prever < "0.3.0" {
 			fmt.Println("upgrading data to version 0.3.0")
 			if to_0_3_0(ea, nodesById) {
@@ -48,7 +48,7 @@ var UpgradeCmd = &cobra.Command{
 	},
 }
 
-func getIPMapper(ea *ExitAction) map[string]*cronsun.Node {
+func getIPMapper(ea *ExitAction, prever string) map[string]*cronsun.Node {
 	nodes, err := cronsun.GetNodes()
 	if err != nil {
 		ea.Exit("failed to fetch nodes from MongoDB: %s", err.Error())
@@ -61,6 +61,9 @@ func getIPMapper(ea *ExitAction) map[string]*cronsun.Node {
 			continue
 		}
 
+		if prever < "0.3.0" {
+			n.RmOldInfo()
+		}
 		ipMapper[n.IP] = n
 	}
 

conf/conf.go

@@ -51,6 +51,7 @@ type Conf struct {
 	Proc    string // 当前执行任务路径
 	Cmd     string // cmd 路径
 	Once    string // 马上执行任务路径
+	Csctl   string // csctl 发送执行命令的路径
 	Lock    string // job lock 路径
 	Group   string // 节点分组
 	Noticer string // 通知
@@ -228,6 +229,7 @@ func (c *Conf) parse(confFile string) error {
 	c.Proc = cleanKeyPrefix(c.Proc)
 	c.Cmd = cleanKeyPrefix(c.Cmd)
 	c.Once = cleanKeyPrefix(c.Once)
+	c.Csctl = cleanKeyPrefix(c.Csctl)
 	c.Lock = cleanKeyPrefix(c.Lock)
 	c.Group = cleanKeyPrefix(c.Group)
 	c.Noticer = cleanKeyPrefix(c.Noticer)
@@ -284,7 +286,7 @@ func (c *Conf) reload(confFile string) {
 	}
 
 	// etcd key 选项需要重启
-	cf.Node, cf.Proc, cf.Cmd, cf.Once, cf.Lock, cf.Group, cf.Noticer = c.Node, c.Proc, c.Cmd, c.Once, c.Lock, c.Group, c.Noticer
+	cf.Node, cf.Proc, cf.Cmd, cf.Once, cf.Csctl, cf.Lock, cf.Group, cf.Noticer = c.Node, c.Proc, c.Cmd, c.Once, c.Csctl, c.Lock, c.Group, c.Noticer
 
 	*c = *cf
 	log.Infof("config file[%s] reload success", confFile)

conf/files/base.json.sample

@@ -4,6 +4,7 @@
     "Proc": "/cronsun/proc/",
     "Cmd": "/cronsun/cmd/",
     "Once": "/cronsun/once/",
+    "Csctl": "/cronsun/csctl/",
     "Lock": "/cronsun/lock/",
     "Group": "/cronsun/group/",
     "Noticer": "/cronsun/noticer/",
@@ -21,8 +22,8 @@
     "Mgo": "@extend:db.json",
     "Mail": "@extend:mail.json",
     "Security": "@extend:security.json",
-    "#comment": "PIDFile and UUIDFile just work for cronnode"
-    "#PIDFile": "Given a none-empty string to write a pid file to the specialed path, or leave it empty to do nothing"
+    "#comment": "PIDFile and UUIDFile just work for cronnode",
+    "#PIDFile": "Given a none-empty string to write a pid file to the specialed path, or leave it empty to do nothing",
     "PIDFile": "/tmp/cronsun/cronnode_pid",
     "UUIDFile": "/etc/cronsun/CRONSUN_UUID"
 }

csctl.go

@@ -0,0 +1,73 @@
+package cronsun
+
+import (
+	"encoding/json"
+	"errors"
+
+	client "github.com/coreos/etcd/clientv3"
+
+	"github.com/shunfei/cronsun/conf"
+)
+
+const (
+	NodeCmdUnknown NodeCmd = iota
+	NodeCmdRmOld
+	NodeCmdSync
+	NodeCmdMax
+)
+
+var (
+	InvalidNodeCmdErr = errors.New("invalid node command")
+
+	NodeCmds = []string{
+		"unknown",
+		"rmold",
+		"sync",
+	}
+)
+
+type NodeCmd int
+
+func (cmd NodeCmd) String() string {
+	if NodeCmdMax <= cmd || cmd <= NodeCmdUnknown {
+		return "unknown"
+	}
+	return NodeCmds[cmd]
+}
+
+func ToNodeCmd(cmd string) (NodeCmd, error) {
+	for nc := NodeCmdUnknown + 1; nc < NodeCmdMax; nc++ {
+		if cmd == NodeCmds[nc] {
+			return nc, nil
+		}
+	}
+	return NodeCmdUnknown, InvalidNodeCmdErr
+}
+
+type CsctlCmd struct {
+	// the command send to node
+	Cmd NodeCmd
+	// the node ids that needs to execute the command, empty means all node
+	Include []string
+	// the node ids that doesn't need to execute the command, empty means none
+	Exclude []string
+}
+
+// 执行 csctl 发送的命令
+// 注册到 /cronsun/csctl/<cmd>
+func PutCsctl(cmd *CsctlCmd) error {
+	if NodeCmdMax <= cmd.Cmd || cmd.Cmd <= NodeCmdUnknown {
+		return InvalidNodeCmdErr
+	}
+
+	params, err := json.Marshal(cmd)
+	if err != nil {
+		return err
+	}
+	_, err = DefalutClient.Put(conf.Config.Csctl+NodeCmds[cmd.Cmd], string(params))
+	return err
+}
+
+func WatchCsctl() client.WatchChan {
+	return DefalutClient.Watch(conf.Config.Csctl, client.WithPrefix())
+}

node.go

@@ -102,7 +102,6 @@ func RemoveNode(query interface{}) error {
 	return mgoDB.WithC(Coll_Node, func(c *mgo.Collection) error {
 		return c.Remove(query)
 	})
-
 }
 
 func ISNodeAlive(id string) (bool, error) {
@@ -142,19 +141,24 @@ func WatchNode() client.WatchChan {
 
 // On 结点实例启动后，在 mongoDB 中记录存活信息
 func (n *Node) On() {
-	// remove old version(< 0.3.0) node info
-	mgoDB.RemoveId(Coll_Node, n.IP)
-
 	n.Alived, n.Version, n.UpTime = true, Version, time.Now()
-	if err := mgoDB.Upsert(Coll_Node, bson.M{"_id": n.ID}, n); err != nil {
-		log.Errorf(err.Error())
-	}
+	n.SyncToMgo()
 }
 
 // On 结点实例停用后，在 mongoDB 中去掉存活信息
 func (n *Node) Down() {
 	n.Alived, n.DownTime = false, time.Now()
+	n.SyncToMgo()
+}
+
+func (n *Node) SyncToMgo() {
 	if err := mgoDB.Upsert(Coll_Node, bson.M{"_id": n.ID}, n); err != nil {
 		log.Errorf(err.Error())
 	}
 }
+
+// RmOldInfo remove old version(< 0.3.0) node info
+func (n *Node) RmOldInfo() {
+	RemoveNode(bson.M{"_id": n.IP})
+	DefalutClient.Delete(conf.Config.Node + n.IP)
+}

node/csctl.go

@@ -0,0 +1,32 @@
+package node
+
+import (
+	"encoding/json"
+
+	"github.com/shunfei/cronsun"
+	"github.com/shunfei/cronsun/log"
+)
+
+func (n *Node) executCsctlCmd(key, value []byte) error {
+	cmd := &cronsun.CsctlCmd{}
+	err := json.Unmarshal(value, cmd)
+	if err != nil {
+		log.Warnf("invalid csctl command[%s] value[%s], err: %s", string(key), string(value), err.Error())
+		return err
+	}
+
+	if cronsun.NodeCmdMax <= cmd.Cmd || cmd.Cmd <= cronsun.NodeCmdUnknown {
+		log.Warnf("invalid csctl command[%s] value[%s], err: %s", string(key), string(value))
+		return cronsun.InvalidNodeCmdErr
+	}
+
+	switch cmd.Cmd {
+	case cronsun.NodeCmdRmOld:
+		n.Node.RmOldInfo()
+	case cronsun.NodeCmdSync:
+		n.Node.SyncToMgo()
+	}
+
+	log.Infof("%s execute csctl command[%s] success", n.String(), cmd.Cmd.String())
+	return nil
+}

node/node.go

@@ -79,9 +79,6 @@ func NewNode(cfg *conf.Conf) (n *Node, err error) {
 
 // 注册到 /cronsun/node/xx
 func (n *Node) Register() (err error) {
-	// remove old version(< 0.3.0) node info
-	cronsun.DefalutClient.Delete(conf.Config.Node + n.IP)
-
 	pid, err := n.Node.Exist()
 	if err != nil {
 		return
@@ -506,6 +503,18 @@ func (n *Node) watchOnce() {
 	}
 }
 
+func (n *Node) watchCsctl() {
+	rch := cronsun.WatchCsctl()
+	for wresp := range rch {
+		for _, ev := range wresp.Events {
+			switch {
+			case ev.IsCreate(), ev.IsModify():
+				n.executCsctlCmd(ev.Kv.Key, ev.Kv.Value)
+			}
+		}
+	}
+}
+
 // 启动服务
 func (n *Node) Run() (err error) {
 	go n.keepAlive()
@@ -524,6 +533,7 @@ func (n *Node) Run() (err error) {
 	go n.watchJobs()
 	go n.watchGroups()
 	go n.watchOnce()
+	go n.watchCsctl()
 	n.Node.On()
 	return
 }