mirror of
https://github.com/alireza0/s-ui.git
synced 2026-10-22 20:01:52 +00:00
cron: apply globalReset on save without restart, validate spec #1269
This commit is contained in:
parent
4437959fb7
commit
fc6d377a2a
5 changed files with 130 additions and 41 deletions
|
|
@ -69,12 +69,7 @@ func (a *APP) Start() error {
|
|||
return err
|
||||
}
|
||||
|
||||
globalReset, err := a.SettingService.GetGlobalReset()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = a.cronJob.Start(loc, trafficAge, statsBucketSeconds, globalReset)
|
||||
err = a.cronJob.Start(loc, trafficAge, statsBucketSeconds)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,17 +4,11 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/alireza0/s-ui/logger"
|
||||
"github.com/alireza0/s-ui/service"
|
||||
|
||||
"github.com/robfig/cron/v3"
|
||||
)
|
||||
|
||||
// cronParser accepts standard 5-field cron, optional leading seconds (6-field)
|
||||
// and descriptors (@daily, @weekly, @every 10s, ...). Used both for the cron
|
||||
// engine and for parsing the user-provided globalReset spec.
|
||||
var cronParser = cron.NewParser(
|
||||
cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
|
||||
)
|
||||
|
||||
type CronJob struct {
|
||||
cron *cron.Cron
|
||||
}
|
||||
|
|
@ -23,14 +17,14 @@ func NewCronJob() *CronJob {
|
|||
return &CronJob{}
|
||||
}
|
||||
|
||||
func (c *CronJob) Start(loc *time.Location, trafficAge int, statsBucketSeconds int64, globalReset string) error {
|
||||
func (c *CronJob) Start(loc *time.Location, trafficAge int, statsBucketSeconds int64) error {
|
||||
// Recover: robfig/cron does not recover panics by default, and gin only
|
||||
// covers the HTTP side. SkipIfStillRunning: the stats job fires every 10s,
|
||||
// can block that long on the write lock, and overlapping runs each drain
|
||||
// the core's traffic counters.
|
||||
c.cron = cron.New(
|
||||
cron.WithLocation(loc),
|
||||
cron.WithParser(cronParser),
|
||||
cron.WithParser(service.CronParser),
|
||||
cron.WithChain(
|
||||
cron.Recover(cron.DefaultLogger),
|
||||
cron.SkipIfStillRunning(cron.DefaultLogger),
|
||||
|
|
@ -48,15 +42,10 @@ func (c *CronJob) Start(loc *time.Location, trafficAge int, statsBucketSeconds i
|
|||
addJob("@every 10s", NewStatsJob(trafficAge > 0, statsBucketSeconds), "stats job")
|
||||
// Start expiry job
|
||||
addJob("@every 1m", NewDepleteJob(), "deplete job")
|
||||
// Periodic global traffic reset, only when a valid cron spec is configured
|
||||
if globalReset != "" && globalReset != "off" {
|
||||
schedule, err := cronParser.Parse(globalReset)
|
||||
if err != nil {
|
||||
logger.Warning("invalid globalReset cron spec <", globalReset, ">: ", err)
|
||||
} else {
|
||||
addJob(globalReset, NewResetTrafficJob(schedule), "traffic reset job")
|
||||
}
|
||||
}
|
||||
// Periodic global traffic reset. Polled rather than scheduled on the spec
|
||||
// itself: the spec was read once at start, so a saved change did nothing
|
||||
// until the panel was restarted.
|
||||
addJob("@every 1m", NewResetTrafficJob(), "traffic reset job")
|
||||
// Start deleting old stats
|
||||
if trafficAge > 0 {
|
||||
addJob("@daily", NewDelStatsJob(trafficAge), "old stats cleanup")
|
||||
|
|
|
|||
|
|
@ -5,22 +5,37 @@ import (
|
|||
|
||||
"github.com/alireza0/s-ui/logger"
|
||||
"github.com/alireza0/s-ui/service"
|
||||
|
||||
"github.com/robfig/cron/v3"
|
||||
)
|
||||
|
||||
type ResetTrafficJob struct {
|
||||
service.ClientService
|
||||
service.ConfigService
|
||||
service.SettingService
|
||||
schedule cron.Schedule
|
||||
}
|
||||
|
||||
func NewResetTrafficJob(schedule cron.Schedule) *ResetTrafficJob {
|
||||
return &ResetTrafficJob{schedule: schedule}
|
||||
func NewResetTrafficJob() *ResetTrafficJob {
|
||||
return &ResetTrafficJob{}
|
||||
}
|
||||
|
||||
// Run reads the spec on every tick, so a saved schedule applies without a
|
||||
// panel restart. globalResetLast holds the next armed boundary; 0 means none.
|
||||
func (s *ResetTrafficJob) Run() {
|
||||
spec, err := s.SettingService.GetGlobalReset()
|
||||
if err != nil {
|
||||
logger.Warning("ResetTrafficJob: get schedule failed: ", err)
|
||||
return
|
||||
}
|
||||
if spec == "" || spec == "off" {
|
||||
return
|
||||
}
|
||||
// Save rejects a bad spec, so this only trips on one written before that
|
||||
// check existed. Debug, not Warning: it would repeat every minute.
|
||||
schedule, err := service.CronParser.Parse(spec)
|
||||
if err != nil {
|
||||
logger.Debug("ResetTrafficJob: invalid cron spec <", spec, ">: ", err)
|
||||
return
|
||||
}
|
||||
|
||||
loc, err := s.SettingService.GetTimeLocation()
|
||||
if err != nil {
|
||||
logger.Warning("ResetTrafficJob: get time location failed: ", err)
|
||||
|
|
@ -28,16 +43,22 @@ func (s *ResetTrafficJob) Run() {
|
|||
}
|
||||
now := time.Now().In(loc)
|
||||
|
||||
last, err := s.SettingService.GetGlobalResetLast()
|
||||
next, err := s.SettingService.GetGlobalResetLast()
|
||||
if err != nil {
|
||||
logger.Warning("ResetTrafficJob: get last reset time failed: ", err)
|
||||
return
|
||||
}
|
||||
// Logged, not a silent return: the setting holds the *next* boundary
|
||||
// despite its name and is not cleared when the schedule changes, so a
|
||||
// monthly-to-daily switch can look dead for up to a month.
|
||||
if last > now.Unix() {
|
||||
logger.Debug("ResetTrafficJob: next reset is at ", time.Unix(last, 0).In(loc).Format(time.RFC3339), ", nothing to do")
|
||||
// A new or changed schedule: arm its first boundary rather than reset now.
|
||||
if next == 0 {
|
||||
next = schedule.Next(now).Unix()
|
||||
if err = s.SettingService.SetGlobalResetLast(next); err != nil {
|
||||
logger.Warning("ResetTrafficJob: set next reset time failed: ", err)
|
||||
return
|
||||
}
|
||||
logger.Info("ResetTrafficJob: next reset at ", time.Unix(next, 0).In(loc).Format(time.RFC3339))
|
||||
return
|
||||
}
|
||||
if next > now.Unix() {
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -57,10 +78,10 @@ func (s *ResetTrafficJob) Run() {
|
|||
// Advance to the next boundary. schedule.Next returns the nearest upcoming
|
||||
// occurrence, so if several periods were missed (e.g. downtime) it snaps
|
||||
// forward instead of resetting once per missed period.
|
||||
next := s.schedule.Next(now)
|
||||
if err = s.SettingService.SetGlobalResetLast(next.Unix()); err != nil {
|
||||
after := schedule.Next(now)
|
||||
if err = s.SettingService.SetGlobalResetLast(after.Unix()); err != nil {
|
||||
logger.Warning("ResetTrafficJob: set last reset time failed: ", err)
|
||||
return
|
||||
}
|
||||
logger.Info("ResetTrafficJob: traffic reset for all clients; next reset at ", next.Format(time.RFC3339))
|
||||
logger.Info("ResetTrafficJob: traffic reset for all clients; next reset at ", after.Format(time.RFC3339))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,9 +14,17 @@ import (
|
|||
"github.com/alireza0/s-ui/logger"
|
||||
"github.com/alireza0/s-ui/util/common"
|
||||
|
||||
"github.com/robfig/cron/v3"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// CronParser accepts standard 5-field cron, optional leading seconds (6-field)
|
||||
// and descriptors (@daily, @weekly, @every 10s, ...). Used both for the cron
|
||||
// engine and for parsing the user-provided globalReset spec.
|
||||
var CronParser = cron.NewParser(
|
||||
cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
|
||||
)
|
||||
|
||||
var defaultConfig = `{
|
||||
"log": {
|
||||
"level": "info"
|
||||
|
|
@ -374,13 +382,14 @@ func (s *SettingService) GetSubURI() (string, error) {
|
|||
return s.getString("subURI")
|
||||
}
|
||||
|
||||
// GetGlobalReset returns the configured period for resetting all clients'
|
||||
// traffic: "off", "weekly", "monthly" or "yearly".
|
||||
// GetGlobalReset returns the cron spec for resetting all clients' traffic;
|
||||
// empty or "off" means disabled.
|
||||
func (s *SettingService) GetGlobalReset() (string, error) {
|
||||
return s.getString("globalReset")
|
||||
}
|
||||
|
||||
// GetGlobalResetLast returns the unix time of the last global traffic reset.
|
||||
// GetGlobalResetLast returns the unix time of the next armed global traffic
|
||||
// reset (despite the name), or 0 when no boundary is armed yet.
|
||||
func (s *SettingService) GetGlobalResetLast() (int64, error) {
|
||||
str, err := s.getString("globalResetLast")
|
||||
if err != nil {
|
||||
|
|
@ -480,6 +489,25 @@ func (s *SettingService) Save(tx *gorm.DB, data json.RawMessage) error {
|
|||
}
|
||||
}
|
||||
|
||||
// A bad spec used to be accepted here and only logged at the next
|
||||
// panel start, so the reset silently never ran.
|
||||
if key == "globalReset" && obj != "" && obj != "off" {
|
||||
if _, err = CronParser.Parse(obj); err != nil {
|
||||
return common.NewError("invalid cron spec <", obj, ">: ", err)
|
||||
}
|
||||
}
|
||||
// globalResetLast holds the next boundary of the old schedule. Clear it
|
||||
// on a change so the reset job arms the new one instead of waiting out
|
||||
// the old period (a monthly-to-daily switch looked dead for a month).
|
||||
if key == "globalReset" {
|
||||
var old model.Setting
|
||||
if tx.Where("key = ?", key).Limit(1).Find(&old).Error == nil && old.Value != obj {
|
||||
if err = tx.Model(model.Setting{}).Where("key = ?", "globalResetLast").Update("value", "0").Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Delete all stats if it is set to 0
|
||||
if key == "trafficAge" && obj == "0" {
|
||||
err = tx.Where("id > 0").Delete(model.Stats{}).Error
|
||||
|
|
|
|||
56
service/setting_global_reset_test.go
Normal file
56
service/setting_global_reset_test.go
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
package service
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A bad spec used to save fine and only log a warning at the next panel
|
||||
// start, so the reset silently never ran.
|
||||
func TestSaveRejectsInvalidGlobalReset(t *testing.T) {
|
||||
db := settingTestDB(t)
|
||||
s := &SettingService{}
|
||||
|
||||
payload, _ := json.Marshal(map[string]string{"globalReset": "every monday"})
|
||||
if err := s.Save(db, payload); err == nil {
|
||||
t.Fatal("Save accepted an invalid cron spec")
|
||||
}
|
||||
|
||||
for _, spec := range []string{"", "off", "0 0 1 * *", "30 0 0 * * *", "@weekly"} {
|
||||
payload, _ := json.Marshal(map[string]string{"globalReset": spec})
|
||||
if err := s.Save(db, payload); err != nil {
|
||||
t.Errorf("Save rejected %q: %v", spec, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// globalResetLast holds the next boundary of the schedule in force. Left
|
||||
// alone on a change, a monthly-to-daily switch waited out the old month.
|
||||
func TestSaveClearsArmedResetOnScheduleChange(t *testing.T) {
|
||||
db := settingTestDB(t)
|
||||
s := &SettingService{}
|
||||
|
||||
save := func(spec string) {
|
||||
t.Helper()
|
||||
payload, _ := json.Marshal(map[string]string{"globalReset": spec})
|
||||
if err := s.Save(db, payload); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
save("0 0 1 * *")
|
||||
if err := s.SetGlobalResetLast(4102444800); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Same spec saved again, as the form does with every other change.
|
||||
save("0 0 1 * *")
|
||||
if last, _ := s.GetGlobalResetLast(); last != 4102444800 {
|
||||
t.Errorf("unchanged spec cleared the armed boundary: %d", last)
|
||||
}
|
||||
|
||||
save("0 0 * * *")
|
||||
if last, _ := s.GetGlobalResetLast(); last != 0 {
|
||||
t.Errorf("changed spec kept the old boundary: %d", last)
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue