Cron Jobs

cmd/chronos-server/main.go

@@ -13,6 +13,7 @@ import (
 
 	apihttp "github.com/chronos-scheduler/chronos/internal/api/http"
 	"github.com/chronos-scheduler/chronos/internal/config"
+	cronengine "github.com/chronos-scheduler/chronos/internal/cron"
 	"github.com/chronos-scheduler/chronos/internal/scheduler"
 	kafkastore "github.com/chronos-scheduler/chronos/internal/store/kafka"
 	mongostore "github.com/chronos-scheduler/chronos/internal/store/mongo"
@@ -79,12 +80,17 @@ func main() {
 		}
 	}()
 
+	cronTick := time.Duration(cfg.CronMaterializerTickSeconds) * time.Second
+	cronLookahead := time.Duration(cfg.CronMaterializerLookaheadMinutes) * time.Minute
+	cronMaterializer := cronengine.NewMaterializer(db, cronTick, cronLookahead, 200, 200)
+
 	router := apihttp.NewRouter(apihttp.RouterDeps{
 		Logger:             logger,
 		PostgresDB:         db,
 		MongoClient:        mongoClient,
 		RedisClient:        redisClient,
 		CORSAllowedOrigins: cfg.CORSAllowedOrigins,
+		CronReconciler:     cronMaterializer,
 	})
 
 	srv := &http.Server{
@@ -96,7 +102,7 @@ func main() {
 		pgstore.NewScheduleBucketStore(db),
 		pgstore.NewJobStore(db),
 		kafkaProducer,
-		cfg.SchedulerBucketSeconds,
+		cfg.SchedulerBucketTickSeconds,
 		0,
 	)
 	backgroundCtx, cancelBackground := context.WithCancel(context.Background())
@@ -109,18 +115,27 @@ func main() {
 			logger.Fatal("http server error", zap.Error(err))
 		}
 	}()
+	go func() {
+		logger.Info("cron materializer started",
+			zap.Duration("tick", cronTick),
+			zap.Duration("lookahead", cronLookahead),
+		)
+		if err := cronMaterializer.Run(backgroundCtx); err != nil {
+			logger.Error("cron materializer exited with error", zap.Error(err))
+		}
+	}()
 	go func() {
 		lockTTL := time.Duration(cfg.LeaderLockTTLSeconds) * time.Second
 		logger.Info("scheduler leader loop started",
-			zap.Int("bucket_seconds", cfg.SchedulerBucketSeconds),
+			zap.Int("bucket_tick_seconds", cfg.SchedulerBucketTickSeconds),
 			zap.Duration("lock_ttl", lockTTL),
 		)
 		scheduler.RunWithLeadership(backgroundCtx, redisClient, lockTTL, func(leaderCtx context.Context) {
 			logger.Info("scheduler leadership acquired")
+			defer logger.Info("scheduler leadership released")
 			if err := dispatcher.Run(leaderCtx); err != nil {
 				logger.Error("scheduler dispatcher exited with error", zap.Error(err))
 			}
-			logger.Info("scheduler leadership released")
 		})
 	}()
 

deployments/kind/app/chronos-server.yaml

@@ -11,7 +11,9 @@ data:
   CHRONOS_KAFKABROKERS: "chronos-kafka.chronos-infra.svc.cluster.local:9092"
   CHRONOS_MINIOEENDPOINT: "chronos-minio.chronos-infra.svc.cluster.local:9000"
   CHRONOS_LEADERLOCKTTLSECONDS: "30"
-  CHRONOS_SCHEDULERBUCKETSECONDS: "60"
+  CHRONOS_SCHEDULERBUCKETTICKSECONDS: "60"
+  CHRONOS_CRONMATERIALIZERTICKSECONDS: "30"
+  CHRONOS_CRONMATERIALIZERLOOKAHEADMINUTES: "15"
 ---
 apiVersion: v1
 kind: Secret

go.mod

@@ -126,4 +126,7 @@ require (
 	github.com/segmentio/kafka-go v0.4.50
 )
 
-require github.com/pierrec/lz4/v4 v4.1.16 // indirect
+require (
+	github.com/pierrec/lz4/v4 v4.1.16 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
+)

go.sum

@@ -176,6 +176,8 @@ github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt
 github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
 github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs=
 github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=