feat: job config

README.md

@@ -210,6 +210,10 @@ lfs:
   # Enable Git SSH transfer.
   ssh_enabled: true
 
+# Cron job configuration
+jobs:
+  mirror_pull: "@every 10m"
+
 # The stats server configuration.
 stats:
   # The address on which the stats server will listen.

server/config/config.go

@@ -105,6 +105,11 @@
 	SSHEnabled bool `env:"SSH_ENABLED" yaml:"ssh_enabled"`
 }
 
+// JobsConfig is the configuration for cron jobs
+type JobsConfig struct {
+	MirrorPull string `env:"MIRROR_PULL" yaml:"mirror_pull"`
+}
+
 // Config is the configuration for Soft Serve.
 type Config struct {
 	// Name is the name of the server.
@@ -131,6 +136,9 @@
 	// LFS is the configuration for Git LFS.
 	LFS LFSConfig `envPrefix:"LFS_" yaml:"lfs"`
 
+	// Jobs is the configuration for cron jobs
+	Jobs JobsConfig `envPrefix:"JOBS_" yaml:"jobs"`
+
 	// InitialAdminKeys is a list of public keys that will be added to the list of admins.
 	InitialAdminKeys []string `env:"INITIAL_ADMIN_KEYS" envSeparator:"\n" yaml:"initial_admin_keys"`
 

server/jobs/jobs.go

@@ -7,9 +7,14 @@
 
 // Job is a job that can be registered with the scheduler.
 type Job struct {
-	ID   int
-	Spec string
-	Func func(context.Context) func()
+	ID     int
+	Runner Runner
+}
+
+// Runner is a job runner
+type Runner interface {
+	Spec(context.Context) string
+	Func(context.Context) func()
 }
 
 var (
@@ -18,10 +23,10 @@
 )
 
 // Register registers a job.
-func Register(name, spec string, fn func(context.Context) func()) {
+func Register(name string, runner Runner) {
 	mtx.Lock()
 	defer mtx.Unlock()
-	jobs[name] = &Job{Spec: spec, Func: fn}
+	jobs[name] = &Job{Runner: runner}
 }
 
 // List returns a map of registered jobs.

server/jobs/mirror.go

@@ -17,87 +17,98 @@
 )
 
 func init() {
-	Register("mirror-pull", "@every 10m", mirrorPull)
+	Register("mirror-pull", mirrorPull{})
 }
 
-// mirrorPull runs the (pull) mirror job task.
-func mirrorPull(ctx context.Context) func() {
+type mirrorPull struct{}
+
+// Spec derives the spec used for pull mirrors and implements Runner
+func (m mirrorPull) Spec(ctx context.Context) string {
+	cfg := config.FromContext(ctx)
+	if cfg.Jobs.MirrorPull != "" {
+		return cfg.Jobs.MirrorPull
+	}
+	return "@every 10m"
+}
+
+// Func runs the (pull) mirror job task and implements Runner
+func (m mirrorPull) Func(ctx context.Context) func() {
 	cfg := config.FromContext(ctx)
 	logger := log.FromContext(ctx).WithPrefix("jobs.mirror")
 	b := backend.FromContext(ctx)
 	dbx := db.FromContext(ctx)
 	datastore := store.FromContext(ctx)
 	return func() {
 		repos, err := b.Repositories(ctx)
 		if err != nil {
 			logger.Error("error getting repositories", "err", err)
 			return
 		}
 
 		// Divide the work up among the number of CPUs.
 		wq := sync.NewWorkPool(ctx, runtime.GOMAXPROCS(0),
 			sync.WithWorkPoolLogger(logger.Errorf),
 		)
 
 		logger.Debug("updating mirror repos")
 		for _, repo := range repos {
 			if repo.IsMirror() {
 				r, err := repo.Open()
 				if err != nil {
 					logger.Error("error opening repository", "repo", repo.Name(), "err", err)
 					continue
 				}
 
 				name := repo.Name()
 				wq.Add(name, func() {
 					repo := repo
 					cmd := git.NewCommand("remote", "update", "--prune").WithContext(ctx)
 					cmd.AddEnvs(
 						fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
 							filepath.Join(cfg.DataPath, "ssh", "known_hosts"),
 							cfg.SSH.ClientKeyPath,
 						),
 					)
 
 					if _, err := cmd.RunInDir(r.Path); err != nil {
 						logger.Error("error running git remote update", "repo", name, "err", err)
 					}
 
 					if cfg.LFS.Enabled {
 						rcfg, err := r.Config()
 						if err != nil {
 							logger.Error("error getting git config", "repo", name, "err", err)
 							return
 						}
 
 						lfsEndpoint := rcfg.Section("lfs").Option("url")
 						if lfsEndpoint == "" {
 							// If there is no LFS url defined, means the repo
 							// doesn't use LFS and we can skip it.
 							return
 						}
 
 						ep, err := lfs.NewEndpoint(lfsEndpoint)
 						if err != nil {
 							logger.Error("error creating LFS endpoint", "repo", name, "err", err)
 							return
 						}
 
 						client := lfs.NewClient(ep)
 						if client == nil {
 							logger.Errorf("failed to create lfs client: unsupported endpoint %s", lfsEndpoint)
 							return
 						}
 
 						if err := backend.StoreRepoMissingLFSObjects(ctx, repo, dbx, datastore, client); err != nil {
 							logger.Error("failed to store missing lfs objects", "err", err, "path", r.Path)
 							return
 						}
 					}
 				})
 			}
 		}
 
 		wq.Run()
 	}
 }

server/server.go

@@ -57,10 +57,10 @@
 	// Add cron jobs.
 	sched := cron.NewScheduler(ctx)
 	for n, j := range jobs.List() {
-		id, err := sched.AddFunc(j.Spec, j.Func(ctx))
+		id, err := sched.AddFunc(j.Runner.Spec(ctx), j.Runner.Func(ctx))
 		if err != nil {
 			logger.Warn("error adding cron job", "job", n, "err", err)
 		}
 
 		j.ID = id
 	}