Merge pull request #49 from shunfei/develop

v0.2.3

bin/web/server.go

@@ -69,6 +69,15 @@ func main() {
  go cronsun.StartNoticer(noticer)
  }
 
+ period := int64(conf.Config.Web.LogCleaner.EveryMinute)
+ var stopCleaner func(interface{})
+ if period > 0 {
+ closeChan := web.RunLogCleaner(time.Duration(period)*time.Minute, time.Duration(conf.Config.Web.LogCleaner.ExpirationDays)*time.Hour*24)
+ stopCleaner = func(i interface{}) {
+ close(closeChan)
+ }
+ }
+
  go func() {
  err := httpServer.Serve(httpL)
  if err != nil {
@@ -80,7 +89,7 @@ func main() {
 
  log.Infof("cronsun web server started on %s, Ctrl+C or send kill sign to exit", conf.Config.Web.BindAddr)
  // 注册退出事件
- event.On(event.EXIT, conf.Exit)
+ event.On(event.EXIT, conf.Exit, stopCleaner)
  // 监听退出信号
  event.Wait()
  event.Emit(event.EXIT, nil)

conf/conf.go

@@ -76,7 +76,11 @@ type webConfig struct {
  Auth struct {
  Enabled bool
  }
- Session SessionConfig
+ Session SessionConfig
+ LogCleaner struct {
+ EveryMinute int
+ ExpirationDays int
+ }
 }
 
 type SessionConfig struct {
@@ -144,6 +148,15 @@ func (c *Conf) parse() error {
  c.Mgo.Timeout *= time.Second
  }
 
+ if c.Web != nil {
+ if c.Web.LogCleaner.EveryMinute < 0 {
+ c.Web.LogCleaner.EveryMinute = 30
+ }
+ if c.Web.LogCleaner.ExpirationDays <= 0 {
+ c.Web.LogCleaner.ExpirationDays = 1
+ }
+ }
+
  c.Node = cleanKeyPrefix(c.Node)
  c.Proc = cleanKeyPrefix(c.Proc)
  c.Cmd = cleanKeyPrefix(c.Cmd)

conf/files/web.json.sample

@@ -7,5 +7,11 @@
  "StorePrefixPath": "/cronsun/sess/",
  "CookieName": "uid",
  "Expiration": 8640000
+ },
+ "#comment": "Delete the expired log (which store in mongodb) periodically",
+ "LogCleaner": {
+ "#comment": "if EveryMinute is 0, the LogCleaner will not run",
+ "EveryMinute": 0,
+ "ExpirationDays": 3
  }
 }

event/event.go

@@ -24,6 +24,10 @@ func On(name string, fs ...func(interface{})) error {
  }
 
  for _, f := range fs {
+ if fs == nil {
+ continue
+ }
+
  fp := reflect.ValueOf(f).Pointer()
  for i := 0; i < len(evs); i++ {
  if reflect.ValueOf(evs[i]).Pointer() == fp {

job.go

@@ -65,6 +65,8 @@ type Job struct {
  FailNotify bool `json:"fail_notify"`
  // 发送通知地址
  To []string `json:"to"`
+ // 单独对任务指定日志清除时间
+ LogExpiration int `json:"log_expiration"`
 
  // 执行任务的结点，用于记录 job log
  runOn string
@@ -526,6 +528,10 @@ func (j *Job) Check() error {
  return ErrIllegalJobGroupName
  }
 
+ if j.LogExpiration < 0 {
+ j.LogExpiration = 0
+ }
+
  j.User = strings.TrimSpace(j.User)
 
  for i := range j.Rules {

job_log.go

@@ -6,6 +6,7 @@ import (
  "gopkg.in/mgo.v2"
  "gopkg.in/mgo.v2/bson"
 
+ "github.com/shunfei/cronsun/conf"
  "github.com/shunfei/cronsun/log"
 )
 
@@ -28,6 +29,7 @@ type JobLog struct {
  Success bool `bson:"success" json:"success"` // 是否执行成功
  BeginTime time.Time `bson:"beginTime" json:"beginTime"` // 任务开始执行时间，精确到毫秒，索引
  EndTime time.Time `bson:"endTime" json:"endTime"` // 任务执行完毕时间，精确到毫秒
+ Cleanup time.Time `bson:"cleanup,omitempty" json:"-"` // 日志清除时间标志
 }
 
 type JobLatestLog struct {
@@ -102,6 +104,17 @@ func CreateJobLog(j *Job, t time.Time, rs string, success bool) {
  BeginTime: t,
  EndTime: et,
  }
+
+ if conf.Config.Web.LogCleaner.EveryMinute > 0 {
+ var expiration int
+ if j.LogExpiration > 0 {
+ expiration = j.LogExpiration
+ } else {
+ expiration = conf.Config.Web.LogCleaner.ExpirationDays
+ }
+ jl.Cleanup = jl.EndTime.Add(time.Duration(expiration) * time.Hour * 24)
+ }
+
  if err := mgoDB.Insert(Coll_JobLog, jl); err != nil {
  log.Errorf(err.Error())
  }
@@ -133,17 +146,32 @@ func CreateJobLog(j *Job, t time.Time, rs string, success bool) {
 }
 
 type StatExecuted struct {
- Total int64 `bson:"total" json:"total"`
- Successed int64 `bson:"successed" json:"successed"`
- Failed int64 `bson:"failed" json:"failed"`
+ Total int64 `bson:"total" json:"total"`
+ Successed int64 `bson:"successed" json:"successed"`
+ Failed int64 `bson:"failed" json:"failed"`
+ Date string `bson:"date" json:"date"`
 }
 
 func JobLogStat() (s *StatExecuted, err error) {
  err = mgoDB.FindOne(Coll_Stat, bson.M{"name": "job"}, &s)
  return
 }
 
-func JobLogDayStat(day time.Time) (s *StatExecuted, err error) {
- err = mgoDB.FindOne(Coll_Stat, bson.M{"name": "job-day", "date": day.Format("2006-01-02")}, &s)
+func JobLogDailyStat(begin, end time.Time) (ls []*StatExecuted, err error) {
+ const oneDay = time.Hour * 24
+ err = mgoDB.WithC(Coll_Stat, func(c *mgo.Collection) error {
+ dateList := make([]string, 0, 8)
+
+ cur := begin
+ for {
+ dateList = append(dateList, cur.Format("2006-01-02"))
+ cur = cur.Add(oneDay)
+ if cur.After(end) {
+ break
+ }
+ }
+ return c.Find(bson.M{"name": "job-day", "date": bson.M{"$in": dateList}}).Sort("date").All(&ls)
+ })
+
  return
 }

node/cron/cron_test.go

@@ -237,7 +237,7 @@ func TestLocalTimezone(t *testing.T) {
  defer cron.Stop()
 
  select {
- case <-time.After(ONE_SECOND * 2):
+ case <-time.After(ONE_SECOND * 3):
  t.FailNow()
  case <-wait(wg):
  }

version.go

@@ -5,7 +5,7 @@ import (
  "runtime"
 )
 
-const Binary = "v0.2.2"
+const Binary = "v0.2.3"
 
 var (
  Version = fmt.Sprintf("%s (build %s)", Binary, runtime.Version())

web/configuration.go

@@ -5,11 +5,18 @@ import "github.com/shunfei/cronsun/conf"
 type Configuration struct{}
 
 func (cnf *Configuration) Configuratios(ctx *Context) {
- outJSON(ctx.W, struct {
- Security *conf.Security `json:"security"`
- Alarm bool `json:"alarm"`
+ r := struct {
+ Security *conf.Security `json:"security"`
+ Alarm bool `json:"alarm"`
+ LogExpirationDays int `json:"log_expiration_days"`
  }{
  Security: conf.Config.Security,
  Alarm: conf.Config.Mail.Enable,
- })
+ }
+
+ if conf.Config.Web.LogCleaner.EveryMinute > 0 {
+ r.LogExpirationDays = conf.Config.Web.LogCleaner.ExpirationDays
+ }
+
+ outJSON(ctx.W, r)
 }

web/gen_bindata.sh

@@ -1,6 +1,6 @@
 #!/bin/sh
-
-cd ui
+BASEDIR=$(dirname "$0")
+cd $BASEDIR/ui
 npm run build
 cd ..
 go-bindata -pkg "web" -prefix "ui/dist/" -o static_assets.go ./ui/dist/

web/info.go

@@ -13,14 +13,43 @@ type Info struct{}
 
 func (inf *Info) Overview(ctx *Context) {
  var overview = struct {
- TotalJobs int64 `json:"totalJobs"`
- JobExecuted *cronsun.StatExecuted `json:"jobExecuted"`
- JobExecutedDaily *cronsun.StatExecuted `json:"jobExecutedDaily"`
+ TotalJobs int64  `json:"totalJobs"`
+ JobExecuted *cronsun.StatExecuted  `json:"jobExecuted"`
+ JobExecutedDaily []*cronsun.StatExecuted `json:"jobExecutedDaily"`
  }{}
 
+ const day = 24 * time.Hour
+ days := 7
+
  overview.JobExecuted, _ = cronsun.JobLogStat()
- overview.JobExecutedDaily, _ = cronsun.JobLogDayStat(time.Now())
+ end := time.Now()
+ begin := end.Add(time.Duration(1-days) * day)
+ statList, _ := cronsun.JobLogDailyStat(begin, end)
+ list := make([]*cronsun.StatExecuted, days)
+ cur := begin
+
+ for i := 0; i < days; i++ {
+ date := cur.Format("2006-01-02")
+ var se *cronsun.StatExecuted
+
+ for j := range statList {
+ if statList[j].Date == date {
+ se = statList[j]
+ statList = statList[1:]
+ break
+ }
+ }
+
+ if se != nil {
+ list[i] = se
+ } else {
+ list[i] = &cronsun.StatExecuted{Date: date}
+ }
+
+ cur = cur.Add(day)
+ }
 
+ overview.JobExecutedDaily = list
  gresp, err := cronsun.DefalutClient.Get(conf.Config.Cmd, v3.WithPrefix(), v3.WithCountOnly())
  if err == nil {
  overview.TotalJobs = gresp.Count

web/job.go

@@ -2,6 +2,7 @@ package web
 
 import (
  "encoding/json"
+ "fmt"
  "net/http"
  "sort"
  "strings"
@@ -56,26 +57,76 @@ func (j *Job) ChangeJobStatus(ctx *Context) {
  ctx.R.Body.Close()
 
  vars := mux.Vars(ctx.R)
- originJob, rev, err := cronsun.GetJobAndRev(vars["group"], vars["id"])
+ job, err = j.updateJobStatus(vars["group"], vars["id"], job.Pause)
  if err != nil {
  outJSONWithCode(ctx.W, http.StatusInternalServerError, err.Error())
  return
  }
 
- originJob.Pause = job.Pause
+ outJSON(ctx.W, job)
+}
+
+func (j *Job) updateJobStatus(group, id string, isPause bool) (*cronsun.Job, error) {
+ originJob, rev, err := cronsun.GetJobAndRev(group, id)
+ if err != nil {
+ return nil, err
+ }
+
+ if originJob.Pause == isPause {
+ return nil, err
+ }
+
+ originJob.Pause = isPause
  b, err := json.Marshal(originJob)
  if err != nil {
- outJSONWithCode(ctx.W, http.StatusInternalServerError, err.Error())
- return
+ return nil, err
  }
 
  _, err = cronsun.DefalutClient.PutWithModRev(originJob.Key(), string(b), rev)
  if err != nil {
- outJSONWithCode(ctx.W, http.StatusInternalServerError, err.Error())
+ return nil, err
+ }
+
+ return originJob, nil
+}
+
+func (j *Job) BatchChangeJobStatus(ctx *Context) {
+ var jobIds []string
+ decoder := json.NewDecoder(ctx.R.Body)
+ err := decoder.Decode(&jobIds)
+ if err != nil {
+ outJSONWithCode(ctx.W, http.StatusBadRequest, err.Error())
+ return
+ }
+ ctx.R.Body.Close()
+
+ vars := mux.Vars(ctx.R)
+ op := vars["op"]
+ var isPause bool
+ switch op {
+ case "pause":
+ isPause = true
+ case "start":
+ default:
+ outJSONWithCode(ctx.W, http.StatusBadRequest, "Unknow batch operation.")
  return
  }
 
- outJSON(ctx.W, originJob)
+ var updated int
+ for i := range jobIds {
+ id := strings.Split(jobIds[i], "/") // [Group, ID]
+ if len(id) != 2 || id[0] == "" || id[1] == "" {
+ continue
+ }
+
+ _, err = j.updateJobStatus(id[0], id[1], isPause)
+ if err != nil {
+ continue
+ }
+ updated++
+ }
+
+ outJSON(ctx.W, fmt.Sprintf("%d of %d updated.", updated, len(jobIds)))
 }
 
 func (j *Job) UpdateJob(ctx *Context) {