Scheduled Jobs

.dockerignore

@@ -21,3 +21,6 @@ AGENTS.md
 CLAUDE.md
 GEMINI.md
 Makefile
+
+/.tmp/
+/.cache/

.gitignore

@@ -12,6 +12,8 @@ profile.cov
 vendor/
 go.work
 go.work.sum
+/.tmp/
+/.cache/
 
 # Built binaries
 bin/
@@ -52,3 +54,4 @@ tilt.log
 /CLAUDE.md
 /GEMINI.md
 /AGENTS.md
+

Tiltfile

@@ -80,7 +80,7 @@ docker_build('job-executor', '.',
 
 k8s_yaml('deployments/kind/app/job-executor.yaml')
 k8s_resource('job-executor',
-             resource_deps=['chronos-postgres', 'chronos-redis', 'chronos-kafka'])
+             resource_deps=['chronos-postgres', 'chronos-mongodb', 'chronos-redis', 'chronos-kafka'])
 
 docker_build('bulk-ingestor', '.',
              dockerfile='build/package/bulk-ingestor/Dockerfile',

cmd/chronos-server/main.go

@@ -13,6 +13,8 @@ import (
 
 	apihttp "github.com/chronos-scheduler/chronos/internal/api/http"
 	"github.com/chronos-scheduler/chronos/internal/config"
+	"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"
 	pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres"
 	redisstore "github.com/chronos-scheduler/chronos/internal/store/redis"
@@ -70,6 +72,13 @@ func main() {
 	}
 	logger.Info("redis connected")
 
+	kafkaProducer := kafkastore.NewProducer(cfg.KafkaBrokers)
+	defer func() {
+		if err := kafkaProducer.Close(); err != nil {
+			logger.Error("kafka producer close error", zap.Error(err))
+		}
+	}()
+
 	router := apihttp.NewRouter(apihttp.RouterDeps{
 		Logger:             logger,
 		PostgresDB:         db,
@@ -83,19 +92,44 @@ func main() {
 		Handler: router,
 	}
 
+	dispatcher := scheduler.NewDispatcher(
+		pgstore.NewScheduleBucketStore(db),
+		pgstore.NewJobStore(db),
+		kafkaProducer,
+		cfg.SchedulerBucketSeconds,
+		0,
+	)
+	backgroundCtx, cancelBackground := context.WithCancel(context.Background())
+	defer cancelBackground()
+
 	// Start HTTP server in background
 	go func() {
 		logger.Info("http server listening", zap.String("addr", srv.Addr))
 		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 			logger.Fatal("http server 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.Duration("lock_ttl", lockTTL),
+		)
+		scheduler.RunWithLeadership(backgroundCtx, redisClient, lockTTL, func(leaderCtx context.Context) {
+			logger.Info("scheduler leadership acquired")
+			if err := dispatcher.Run(leaderCtx); err != nil {
+				logger.Error("scheduler dispatcher exited with error", zap.Error(err))
+			}
+			logger.Info("scheduler leadership released")
+		})
+	}()
 
 	// Block until SIGINT or SIGTERM
 	quit := make(chan os.Signal, 1)
 	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
 	sig := <-quit
 	logger.Info("shutdown signal received", zap.String("signal", sig.String()))
+	cancelBackground()
 
 	// Graceful shutdown with 10s timeout
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

cmd/job-executor/main.go

@@ -6,12 +6,19 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"
 
 	"go.uber.org/zap"
 
 	"github.com/chronos-scheduler/chronos/internal/config"
+	"github.com/chronos-scheduler/chronos/internal/executor"
+	"github.com/chronos-scheduler/chronos/internal/executor/dispatch"
+	kafkastore "github.com/chronos-scheduler/chronos/internal/store/kafka"
+	mongostore "github.com/chronos-scheduler/chronos/internal/store/mongo"
+	pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres"
+	redisstore "github.com/chronos-scheduler/chronos/internal/store/redis"
 	"github.com/chronos-scheduler/chronos/internal/telemetry"
 )
 
@@ -34,6 +41,75 @@ func main() {
 		zap.Strings("kafka_brokers", cfg.KafkaBrokers),
 	)
 
+	db, err := pgstore.Connect(cfg.PostgresDSN)
+	if err != nil {
+		logger.Fatal("postgres connection failed", zap.Error(err))
+	}
+	logger.Info("postgres connected")
+
+	mongoClient, err := mongostore.Connect(cfg.MongoURI)
+	if err != nil {
+		logger.Fatal("mongodb connection failed", zap.Error(err))
+	}
+	defer func() {
+		_ = mongoClient.Disconnect(context.Background())
+	}()
+	logger.Info("mongodb connected")
+
+	redisClient, err := redisstore.Connect(cfg.RedisAddr)
+	if err != nil {
+		logger.Fatal("redis connection failed", zap.Error(err))
+	}
+	logger.Info("redis connected")
+
+	kafkaProducer := kafkastore.NewProducer(cfg.KafkaBrokers)
+	defer func() {
+		if err := kafkaProducer.Close(); err != nil {
+			logger.Error("kafka producer close error", zap.Error(err))
+		}
+	}()
+	kafkaConsumer := kafkastore.NewConsumer(cfg.KafkaBrokers, "chronos-executor-fanout", "bucket-triggers")
+	defer func() {
+		if err := kafkaConsumer.Close(); err != nil {
+			logger.Error("kafka consumer close error", zap.Error(err))
+		}
+	}()
+	dispatchConsumerClient := kafkastore.NewConsumer(cfg.KafkaBrokers, "chronos-executor-dispatch", "job-dispatch")
+	defer func() {
+		if err := dispatchConsumerClient.Close(); err != nil {
+			logger.Error("dispatch kafka consumer close error", zap.Error(err))
+		}
+	}()
+
+	fanout := executor.NewFanoutConsumer(
+		logger,
+		kafkaConsumer,
+		kafkaProducer,
+		pgstore.NewJobStore(db),
+		redisClient,
+	)
+
+	jobStore := pgstore.NewJobStore(db)
+	executionStore := pgstore.NewJobExecutionStore(db)
+	logStore := mongostore.NewLogStore(mongoClient, "chronos")
+	kafkaDispatcher := dispatch.NewKafkaDispatcher(kafkaProducer)
+	workerID, _ := os.Hostname()
+	reporter := executor.NewResultReporter(logger, jobStore, executionStore, logStore, kafkaProducer, workerID)
+	taskProcessor := executor.NewTaskProcessor(logger, jobStore, executionStore, kafkaDispatcher, reporter)
+
+	workers := executorWorkersFromEnv()
+	queueSize := executorQueueSizeFromEnv(workers)
+	dispatchPool := executor.NewWorkerPool(workers, queueSize, func(ctx context.Context, task *executor.DispatchTask) error {
+		if err := taskProcessor.Process(ctx, task); err != nil {
+			logger.Error("dispatch task failed",
+				zap.String("job_id", task.Event.JobID),
+				zap.Error(err),
+			)
+		}
+		return nil
+	})
+	dispatchConsumer := executor.NewDispatchConsumer(logger, dispatchConsumerClient, dispatchPool)
+
 	mux := http.NewServeMux()
 	mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json")
@@ -46,12 +122,36 @@ func main() {
 		Handler: mux,
 	}
 
+	workerCtx, workerCancel := context.WithCancel(context.Background())
+	defer workerCancel()
+
 	go func() {
 		logger.Info("http server listening", zap.String("addr", srv.Addr))
 		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 			logger.Fatal("http server error", zap.Error(err))
 		}
 	}()
+	go func() {
+		logger.Info("dispatch consumer started",
+			zap.String("consumer_group", "chronos-executor-dispatch"),
+			zap.String("topic", "job-dispatch"),
+			zap.Int("workers", workers),
+			zap.Int("queue_size", queueSize),
+		)
+		if err := dispatchConsumer.Run(workerCtx); err != nil {
+			logger.Fatal("dispatch consumer exited with error", zap.Error(err))
+		}
+	}()
+
+	go func() {
+		logger.Info("fanout consumer started",
+			zap.String("consumer_group", "chronos-executor-fanout"),
+			zap.String("topic", "bucket-triggers"),
+		)
+		if err := fanout.Run(workerCtx); err != nil {
+			logger.Fatal("fanout consumer exited with error", zap.Error(err))
+		}
+	}()
 
 	quit := make(chan os.Signal, 1)
 	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
@@ -60,10 +160,35 @@ func main() {
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
+	workerCancel()
 
 	if err := srv.Shutdown(ctx); err != nil {
 		logger.Error("http server shutdown error", zap.Error(err))
 	}
 
 	logger.Info("job-executor stopped")
 }
+
+func executorWorkersFromEnv() int {
+	raw := os.Getenv("EXECUTOR_WORKERS")
+	if raw == "" {
+		return 10
+	}
+	n, err := strconv.Atoi(raw)
+	if err != nil || n <= 0 {
+		return 10
+	}
+	return n
+}
+
+func executorQueueSizeFromEnv(workers int) int {
+	raw := os.Getenv("EXECUTOR_QUEUE_SIZE")
+	if raw == "" {
+		return 10
+	}
+	n, err := strconv.Atoi(raw)
+	if err != nil || n <= 0 {
+		return 10
+	}
+	return n
+}

deployments/docker-compose/docker-compose.yml

@@ -94,6 +94,7 @@ services:
     entrypoint: >
       bash -c "
         kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic bucket-triggers  --partitions 12 --replication-factor 1 --config retention.ms=3600000  &&
+        kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic job-dispatch     --partitions 24 --replication-factor 1 --config retention.ms=86400000 &&
         kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic bulk-records     --partitions 24 --replication-factor 1 --config retention.ms=86400000 &&
         kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic bulk-results     --partitions 24 --replication-factor 1 --config retention.ms=86400000 &&
         kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic job-events       --partitions 12 --replication-factor 1 --config retention.ms=604800000 &&

deployments/kind/app/job-executor.yaml

@@ -9,6 +9,8 @@ data:
   CHRONOS_LOGFORMAT: "json"
   CHRONOS_REDISADDR: "chronos-redis-master.chronos-infra.svc.cluster.local:6379"
   CHRONOS_KAFKABROKERS: "chronos-kafka.chronos-infra.svc.cluster.local:9092"
+  EXECUTOR_WORKERS: "10"
+  EXECUTOR_QUEUE_SIZE: "10"
   CHRONOS_LEADERLOCKTTLSECONDS: "30"
 ---
 apiVersion: v1

deployments/kind/infra-manifests/kafka.yaml

@@ -108,6 +108,7 @@ spec:
               echo "Waiting for Kafka..."
               until nc -z chronos-kafka.chronos-infra.svc.cluster.local 9092; do sleep 3; done
               kafka-topics --bootstrap-server chronos-kafka.chronos-infra.svc.cluster.local:9092 --create --if-not-exists --topic bucket-triggers  --partitions 12 --replication-factor 1 --config retention.ms=3600000
+              kafka-topics --bootstrap-server chronos-kafka.chronos-infra.svc.cluster.local:9092 --create --if-not-exists --topic job-dispatch     --partitions 24 --replication-factor 1 --config retention.ms=86400000
               kafka-topics --bootstrap-server chronos-kafka.chronos-infra.svc.cluster.local:9092 --create --if-not-exists --topic bulk-records     --partitions 24 --replication-factor 1 --config retention.ms=86400000
               kafka-topics --bootstrap-server chronos-kafka.chronos-infra.svc.cluster.local:9092 --create --if-not-exists --topic bulk-results     --partitions 24 --replication-factor 1 --config retention.ms=86400000
               kafka-topics --bootstrap-server chronos-kafka.chronos-infra.svc.cluster.local:9092 --create --if-not-exists --topic job-events       --partitions 12 --replication-factor 1 --config retention.ms=604800000

go.mod

@@ -86,14 +86,14 @@ require (
 	github.com/containerd/platforms v0.2.1 // indirect
 	github.com/cpuguy83/dockercfg v0.3.2 // indirect
 	github.com/distribution/reference v0.6.0 // indirect
-	github.com/docker/go-connections v0.6.0 // indirect
+	github.com/docker/go-connections v0.6.0
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/ebitengine/purego v0.10.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
-	github.com/google/uuid v1.6.0 // indirect
+	github.com/google/uuid v1.6.0
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/mdelapenya/tlscert v0.2.0 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
@@ -120,3 +120,10 @@ require (
 	go.opentelemetry.io/otel/metric v1.41.0 // indirect
 	go.opentelemetry.io/otel/trace v1.41.0 // indirect
 )
+
+require (
+	github.com/docker/docker v28.3.3+incompatible
+	github.com/segmentio/kafka-go v0.4.50
+)
+
+require github.com/pierrec/lz4/v4 v4.1.16 // indirect