Merge pull request #45 from shunfei/develop

v0.2.2

job.go

@@ -21,6 +21,7 @@ import (
 	"github.com/shunfei/cronsun/conf"
 	"github.com/shunfei/cronsun/log"
 	"github.com/shunfei/cronsun/node/cron"
+	"github.com/shunfei/cronsun/utils"
 )
 
 const (
@@ -387,7 +388,15 @@ func (j *Job) alone() {
 }
 
 func (j *Job) splitCmd() {
-	j.cmd = strings.Split(j.Command, " ")
+	ps := strings.SplitN(j.Command, " ", 2)
+	if len(ps) == 1 {
+		j.cmd = ps
+		return
+	}
+
+	j.cmd = make([]string, 0, 2)
+	j.cmd = append(j.cmd, ps[0])
+	j.cmd = append(j.cmd, utils.ParseCmdArguments(ps[1])...)
 }
 
 func (j *Job) String() string {

node.go

@@ -90,7 +90,14 @@ func GetNodesBy(query interface{}) (nodes []*Node, err error) {
 	return
 }
 
-func ISNodeFault(id string) (bool, error) {
+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) {
 	n := 0
 	err := mgoDB.WithC(Coll_Node, func(c *mgo.Collection) error {
 		var e error

noticer.go

@@ -182,15 +182,16 @@ func monitorNodes(n Noticer) {
 			switch {
 			case ev.Type == client.EventTypeDelete:
 				id = GetIDFromKey(string(ev.Kv.Key))
-				ok, err = ISNodeFault(id)
+				ok, err = ISNodeAlive(id)
 				if err != nil {
 					log.Warnf("query node[%s] err: %s", id, err.Error())
 					continue
 				}
 
 				if ok {
 					n.Send(&Message{
-						Subject: "node[" + id + "] fault at time[" + time.Now().Format(time.RFC3339) + "]",
+						Subject: "Node[" + id + "] break away cluster, this happed at " + time.Now().Format(time.RFC3339),
+						Body:    "Node breaked away cluster, this might happed when node crash or network problems.",
 						To:      conf.Config.Mail.To,
 					})
 				}

utils/argument_parser.go

@@ -0,0 +1,144 @@
+package utils
+
+import (
+	"errors"
+)
+
+type fmsState int
+
+const (
+	stateArgumentOutside fmsState = iota
+	stateArgumentStart
+	stateArgumentEnd
+)
+
+var errEndOfLine = errors.New("End of line")
+
+type cmdArgumentParser struct {
+	s            string
+	i            int
+	length       int
+	state        fmsState
+	startToken   byte
+	shouldEscape bool
+	currArgument []byte
+	err          error
+}
+
+func newCmdArgumentParser(s string) *cmdArgumentParser {
+	return &cmdArgumentParser{
+		s:            s,
+		i:            -1,
+		length:       len(s),
+		currArgument: make([]byte, 0, 16),
+	}
+}
+
+func (cap *cmdArgumentParser) parse() (arguments []string) {
+	for {
+		cap.next()
+
+		if cap.err != nil {
+			if cap.shouldEscape {
+				cap.currArgument = append(cap.currArgument, '\\')
+			}
+
+			if len(cap.currArgument) > 0 {
+				arguments = append(arguments, string(cap.currArgument))
+			}
+
+			return
+		}
+
+		switch cap.state {
+		case stateArgumentOutside:
+			cap.detectStartToken()
+		case stateArgumentStart:
+			if !cap.detectEnd() {
+				cap.detectContent()
+			}
+		case stateArgumentEnd:
+			cap.state = stateArgumentOutside
+			arguments = append(arguments, string(cap.currArgument))
+			cap.currArgument = cap.currArgument[:0]
+		}
+	}
+}
+
+func (cap *cmdArgumentParser) previous() {
+	if cap.i >= 0 {
+		cap.i--
+	}
+}
+
+func (cap *cmdArgumentParser) next() {
+	if cap.length-cap.i == 1 {
+		cap.err = errEndOfLine
+		return
+	}
+	cap.i++
+}
+
+func (cap *cmdArgumentParser) detectStartToken() {
+	c := cap.s[cap.i]
+	if c == ' ' {
+		return
+	}
+
+	switch c {
+	case '\\':
+		cap.startToken = 0
+		cap.shouldEscape = true
+	case '"', '\'':
+		cap.startToken = c
+	default:
+		cap.startToken = 0
+		cap.previous()
+	}
+	cap.state = stateArgumentStart
+}
+
+func (cap *cmdArgumentParser) detectContent() {
+	c := cap.s[cap.i]
+
+	if cap.shouldEscape {
+		switch c {
+		case ' ', '\\', cap.startToken:
+			cap.currArgument = append(cap.currArgument, c)
+		default:
+			cap.currArgument = append(cap.currArgument, '\\', c)
+		}
+		cap.shouldEscape = false
+		return
+	}
+
+	if c == '\\' {
+		cap.shouldEscape = true
+	} else {
+		cap.currArgument = append(cap.currArgument, c)
+	}
+}
+
+func (cap *cmdArgumentParser) detectEnd() (detected bool) {
+	c := cap.s[cap.i]
+
+	if cap.startToken == 0 {
+		if c == ' ' && !cap.shouldEscape {
+			cap.state = stateArgumentEnd
+			cap.previous()
+			return true
+		}
+		return false
+	}
+
+	if c == cap.startToken && !cap.shouldEscape {
+		cap.state = stateArgumentEnd
+		return true
+	}
+
+	return false
+}
+
+func ParseCmdArguments(s string) (arguments []string) {
+	return newCmdArgumentParser(s).parse()
+}

utils/argument_parser_test.go

@@ -0,0 +1,78 @@
+package utils
+
+import (
+	"testing"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestCmdArgumentParser(t *testing.T) {
+	var args []string
+	var str string
+
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 0)
+	})
+
+	str = "    "
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 0)
+	})
+
+	str = "aa bbb  ccc "
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 3)
+		So(args[0], ShouldEqual, "aa")
+		So(args[1], ShouldEqual, "bbb")
+		So(args[2], ShouldEqual, "ccc")
+	})
+
+	str = "' \\\""
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 1)
+		So(args[0], ShouldEqual, " \\\"")
+	})
+
+	str = `a "b c"` // a "b c"
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 2)
+		So(args[0], ShouldEqual, "a")
+		So(args[1], ShouldEqual, "b c")
+	})
+
+	str = `a '\''"`
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 2)
+		So(args[0], ShouldEqual, "a")
+		So(args[1], ShouldEqual, "'")
+	})
+
+	str = `   \\a   'b c'   c\ d\  `
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 3)
+		So(args[0], ShouldEqual, "\\a")
+		So(args[1], ShouldEqual, "b c")
+		So(args[2], ShouldEqual, "c d ")
+	})
+
+	str = `\`
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 1)
+		So(args[0], ShouldEqual, "\\")
+	})
+
+	str = `  \   ` // \SPACE
+	Convey("Parse Cmd Arguments ["+str+"]", t, func() {
+		args = ParseCmdArguments(str)
+		So(len(args), ShouldEqual, 1)
+		So(args[0], ShouldEqual, " ")
+	})
+}

web/node.go

@@ -8,6 +8,7 @@ import (
 
 	v3 "github.com/coreos/etcd/clientv3"
 	"github.com/gorilla/mux"
+	"gopkg.in/mgo.v2/bson"
 
 	"github.com/shunfei/cronsun"
 	"github.com/shunfei/cronsun/conf"
@@ -163,3 +164,62 @@ func (n *Node) GetNodes(ctx *Context) {
 
 	outJSONWithCode(ctx.W, http.StatusOK, nodes)
 }
+
+// DeleteNode force remove node (by ip) which state in offline or damaged.
+func (n *Node) DeleteNode(ctx *Context) {
+	vars := mux.Vars(ctx.R)
+	ip := strings.TrimSpace(vars["ip"])
+	if len(ip) == 0 {
+		outJSONWithCode(ctx.W, http.StatusBadRequest, "node ip is required.")
+		return
+	}
+
+	resp, err := cronsun.DefalutClient.Get(conf.Config.Node + ip)
+	if err != nil {
+		outJSONWithCode(ctx.W, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	if len(resp.Kvs) > 0 {
+		outJSONWithCode(ctx.W, http.StatusBadRequest, "can not remove a running node.")
+		return
+	}
+
+	err = cronsun.RemoveNode(bson.M{"_id": ip})
+	if err != nil {
+		outJSONWithCode(ctx.W, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	// remove node from group
+	var errmsg = "failed to remove node %s from groups, please remove it manually: %s"
+	resp, err = cronsun.DefalutClient.Get(conf.Config.Group, v3.WithPrefix())
+	if err != nil {
+		outJSONWithCode(ctx.W, http.StatusInternalServerError, fmt.Sprintf(errmsg, ip, err.Error()))
+		return
+	}
+
+	for i := range resp.Kvs {
+		g := cronsun.Group{}
+		err = json.Unmarshal(resp.Kvs[i].Value, &g)
+		if err != nil {
+			outJSONWithCode(ctx.W, http.StatusInternalServerError, fmt.Sprintf(errmsg, ip, err.Error()))
+			return
+		}
+
+		var nids = make([]string, 0, len(g.NodeIDs))
+		for _, nid := range g.NodeIDs {
+			if nid != ip {
+				nids = append(nids, nid)
+			}
+		}
+		g.NodeIDs = nids
+
+		if _, err = g.Put(resp.Kvs[i].ModRevision); err != nil {
+			outJSONWithCode(ctx.W, http.StatusInternalServerError, fmt.Sprintf(errmsg, ip, err.Error()))
+			return
+		}
+	}
+
+	outJSONWithCode(ctx.W, http.StatusNoContent, nil)
+}

web/routers.go

@@ -82,6 +82,8 @@ func initRouters() (s *http.Server, err error) {
 
 	h = NewAuthHandler(nodeHandler.GetNodes)
 	subrouter.Handle("/nodes", h).Methods("GET")
+	h = NewAuthHandler(nodeHandler.DeleteNode)
+	subrouter.Handle("/node/{ip}", h).Methods("DELETE")
 	// get node group list
 	h = NewAuthHandler(nodeHandler.GetGroups)
 	subrouter.Handle("/node/groups", h).Methods("GET")

web/ui/src/components/Job.vue

@@ -169,7 +169,7 @@ export default {
     },
 
     formatLatest: function(latest){
-      return this.$L('{begin ~ end}, on {node} took {times}', formatTime(latest.beginTime, latest.endTime), latest.node, formatDuration(latest.beginTime, latest.endTime))
+      return this.$L('on {node} took {times}, {begin ~ end}', latest.node, formatDuration(latest.beginTime, latest.endTime), formatTime(latest.beginTime, latest.endTime));
     },
 
     showExecuteJobModal: function(jobName, jobGroup, jobId){
@@ -182,4 +182,4 @@ export default {
     ExecuteJob
   }
 }
-</script>
+</script>

web/ui/src/components/Log.vue

@@ -165,7 +165,7 @@ export default {
     },
 
     formatTime: function(log){
-      return this.$L('{begin ~ end}, took {times}', formatTime(log.beginTime, log.endTime), formatDuration(log.beginTime, log.endTime));
+      return this.$L('took {times}, {begin ~ end}', formatDuration(log.beginTime, log.endTime), formatTime(log.beginTime, log.endTime));
     },
 
     showExecuteJobModal: function(jobName, jobGroup, jobId){
@@ -177,4 +177,4 @@ export default {
     ExecuteJob
   }
 }
-</script>
+</script>