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) {