Add cron schedules and sync docs

agent/internal/app/app.go

@@ -47,9 +47,7 @@ func (a *App) Run(ctx context.Context) error {
 	a.Metrics.ConfigLoadSuccess.Set(1)
 	a.Metrics.ChecksConfigured.Set(float64(len(cfg.Checks)))
 	a.Metrics.BuildInfo.WithLabelValues(a.Version, boolLabel(cfg.PushesToServer()), boolLabel(cfg.ExposesMetrics())).Set(1)
-	for _, c := range cfg.Checks {
-		a.Metrics.CheckInterval.WithLabelValues(c.ID).Set(c.Interval.Duration.Seconds())
-	}
+	a.recordCheckScheduleMetrics(cfg.Checks)
 
 	var wg sync.WaitGroup
 
@@ -256,9 +254,7 @@ func (a *App) Reload(newCfg *config.Config, newAgentID string) error {
 	a.Metrics.ConfigReloadTotal.WithLabelValues("success").Inc()
 	a.Metrics.ConfigLastReloadSuccess.Set(float64(time.Now().Unix()))
 	a.Metrics.ChecksConfigured.Set(float64(len(newCfg.Checks)))
-	for _, c := range newCfg.Checks {
-		a.Metrics.CheckInterval.WithLabelValues(c.ID).Set(c.Interval.Duration.Seconds())
-	}
+	a.recordCheckScheduleMetrics(newCfg.Checks)
 
 	a.schedulerMu.RLock()
 	mgr := a.scheduler
@@ -275,6 +271,16 @@ func (a *App) RecordReloadFailure() {
 	a.Metrics.ConfigReloadTotal.WithLabelValues("failure").Inc()
 }
 
+func (a *App) recordCheckScheduleMetrics(checks []config.CheckConfig) {
+	for _, c := range checks {
+		if c.UsesInterval() {
+			a.Metrics.CheckInterval.WithLabelValues(c.ID).Set(c.Interval.Duration.Seconds())
+		} else {
+			a.Metrics.CheckInterval.DeleteLabelValues(c.ID)
+		}
+	}
+}
+
 func (a *App) config() *config.Config {
 	a.cfgMu.RLock()
 	defer a.cfgMu.RUnlock()

agent/internal/app/app_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/monlet/agent/internal/config"
 	"github.com/monlet/agent/internal/metrics"
 	"github.com/monlet/agent/internal/spool"
+	dto "github.com/prometheus/client_model/go"
 )
 
 func TestPrometheusOnlyDoesNotPush(t *testing.T) {
@@ -257,6 +258,92 @@ func TestReloadAcceptsLabelsAndChecks(t *testing.T) {
 	}
 }
 
+func TestReloadUpdatesIntervalMetricForCron(t *testing.T) {
+	oldCfg := &config.Config{
+		AgentID:  "a1",
+		Hostname: "h",
+		StateDir: "/tmp/old",
+		Server: config.ServerConfig{
+			Enabled:           boolPtr(true),
+			URL:               "http://server",
+			HeartbeatInterval: config.Duration{Duration: 10 * time.Second},
+			BatchInterval:     config.Duration{Duration: 10 * time.Second},
+		},
+		Metrics: config.MetricsConfig{Enabled: false},
+		Checks: []config.CheckConfig{{
+			ID:       "c1",
+			Command:  "true",
+			Interval: config.Duration{Duration: time.Second},
+			Timeout:  config.Duration{Duration: time.Second},
+		}},
+	}
+	a := &App{
+		Cfg:     oldCfg,
+		AgentID: "a1",
+		Metrics: metrics.New(),
+	}
+	a.recordCheckScheduleMetrics(oldCfg.Checks)
+	if got, ok := checkIntervalMetric(t, a, "c1"); !ok || got != 1 {
+		t.Fatalf("interval metric missing before reload: value=%v ok=%v", got, ok)
+	}
+
+	cronCfg := *oldCfg
+	cronCfg.Checks = []config.CheckConfig{{
+		ID:      "c1",
+		Command: "true",
+		Cron:    "0 * * * *",
+		Timeout: config.Duration{Duration: time.Second},
+	}}
+	if err := a.Reload(&cronCfg, "a1"); err != nil {
+		t.Fatal(err)
+	}
+	if got, ok := checkIntervalMetric(t, a, "c1"); ok {
+		t.Fatalf("interval metric must be deleted for cron check, got %v", got)
+	}
+
+	intervalCfg := cronCfg
+	intervalCfg.Checks = []config.CheckConfig{{
+		ID:       "c1",
+		Command:  "true",
+		Interval: config.Duration{Duration: 2 * time.Second},
+		Timeout:  config.Duration{Duration: time.Second},
+	}}
+	if err := a.Reload(&intervalCfg, "a1"); err != nil {
+		t.Fatal(err)
+	}
+	if got, ok := checkIntervalMetric(t, a, "c1"); !ok || got != 2 {
+		t.Fatalf("interval metric missing after interval reload: value=%v ok=%v", got, ok)
+	}
+}
+
+func checkIntervalMetric(t *testing.T, a *App, checkID string) (float64, bool) {
+	t.Helper()
+	families, err := a.Metrics.Registry.Gather()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, family := range families {
+		if family.GetName() != "monlet_agent_check_interval_seconds" {
+			continue
+		}
+		for _, metric := range family.GetMetric() {
+			if labelValue(metric.GetLabel(), "check_id") == checkID && metric.GetGauge() != nil {
+				return metric.GetGauge().GetValue(), true
+			}
+		}
+	}
+	return 0, false
+}
+
+func labelValue(labels []*dto.LabelPair, name string) string {
+	for _, label := range labels {
+		if label.GetName() == name {
+			return label.GetValue()
+		}
+	}
+	return ""
+}
+
 func boolPtr(v bool) *bool {
 	return &v
 }

agent/internal/config/config.go

@@ -11,6 +11,7 @@ import (
 	"unicode/utf8"
 
 	"github.com/BurntSushi/toml"
+	"github.com/robfig/cron/v3"
 )
 
 var idPattern = regexp.MustCompile(`^[A-Za-z0-9._:-]+$`)
@@ -58,6 +59,7 @@ type CheckConfig struct {
 	Name                 string         `toml:"name"`
 	Command              string         `toml:"command"`
 	Interval             Duration       `toml:"interval"`
+	Cron                 string         `toml:"cron"`
 	Timeout              Duration       `toml:"timeout"`
 	NotificationsEnabled *bool          `toml:"notifications_enabled"`
 	DedupeKey            string         `toml:"dedupe_key"`
@@ -211,14 +213,24 @@ func (c *Config) Validate() error {
 		if strings.TrimSpace(ch.Command) == "" {
 			return fmt.Errorf("check %q: command is required", ch.ID)
 		}
-		if ch.Interval.Duration <= 0 {
-			return fmt.Errorf("check %q: interval must be > 0", ch.ID)
-		}
 		if ch.Timeout.Duration <= 0 {
 			return fmt.Errorf("check %q: timeout must be > 0", ch.ID)
 		}
-		if ch.Timeout.Duration > ch.Interval.Duration {
+		if ch.Interval.Duration < 0 {
+			return fmt.Errorf("check %q: interval must be > 0", ch.ID)
+		}
+		ch.Cron = strings.TrimSpace(ch.Cron)
+		hasInterval := ch.Interval.Duration > 0
+		hasCron := ch.Cron != ""
+		switch {
+		case hasInterval == hasCron:
+			return fmt.Errorf("check %q: exactly one of interval or cron is required", ch.ID)
+		case hasInterval && ch.Timeout.Duration > ch.Interval.Duration:
 			return fmt.Errorf("check %q: timeout must be <= interval", ch.ID)
+		case hasCron:
+			if _, err := ParseCronSchedule(ch.Cron); err != nil {
+				return fmt.Errorf("check %q: invalid cron: %w", ch.ID, err)
+			}
 		}
 		if len(ch.DedupeKey) > 256 {
 			return fmt.Errorf("check %q: dedupe_key too long", ch.ID)
@@ -253,6 +265,42 @@ func (c CheckConfig) NotificationsOn() bool {
 	return c.NotificationsEnabled == nil || *c.NotificationsEnabled
 }
 
+func (c CheckConfig) UsesInterval() bool {
+	return c.Interval.Duration > 0
+}
+
+func (c CheckConfig) UsesCron() bool {
+	return strings.TrimSpace(c.Cron) != ""
+}
+
+func ParseCronSchedule(spec string) (cron.Schedule, error) {
+	spec = strings.TrimSpace(spec)
+	if spec == "" {
+		return nil, fmt.Errorf("cron is empty")
+	}
+	withoutTZ := specWithoutTZ(spec)
+	lower := strings.ToLower(withoutTZ)
+	if strings.HasPrefix(lower, "@every") {
+		return nil, fmt.Errorf("@every is not supported; use interval")
+	}
+	if lower == "@reboot" {
+		return nil, fmt.Errorf("@reboot is not supported")
+	}
+	return cron.ParseStandard(spec)
+}
+
+func specWithoutTZ(spec string) string {
+	fields := strings.Fields(spec)
+	if len(fields) == 0 {
+		return ""
+	}
+	first := fields[0]
+	if strings.HasPrefix(first, "CRON_TZ=") || strings.HasPrefix(first, "TZ=") {
+		return strings.ToLower(strings.Join(fields[1:], " "))
+	}
+	return strings.ToLower(strings.Join(fields, " "))
+}
+
 func validateResourceLimits(checkID string, limits ResourceLimits) error {
 	if limits.CPUTime.Duration < 0 {
 		return fmt.Errorf("check %q: resource_limits.cpu_time must be > 0", checkID)

agent/internal/config/config_test.go

@@ -48,9 +48,84 @@ func TestLoadMinimal(t *testing.T) {
 	}
 }
 
+func TestLoadCronCheck(t *testing.T) {
+	c, err := Load(writeTmp(t, `
+	agent_id = "host-1"
+	state_dir = "/tmp/x"
+
+	[server]
+	enabled = true
+	url = "http://localhost"
+
+	[[checks]]
+	id = "c1"
+	command = "true"
+	cron = "CRON_TZ=UTC 0 6 * * *"
+	timeout = "5m"
+	`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !c.Checks[0].UsesCron() || c.Checks[0].UsesInterval() {
+		t.Fatalf("unexpected schedule mode: %+v", c.Checks[0])
+	}
+}
+
+func TestValidateCheckScheduleChoice(t *testing.T) {
+	for name, body := range map[string]string{
+		"both_interval_and_cron": `
+	interval = "10s"
+	cron = "0 * * * *"
+	timeout = "5s"
+	`,
+		"neither_interval_nor_cron": `
+	timeout = "5s"
+	`,
+		"invalid_cron": `
+	cron = "not a cron"
+	timeout = "5s"
+	`,
+		"seconds_cron": `
+	cron = "0 0 * * * *"
+	timeout = "5s"
+	`,
+		"every_cron": `
+	cron = "@every 10s"
+	timeout = "5s"
+	`,
+		"tz_every_cron": `
+	cron = "CRON_TZ=UTC @every 10s"
+	timeout = "5s"
+	`,
+		"reboot_cron": `
+	cron = "@reboot"
+	timeout = "5s"
+	`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			_, err := Load(writeTmp(t, `
+	agent_id = "host-1"
+	state_dir = "/tmp/x"
+
+	[server]
+	enabled = true
+	url = "http://localhost"
+
+	[[checks]]
+	id = "c1"
+	command = "true"
+	`+body+`
+	`))
+			if err == nil {
+				t.Fatal("expected schedule validation error")
+			}
+		})
+	}
+}
+
 func TestLoadResourceLimits(t *testing.T) {
 	c, err := Load(writeTmp(t, `
-agent_id = "host-1"
+	agent_id = "host-1"
 state_dir = "/tmp/x"
 
 [server]

agent/internal/scheduler/scheduler.go

@@ -11,6 +11,8 @@ import (
 	"github.com/monlet/agent/internal/runner"
 )
 
+var parseCronSchedule = config.ParseCronSchedule
+
 // Event is a contract-shaped check result emitted by the scheduler.
 type Event struct {
 	EventID              string    `json:"event_id"`
@@ -207,8 +209,10 @@ func (w *checkWorker) loop(initial config.CheckConfig) {
 		w.wg.Done()
 	}()
 	cfg := initial
-	t := time.NewTicker(cfg.Interval.Duration)
+	t := time.NewTimer(time.Hour)
+	stopTimer(t)
 	defer t.Stop()
+	var timerCh <-chan time.Time
 
 	running := false
 	stopping := false
@@ -230,7 +234,26 @@ func (w *checkWorker) loop(initial config.CheckConfig) {
 		}()
 	}
 
-	start(cfg)
+	arm := func(c config.CheckConfig, immediateInterval bool) bool {
+		delay, err := nextDelay(c, time.Now(), immediateInterval)
+		if err != nil {
+			return false
+		}
+		resetTimer(t, delay)
+		timerCh = t.C
+		return true
+	}
+	disarm := func() {
+		stopTimer(t)
+		timerCh = nil
+	}
+
+	if !arm(cfg, true) {
+		if w.hooks.OnStopped != nil {
+			w.hooks.OnStopped(cfg.ID)
+		}
+		return
+	}
 
 	for {
 		select {
@@ -257,11 +280,21 @@ func (w *checkWorker) loop(initial config.CheckConfig) {
 				continue
 			}
 			cfg = next
-			t.Reset(cfg.Interval.Duration)
+			if cfg.UsesCron() {
+				runAfterCurrent = false
+				if !arm(cfg, false) {
+					return
+				}
+				continue
+			}
 			if running {
 				runAfterCurrent = true
+				disarm()
 			} else {
 				start(cfg)
+				if !arm(cfg, false) {
+					return
+				}
 			}
 		case <-w.doneCh:
 			running = false
@@ -281,26 +314,35 @@ func (w *checkWorker) loop(initial config.CheckConfig) {
 			}
 			if runAfterCurrent {
 				runAfterCurrent = false
+				startAfterDrain := true
 				for {
 					select {
 					case next := <-w.updateCh:
 						if !reflect.DeepEqual(cfg, next) {
 							cfg = next
-							t.Reset(cfg.Interval.Duration)
+							startAfterDrain = !cfg.UsesCron()
 						}
 					default:
-						start(cfg)
+						if startAfterDrain {
+							start(cfg)
+						}
+						if !arm(cfg, false) {
+							return
+						}
 						goto nextLoop
 					}
 				}
 			nextLoop:
 				continue
 			}
-		case <-t.C:
+		case <-timerCh:
 			if stopping {
 				continue
 			}
 			start(cfg)
+			if !arm(cfg, false) {
+				return
+			}
 		}
 	}
 }
@@ -358,3 +400,41 @@ func incidentKey(agentID string, c config.CheckConfig) string {
 	}
 	return agentID + ":" + c.ID
 }
+
+func nextDelay(c config.CheckConfig, now time.Time, immediateInterval bool) (time.Duration, error) {
+	if c.UsesCron() {
+		schedule, err := parseCronSchedule(c.Cron)
+		if err != nil {
+			return 0, err
+		}
+		return delayUntil(now, schedule.Next(now)), nil
+	}
+	if immediateInterval {
+		return 0, nil
+	}
+	return c.Interval.Duration, nil
+}
+
+func delayUntil(now, next time.Time) time.Duration {
+	if next.Before(now) {
+		return 0
+	}
+	return next.Sub(now)
+}
+
+func resetTimer(t *time.Timer, d time.Duration) {
+	if d < 0 {
+		d = 0
+	}
+	stopTimer(t)
+	t.Reset(d)
+}
+
+func stopTimer(t *time.Timer) {
+	if !t.Stop() {
+		select {
+		case <-t.C:
+		default:
+		}
+	}
+}

agent/internal/scheduler/scheduler_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/monlet/agent/internal/config"
+	"github.com/robfig/cron/v3"
 )
 
 func runScheduler(ctx context.Context, agentID string, checks []config.CheckConfig, out chan<- Event, h Hooks) {
@@ -27,6 +28,32 @@ func mkCheck(id, cmd string, interval, timeout time.Duration) config.CheckConfig
 	}
 }
 
+type fakeCronSchedule struct {
+	delay time.Duration
+}
+
+func (s fakeCronSchedule) Next(t time.Time) time.Time {
+	return t.Add(s.delay)
+}
+
+func withFakeCron(t *testing.T, delay time.Duration) {
+	t.Helper()
+	old := parseCronSchedule
+	parseCronSchedule = func(string) (cron.Schedule, error) {
+		return fakeCronSchedule{delay: delay}, nil
+	}
+	t.Cleanup(func() { parseCronSchedule = old })
+}
+
+func mkCronCheck(id, cmd string, timeout time.Duration) config.CheckConfig {
+	return config.CheckConfig{
+		ID:      id,
+		Command: cmd,
+		Cron:    "0 * * * *",
+		Timeout: config.Duration{Duration: timeout},
+	}
+}
+
 func TestEmitsEvents(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
 	defer cancel()
@@ -45,6 +72,41 @@ func TestEmitsEvents(t *testing.T) {
 	}
 }
 
+func TestCronDoesNotRunImmediately(t *testing.T) {
+	withFakeCron(t, 200*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	out := make(chan Event, 10)
+	m := NewManager(ctx, "a1", out, Hooks{})
+	m.Update([]config.CheckConfig{mkCronCheck("cron", "echo cron", 100*time.Millisecond)})
+	select {
+	case ev := <-out:
+		t.Fatalf("cron check ran before next slot: %+v", ev)
+	case <-ctx.Done():
+	}
+	m.Stop()
+	m.Wait()
+}
+
+func TestCronEmitsOnDueSlot(t *testing.T) {
+	withFakeCron(t, 30*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	out := make(chan Event, 10)
+	m := NewManager(ctx, "a1", out, Hooks{})
+	m.Update([]config.CheckConfig{mkCronCheck("cron", "echo cron", 100*time.Millisecond)})
+	select {
+	case ev := <-out:
+		if ev.CheckID != "cron" || ev.Status != "ok" {
+			t.Fatalf("unexpected event: %+v", ev)
+		}
+	case <-ctx.Done():
+		t.Fatal("no cron event")
+	}
+	m.Stop()
+	m.Wait()
+}
+
 func TestNoOverlapSkips(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 600*time.Millisecond)
 	defer cancel()
@@ -59,6 +121,23 @@ func TestNoOverlapSkips(t *testing.T) {
 	}
 }
 
+func TestCronNoOverlapSkips(t *testing.T) {
+	withFakeCron(t, 50*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
+	defer cancel()
+	out := make(chan Event, 10)
+	var skipped int32
+	hooks := Hooks{OnSkipped: func(string) { atomic.AddInt32(&skipped, 1) }}
+	m := NewManager(ctx, "a1", out, hooks)
+	m.Update([]config.CheckConfig{mkCronCheck("cron", "sleep 0.2", 500*time.Millisecond)})
+	<-ctx.Done()
+	m.Stop()
+	m.Wait()
+	if atomic.LoadInt32(&skipped) == 0 {
+		t.Fatal("expected at least one skipped cron slot")
+	}
+}
+
 func TestReportsFullEventChannel(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel()