Job runner and agent: pipelines run end to end

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,
-> 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).
+> **Status:** early development. A pipeline runs end to end: the controller schedules its jobs, a worker checks
+> out the exact commit and runs each job's commands through the agent, and the output is stored. Real VMs,
+> secrets and Git providers come next; see [docs/ROADMAP.md](docs/ROADMAP.md).
 > See the [organisation page](https://github.com/opifica) for the architecture.
 
 ## What lives here
@@ -16,12 +16,14 @@ where every job runs in a fresh virtual machine.
 | `schema/` | JSON Schema for pipeline files, for editor autocompletion | Working |
 | `cmd/opifica` | CLI: `validate`, `plan`, `schema` | Working |
 | `proto/` | API definitions: worker ↔ controller, worker ↔ guest agent (Protobuf, ConnectRPC) | Draft |
-| `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, image verification, controller client | Working |
+| `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, controller client, Git mirrors, job runner, `worker.Run` | 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 |
 | `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 |
+| `internal/controller/logs` | Job output on the controller's disk | Working |
+| | Secrets, Git providers, admin API | Not started |
+| `cmd/agent`, `internal/agent` | Runs inside each VM: checkout, running steps, masking secrets in output, killing process trees on cancel and timeout | Working |
+| `cmd/devworker` | A worker without VMs for trying Opifica on one machine (no isolation) | Working |
 | `cmd/signer` | Signing service for release artifacts | Not started |
 
 The workers live in their own repositories, so each one contains only its own OS-specific code:
@@ -123,18 +125,25 @@ Database tests need PostgreSQL. Either set `OPIFICA_TEST_DATABASE_URL` to a serv
 on, or install the PostgreSQL server binaries (e.g. `apt install postgresql-17`): the tests then start their own
 throwaway server. Without either, those tests are skipped.
 
-### Trying the controller
+### Running a pipeline on one machine
+
+`devworker` runs jobs without VMs, directly on your machine, with **no isolation**. Use it only with code you trust.
 
 ```sh
 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
+go build -o bin/ ./cmd/...
+
+bin/controller token create -pool trusted        # prints a join token and the CA fingerprint
+bin/controller serve -addr 127.0.0.1:8443 &       # TLS; keep it running
+
+bin/devworker join -controller https://127.0.0.1:8443 -token opj_... -ca-fingerprint sha256:...
+bin/devworker run  -controller https://127.0.0.1:8443 &
 
 # 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>
+bin/controller run create -repo acme/app -clone-url "$PWD" -pool trusted \
+    -sha "$(git rev-parse HEAD)" -file .opifica/pipeline.yaml
+bin/controller run show <run-id>
+bin/controller run logs <run-id>
 ```
 
 ## Stack

cmd/agent/main.go

@@ -2,41 +2,59 @@
 // worker on the host: checkout, running steps, streaming output and
 // collecting files. It never talks to the controller.
 //
-// This is a skeleton: every method returns "unimplemented". In production the
-// agent listens on vsock or virtio-serial; -listen is for development.
+// In a VM the agent listens on vsock (not implemented yet; see the worker
+// repositories). -listen with a TCP address or unix:/path is for development.
 package main
 
 import (
 	"context"
 	"flag"
 	"log/slog"
-	"net/http"
+	"net"
 	"os"
 	"os/signal"
+	"path/filepath"
+	"strings"
 	"syscall"
 
-	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/agent"
 	"github.com/opifica/opifica/internal/server"
 	"github.com/opifica/opifica/internal/version"
 )
 
 func main() {
-	listen := flag.String("listen", "127.0.0.1:7070", "address to listen on (development only)")
+	listen := flag.String("listen", "127.0.0.1:7070", "TCP address or unix:/path/to/socket (development)")
+	root := flag.String("root", defaultRoot(), "work directory; every request path stays inside it")
 	flag.Parse()
 	log := slog.New(slog.NewTextHandler(os.Stderr, nil)).With("service", "agent")
-	log.Info("starting", "version", version.Version)
+	log.Info("starting", "version", version.Version, "root", *root)
+
+	if err := os.MkdirAll(*root, 0o755); err != nil {
+		log.Error("creating work directory", "err", err)
+		os.Exit(1)
+	}
+	network, addr := "tcp", *listen
+	if p, ok := strings.CutPrefix(*listen, "unix:"); ok {
+		network, addr = "unix", p
+		os.Remove(p)
+	}
+	ln, err := net.Listen(network, addr)
+	if err != nil {
+		log.Error("listening", "err", err)
+		os.Exit(1)
+	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
-	mux := http.NewServeMux()
-	mux.HandleFunc("GET /healthz", server.Health)
-	mux.Handle(opificav1connect.NewAgentServiceHandler(&agentService{}))
-	if err := server.Run(ctx, *listen, mux, log); err != nil {
+	if err := server.Serve(ctx, ln, agent.Handler(&agent.Server{Root: *root, Log: log}), log); err != nil {
 		log.Error("server stopped", "err", err)
 		os.Exit(1)
 	}
 }
 
-type agentService struct {
-	opificav1connect.UnimplementedAgentServiceHandler
+func defaultRoot() string {
+	if d, err := os.UserHomeDir(); err == nil {
+		return filepath.Join(d, "opifica-work")
+	}
+	return "opifica-work"
 }

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, the Connect stream, and the scheduler (runs, jobs, leases,
-// retries, cancellation). Logs, secrets and Git providers come next.
+// tokens, Join, the Connect stream, the scheduler (runs, jobs, leases,
+// retries, cancellation) and job logs. Secrets and Git providers come next.
 package main
 
 import (
@@ -15,6 +15,7 @@ import (
 	"log/slog"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"strings"
 	"syscall"
 	"text/tabwriter"
@@ -24,6 +25,7 @@ import (
 
 	"github.com/opifica/opifica/internal/controller/api"
 	"github.com/opifica/opifica/internal/controller/ca"
+	"github.com/opifica/opifica/internal/controller/logs"
 	"github.com/opifica/opifica/internal/controller/scheduler"
 	"github.com/opifica/opifica/internal/controller/store"
 	"github.com/opifica/opifica/internal/controller/workers"
@@ -42,6 +44,7 @@ Commands:
   run list       List recent runs
   run show       Show a run and its jobs
   run cancel     Cancel a run
+  run logs       Print the output of a run's jobs
   version        Print the version
 
 Common options (or environment variables):
@@ -139,7 +142,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{})
+	sched := scheduler.New(svc.Store, svc.Sessions(), log.With("component", "scheduler"), scheduler.Config{
+		Logs: &logs.Store{Dir: filepath.Join(c.dataDir, "logs")},
+	})
 	svc.Messages = sched
 	go sched.Run(ctx)
 	if err := server.RunTLS(ctx, *addr, api.NewHandler(svc), tlsCfg, log); err != nil {

cmd/controller/runs.go

@@ -2,17 +2,20 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
 	"log/slog"
 	"os"
+	"path/filepath"
 	"strings"
 	"text/tabwriter"
 	"time"
 
 	"github.com/google/uuid"
 
+	"github.com/opifica/opifica/internal/controller/logs"
 	"github.com/opifica/opifica/internal/controller/scheduler"
 	"github.com/opifica/opifica/internal/controller/store/db"
 	"github.com/opifica/opifica/internal/pipeline"
@@ -28,6 +31,8 @@ func runCommand(sub string, args []string, stdout, stderr io.Writer) int {
 		return runShow(args, stdout, stderr)
 	case "cancel":
 		return runCancel(args, stdout, stderr)
+	case "logs":
+		return runLogs(args, stdout, stderr)
 	}
 	fmt.Fprintf(stderr, "controller: unknown command \"run %s\"\n\n%s", sub, usage)
 	return 2
@@ -233,3 +238,47 @@ func short(sha string) string {
 	}
 	return sha
 }
+
+func runLogs(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run logs", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	only := fs.String("job", "", "show only this job (its key, e.g. test[os=linux], or its name)")
+	id, ok := runIDArg(fs, args, stderr, "logs")
+	if !ok {
+		return 2
+	}
+	store := &logs.Store{Dir: filepath.Join(c.dataDir, "logs")}
+	return withScheduler(c, stderr, func(ctx context.Context, _ *scheduler.Scheduler, q *db.Queries) error {
+		jobs, err := q.ListRunJobs(ctx, id)
+		if err != nil {
+			return err
+		}
+		shown := 0
+		for _, j := range jobs {
+			if *only != "" && j.Key != *only && j.Name != *only {
+				continue
+			}
+			shown++
+			fmt.Fprintf(stdout, "---- %s (%s, %s) ----\n", j.Name, j.Os, j.State)
+			r, err := store.Open(j.ID)
+			if errors.Is(err, logs.ErrNoLog) {
+				fmt.Fprintln(stdout, "(no output)")
+				continue
+			}
+			if err != nil {
+				return err
+			}
+			_, err = io.Copy(stdout, r)
+			r.Close()
+			if err != nil {
+				return err
+			}
+		}
+		if *only != "" && shown == 0 {
+			return fmt.Errorf("run %s has no job %q", id, *only)
+		}
+		return nil
+	})
+}

cmd/devworker/main.go

@@ -0,0 +1,129 @@
+// Command devworker is a worker for trying Opifica on one machine, without
+// VMs. Every job gets a fresh directory and its own agent, but runs directly
+// on this machine as the current user.
+//
+// THERE IS NO ISOLATION. Run only code you trust. Real workers live in the
+// worker-linux, worker-windows and worker-macos repositories.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+
+	"github.com/opifica/opifica/internal/version"
+	"github.com/opifica/opifica/pkg/worker"
+	"github.com/opifica/opifica/pkg/worker/workertest"
+)
+
+const usage = `Usage: devworker <command> [options]
+
+A worker WITHOUT VMs, for development: jobs run on this machine, unisolated.
+
+Commands:
+  join   Join a controller with a token from "controller token create"
+  run    Run jobs
+`
+
+func main() {
+	os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
+}
+
+func run(args []string, stdout, stderr io.Writer) int {
+	if len(args) == 0 {
+		fmt.Fprint(stderr, usage)
+		return 2
+	}
+	switch args[0] {
+	case "join":
+		return join(args[1:], stdout, stderr)
+	case "run":
+		return runJobs(args[1:], stderr)
+	case "version":
+		fmt.Fprintln(stdout, version.Version)
+		return 0
+	}
+	fmt.Fprint(stderr, usage)
+	return 2
+}
+
+type common struct {
+	controller string
+	stateDir   string
+	capacity   int
+}
+
+func (c *common) register(fs *flag.FlagSet) {
+	fs.StringVar(&c.controller, "controller", "https://localhost:8443", "controller URL")
+	home, _ := os.UserHomeDir()
+	fs.StringVar(&c.stateDir, "state-dir", filepath.Join(home, ".opifica-devworker"), "identity and repository mirrors")
+	fs.IntVar(&c.capacity, "capacity", 2, "jobs to run at once")
+}
+
+func (c *common) pool() worker.PoolConfig {
+	return worker.PoolConfig{
+		Capacity: c.capacity,
+		Images:   map[string]worker.Image{"default": {Name: "default"}},
+		Warm:     map[string]int{"default": 1},
+	}
+}
+
+func join(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("join", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	token := fs.String("token", "", "join token (required)")
+	fp := fs.String("ca-fingerprint", "", "CA fingerprint printed with the token (required)")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if *token == "" || *fp == "" {
+		fmt.Fprintln(stderr, "devworker: -token and -ca-fingerprint are required")
+		return 2
+	}
+	id, err := worker.Join(context.Background(), worker.JoinConfig{
+		ControllerURL: c.controller, Token: *token, CAFingerprint: *fp, StateDir: c.stateDir,
+		Info: worker.Info(c.pool(), nil),
+	})
+	if err != nil {
+		fmt.Fprintf(stderr, "devworker: %v\n", err)
+		return 1
+	}
+	fmt.Fprintf(stdout, "Joined as worker %s. Identity saved in %s.\n", id.WorkerID, c.stateDir)
+	return 0
+}
+
+func runJobs(args []string, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	log := slog.New(slog.NewTextHandler(stderr, nil)).With("service", "devworker")
+	log.Warn("jobs run on this machine WITHOUT isolation; use only with code you trust")
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	err := worker.Run(ctx, worker.Config{
+		ControllerURL:   c.controller,
+		StateDir:        c.stateDir,
+		Hypervisor:      &workertest.Hypervisor{Agent: true},
+		Pool:            c.pool(),
+		AllowLocalRepos: true,
+		Logger:          log,
+	})
+	if err != nil && ctx.Err() == nil {
+		log.Error("worker stopped", "err", err)
+		return 1
+	}
+	return 0
+}