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>