Scheduler: runs, jobs, leases, retries and cancellation

README.md

@@ -3,9 +3,9 @@
 Core of [Opifica](https://github.com/opifica), an open-source CI/CD system built in the EU
 where every job runs in a fresh virtual machine.
 
-> **Status:** early development. Working and tested: the pipeline language, the worker kit's warm pool, and
-> workers joining and staying connected to the controller. Scheduling and the agent come next; see
-> [docs/ROADMAP.md](docs/ROADMAP.md).
+> **Status:** early development. Working and tested: the pipeline language, the worker kit's warm pool,
+> workers joining and staying connected to the controller, and scheduling runs onto them. The job runner and
+> agent come next; see [docs/ROADMAP.md](docs/ROADMAP.md).
 > See the [organisation page](https://github.com/opifica) for the architecture.
 
 ## What lives here
@@ -19,7 +19,8 @@ where every job runs in a fresh virtual machine.
 | `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, image verification, controller client | Working |
 | `pkg/worker/workertest` | Fake hypervisor for testing worker code without VMs | Working |
 | `cmd/controller` | Join tokens, worker joining over TLS, the Connect stream, worker revocation | Working |
-| | Scheduler, runs, logs, secrets, Git providers | Not started |
+| `internal/controller/scheduler` | Runs and jobs, matching jobs to workers, leases, retries, cancellation | Working |
+| | Logs, secrets, Git providers, admin API | Not started |
 | `cmd/agent` | Runs inside each VM, executes steps, streams logs and results | Skeleton |
 | `cmd/signer` | Signing service for release artifacts | Not started |
 
@@ -129,6 +130,11 @@ export OPIFICA_DATABASE_URL=postgres://localhost/opifica
 go run ./cmd/controller token create -pool trusted   # prints a join token and the CA fingerprint
 go run ./cmd/controller serve                        # TLS on :8443
 go run ./cmd/controller worker list
+
+# Until Git providers exist, start runs by hand:
+go run ./cmd/controller run create -repo acme/app -pool trusted -sha $(git rev-parse HEAD) -file .opifica/pipeline.yaml
+go run ./cmd/controller run list
+go run ./cmd/controller run show <run-id>
 ```
 
 ## Stack

cmd/controller/main.go

@@ -2,8 +2,8 @@
 // integration, pipeline engine, scheduler and secrets broker.
 //
 // Implemented so far: the database, the built-in certificate authority, join
-// tokens, Join and the Connect stream (sessions, heartbeats, revocation).
-// Scheduling comes next.
+// tokens, Join, the Connect stream, and the scheduler (runs, jobs, leases,
+// retries, cancellation). Logs, secrets and Git providers come next.
 package main
 
 import (
@@ -24,6 +24,7 @@ import (
 
 	"github.com/opifica/opifica/internal/controller/api"
 	"github.com/opifica/opifica/internal/controller/ca"
+	"github.com/opifica/opifica/internal/controller/scheduler"
 	"github.com/opifica/opifica/internal/controller/store"
 	"github.com/opifica/opifica/internal/controller/workers"
 	"github.com/opifica/opifica/internal/server"
@@ -37,6 +38,10 @@ Commands:
   token create   Create a one-time join token for a new build machine
   worker list    List workers
   worker revoke  Revoke a worker; it is disconnected and can't reconnect
+  run create     Start a run from a pipeline file (until Git providers exist)
+  run list       List recent runs
+  run show       Show a run and its jobs
+  run cancel     Cancel a run
   version        Print the version
 
 Common options (or environment variables):
@@ -62,6 +67,8 @@ func run(args []string, stdout, stderr io.Writer) int {
 		return workerList(args[2:], stdout, stderr)
 	case args[0] == "worker" && len(args) > 1 && args[1] == "revoke":
 		return workerRevoke(args[2:], stdout, stderr)
+	case args[0] == "run" && len(args) > 1:
+		return runCommand(args[1], args[2:], stdout, stderr)
 	case args[0] == "version":
 		fmt.Fprintln(stdout, version.Version)
 		return 0
@@ -132,6 +139,9 @@ func serve(args []string, stderr io.Writer) int {
 		return 1
 	}
 	log.Info("certificate authority ready", "fingerprint", svc.CA.Fingerprint())
+	sched := scheduler.New(svc.Store, svc.Sessions(), log.With("component", "scheduler"), scheduler.Config{})
+	svc.Messages = sched
+	go sched.Run(ctx)
 	if err := server.RunTLS(ctx, *addr, api.NewHandler(svc), tlsCfg, log); err != nil {
 		log.Error("server stopped", "err", err)
 		return 1
@@ -257,16 +267,17 @@ func workerRevoke(args []string, stdout, stderr io.Writer) int {
 	var c common
 	c.register(fs)
 	fs.Usage = func() { fmt.Fprintln(stderr, "Usage: controller worker revoke [options] <worker-id>") }
-	if err := fs.Parse(args); err != nil {
+	pos, err := parseWithArgs(fs, args)
+	if err != nil {
 		return 2
 	}
-	if fs.NArg() != 1 {
+	if len(pos) != 1 {
 		fs.Usage()
 		return 2
 	}
-	id, err := uuid.Parse(fs.Arg(0))
+	id, err := uuid.Parse(pos[0])
 	if err != nil {
-		fmt.Fprintf(stderr, "controller: invalid worker id %q\n", fs.Arg(0))
+		fmt.Fprintf(stderr, "controller: invalid worker id %q\n", pos[0])
 		return 2
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)

cmd/controller/main_test.go

@@ -140,3 +140,39 @@ func TestUsage(t *testing.T) {
 		t.Errorf("token create without -pool: exit %d", code)
 	}
 }
+
+func TestRunCommands(t *testing.T) {
+	dbURL := pgtest.New(t)
+	dir := t.TempDir()
+	file := filepath.Join(dir, "pipeline.yaml")
+	os.WriteFile(file, []byte("on: push\njobs:\n  build:\n    runs-on: linux\n    steps: [{ run: make }]\n  deploy:\n    needs: build\n    runs-on: linux\n    steps: [{ run: make deploy }]\n"), 0o644)
+	common := []string{"-database-url", dbURL, "-data-dir", dir}
+
+	var out, errOut bytes.Buffer
+	if code := run(append([]string{"run", "create", "-repo", "acme/app", "-pool", "trusted", "-sha", "abc123", "-file", file}, common...), &out, &errOut); code != 0 {
+		t.Fatalf("run create: exit %d: %s", code, errOut.String())
+	}
+	id := regexp.MustCompile(`Run (\S+) created with 2 job\(s\)`).FindStringSubmatch(out.String())
+	if id == nil {
+		t.Fatalf("run create output: %s", out.String())
+	}
+
+	out.Reset()
+	if code := run(append([]string{"run", "list"}, common...), &out, &errOut); code != 0 || !regexp.MustCompile(`acme/app\s+push\s+main\s+abc123\s+trusted\s+running`).MatchString(out.String()) {
+		t.Errorf("run list (exit %d):\n%s", code, out.String())
+	}
+	out.Reset()
+	if code := run(append([]string{"run", "show", id[1]}, common...), &out, &errOut); code != 0 ||
+		!regexp.MustCompile(`build\s+linux\s+queued`).MatchString(out.String()) || !regexp.MustCompile(`deploy\s+linux\s+pending`).MatchString(out.String()) {
+		t.Errorf("run show (exit %d):\n%s", code, out.String())
+	}
+	out.Reset()
+	if code := run(append([]string{"run", "cancel", id[1]}, common...), &out, &errOut); code != 0 {
+		t.Errorf("run cancel: exit %d: %s", code, errOut.String())
+	}
+	out.Reset()
+	run(append([]string{"run", "show", id[1]}, common...), &out, &errOut)
+	if !regexp.MustCompile(`: cancelled`).MatchString(out.String()) {
+		t.Errorf("after cancel:\n%s", out.String())
+	}
+}

cmd/controller/runs.go

@@ -0,0 +1,235 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/opifica/opifica/internal/controller/scheduler"
+	"github.com/opifica/opifica/internal/controller/store/db"
+	"github.com/opifica/opifica/internal/pipeline"
+)
+
+func runCommand(sub string, args []string, stdout, stderr io.Writer) int {
+	switch sub {
+	case "create":
+		return runCreate(args, stdout, stderr)
+	case "list":
+		return runList(args, stdout, stderr)
+	case "show":
+		return runShow(args, stdout, stderr)
+	case "cancel":
+		return runCancel(args, stdout, stderr)
+	}
+	fmt.Fprintf(stderr, "controller: unknown command \"run %s\"\n\n%s", sub, usage)
+	return 2
+}
+
+// withScheduler opens the store and builds a scheduler without running its
+// loop: the serving controller picks up what this process writes.
+func withScheduler(c common, stderr io.Writer, fn func(ctx context.Context, sched *scheduler.Scheduler, q *db.Queries) error) int {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	svc, closeFn, err := c.open(ctx)
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	defer closeFn()
+	quiet := slog.New(slog.NewTextHandler(io.Discard, nil))
+	sched := scheduler.New(svc.Store, svc.Sessions(), quiet, scheduler.Config{})
+	if err := fn(ctx, sched, svc.Store.Queries); err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	return 0
+}
+
+func runCreate(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run create", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	repo := fs.String("repo", "", "repository name, e.g. acme/app (required)")
+	cloneURL := fs.String("clone-url", "", "URL the worker clones from")
+	file := fs.String("file", pipeline.DefaultPath, "pipeline file")
+	pool := fs.String("pool", "", "pool to run in (required); later decided by routing policy")
+	event := fs.String("event", "push", "push, pull_request or tag")
+	branch := fs.String("branch", "main", "branch (push) or source branch (pull request)")
+	base := fs.String("base", "main", "pull request target branch")
+	tag := fs.String("tag", "", "tag name, for -event tag")
+	sha := fs.String("sha", "", "commit SHA (required)")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if *repo == "" || *pool == "" || *sha == "" {
+		fmt.Fprintln(stderr, "controller: -repo, -pool and -sha are required")
+		return 2
+	}
+	src, err := os.ReadFile(*file)
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	ev := pipeline.Event{Type: pipeline.EventType(*event), SHA: *sha}
+	switch ev.Type {
+	case pipeline.EventPush:
+		ev.Branch, ev.Ref = *branch, "refs/heads/"+*branch
+	case pipeline.EventPullRequest:
+		ev.Branch, ev.BaseBranch = *branch, *base
+	case pipeline.EventTag:
+		ev.Tag, ev.Ref = *tag, "refs/tags/"+*tag
+	default:
+		fmt.Fprintf(stderr, "controller: unknown event %q\n", *event)
+		return 2
+	}
+	return withScheduler(c, stderr, func(ctx context.Context, sched *scheduler.Scheduler, q *db.Queries) error {
+		run, err := sched.CreateRun(ctx, scheduler.RunRequest{
+			Repo: *repo, CloneURL: *cloneURL, Event: ev, Pool: *pool, Pipeline: src,
+		})
+		if err != nil {
+			return err
+		}
+		jobs, err := q.ListRunJobs(ctx, run.ID)
+		if err != nil {
+			return err
+		}
+		fmt.Fprintf(stdout, "Run %s created with %d job(s).\n", run.ID, len(jobs))
+		return nil
+	})
+}
+
+func runList(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run list", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	limit := fs.Int("n", 20, "number of runs to show")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	return withScheduler(c, stderr, func(ctx context.Context, _ *scheduler.Scheduler, q *db.Queries) error {
+		runs, err := q.ListRuns(ctx, int32(*limit))
+		if err != nil {
+			return err
+		}
+		tw := tabwriter.NewWriter(stdout, 0, 4, 2, ' ', 0)
+		fmt.Fprintln(tw, "ID\tREPO\tEVENT\tREF\tCOMMIT\tPOOL\tSTATE\tCREATED")
+		for _, r := range runs {
+			fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", r.ID, r.Repo, r.EventType, refName(r),
+				short(r.CommitSha), r.Pool, r.State, r.CreatedAt.Local().Format(time.DateTime))
+		}
+		return tw.Flush()
+	})
+}
+
+func runShow(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run show", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	id, ok := runIDArg(fs, args, stderr, "show")
+	if !ok {
+		return 2
+	}
+	return withScheduler(c, stderr, func(ctx context.Context, _ *scheduler.Scheduler, q *db.Queries) error {
+		r, err := q.GetRun(ctx, id)
+		if err != nil {
+			return fmt.Errorf("run %s: %w", id, err)
+		}
+		fmt.Fprintf(stdout, "Run %s\n  %s %s @ %s in pool %s: %s\n\n", r.ID, r.Repo, refName(r), short(r.CommitSha), r.Pool, r.State)
+		jobs, err := q.ListRunJobs(ctx, id)
+		if err != nil {
+			return err
+		}
+		tw := tabwriter.NewWriter(stdout, 0, 4, 2, ' ', 0)
+		fmt.Fprintln(tw, "JOB\tOS\tSTATE\tATTEMPT\tWORKER\tEXIT\tNOTE")
+		for _, j := range jobs {
+			worker, exit := "-", "-"
+			if j.WorkerID != nil {
+				worker = j.WorkerID.String()[:8]
+			}
+			if j.ExitCode != nil {
+				exit = fmt.Sprint(*j.ExitCode)
+			}
+			fmt.Fprintf(tw, "%s\t%s\t%s\t%d/%d\t%s\t%s\t%s\n", j.Name, j.Os, j.State, j.Attempt, j.MaxAttempts, worker, exit, j.Error)
+		}
+		return tw.Flush()
+	})
+}
+
+func runCancel(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run cancel", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	id, ok := runIDArg(fs, args, stderr, "cancel")
+	if !ok {
+		return 2
+	}
+	return withScheduler(c, stderr, func(ctx context.Context, sched *scheduler.Scheduler, _ *db.Queries) error {
+		if err := sched.CancelRun(ctx, id); err != nil {
+			return err
+		}
+		fmt.Fprintf(stdout, "Run %s is being cancelled; running jobs are stopped on their workers.\n", id)
+		return nil
+	})
+}
+
+func runIDArg(fs *flag.FlagSet, args []string, stderr io.Writer, cmd string) (uuid.UUID, bool) {
+	pos, err := parseWithArgs(fs, args)
+	if err != nil {
+		return uuid.UUID{}, false
+	}
+	if len(pos) != 1 {
+		fmt.Fprintf(stderr, "Usage: controller run %s [options] <run-id>\n", cmd)
+		return uuid.UUID{}, false
+	}
+	id, err := uuid.Parse(pos[0])
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: invalid run id %q\n", pos[0])
+		return uuid.UUID{}, false
+	}
+	return id, true
+}
+
+// parseWithArgs parses flags that may come before or after positional
+// arguments, e.g. both "show -data-dir x ID" and "show ID -data-dir x".
+func parseWithArgs(fs *flag.FlagSet, args []string) ([]string, error) {
+	var pos []string
+	for {
+		if err := fs.Parse(args); err != nil {
+			return nil, err
+		}
+		if fs.NArg() == 0 {
+			return pos, nil
+		}
+		pos = append(pos, fs.Arg(0))
+		args = fs.Args()[1:]
+	}
+}
+
+func refName(r db.Run) string {
+	switch {
+	case r.Tag != "":
+		return r.Tag
+	case r.EventType == "pull_request":
+		return r.Branch + " -> " + r.BaseBranch
+	}
+	return strings.TrimPrefix(r.Branch, "refs/heads/")
+}
+
+func short(sha string) string {
+	if len(sha) > 12 {
+		return sha[:12]
+	}
+	return sha
+}