feat: add cronjob ssh command

Author: linsein

cmd/soft/serve/server.go

@@ -59,11 +59,16 @@ func NewServer(ctx context.Context) (*Server, error) {
 	// Add cron jobs.
 	sched := cron.NewScheduler(ctx)
 	for n, j := range jobs.List() {
-		spec := j.Runner.Spec(ctx)
+		jobCfg, err := j.Runner.Config(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("parse cronjob [%s] config: %w", err)
+		}
+		spec := jobCfg.Spec()
 		if spec == "" {
 			continue
 		}
-		id, err := sched.AddFunc(spec, j.Runner.Func(ctx))
+
+		id, err := sched.AddFunc(spec, j.Runner.Func(ctx, jobCfg))
 		if err != nil {
 			logger.Warn("error adding cron job", "job", n, "err", err)
 		}

pkg/config/config.go

@@ -221,6 +221,7 @@ func (c *Config) Environ() []string {
 		fmt.Sprintf("SOFT_SERVE_LFS_ENABLED=%t", c.LFS.Enabled),
 		fmt.Sprintf("SOFT_SERVE_LFS_SSH_ENABLED=%t", c.LFS.SSHEnabled),
 		fmt.Sprintf("SOFT_SERVE_JOBS_MIRROR_PULL=%s", c.Jobs.MirrorPull),
+		fmt.Sprintf("SOFT_SERVE_JOBS_GIT_GC=%s", c.Jobs.GitGC),
 	}...)
 
 	return envs

pkg/config/file.go

@@ -143,6 +143,7 @@ lfs:
 # Cron job configuration
 jobs:
   mirror_pull: "{{ .Jobs.MirrorPull }}"
+  git_gc: "{{ .Jobs.GitGC }}"
 
 # Additional admin keys.
 #initial_admin_keys:

pkg/jobs/garbage_collector.go

@@ -1,37 +1,64 @@
 package jobs
 
 import (
+	"bytes"
 	"context"
+	"fmt"
 	"runtime"
 
 	"charm.land/log/v2"
 	"github.com/charmbracelet/soft-serve/git"
 	"github.com/charmbracelet/soft-serve/pkg/backend"
 	"github.com/charmbracelet/soft-serve/pkg/config"
 	"github.com/charmbracelet/soft-serve/pkg/sync"
+	"github.com/spf13/pflag"
 )
 
 func init() {
 	Register("git-gc", gitGC{})
 }
 
-type gitGC struct{}
+type (
+	gitGC       struct{}
+	gitGCConfig struct {
+		baseRunnerConfig
 
-func (g gitGC) Spec(ctx context.Context) string {
-	cfg := config.FromContext(ctx)
-	if cfg.Jobs.GitGC != "" {
-		return cfg.Jobs.GitGC
+		RepoConfig map[string]string
+		Aggressive bool
 	}
-	return ""
+)
+
+// Description return the description of garbage collector job task and implements Runner.
+func (m gitGC) Description() string {
+	return "clean up the garbage in repositories"
 }
 
-func (g gitGC) Func(ctx context.Context) func() {
+// Config returns the garbage collector job task configuration and implements Runner.
+func (m gitGC) Config(ctx context.Context) (RunnerConfig, error) {
+	cfg := gitGCConfig{
+		baseRunnerConfig: baseRunnerConfig{CronSpec: ""},
+		Aggressive:       false,
+		RepoConfig:       make(map[string]string),
+	}
+
+	if spec := config.FromContext(ctx).Jobs.GitGC; spec != "" {
+		cfg.CronSpec = spec
+	}
+
+	return &cfg, nil
+}
+
+// Func runs the garbage collector job task and implements Runner.
+func (g gitGC) Func(ctx context.Context, cronCfg RunnerConfig) func() {
 	b := backend.FromContext(ctx)
 	logger := log.FromContext(ctx).WithPrefix("jobs.gitgc")
+	jobcfg := cronCfg.(*gitGCConfig)
+
 	return func() {
 		repos, err := b.Repositories(ctx)
 		if err != nil {
 			logger.Error("error getting repositories", "err", err)
+			fmt.Fprintf(jobcfg.Error(), "error getting repositories: %v\n", err)
 			return
 		}
 
@@ -44,20 +71,56 @@ func (g gitGC) Func(ctx context.Context) func() {
 			r, err := repo.Open()
 			if err != nil {
 				logger.Error("error opening repository", "repo", repo.Name(), "err", err)
+				fmt.Fprintf(jobcfg.Error(), "[%s] error opening repository: %v\n", repo.Name(), err)
 				continue
 			}
 
 			name := repo.Name()
 			wq.Add(name, func() {
-				cmd := git.NewCommand("gc").WithContext(ctx)
+				// buffer and write to stdout/stderr in one go,
+				// avoiding output confusion through parallel writing.
+				var (
+					stdout = bytes.NewBuffer(nil)
+					stderr = bytes.NewBuffer(nil)
+				)
+				defer func() {
+					jobcfg.Output().Write(stdout.Bytes())
+					jobcfg.Error().Write(stderr.Bytes())
+				}()
+
+				var cmdArgs []string = nil
+				for key, val := range jobcfg.RepoConfig {
+					cmdArgs = append(cmdArgs, "-c", key+"="+val)
+				}
+
+				cmdArgs = append(cmdArgs, "gc")
+
+				if jobcfg.Aggressive {
+					cmdArgs = append(cmdArgs, "--aggressive")
+				}
+
+				// `git gc` would not output anything if no tty
+				cmd := git.NewCommand(cmdArgs...).WithContext(ctx)
 				if _, err := cmd.RunInDir(r.Path); err != nil {
 					logger.Error("error running git remote update", "repo", name, "err", err)
+					fmt.Fprintf(stderr, "[%s] git gc failed: %v\n", name, err)
+				} else {
+					fmt.Fprintf(stdout, "[%s] git gc succeed\n", name)
 				}
-
 			})
 
 			// TODO: clean up lfs
 		}
-
+		wq.Run()
 	}
 }
+
+// FlagSet returns the flag set that can modify configuration values and implements RunnerConfig
+func (cfg *gitGCConfig) FlagSet() *pflag.FlagSet {
+	flags := pflag.NewFlagSet("git-gc", pflag.ContinueOnError)
+	flags.StringToStringVarP(&cfg.RepoConfig, "config", "c", cfg.RepoConfig, "Override values from git repository configuration files")
+
+	flags.BoolVar(&cfg.Aggressive, "aggressive", cfg.Aggressive, "Optimize the repository more aggressively, see git-gc(1) for more details")
+
+	return flags
+}

pkg/jobs/jobs.go

@@ -2,7 +2,10 @@ package jobs
 
 import (
 	"context"
+	"io"
 	"sync"
+
+	"github.com/spf13/pflag"
 )
 
 // Job is a job that can be registered with the scheduler.
@@ -13,8 +16,56 @@ type Job struct {
 
 // Runner is a job runner.
 type Runner interface {
-	Spec(context.Context) string
-	Func(context.Context) func()
+	Description() string
+	Config(context.Context) (RunnerConfig, error)
+
+	Func(context.Context, RunnerConfig) func()
+}
+
+// JobConfig is the configuration for Runner, passed to Runner.Func
+type RunnerConfig interface {
+	FlagSet() *pflag.FlagSet
+	Spec() string
+
+	SetOut(out io.Writer)
+	Output() io.Writer
+	SetErr(err io.Writer)
+	Error() io.Writer
+}
+
+// baseRunnerConfig implements the common part of job tasks' RunnerConfig
+type baseRunnerConfig struct {
+	CronSpec string `yaml:"spec"`
+
+	output io.Writer
+	error  io.Writer
+}
+
+// SetOut sets the stdout of cron job
+func (cfg *baseRunnerConfig) SetOut(out io.Writer) { cfg.output = out }
+
+// Output return the stdout of cron job
+func (cfg *baseRunnerConfig) Output() io.Writer {
+	if cfg.output == nil {
+		return io.Discard
+	}
+	return cfg.output
+}
+
+// SetErr sets the stderr of cron job
+func (cfg *baseRunnerConfig) SetErr(err io.Writer) { cfg.error = err }
+
+// Error return the stderr of cron job
+func (cfg *baseRunnerConfig) Error() io.Writer {
+	if cfg.error == nil {
+		return io.Discard
+	}
+	return cfg.error
+}
+
+// Spec derives the spec for built-in job scheduler and implements RunnerConfig.
+func (cfg *baseRunnerConfig) Spec() string {
+	return cfg.CronSpec
 }
 
 var (

pkg/jobs/mirror.go

@@ -1,6 +1,7 @@
 package jobs
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"path/filepath"
@@ -15,30 +16,47 @@ import (
 	"github.com/charmbracelet/soft-serve/pkg/lfs"
 	"github.com/charmbracelet/soft-serve/pkg/store"
 	"github.com/charmbracelet/soft-serve/pkg/sync"
+	"github.com/spf13/pflag"
 )
 
 func init() {
 	Register("mirror-pull", mirrorPull{})
 }
 
-type mirrorPull struct{}
+type (
+	mirrorPull       struct{}
+	mirrorPullConfig struct {
+		baseRunnerConfig
+		RepoConfig map[string]string
+	}
+)
 
-// 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
+// Description return the description of (pull) mirror job task and implements Runner.
+func (m mirrorPull) Description() string {
+	return "fetch upstream for mirror repositories"
+}
+
+// Config returns the (pull) mirror cronjob configuration and implements Runner.
+func (m mirrorPull) Config(ctx context.Context) (RunnerConfig, error) {
+	cfg := mirrorPullConfig{
+		baseRunnerConfig: baseRunnerConfig{CronSpec: "@every 10m"},
+		RepoConfig:       make(map[string]string),
+	}
+	if spec := config.FromContext(ctx).Jobs.MirrorPull; spec != "" {
+		cfg.CronSpec = spec
 	}
-	return "@every 10m"
+
+	return &cfg, nil
 }
 
 // Func runs the (pull) mirror job task and implements Runner.
-func (m mirrorPull) Func(ctx context.Context) func() {
+func (m mirrorPull) Func(ctx context.Context, cronCfg RunnerConfig) func() {
 	cfg := config.FromContext(ctx)
 	logger := log.FromContext(ctx).WithPrefix("jobs.mirror")
 	b := backend.FromContext(ctx)
 	dbx := db.FromContext(ctx)
 	datastore := store.FromContext(ctx)
+	jobcfg := cronCfg.(*mirrorPullConfig)
 	return func() {
 		repos, err := b.Repositories(ctx)
 		if err != nil {
@@ -64,13 +82,31 @@ func (m mirrorPull) Func(ctx context.Context) func() {
 				wq.Add(name, func() {
 					repo := repo
 
+					// buffer and write to stdout/stderr in one go,
+					// avoiding output confusion through parallel writing.
+					var (
+						stdout = bytes.NewBuffer(nil)
+						stderr = bytes.NewBuffer(nil)
+					)
+					defer func() {
+						jobcfg.Output().Write(stdout.Bytes())
+						jobcfg.Error().Write(stderr.Bytes())
+					}()
+
 					cmds := []string{
 						"fetch --prune",         // fetch prune before updating remote
 						"remote update --prune", // update remote and prune remote refs
 					}
 
+					gitFlags := []string{}
+					for key, val := range jobcfg.RepoConfig {
+						gitFlags = append(gitFlags, "-c", key+"="+val)
+					}
+
 					for _, c := range cmds {
 						args := strings.Split(c, " ")
+						args = append(gitFlags, args...)
+
 						cmd := git.NewCommand(args...).WithContext(ctx)
 						cmd.AddEnvs(
 							fmt.Sprintf(`GIT_SSH_COMMAND=ssh -o UserKnownHostsFile="%s" -o StrictHostKeyChecking=no -i "%s"`,
@@ -80,6 +116,7 @@ func (m mirrorPull) Func(ctx context.Context) func() {
 						)
 
 						if _, err := cmd.RunInDir(r.Path); err != nil {
+							fmt.Fprintf(stderr, "[%s]: error running git remote update: %v\n", name, err)
 							logger.Error("error running git remote update", "repo", name, "err", err)
 						}
 					}
@@ -88,6 +125,7 @@ func (m mirrorPull) Func(ctx context.Context) func() {
 						rcfg, err := r.Config()
 						if err != nil {
 							logger.Error("error getting git config", "repo", name, "err", err)
+							fmt.Fprintf(stderr, "[%s]: lfs pull: error getting git config: %v", name, err)
 							return
 						}
 
@@ -101,24 +139,38 @@ func (m mirrorPull) Func(ctx context.Context) func() {
 						ep, err := lfs.NewEndpoint(lfsEndpoint)
 						if err != nil {
 							logger.Error("error creating LFS endpoint", "repo", name, "err", err)
+							fmt.Fprintf(stderr, "[%s]: lfs pull: creating LFS endpoint: %v", name, err)
 							return
 						}
 
 						client := lfs.NewClient(ep)
 						if client == nil {
+							fmt.Fprintf(stderr,
+								"[%s]: lfs pull: failed to create lfs client: unsupported endpoint %s",
+								name, lfsEndpoint)
 							logger.Errorf("failed to create lfs client: unsupported endpoint %s", lfsEndpoint)
 							return
 						}
 
 						if err := backend.StoreRepoMissingLFSObjects(ctx, repo, dbx, datastore, client); err != nil {
+							fmt.Fprintf(stderr, "[%s]: lfs pull: failed to store missing lfs objects: %v", name, err)
 							logger.Error("failed to store missing lfs objects", "err", err, "path", r.Path)
 							return
 						}
 					}
+					fmt.Fprintf(stdout, "[%s]: mirror pull succeed\n", name)
 				})
 			}
 		}
 
 		wq.Run()
 	}
 }
+
+// FlagSet returns the flag set that can modify configuration values and implements RunnerConfig
+func (cfg *mirrorPullConfig) FlagSet() *pflag.FlagSet {
+	flags := pflag.NewFlagSet("mirror-pull", pflag.ContinueOnError)
+	flags.StringToStringVarP(&cfg.RepoConfig, "config", "c", cfg.RepoConfig, "Override values from git repository configuration files")
+
+	return flags
+}