From a023d6909188e608f131849f7cf8cb97187a8367 Mon Sep 17 00:00:00 2001 From: "K.Milchev" <49519924+KristiforMilchev@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:02:30 +0200 Subject: [PATCH] Job runner and agent: pipelines run end to end - internal/agent: checkout of the exact commit, running steps (bash with pipefail, sh, pwsh, cmd), secrets masked in output even when split across chunks, the whole process tree killed on cancel and timeout, collecting files without following symlinks; paths can't leave the work directory - pkg/worker: Git mirrors (local paths only when allowed; tokens passed via the environment), the job runner (fresh VM per job, infrastructure errors retried, job failures not), an outbox that carries updates and logs across reconnects, and worker.Run for the OS-specific workers - workertest: optional real agent per fake VM - Controller: logs stored per job, deduplicated, accepted only from the worker holding the job; controller run logs - cmd/devworker: try it all on one machine without VMs (no isolation) - Reconnect backoff starts at 100ms so a worker is back within its lease - Pipeline errors: a hint for values containing ': ' - End-to-end tests with real commands, a real repository and real agents Co-Authored-By: Claude Opus 5.5 --- README.md | 35 +- cmd/agent/main.go | 42 +- cmd/controller/main.go | 11 +- cmd/controller/runs.go | 49 +++ cmd/devworker/main.go | 129 ++++++ docs/ROADMAP.md | 18 +- internal/agent/agent.go | 321 +++++++++++++++ internal/agent/agent_test.go | 337 ++++++++++++++++ internal/agent/client.go | 31 ++ internal/agent/proc_unix.go | 20 + internal/agent/proc_windows.go | 19 + internal/controller/logs/logs.go | 101 +++++ internal/controller/logs/logs_test.go | 38 ++ internal/controller/scheduler/scheduler.go | 61 ++- internal/integration/connect_test.go | 14 +- internal/integration/e2e_test.go | 335 ++++++++++++++++ internal/mask/mask.go | 133 +++++++ internal/mask/mask_test.go | 69 ++++ internal/pipeline/parse.go | 11 + internal/pipeline/pipeline_test.go | 2 + pkg/worker/client.go | 6 +- pkg/worker/mirror.go | 160 ++++++++ pkg/worker/runner.go | 431 +++++++++++++++++++++ pkg/worker/worker.go | 206 ++++++++++ pkg/worker/workertest/hypervisor.go | 60 ++- 25 files changed, 2594 insertions(+), 45 deletions(-) create mode 100644 cmd/devworker/main.go create mode 100644 internal/agent/agent.go create mode 100644 internal/agent/agent_test.go create mode 100644 internal/agent/client.go create mode 100644 internal/agent/proc_unix.go create mode 100644 internal/agent/proc_windows.go create mode 100644 internal/controller/logs/logs.go create mode 100644 internal/controller/logs/logs_test.go create mode 100644 internal/integration/e2e_test.go create mode 100644 internal/mask/mask.go create mode 100644 internal/mask/mask_test.go create mode 100644 pkg/worker/mirror.go create mode 100644 pkg/worker/runner.go create mode 100644 pkg/worker/worker.go diff --git a/README.md b/README.md index e9217a4..1ebc3e9 100644 --- a/README.md +++ b/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 +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 +bin/controller run logs ``` ## Stack diff --git a/cmd/agent/main.go b/cmd/agent/main.go index f934a2a..2204144 100644 --- a/cmd/agent/main.go +++ b/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" } diff --git a/cmd/controller/main.go b/cmd/controller/main.go index d845eeb..752e676 100644 --- a/cmd/controller/main.go +++ b/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 { diff --git a/cmd/controller/runs.go b/cmd/controller/runs.go index 058a245..c5b3709 100644 --- a/cmd/controller/runs.go +++ b/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 + }) +} diff --git a/cmd/devworker/main.go b/cmd/devworker/main.go new file mode 100644 index 0000000..ae6e740 --- /dev/null +++ b/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 [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 +} diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 242f0b0..6d10433 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -36,7 +36,7 @@ result are stored. Tested in ordinary CI, using the fake hypervisor with a real - [x] **Scheduler:** match planned jobs to workers by OS, architecture, image, labels and pool; respect `needs`; safe with several controllers (`FOR UPDATE SKIP LOCKED`) - [x] **Runs:** create a run from a pipeline file and an event; track job and run state; cancel -- [ ] **Logs:** receive log chunks, drop duplicates after reconnects, store them on disk +- [x] **Logs:** receive log chunks, drop duplicates after reconnects, store them on disk (`controller run logs`) - [ ] **Secrets:** encrypted store; requests checked against the job token and policy - [ ] **Admin config:** static file for pools, routing (event → pool) and secret policy - [x] **TLS** on the worker API, with worker client certificates accepted @@ -48,16 +48,18 @@ result are stored. Tested in ordinary CI, using the fake hypervisor with a real ### Worker kit and agent - [x] Controller client: join (saving the identity to disk), connect, reconnect with backoff, heartbeats -- [ ] Heartbeats filled from the warm pool and job runner -- [ ] Job runner: lease a VM → checkout → run steps through the agent → stream masked logs → collect reports - and artifacts → destroy the VM -- [ ] Local Git mirror per repository -- [ ] Agent: `Ready`, `Checkout`, `Exec`, `Collect` +- [x] Heartbeats filled from the warm pool and job runner +- [x] Job runner: lease a VM → checkout → run steps through the agent → stream masked logs → destroy the VM; + updates survive reconnects through an outbox +- [ ] Collect and store test reports and artifacts (steps are skipped with a note for now) +- [x] Local Git mirror per repository +- [x] Agent: `Ready`, `Checkout`, `Exec`, `Collect`; secrets masked in output; process trees killed on cancel and timeout +- [x] `devworker`: try everything on one machine without VMs (no isolation) ### Testing -- [ ] Integration test: controller + PostgreSQL + worker (fake hypervisor) + real agent, in CI -- [ ] Failure tests: worker killed mid-job, lease expiry, reconnect, denied secret +- [x] Integration test: controller + PostgreSQL + worker (fake hypervisor) + real agent, in CI +- [x] Failure tests: worker killed mid-job, lease expiry, reconnect, denied secret ## M2 · Triggered by Git diff --git a/internal/agent/agent.go b/internal/agent/agent.go new file mode 100644 index 0000000..e1b0689 --- /dev/null +++ b/internal/agent/agent.go @@ -0,0 +1,321 @@ +// Package agent is the program that runs inside every job VM. It serves the +// AgentService to the worker on the host: checking out code, running steps +// and collecting files. It never talks to the controller, and it only ever +// sees the secrets of the step it is running. +package agent + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "io/fs" + "log/slog" + "os" + "os/exec" + "path/filepath" + "regexp" + "runtime" + "strings" + "sync" + "time" + + "connectrpc.com/connect" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" + "github.com/opifica/opifica/internal/mask" + "github.com/opifica/opifica/internal/version" +) + +// TimeoutExitCode is reported when a command is stopped because it ran out +// of time, matching the timeout(1) convention. +const TimeoutExitCode = 124 + +const ( + chunkSize = 64 << 10 + maxCollected = 2 << 30 // per Collect call +) + +// Server implements opificav1connect.AgentServiceHandler. +type Server struct { + // Root is the work directory. Every path in a request is relative to it + // and may not leave it. + Root string + Log *slog.Logger +} + +var _ opificav1connect.AgentServiceHandler = (*Server)(nil) + +func (s *Server) log() *slog.Logger { + if s.Log != nil { + return s.Log + } + return slog.Default() +} + +// resolve turns a request path into a path under Root. +func (s *Server) resolve(p string) (string, error) { + if p == "" { + return s.Root, nil + } + if filepath.IsAbs(p) || strings.HasPrefix(p, "/") || strings.HasPrefix(p, `\`) { + return "", fmt.Errorf("path %q must be relative to the work directory", p) + } + full := filepath.Join(s.Root, p) + rel, err := filepath.Rel(s.Root, full) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("path %q leaves the work directory", p) + } + return full, nil +} + +func invalid(err error) error { return connect.NewError(connect.CodeInvalidArgument, err) } + +// Ready implements AgentServiceHandler. +func (s *Server) Ready(context.Context, *connect.Request[opificav1.ReadyRequest]) (*connect.Response[opificav1.ReadyResponse], error) { + return connect.NewResponse(&opificav1.ReadyResponse{ + AgentVersion: version.Version, Os: runtime.GOOS, Arch: runtime.GOARCH, + }), nil +} + +var shaRe = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`) + +// Checkout implements AgentServiceHandler. +func (s *Server) Checkout(ctx context.Context, req *connect.Request[opificav1.CheckoutRequest]) (*connect.Response[opificav1.CheckoutResponse], error) { + m := req.Msg + if !shaRe.MatchString(m.GetCommitSha()) { + return nil, invalid(fmt.Errorf("commit_sha must be a full hex commit id, got %q", m.GetCommitSha())) + } + if m.GetMirrorUrl() == "" { + return nil, invalid(errors.New("mirror_url is required")) + } + dir, err := s.resolve(m.GetDirectory()) + if err != nil { + return nil, invalid(err) + } + if entries, err := os.ReadDir(dir); err == nil && len(entries) > 0 { + return nil, invalid(fmt.Errorf("checkout directory %q is not empty", m.GetDirectory())) + } + + steps := [][]string{ + {"clone", "--quiet", "--no-checkout", "--", m.GetMirrorUrl(), dir}, + {"-C", dir, "checkout", "--quiet", "--detach", m.GetCommitSha()}, + {"-C", dir, "rev-parse", "HEAD"}, + } + var head string + for _, args := range steps { + out, err := git(ctx, args...) + if err != nil { + return nil, connect.NewError(connect.CodeFailedPrecondition, err) + } + head = strings.TrimSpace(out) + } + if head != m.GetCommitSha() { + return nil, connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("checked out %s, wanted %s", head, m.GetCommitSha())) + } + return connect.NewResponse(&opificav1.CheckoutResponse{HeadSha: head}), nil +} + +func git(ctx context.Context, args ...string) (string, error) { + cmd := exec.CommandContext(ctx, "git", args...) + // Never prompt, never read the host's or user's git config. + cmd.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL="+os.DevNull) + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + if err := cmd.Run(); err != nil { + return "", fmt.Errorf("git %s: %v: %s", args[0], err, strings.TrimSpace(stderr.String())) + } + return stdout.String(), nil +} + +// shellCommand returns the program and arguments that run script. +func shellCommand(shell, script string) ([]string, error) { + if shell == "" { + if runtime.GOOS == "windows" { + shell = "pwsh" + if _, err := exec.LookPath("pwsh"); err != nil { + shell = "powershell" + } + } else { + shell = "bash" + if _, err := exec.LookPath("bash"); err != nil { + shell = "sh" + } + } + } + switch shell { + case "bash": + // Stop at the first failing command, including inside pipelines. + return []string{"bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", script}, nil + case "sh": + return []string{"sh", "-e", "-c", script}, nil + case "pwsh", "powershell": + return []string{shell, "-NoProfile", "-NonInteractive", "-Command", + "$ErrorActionPreference = 'Stop'\n" + script + "\nif ($LASTEXITCODE) { exit $LASTEXITCODE }"}, nil + case "cmd": + return []string{"cmd", "/D", "/S", "/C", script}, nil + } + return nil, fmt.Errorf("unknown shell %q (bash, sh, pwsh, powershell or cmd)", shell) +} + +// streamWriter sends output as ExecResponse chunks. stdout and stderr write +// from different goroutines, so sends are serialized. +type streamWriter struct { + mu *sync.Mutex + stream *connect.ServerStream[opificav1.ExecResponse] + stderr bool +} + +func (w *streamWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + data := bytes.Clone(p) + msg := &opificav1.ExecResponse{Event: &opificav1.ExecResponse_Stdout{Stdout: data}} + if w.stderr { + msg.Event = &opificav1.ExecResponse_Stderr{Stderr: data} + } + if err := w.stream.Send(msg); err != nil { + return 0, err + } + return len(p), nil +} + +// Exec implements AgentServiceHandler. Output is masked before it leaves the +// agent. The command and everything it started are killed when the step +// times out or the worker cancels the call. +func (s *Server) Exec(ctx context.Context, req *connect.Request[opificav1.ExecRequest], stream *connect.ServerStream[opificav1.ExecResponse]) error { + m := req.Msg + if strings.TrimSpace(m.GetCommand()) == "" { + return invalid(errors.New("command is required")) + } + argv, err := shellCommand(m.GetShell(), m.GetCommand()) + if err != nil { + return invalid(err) + } + dir, err := s.resolve(m.GetWorkingDirectory()) + if err != nil { + return invalid(err) + } + if info, err := os.Stat(dir); err != nil || !info.IsDir() { + return invalid(fmt.Errorf("working directory %q does not exist", m.GetWorkingDirectory())) + } + + runCtx := ctx + var cancelTimeout context.CancelFunc = func() {} + if t := m.GetTimeoutSeconds(); t > 0 { + runCtx, cancelTimeout = context.WithTimeout(ctx, time.Duration(t)*time.Second) + } + defer cancelTimeout() + + cmd := exec.CommandContext(runCtx, argv[0], argv[1:]...) + cmd.Dir = dir + cmd.Env = os.Environ() + for k, v := range m.GetEnv() { + cmd.Env = append(cmd.Env, k+"="+v) + } + killTree(cmd) + // If a background process keeps the output open, don't wait forever. + cmd.WaitDelay = 5 * time.Second + + var mu sync.Mutex + stdout := mask.NewWriter(&streamWriter{mu: &mu, stream: stream}, m.GetMask()) + stderr := mask.NewWriter(&streamWriter{mu: &mu, stream: stream, stderr: true}, m.GetMask()) + cmd.Stdout, cmd.Stderr = stdout, stderr + + runErr := cmd.Run() + stdout.Flush() + stderr.Flush() + + if ctx.Err() != nil { + return connect.NewError(connect.CodeCanceled, errors.New("cancelled by the worker")) + } + code := 0 + var exitErr *exec.ExitError + switch { + case runErr == nil: + case errors.Is(runCtx.Err(), context.DeadlineExceeded): + code = TimeoutExitCode + fmt.Fprintf(stderr, "\nopifica: step timed out after %ds\n", m.GetTimeoutSeconds()) + stderr.Flush() + case errors.As(runErr, &exitErr): + code = exitErr.ExitCode() + if code < 0 { + code = 1 // killed by a signal + } + default: + // The command couldn't start, e.g. the shell is missing. + return connect.NewError(connect.CodeFailedPrecondition, runErr) + } + return stream.Send(&opificav1.ExecResponse{Event: &opificav1.ExecResponse_ExitCode{ExitCode: int32(code)}}) +} + +// Collect implements AgentServiceHandler. Patterns are globs relative to +// base_directory. Symbolic links are skipped, so a job can't make the agent +// read files outside the work directory. +func (s *Server) Collect(ctx context.Context, req *connect.Request[opificav1.CollectRequest], stream *connect.ServerStream[opificav1.CollectResponse]) error { + base, err := s.resolve(req.Msg.GetBaseDirectory()) + if err != nil { + return invalid(err) + } + seen := map[string]bool{} + var total int64 + for _, pattern := range req.Msg.GetPatterns() { + if filepath.IsAbs(pattern) || strings.Contains(filepath.ToSlash(pattern), "..") { + return invalid(fmt.Errorf("pattern %q must stay inside the base directory", pattern)) + } + matches, err := filepath.Glob(filepath.Join(base, pattern)) + if err != nil { + return invalid(fmt.Errorf("pattern %q: %v", pattern, err)) + } + for _, path := range matches { + if seen[path] { + continue + } + seen[path] = true + info, err := os.Lstat(path) + if err != nil || !info.Mode().IsRegular() { + continue // directories, symlinks, devices + } + if total += info.Size(); total > maxCollected { + return connect.NewError(connect.CodeResourceExhausted, fmt.Errorf("collected files exceed %d bytes", maxCollected)) + } + rel, _ := filepath.Rel(base, path) + if err := sendFile(ctx, stream, path, filepath.ToSlash(rel)); err != nil { + return err + } + } + } + return nil +} + +func sendFile(ctx context.Context, stream *connect.ServerStream[opificav1.CollectResponse], path, rel string) error { + f, err := os.Open(path) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + return nil + } + return err + } + defer f.Close() + buf := make([]byte, chunkSize) + for { + if err := ctx.Err(); err != nil { + return err + } + n, err := f.Read(buf) + eof := errors.Is(err, io.EOF) + if err != nil && !eof { + return err + } + if n > 0 || eof { + if err := stream.Send(&opificav1.CollectResponse{Path: rel, Data: bytes.Clone(buf[:n]), Eof: eof}); err != nil { + return err + } + } + if eof { + return nil + } + } +} diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go new file mode 100644 index 0000000..2a82a95 --- /dev/null +++ b/internal/agent/agent_test.go @@ -0,0 +1,337 @@ +package agent_test + +import ( + "context" + "errors" + "io" + "log/slog" + "net" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "connectrpc.com/connect" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" + "github.com/opifica/opifica/internal/agent" + "github.com/opifica/opifica/internal/server" +) + +func start(t *testing.T) (opificav1connect.AgentServiceClient, string) { + t.Helper() + root := t.TempDir() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + done := make(chan struct{}) + go func() { + server.Serve(ctx, ln, agent.Handler(&agent.Server{Root: root, Log: log}), log) + close(done) + }() + t.Cleanup(func() { cancel(); <-done }) + addr := ln.Addr().String() + client := agent.NewClient(func(ctx context.Context) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "tcp", addr) + }) + return client, root +} + +type result struct { + stdout, stderr string + exit int32 + err error +} + +func run(t *testing.T, c opificav1connect.AgentServiceClient, req *opificav1.ExecRequest) result { + t.Helper() + return runCtx(context.Background(), t, c, req) +} + +func runCtx(ctx context.Context, t *testing.T, c opificav1connect.AgentServiceClient, req *opificav1.ExecRequest) result { + t.Helper() + stream, err := c.Exec(ctx, connect.NewRequest(req)) + if err != nil { + return result{err: err} + } + var r result + r.exit = -1 + for stream.Receive() { + switch e := stream.Msg().GetEvent().(type) { + case *opificav1.ExecResponse_Stdout: + r.stdout += string(e.Stdout) + case *opificav1.ExecResponse_Stderr: + r.stderr += string(e.Stderr) + case *opificav1.ExecResponse_ExitCode: + r.exit = e.ExitCode + } + } + r.err = stream.Err() + return r +} + +func unixOnly(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("uses a POSIX shell") + } +} + +func TestReady(t *testing.T) { + c, _ := start(t) + resp, err := c.Ready(context.Background(), connect.NewRequest(&opificav1.ReadyRequest{})) + if err != nil { + t.Fatal(err) + } + if resp.Msg.GetOs() != runtime.GOOS || resp.Msg.GetArch() != runtime.GOARCH { + t.Errorf("ready = %v", resp.Msg) + } +} + +func TestExec(t *testing.T) { + unixOnly(t) + c, root := start(t) + os.MkdirAll(filepath.Join(root, "src", "sub"), 0o755) + + r := run(t, c, &opificav1.ExecRequest{ + Command: `echo "out $GREETING"; echo err >&2; pwd`, + WorkingDirectory: "src/sub", + Env: map[string]string{"GREETING": "hello"}, + }) + if r.err != nil || r.exit != 0 { + t.Fatalf("exit %d, err %v", r.exit, r.err) + } + if !strings.HasPrefix(r.stdout, "out hello\n") || r.stderr != "err\n" { + t.Errorf("stdout %q, stderr %q", r.stdout, r.stderr) + } + if !strings.HasSuffix(strings.TrimSpace(r.stdout), filepath.Join("src", "sub")) { + t.Errorf("ran in the wrong directory: %q", r.stdout) + } + + if r := run(t, c, &opificav1.ExecRequest{Command: "exit 3"}); r.exit != 3 || r.err != nil { + t.Errorf("exit 3: got exit %d, err %v", r.exit, r.err) + } +} + +func TestBashStopsAtFirstFailure(t *testing.T) { + unixOnly(t) + if _, err := exec.LookPath("bash"); err != nil { + t.Skip("no bash") + } + c, _ := start(t) + r := run(t, c, &opificav1.ExecRequest{Command: "false | true\necho should-not-run"}) + if r.exit == 0 || strings.Contains(r.stdout, "should-not-run") { + t.Errorf("a failing pipeline didn't stop the step: exit %d, stdout %q", r.exit, r.stdout) + } + r = run(t, c, &opificav1.ExecRequest{Command: "echo ok", Shell: "sh"}) + if r.exit != 0 || r.stdout != "ok\n" { + t.Errorf("sh: exit %d, stdout %q", r.exit, r.stdout) + } +} + +func TestSecretsAreMasked(t *testing.T) { + unixOnly(t) + c, _ := start(t) + // printf in pieces so the secret reaches the agent split across writes. + r := run(t, c, &opificav1.ExecRequest{ + Command: `printf 'token=s3cr'; sleep 0.05; printf '3t-value!\n'; echo "$TOKEN" >&2`, + Env: map[string]string{"TOKEN": "s3cr3t-value"}, + Mask: []string{"s3cr3t-value"}, + }) + if strings.Contains(r.stdout+r.stderr, "s3cr3t") { + t.Errorf("secret leaked: stdout %q, stderr %q", r.stdout, r.stderr) + } + if r.stdout != "token=***!\n" || r.stderr != "***\n" { + t.Errorf("stdout %q, stderr %q", r.stdout, r.stderr) + } +} + +func TestTimeoutKillsTheWholeTree(t *testing.T) { + unixOnly(t) + c, root := start(t) + pidFile := filepath.Join(root, "child.pid") + start := time.Now() + r := run(t, c, &opificav1.ExecRequest{ + Command: `sleep 60 & echo $! > child.pid; wait`, + TimeoutSeconds: 1, + }) + if r.exit != agent.TimeoutExitCode || !strings.Contains(r.stderr, "timed out after 1s") { + t.Errorf("exit %d, stderr %q", r.exit, r.stderr) + } + if d := time.Since(start); d > 10*time.Second { + t.Errorf("took %v", d) + } + assertDead(t, pidFile) +} + +func TestCancelKillsTheWholeTree(t *testing.T) { + unixOnly(t) + c, root := start(t) + pidFile := filepath.Join(root, "child.pid") + ctx, cancel := context.WithCancel(context.Background()) + go func() { + for { + if _, err := os.Stat(pidFile); err == nil { + cancel() + return + } + time.Sleep(10 * time.Millisecond) + } + }() + r := runCtx(ctx, t, c, &opificav1.ExecRequest{Command: `sleep 60 & echo $! > child.pid; wait`}) + if r.err == nil { + t.Error("cancelled exec returned no error") + } + assertDead(t, pidFile) +} + +// assertDead waits for the process whose PID is in file to be gone. +func assertDead(t *testing.T, file string) { + t.Helper() + data, err := os.ReadFile(file) + if err != nil { + t.Fatalf("reading pid: %v", err) + } + pid, _ := strconv.Atoi(strings.TrimSpace(string(data))) + deadline := time.Now().Add(5 * time.Second) + for { + if err := syscall.Kill(pid, 0); errors.Is(err, syscall.ESRCH) { + return + } + if time.Now().After(deadline) { + syscall.Kill(pid, syscall.SIGKILL) + t.Fatalf("background process %d survived", pid) + } + time.Sleep(20 * time.Millisecond) + } +} + +func TestExecRejects(t *testing.T) { + c, _ := start(t) + for name, req := range map[string]*opificav1.ExecRequest{ + "empty": {Command: " "}, + "unknown shell": {Command: "x", Shell: "fish"}, + "absolute dir": {Command: "x", WorkingDirectory: "/etc"}, + "escaping dir": {Command: "x", WorkingDirectory: "../.."}, + "missing dir": {Command: "x", WorkingDirectory: "nope"}, + } { + if r := run(t, c, req); connect.CodeOf(r.err) != connect.CodeInvalidArgument { + t.Errorf("%s: err = %v, want invalid argument", name, r.err) + } + } +} + +// newRepo creates a git repository with one commit and returns its path and +// the commit id. +func newRepo(t *testing.T) (string, string) { + t.Helper() + dir := t.TempDir() + gitIn := func(args ...string) string { + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@example.com", + "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@example.com", "GIT_CONFIG_GLOBAL="+os.DevNull) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v: %s", args, err, out) + } + return strings.TrimSpace(string(out)) + } + gitIn("init", "-q", "-b", "main") + os.WriteFile(filepath.Join(dir, "README"), []byte("hello\n"), 0o644) + gitIn("add", ".") + gitIn("commit", "-q", "-m", "first") + first := gitIn("rev-parse", "HEAD") + os.WriteFile(filepath.Join(dir, "README"), []byte("changed\n"), 0o644) + gitIn("commit", "-q", "-am", "second") + return dir, first +} + +func TestCheckout(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("no git") + } + c, root := start(t) + repo, first := newRepo(t) + ctx := context.Background() + + // An older commit, not the branch head: the exact SHA is what's built. + resp, err := c.Checkout(ctx, connect.NewRequest(&opificav1.CheckoutRequest{MirrorUrl: repo, CommitSha: first, Directory: "src"})) + if err != nil { + t.Fatal(err) + } + if resp.Msg.GetHeadSha() != first { + t.Errorf("head = %s, want %s", resp.Msg.GetHeadSha(), first) + } + if data, _ := os.ReadFile(filepath.Join(root, "src", "README")); string(data) != "hello\n" { + t.Errorf("README = %q, want the first commit's", data) + } + + for name, req := range map[string]*opificav1.CheckoutRequest{ + "not empty": {MirrorUrl: repo, CommitSha: first, Directory: "src"}, + "short sha": {MirrorUrl: repo, CommitSha: first[:8], Directory: "b"}, + "escaping dir": {MirrorUrl: repo, CommitSha: first, Directory: "../x"}, + "no mirror": {CommitSha: first, Directory: "c"}, + } { + if _, err := c.Checkout(ctx, connect.NewRequest(req)); connect.CodeOf(err) != connect.CodeInvalidArgument { + t.Errorf("%s: err = %v, want invalid argument", name, err) + } + } + missing := strings.Repeat("ab", 20) + if _, err := c.Checkout(ctx, connect.NewRequest(&opificav1.CheckoutRequest{MirrorUrl: repo, CommitSha: missing, Directory: "d"})); connect.CodeOf(err) != connect.CodeFailedPrecondition { + t.Errorf("unknown commit: err = %v, want failed precondition", err) + } +} + +func TestCollect(t *testing.T) { + c, root := start(t) + os.MkdirAll(filepath.Join(root, "src", "build"), 0o755) + big := strings.Repeat("x", 200<<10) // several chunks + os.WriteFile(filepath.Join(root, "src", "build", "a.xml"), []byte(""), 0o644) + os.WriteFile(filepath.Join(root, "src", "build", "b.xml"), []byte(big), 0o644) + os.WriteFile(filepath.Join(root, "src", "build", "c.txt"), []byte("no"), 0o644) + secret := filepath.Join(t.TempDir(), "secret") + os.WriteFile(secret, []byte("outside"), 0o644) + os.Symlink(secret, filepath.Join(root, "src", "build", "link.xml")) + + stream, err := c.Collect(context.Background(), connect.NewRequest(&opificav1.CollectRequest{ + BaseDirectory: "src", Patterns: []string{"build/*.xml", "build/a.xml"}, + })) + if err != nil { + t.Fatal(err) + } + files := map[string]string{} + for stream.Receive() { + files[stream.Msg().GetPath()] += string(stream.Msg().GetData()) + } + if err := stream.Err(); err != nil { + t.Fatal(err) + } + if len(files) != 2 || files["build/a.xml"] != "" || files["build/b.xml"] != big { + t.Errorf("collected %d files: %v", len(files), keys(files)) + } + + bad, _ := c.Collect(context.Background(), connect.NewRequest(&opificav1.CollectRequest{Patterns: []string{"../../etc/passwd"}})) + for bad.Receive() { + } + if connect.CodeOf(bad.Err()) != connect.CodeInvalidArgument { + t.Errorf("escaping pattern: err = %v", bad.Err()) + } +} + +func keys(m map[string]string) []string { + var ks []string + for k := range m { + ks = append(ks, k) + } + return ks +} diff --git a/internal/agent/client.go b/internal/agent/client.go new file mode 100644 index 0000000..830ed04 --- /dev/null +++ b/internal/agent/client.go @@ -0,0 +1,31 @@ +package agent + +import ( + "context" + "net" + "net/http" + + "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" +) + +// NewClient returns an AgentService client that reaches the agent through +// dial, e.g. a VM's vsock connection. The channel is private to the host and +// the guest, so it is plain HTTP/2 without TLS. +func NewClient(dial func(ctx context.Context) (net.Conn, error)) opificav1connect.AgentServiceClient { + protocols := new(http.Protocols) + protocols.SetUnencryptedHTTP2(true) + transport := &http.Transport{ + Protocols: protocols, + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + return dial(ctx) + }, + } + return opificav1connect.NewAgentServiceClient(&http.Client{Transport: transport}, "http://agent") +} + +// Handler returns the agent's HTTP handler. +func Handler(s *Server) http.Handler { + mux := http.NewServeMux() + mux.Handle(opificav1connect.NewAgentServiceHandler(s)) + return mux +} diff --git a/internal/agent/proc_unix.go b/internal/agent/proc_unix.go new file mode 100644 index 0000000..514668a --- /dev/null +++ b/internal/agent/proc_unix.go @@ -0,0 +1,20 @@ +//go:build unix + +package agent + +import ( + "os/exec" + "syscall" +) + +// killTree runs the command in its own process group and makes cancellation +// kill the whole group, so background processes a step started die with it. +func killTree(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.Cancel = func() error { + if cmd.Process == nil { + return nil + } + return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + } +} diff --git a/internal/agent/proc_windows.go b/internal/agent/proc_windows.go new file mode 100644 index 0000000..6023f00 --- /dev/null +++ b/internal/agent/proc_windows.go @@ -0,0 +1,19 @@ +//go:build windows + +package agent + +import "os/exec" + +// killTree kills the command on cancellation. +// +// TODO: put the process in a job object so its children are killed too. The +// VM is destroyed after every job, so leftovers can't outlive the job, but +// a later step in the same job could still see them. +func killTree(cmd *exec.Cmd) { + cmd.Cancel = func() error { + if cmd.Process == nil { + return nil + } + return cmd.Process.Kill() + } +} diff --git a/internal/controller/logs/logs.go b/internal/controller/logs/logs.go new file mode 100644 index 0000000..80345bb --- /dev/null +++ b/internal/controller/logs/logs.go @@ -0,0 +1,101 @@ +// Package logs stores job output on the controller's disk, one file per job. +// +// Workers resend chunks after a reconnect; chunks are numbered per job, and +// anything not newer than the last stored chunk is dropped. The last number +// is kept in memory, so after a controller restart a resent chunk may appear +// twice. S3-compatible storage is planned (docs/ROADMAP.md, M7). +package logs + +import ( + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sync" + + "github.com/google/uuid" +) + +// Store appends and reads job logs. +type Store struct { + Dir string + + mu sync.Mutex + open map[uuid.UUID]*jobLog +} + +type jobLog struct { + mu sync.Mutex + f *os.File + lastSeq uint64 +} + +// ErrNoLog means a job has no output yet. +var ErrNoLog = errors.New("no log for this job yet") + +func (s *Store) path(job uuid.UUID) string { + return filepath.Join(s.Dir, job.String()+".log") +} + +func (s *Store) get(job uuid.UUID) (*jobLog, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.open == nil { + s.open = map[uuid.UUID]*jobLog{} + } + if l, ok := s.open[job]; ok { + return l, nil + } + if err := os.MkdirAll(s.Dir, 0o750); err != nil { + return nil, err + } + f, err := os.OpenFile(s.path(job), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o640) + if err != nil { + return nil, err + } + l := &jobLog{f: f} + s.open[job] = l + return l, nil +} + +// Append stores chunk seq of a job's log. It reports whether the chunk was +// new. +func (s *Store) Append(job uuid.UUID, seq uint64, data []byte) (bool, error) { + l, err := s.get(job) + if err != nil { + return false, err + } + l.mu.Lock() + defer l.mu.Unlock() + if seq <= l.lastSeq { + return false, nil + } + if _, err := l.f.Write(data); err != nil { + return false, fmt.Errorf("writing log: %w", err) + } + l.lastSeq = seq + return true, nil +} + +// Finish closes a job's log file once the job is done. +func (s *Store) Finish(job uuid.UUID) { + s.mu.Lock() + l, ok := s.open[job] + delete(s.open, job) + s.mu.Unlock() + if ok { + l.mu.Lock() + l.f.Close() + l.mu.Unlock() + } +} + +// Open returns a job's log. +func (s *Store) Open(job uuid.UUID) (io.ReadCloser, error) { + f, err := os.Open(s.path(job)) + if errors.Is(err, os.ErrNotExist) { + return nil, ErrNoLog + } + return f, err +} diff --git a/internal/controller/logs/logs_test.go b/internal/controller/logs/logs_test.go new file mode 100644 index 0000000..344dff5 --- /dev/null +++ b/internal/controller/logs/logs_test.go @@ -0,0 +1,38 @@ +package logs + +import ( + "errors" + "io" + "testing" + + "github.com/google/uuid" +) + +func TestAppendDropsDuplicates(t *testing.T) { + s := &Store{Dir: t.TempDir()} + job := uuid.New() + if _, err := s.Open(job); !errors.Is(err, ErrNoLog) { + t.Errorf("Open before any output: err = %v", err) + } + for _, c := range []struct { + seq uint64 + data string + want bool + }{ + {1, "a", true}, {2, "b", true}, {2, "b", false}, {1, "a", false}, {3, "c", true}, + } { + if got, err := s.Append(job, c.seq, []byte(c.data)); err != nil || got != c.want { + t.Errorf("Append(%d) = %v, %v; want %v", c.seq, got, err, c.want) + } + } + s.Finish(job) + r, err := s.Open(job) + if err != nil { + t.Fatal(err) + } + defer r.Close() + data, _ := io.ReadAll(r) + if string(data) != "abc" { + t.Errorf("log = %q, want abc", data) + } +} diff --git a/internal/controller/scheduler/scheduler.go b/internal/controller/scheduler/scheduler.go index d31a619..45e6a5f 100644 --- a/internal/controller/scheduler/scheduler.go +++ b/internal/controller/scheduler/scheduler.go @@ -20,6 +20,7 @@ import ( "fmt" "log/slog" mrand "math/rand/v2" + "sync" "time" "github.com/google/uuid" @@ -27,6 +28,7 @@ import ( "google.golang.org/protobuf/types/known/durationpb" opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/controller/logs" "github.com/opifica/opifica/internal/controller/store" "github.com/opifica/opifica/internal/controller/store/db" "github.com/opifica/opifica/internal/controller/workers" @@ -47,6 +49,8 @@ type Config struct { // Interval is how often the scheduler looks for work and expired leases // even when nothing woke it. Interval time.Duration + // Logs stores job output. Nil drops it. + Logs *logs.Store } // Scheduler implements workers.MessageHandler. @@ -56,6 +60,11 @@ type Scheduler struct { log *slog.Logger cfg Config kick chan struct{} + + // owners caches which worker holds a job, so log chunks don't each + // need a database query to check they come from the right worker. + ownersMu sync.Mutex + owners map[uuid.UUID]uuid.UUID } var _ workers.MessageHandler = (*Scheduler)(nil) @@ -71,7 +80,7 @@ func New(st *store.Store, sessions *workers.Registry, log *slog.Logger, cfg Conf if log == nil { log = slog.Default() } - return &Scheduler{store: st, sessions: sessions, log: log, cfg: cfg, kick: make(chan struct{}, 1)} + return &Scheduler{store: st, sessions: sessions, log: log, cfg: cfg, kick: make(chan struct{}, 1), owners: map[uuid.UUID]uuid.UUID{}} } // Kick makes the scheduler look for work now. @@ -397,6 +406,7 @@ func (s *Scheduler) reapLeases(ctx context.Context) { continue } s.log.Warn("lease expired", "job", job.ID, "worker", e.WorkerID, "attempt", job.Attempt, "now", job.State) + s.released(job.ID) if e.WorkerID != nil { // In case the worker is alive but lost track of the job. s.cancelOnWorker(ctx, *e.WorkerID, job.ID, "lease expired") @@ -454,6 +464,9 @@ func (s *Scheduler) WorkerDisconnected(worker uuid.UUID) { s.log.Error("releasing unstarted jobs", "worker", worker, "err", err) return } + for _, id := range ids { + s.released(id) + } if len(ids) > 0 { s.log.Info("worker disconnected before starting jobs; queued them again", "worker", worker, "jobs", len(ids)) s.Kick() @@ -477,11 +490,53 @@ func (s *Scheduler) HandleWorkerMessage(ctx context.Context, worker uuid.UUID, m }, }}) case *opificav1.ConnectRequest_LogChunk: - // Log storage comes next; drop for now. + s.logChunk(ctx, worker, m.LogChunk) } return nil } +func (s *Scheduler) logChunk(ctx context.Context, worker uuid.UUID, c *opificav1.LogChunk) { + if s.cfg.Logs == nil { + return + } + jobID, err := uuid.Parse(c.GetJobId()) + if err != nil || !s.holds(ctx, worker, jobID) { + return + } + if _, err := s.cfg.Logs.Append(jobID, c.GetSequence(), c.GetData()); err != nil { + s.log.Error("storing log", "job", jobID, "err", err) + } +} + +// holds reports whether worker currently holds the job. +func (s *Scheduler) holds(ctx context.Context, worker, job uuid.UUID) bool { + s.ownersMu.Lock() + owner, ok := s.owners[job] + s.ownersMu.Unlock() + if ok { + return owner == worker + } + j, err := s.store.GetJob(ctx, job) + if err != nil || j.WorkerID == nil || *j.WorkerID != worker || (j.State != "assigned" && j.State != "running") { + return false + } + s.ownersMu.Lock() + s.owners[job] = worker + s.ownersMu.Unlock() + return true +} + +// released forgets a job's owner and closes its log once the worker no +// longer holds it. +func (s *Scheduler) released(job uuid.UUID) { + s.ownersMu.Lock() + delete(s.owners, job) + s.ownersMu.Unlock() + if s.cfg.Logs != nil { + s.cfg.Logs.Finish(job) + } +} + func (s *Scheduler) renew(ctx context.Context, worker uuid.UUID, ids []string) { if len(ids) == 0 { return @@ -526,6 +581,7 @@ func (s *Scheduler) jobUpdate(ctx context.Context, worker uuid.UUID, u *opificav return } log.Info("job finished", "key", job.Key, "state", state) + s.released(jobID) s.advance(ctx, job.RunID) case opificav1.JobState_JOB_STATE_ERRORED: @@ -535,6 +591,7 @@ func (s *Scheduler) jobUpdate(ctx context.Context, worker uuid.UUID, u *opificav return } log.Warn("job errored on the worker", "err", u.GetError(), "attempt", job.Attempt, "now", job.State) + s.released(jobID) s.settled(ctx, job) default: diff --git a/internal/integration/connect_test.go b/internal/integration/connect_test.go index a7d31e2..fde168a 100644 --- a/internal/integration/connect_test.go +++ b/internal/integration/connect_test.go @@ -21,6 +21,7 @@ import ( "github.com/opifica/opifica/gen/opifica/v1/opificav1connect" "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 +43,7 @@ type env struct { t *testing.T svc *workers.Service sched *scheduler.Scheduler + logs *logs.Store url string } @@ -65,8 +67,9 @@ func newEnv(t *testing.T) *env { TouchInterval: 50 * time.Millisecond, }, } + logStore := &logs.Store{Dir: t.TempDir()} sched := scheduler.New(st, svc.Sessions(), quiet(), scheduler.Config{ - Lease: 400 * time.Millisecond, Interval: 20 * time.Millisecond, + Lease: 400 * time.Millisecond, Interval: 20 * time.Millisecond, Logs: logStore, }) svc.Messages = sched go sched.Run(ctx) @@ -88,7 +91,7 @@ func newEnv(t *testing.T) *env { <-done st.Close() }) - return &env{t: t, svc: svc, sched: sched, url: "https://" + ln.Addr().String()} + return &env{t: t, svc: svc, sched: sched, logs: logStore, url: "https://" + ln.Addr().String()} } func info() *opificav1.WorkerInfo { @@ -107,6 +110,11 @@ func (e *env) join(pool string) *worker.Identity { } func (e *env) joinWith(pool string, wi *opificav1.WorkerInfo) *worker.Identity { + e.t.Helper() + return e.joinInto(pool, wi, e.t.TempDir()) +} + +func (e *env) joinInto(pool string, wi *opificav1.WorkerInfo, stateDir string) *worker.Identity { e.t.Helper() tok, err := e.svc.CreateToken(context.Background(), workers.TokenRequest{Pool: pool}) if err != nil { @@ -114,7 +122,7 @@ func (e *env) joinWith(pool string, wi *opificav1.WorkerInfo) *worker.Identity { } id, err := worker.Join(context.Background(), worker.JoinConfig{ ControllerURL: e.url, Token: tok.Secret, CAFingerprint: tok.CAFingerprint, - StateDir: e.t.TempDir(), Info: wi, + StateDir: stateDir, Info: wi, }) if err != nil { e.t.Fatalf("Join: %v", err) diff --git a/internal/integration/e2e_test.go b/internal/integration/e2e_test.go new file mode 100644 index 0000000..fe35276 --- /dev/null +++ b/internal/integration/e2e_test.go @@ -0,0 +1,335 @@ +package integration + +import ( + "context" + "io" + "os" + "os/exec" + "path/filepath" + "regexp" + "runtime" + "strings" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/opifica/opifica/internal/controller/scheduler" + "github.com/opifica/opifica/internal/pipeline" + "github.com/opifica/opifica/pkg/worker" + "github.com/opifica/opifica/pkg/worker/workertest" +) + +// These tests run the whole path for real: controller, scheduler, the worker +// runtime, a Git mirror, and a real agent per job running real commands. Only +// the VM is faked: each "VM" is a fresh directory with its own agent. + +func hostOS() string { + if runtime.GOOS == "darwin" { + return "macos" + } + return runtime.GOOS +} + +type realWorker struct { + hv *workertest.Hypervisor + cancel context.CancelFunc + done chan error +} + +func (e *env) realWorker(capacity int) *realWorker { + e.t.Helper() + if runtime.GOOS == "windows" { + e.t.Skip("the test pipelines use a POSIX shell") + } + if _, err := exec.LookPath("git"); err != nil { + e.t.Skip("git is not installed") + } + pool := worker.PoolConfig{ + Capacity: capacity, + Images: map[string]worker.Image{"default": {Name: "default"}}, + Warm: map[string]int{"default": 1}, + } + dir := e.t.TempDir() + e.joinInto("trusted", worker.Info(pool, nil), dir) + + rw := &realWorker{hv: &workertest.Hypervisor{Agent: true}, done: make(chan error, 1)} + ctx, cancel := context.WithCancel(context.Background()) + rw.cancel = cancel + go func() { + rw.done <- worker.Run(ctx, worker.Config{ + ControllerURL: e.url, StateDir: dir, Hypervisor: rw.hv, Pool: pool, + AllowLocalRepos: true, HeartbeatInterval: 50 * time.Millisecond, Logger: quiet(), + }) + }() + e.t.Cleanup(func() { + cancel() + <-rw.done + if n := rw.hv.Alive(); n != 0 { + e.t.Errorf("%d VMs still alive after the worker stopped", n) + } + }) + return rw +} + +// gitRepo creates a repository with the given files and returns its path +// and commit. +func gitRepo(t *testing.T, files map[string]string) (string, string) { + t.Helper() + dir := t.TempDir() + for name, content := range files { + p := filepath.Join(dir, name) + os.MkdirAll(filepath.Dir(p), 0o755) + if err := os.WriteFile(p, []byte(content), 0o644); err != nil { + t.Fatal(err) + } + } + gitIn := func(args ...string) string { + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@example.com", + "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@example.com", "GIT_CONFIG_GLOBAL="+os.DevNull) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v: %s", args, err, out) + } + return strings.TrimSpace(string(out)) + } + gitIn("init", "-q", "-b", "main") + gitIn("add", ".") + gitIn("commit", "-q", "-m", "test") + return dir, gitIn("rev-parse", "HEAD") +} + +func (e *env) startRun(repo, sha, src string) uuid.UUID { + e.t.Helper() + run, err := e.sched.CreateRun(context.Background(), runRequest(repo, sha, src)) + if err != nil { + e.t.Fatalf("CreateRun: %v", err) + } + return run.ID +} + +func (e *env) jobLog(runID uuid.UUID, key string) string { + e.t.Helper() + jobs, err := e.svc.Store.ListRunJobs(context.Background(), runID) + if err != nil { + e.t.Fatal(err) + } + for _, j := range jobs { + if j.Key == key { + r, err := e.logs.Open(j.ID) + if err != nil { + return "" + } + defer r.Close() + data, _ := io.ReadAll(r) + return string(data) + } + } + e.t.Fatalf("no job %q", key) + return "" +} + +func wantIn(t *testing.T, log string, parts ...string) { + t.Helper() + for _, p := range parts { + if !strings.Contains(log, p) { + t.Errorf("log lacks %q:\n%s", p, log) + } + } +} + +func TestE2ERealCommands(t *testing.T) { + e := newEnv(t) + e.realWorker(2) + repo, sha := gitRepo(t, map[string]string{"README": "hello world\n", "scripts/check.sh": "echo checked in $(pwd)\n"}) + + runID := e.startRun(repo, sha, ` +on: push +env: + GREETING: hi +jobs: + build: + runs-on: `+hostOS()+` + steps: + - run: | + echo "$GREETING, the readme says: $(cat README)" + test "$OPIFICA_COMMIT_SHA" = "`+sha+`" + test "$CI" = true + - run: echo to-stderr >&2 + - run: sh check.sh + working-directory: scripts + check: + needs: build + runs-on: `+hostOS()+` + steps: + - name: allowed to fail + run: exit 7 + continue-on-error: true + - run: echo still running + - report: junit build/*.xml +`) + run, jobs := e.waitRun(runID) + if run.State != "succeeded" { + t.Fatalf("run %s; build %s (%q), check %s (%q)\n%s", run.State, jobs["build"].State, jobs["build"].Error, + jobs["check"].State, jobs["check"].Error, e.jobLog(runID, "build")) + } + wantIn(t, e.jobLog(runID, "build"), + "==> Fetching", "==> Starting a fresh VM", "==> Step 1/3", + "hi, the readme says: hello world", "to-stderr", "checked in ", "/src/scripts") + wantIn(t, e.jobLog(runID, "check"), + "Step failed with exit code 7; continuing", "still running", "Test reports are not stored yet") +} + +func TestE2EFailure(t *testing.T) { + e := newEnv(t) + e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, sha, ` +on: push +jobs: + test: + runs-on: `+hostOS()+` + steps: + - run: echo before; exit 7 + - run: echo never + deploy: + needs: test + runs-on: `+hostOS()+` + steps: [{ run: echo deploying }] +`) + run, jobs := e.waitRun(runID) + if run.State != "failed" || jobs["test"].State != "failed" || *jobs["test"].ExitCode != 7 || jobs["deploy"].State != "skipped" { + t.Fatalf("run %s; test %s exit %v; deploy %s", run.State, jobs["test"].State, jobs["test"].ExitCode, jobs["deploy"].State) + } + if jobs["test"].Attempt != 1 { + t.Errorf("a failing command was retried (attempt %d); only infrastructure errors should be", jobs["test"].Attempt) + } + log := e.jobLog(runID, "test") + wantIn(t, log, "before", "Step failed with exit code 7") + if strings.Contains(log, "never") { + t.Error("ran the step after the failing one") + } +} + +// Every job gets a fresh VM: nothing one job writes is visible to the next. +func TestE2EFreshVMPerJob(t *testing.T) { + e := newEnv(t) + rw := e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, sha, ` +on: push +jobs: + first: + runs-on: `+hostOS()+` + steps: [{ run: "test ! -e ../left-behind && touch ../left-behind README.modified" }] + second: + needs: first + runs-on: `+hostOS()+` + steps: [{ run: "test ! -e ../left-behind && test ! -e README.modified" }] +`) + run, jobs := e.waitRun(runID) + if run.State != "succeeded" { + t.Fatalf("run %s: first %s, second %s\n%s", run.State, jobs["first"].State, jobs["second"].State, e.jobLog(runID, "second")) + } + if created, _ := rw.hv.Counts(); created < 2 { + t.Errorf("only %d VMs created for 2 jobs", created) + } +} + +func TestE2ESecretDenied(t *testing.T) { + e := newEnv(t) + e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, sha, ` +on: push +jobs: + deploy: + runs-on: `+hostOS()+` + secrets: [DEPLOY_TOKEN] + steps: + - run: echo deploying + secrets: [DEPLOY_TOKEN] +`) + run, jobs := e.waitRun(runID) + j := jobs["deploy"] + if run.State != "failed" || !strings.Contains(j.Error, "secret DEPLOY_TOKEN was denied") || j.Attempt != 1 { + t.Errorf("run %s; job %s, attempt %d, error %q", run.State, j.State, j.Attempt, j.Error) + } + // "deploying" alone on a line is the command's output; the step's header + // line ("==> Step 1/1: echo deploying") doesn't count. + if regexp.MustCompile(`(?m)^deploying$`).MatchString(e.jobLog(runID, "deploy")) { + t.Error("the step ran without its secret") + } +} + +func TestE2ECancelStopsTheCommand(t *testing.T) { + e := newEnv(t) + rw := e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, sha, "on: push\njobs:\n long:\n runs-on: "+hostOS()+"\n steps: [{ run: echo started; sleep 300 }]\n") + eventually(t, "the command to start", func() bool { return strings.Contains(e.jobLog(runID, "long"), "started") }) + + start := time.Now() + if err := e.sched.CancelRun(context.Background(), runID); err != nil { + t.Fatal(err) + } + run, jobs := e.waitRun(runID) + if run.State != "cancelled" || jobs["long"].State != "cancelled" { + t.Errorf("run %s, job %s", run.State, jobs["long"].State) + } + if d := time.Since(start); d > 5*time.Second { + t.Errorf("cancelling took %v", d) + } + eventually(t, "the job's VM to be destroyed", func() bool { return rw.hv.Alive() <= 1 }) // the warm one may remain +} + +func TestE2EJobTimeout(t *testing.T) { + e := newEnv(t) + e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, sha, "on: push\njobs:\n slow:\n runs-on: "+hostOS()+"\n timeout: 2s\n steps: [{ run: sleep 60 }]\n") + run, jobs := e.waitRun(runID) + j := jobs["slow"] + if run.State != "failed" || j.State != "failed" || !strings.Contains(j.Error+e.jobLog(runID, "slow"), "timed out") { + t.Errorf("run %s; job %s, error %q\n%s", run.State, j.State, j.Error, e.jobLog(runID, "slow")) + } +} + +func TestE2EMissingCommitFailsWithoutRetry(t *testing.T) { + e := newEnv(t) + e.realWorker(1) + repo, _ := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, strings.Repeat("ab", 20), "on: push\njobs:\n a:\n runs-on: "+hostOS()+"\n steps: [{ run: echo hi }]\n") + run, jobs := e.waitRun(runID) + if run.State != "failed" || jobs["a"].Attempt != 1 || !strings.Contains(jobs["a"].Error, "commit not found") { + t.Errorf("run %s; attempt %d; error %q", run.State, jobs["a"].Attempt, jobs["a"].Error) + } +} + +func runRequest(repo, sha, src string) scheduler.RunRequest { + return scheduler.RunRequest{ + Repo: "acme/app", CloneURL: repo, Pool: "trusted", Pipeline: []byte(src), + Event: pipeline.Event{Type: pipeline.EventPush, Branch: "main", Ref: "refs/heads/main", SHA: sha}, + } +} + +// The connection drops while a job runs. The job keeps running, the worker +// reconnects within the lease, and the output and result arrive: no retry. +func TestE2EJobSurvivesReconnect(t *testing.T) { + e := newEnv(t) + e.realWorker(1) + repo, sha := gitRepo(t, map[string]string{"README": "x"}) + runID := e.startRun(repo, sha, "on: push\njobs:\n a:\n runs-on: "+hostOS()+"\n steps: [{ run: echo before; sleep 1; echo after }]\n") + eventually(t, "the command to start", func() bool { return strings.Contains(e.jobLog(runID, "a"), "before") }) + + for _, s := range e.svc.Sessions().List() { + e.svc.Sessions().Disconnect(s.WorkerID, context.Canceled) + } + run, jobs := e.waitRun(runID) + if run.State != "succeeded" || jobs["a"].Attempt != 1 { + t.Fatalf("run %s, attempt %d (%q)", run.State, jobs["a"].Attempt, jobs["a"].Error) + } + wantIn(t, e.jobLog(runID, "a"), "before", "after") +} diff --git a/internal/mask/mask.go b/internal/mask/mask.go new file mode 100644 index 0000000..1b13c6c --- /dev/null +++ b/internal/mask/mask.go @@ -0,0 +1,133 @@ +// Package mask replaces secret values in streamed output. +// +// Output arrives in arbitrary chunks, so a secret can be split across two of +// them. Writer holds back the tail of each chunk that could be the start of +// a secret until the next chunk (or Flush) shows whether it is. +package mask + +import ( + "bytes" + "io" + "sort" + "sync" +) + +// Replacement is written in place of every secret value. +const Replacement = "***" + +// minLen is the shortest value that is masked. Shorter values ("1", "true") +// would mangle ordinary output while protecting nothing. +const minLen = 4 + +// Writer masks secrets in everything written to it before passing it on. +type Writer struct { + mu sync.Mutex + out io.Writer + secrets [][]byte // longest first, so overlapping secrets mask fully + maxLen int + pending []byte +} + +// NewWriter returns a Writer that masks secrets in what it writes to out. +// Values shorter than 4 bytes are not masked. +func NewWriter(out io.Writer, secrets []string) *Writer { + w := &Writer{out: out} + seen := map[string]bool{} + for _, s := range secrets { + if len(s) < minLen || seen[s] { + continue + } + seen[s] = true + w.secrets = append(w.secrets, []byte(s)) + w.maxLen = max(w.maxLen, len(s)) + } + sort.Slice(w.secrets, func(i, j int) bool { return len(w.secrets[i]) > len(w.secrets[j]) }) + return w +} + +// Write masks and forwards p, except for a tail that might be the start of a +// secret, which waits for the next Write or Flush. +func (w *Writer) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + if len(w.secrets) == 0 { + return len(p), w.write(p) + } + buf := append(w.pending, p...) + masked, keep := w.mask(buf, false) + w.pending = append(w.pending[:0:0], buf[len(buf)-keep:]...) + return len(p), w.write(masked) +} + +// Flush writes whatever is being held back. Call it when the stream ends. +func (w *Writer) Flush() error { + w.mu.Lock() + defer w.mu.Unlock() + if len(w.pending) == 0 { + return nil + } + masked, _ := w.mask(w.pending, true) + w.pending = nil + return w.write(masked) +} + +func (w *Writer) write(p []byte) error { + if len(p) == 0 { + return nil + } + _, err := w.out.Write(p) + return err +} + +// mask returns buf with secrets replaced, and how many bytes at the end of +// buf were held back (always 0 when final). +func (w *Writer) mask(buf []byte, final bool) ([]byte, int) { + var out bytes.Buffer + i := 0 + for i < len(buf) { + if n := w.matchAt(buf[i:]); n > 0 { + out.WriteString(Replacement) + i += n + continue + } + if !final && w.couldStartSecret(buf[i:]) { + return out.Bytes(), len(buf) - i + } + out.WriteByte(buf[i]) + i++ + } + return out.Bytes(), 0 +} + +// matchAt returns the length of the secret that b starts with, or 0. +func (w *Writer) matchAt(b []byte) int { + for _, s := range w.secrets { + if bytes.HasPrefix(b, s) { + return len(s) + } + } + return 0 +} + +// couldStartSecret reports whether b (the rest of the buffer) is a proper +// prefix of some secret, i.e. more input could complete it. +func (w *Writer) couldStartSecret(b []byte) bool { + if len(b) >= w.maxLen { + return false + } + for _, s := range w.secrets { + if len(b) < len(s) && bytes.HasPrefix(s, b) { + return true + } + } + return false +} + +// String masks secrets in a complete string. +func String(s string, secrets []string) string { + var b bytes.Buffer + w := NewWriter(&b, secrets) + w.Write([]byte(s)) + w.Flush() + return b.String() +} diff --git a/internal/mask/mask_test.go b/internal/mask/mask_test.go new file mode 100644 index 0000000..16aafa2 --- /dev/null +++ b/internal/mask/mask_test.go @@ -0,0 +1,69 @@ +package mask + +import ( + "bytes" + "strings" + "testing" +) + +func TestString(t *testing.T) { + secrets := []string{"hunter2!", "s3cr3t-token", "abc"} + tests := map[string]string{ + "password is hunter2!": "password is ***", + "hunter2!hunter2!": "******", + "token=s3cr3t-token;": "token=***;", + "abc is too short to mask": "abc is too short to mask", + "nothing here": "nothing here", + "almost hunter2 but not": "almost hunter2 but not", + "ends with a prefix: s3cr3t-tok": "ends with a prefix: s3cr3t-tok", + } + for in, want := range tests { + if got := String(in, secrets); got != want { + t.Errorf("String(%q) = %q, want %q", in, got, want) + } + } +} + +// Every possible split of the input into two writes must give the same result. +func TestSecretSplitAcrossWrites(t *testing.T) { + secrets := []string{"correct-horse-battery", "horse-battery-staple"} + in := "a correct-horse-battery-staple b horse-battery-staple c correct-hors" + want := String(in, secrets) + if strings.Contains(want, "battery") && !strings.HasSuffix(want, "correct-hors") { + t.Fatalf("baseline masking failed: %q", want) + } + for i := 0; i <= len(in); i++ { + for j := i; j <= len(in); j++ { + var b bytes.Buffer + w := NewWriter(&b, secrets) + w.Write([]byte(in[:i])) + w.Write([]byte(in[i:j])) + w.Write([]byte(in[j:])) + w.Flush() + if b.String() != want { + t.Fatalf("split at %d,%d: got %q, want %q", i, j, b.String(), want) + } + } + } +} + +func TestByteAtATime(t *testing.T) { + var b bytes.Buffer + w := NewWriter(&b, []string{"TOPSECRET"}) + for _, c := range []byte("x TOPSECRET y TOPSECRE") { + w.Write([]byte{c}) + } + w.Flush() + if got := b.String(); got != "x *** y TOPSECRE" { + t.Errorf("got %q", got) + } +} + +func TestNoSecretsPassesThrough(t *testing.T) { + var b bytes.Buffer + w := NewWriter(&b, nil) + w.Write([]byte("hello")) + if b.String() != "hello" { + t.Errorf("got %q without Flush; nothing should be held back", b.String()) + } +} diff --git a/internal/pipeline/parse.go b/internal/pipeline/parse.go index f8226bf..6548a0f 100644 --- a/internal/pipeline/parse.go +++ b/internal/pipeline/parse.go @@ -45,6 +45,7 @@ func Parse(data []byte) (*Pipeline, error) { if err := yaml.Unmarshal(data, &root); err != nil { errs := convertYAMLError(err) addFlowHint(errs, data) + addColonHint(errs) return nil, errs } @@ -120,6 +121,16 @@ func addFlowHint(errs Errors, data []byte) { } } +// addColonHint explains "mapping values are not allowed": usually a command +// or value containing ": ", which YAML reads as a nested key. +func addColonHint(errs Errors) { + for _, e := range errs { + if strings.Contains(e.Msg, "mapping values are not allowed") && !strings.Contains(e.Msg, "hint:") { + e.Msg += ` (hint: a value containing ": " must be quoted, or written as a block, e.g. run: | on its own line with the command indented below)` + } + } +} + // addColumns fills in columns yaml.v3 leaves out of decode errors, using the // key on the reported line. func addColumns(errs Errors, root *yaml.Node) { diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index 1366fc5..fce6b75 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -204,6 +204,8 @@ func TestErrors(t *testing.T) { []string{"this job has no matrix"}}, {"unknown event field", "on: push\njobs:\n a:\n runs-on: linux\n steps:\n - run: echo ${{ event.author }}\n", []string{`unknown event field "author"`}}, + {"colon in a command", "on: push\njobs:\n a:\n runs-on: linux\n steps:\n - run: git log --format='%h: %s'\n", + []string{`hint: a value containing ": " must be quoted`}}, {"expression in flow mapping", "on: push\njobs:\n a:\n matrix: { os: [linux] }\n runs-on: { os: ${{ matrix.os }} }\n steps: [{run: x}]\n", []string{"hint: ${{ }} can't be used inside inline"}}, {"quoted expression in flow mapping is fine", "on: push\njobs:\n a:\n matrix: { os: [linux] }\n runs-on: { os: \"${{ matrix.os }}\", arch: sparc }\n steps: [{run: x}]\n", diff --git a/pkg/worker/client.go b/pkg/worker/client.go index bcd40fd..c492928 100644 --- a/pkg/worker/client.go +++ b/pkg/worker/client.go @@ -273,9 +273,11 @@ func (c *Client) Run(ctx context.Context) error { } } -// backoff is exponential from 500ms with ±25% jitter, capped at max. +// backoff is exponential from 100ms with ±25% jitter, capped at max. The +// first retry is quick because most drops are momentary, and a worker that +// is back within its lease keeps its running jobs. func backoff(attempt int, max time.Duration) time.Duration { - d := time.Duration(float64(500*time.Millisecond) * math.Pow(2, float64(min(attempt, 20)))) + d := time.Duration(float64(100*time.Millisecond) * math.Pow(2, float64(min(attempt, 20)))) d = min(d, max) return time.Duration(float64(d) * (0.75 + 0.5*mrand.Float64())) } diff --git a/pkg/worker/mirror.go b/pkg/worker/mirror.go new file mode 100644 index 0000000..8e2c2c4 --- /dev/null +++ b/pkg/worker/mirror.go @@ -0,0 +1,160 @@ +package worker + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "sync" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" +) + +// Mirrors keeps a bare mirror of each repository on the worker's disk, so +// jobs check out from a local copy: fast, light on the forge, and the VM +// never needs access to the forge itself. +type Mirrors struct { + // Dir holds the mirrors. + Dir string + // AllowLocal permits local paths and file:// URLs as clone URLs. Only for + // development: a controller could otherwise make the worker read any + // repository on its disk. + AllowLocal bool + + mu sync.Mutex + locks map[string]*sync.Mutex +} + +// ErrCommitNotFound means the repository doesn't contain the commit. +var ErrCommitNotFound = errors.New("commit not found in the repository") + +var commitRe = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`) + +func (m *Mirrors) lock(key string) *sync.Mutex { + m.mu.Lock() + defer m.mu.Unlock() + if m.locks == nil { + m.locks = map[string]*sync.Mutex{} + } + l, ok := m.locks[key] + if !ok { + l = &sync.Mutex{} + m.locks[key] = l + } + return l +} + +func (m *Mirrors) checkURL(u string) error { + switch { + case strings.HasPrefix(u, "https://"), strings.HasPrefix(u, "ssh://"): + return nil + case regexp.MustCompile(`^[\w.-]+@[\w.-]+:`).MatchString(u): // git@host:org/repo + return nil + case strings.HasPrefix(u, "http://"): + return fmt.Errorf("clone URL %q: plain http is not allowed", u) + case m.AllowLocal: + return nil + } + return fmt.Errorf("clone URL %q: local repositories are not allowed on this worker", u) +} + +// Ensure makes sure the mirror of cloneURL contains sha and returns the +// mirror's path. A commit already in the mirror is used without fetching: +// commits never change. +func (m *Mirrors) Ensure(ctx context.Context, cloneURL string, cred *opificav1.Credential, sha string) (string, error) { + if !commitRe.MatchString(sha) { + return "", fmt.Errorf("invalid commit id %q", sha) + } + if err := m.checkURL(cloneURL); err != nil { + return "", err + } + env, cleanup, err := credentialEnv(cred) + if err != nil { + return "", err + } + defer cleanup() + + sum := sha256.Sum256([]byte(cloneURL)) + key := hex.EncodeToString(sum[:12]) + path := filepath.Join(m.Dir, key+".git") + l := m.lock(key) + l.Lock() + defer l.Unlock() + + if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) { + if err := os.MkdirAll(m.Dir, 0o700); err != nil { + return "", err + } + tmp, err := os.MkdirTemp(m.Dir, key+".tmp-") + if err != nil { + return "", err + } + defer os.RemoveAll(tmp) + if _, err := runGit(ctx, env, "clone", "--quiet", "--mirror", "--", cloneURL, tmp); err != nil { + return "", err + } + if err := os.Rename(tmp, path); err != nil { + return "", err + } + } + + if hasCommit(ctx, path, sha) { + return path, nil + } + if _, err := runGit(ctx, env, "-C", path, "fetch", "--quiet", "--prune", "origin"); err != nil { + return "", err + } + if hasCommit(ctx, path, sha) { + return path, nil + } + // Not on any branch or tag (e.g. a pull request ref): ask for it directly. + if _, err := runGit(ctx, env, "-C", path, "fetch", "--quiet", "origin", sha); err == nil && hasCommit(ctx, path, sha) { + return path, nil + } + return "", fmt.Errorf("%w: %s", ErrCommitNotFound, sha) +} + +func hasCommit(ctx context.Context, repo, sha string) bool { + _, err := runGit(ctx, nil, "-C", repo, "cat-file", "-e", sha+"^{commit}") + return err == nil +} + +// credentialEnv passes a token to git through the environment rather than +// the command line, where other users of the host could see it. +func credentialEnv(cred *opificav1.Credential) ([]string, func(), error) { + noop := func() {} + switch k := cred.GetKind().(type) { + case nil: + return nil, noop, nil + case *opificav1.Credential_Token: + basic := base64.StdEncoding.EncodeToString([]byte("x-access-token:" + k.Token)) + return []string{ + "GIT_CONFIG_COUNT=1", + "GIT_CONFIG_KEY_0=http.extraHeader", + "GIT_CONFIG_VALUE_0=Authorization: Basic " + basic, + }, noop, nil + case *opificav1.Credential_SshPrivateKey: + return nil, noop, errors.New("SSH key credentials are not supported yet") + } + return nil, noop, fmt.Errorf("unknown credential type %T", cred.GetKind()) +} + +func runGit(ctx context.Context, env []string, args ...string) (string, error) { + cmd := exec.CommandContext(ctx, "git", args...) + cmd.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL="+os.DevNull) + cmd.Env = append(cmd.Env, env...) + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + if err := cmd.Run(); err != nil { + return "", fmt.Errorf("git %s: %v: %s", args[0], err, strings.TrimSpace(stderr.String())) + } + return stdout.String(), nil +} diff --git a/pkg/worker/runner.go b/pkg/worker/runner.go new file mode 100644 index 0000000..0e3dc9b --- /dev/null +++ b/pkg/worker/runner.go @@ -0,0 +1,431 @@ +package worker + +import ( + "context" + "errors" + "fmt" + "log/slog" + "maps" + "math" + "net" + "slices" + "strings" + "sync" + "sync/atomic" + "time" + + "connectrpc.com/connect" + "github.com/google/uuid" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/agent" +) + +// checkoutDir is where the agent checks out the repository, relative to its +// work directory. Step working directories are relative to it. +const checkoutDir = "src" + +// RunnerConfig configures a Runner. +type RunnerConfig struct { + Pool *Pool + Mirrors *Mirrors + // Send delivers a message to the controller. It should keep trying + // across reconnects; see NewOutbox. + Send func(msg *opificav1.ConnectRequest, onSent func()) + // GuestMirrorURL turns the mirror's path on the host into the URL the + // agent in the VM clones from. Hypervisor drivers expose the mirror to + // the guest (e.g. over vsock); the default, the host path itself, only + // works when the agent runs on the host (development). + GuestMirrorURL func(vm VM, hostPath string) (string, error) + // SecretTimeout bounds waiting for the controller to answer a secret + // request. Defaults to 30s. + SecretTimeout time.Duration + Logger *slog.Logger +} + +// Runner runs the jobs the controller assigns, each in a fresh VM. +type Runner struct { + cfg RunnerConfig + log *slog.Logger + + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + + mu sync.Mutex + jobs map[string]*runningJob + secrets map[string]chan *opificav1.SecretResponse + draining bool +} + +type runningJob struct { + spec *opificav1.Job + cancel context.CancelCauseFunc + seq atomic.Uint64 +} + +var errCancelledByController = errors.New("cancelled by the controller") + +// NewRunner creates a runner. Close it to stop every job. +func NewRunner(cfg RunnerConfig) *Runner { + if cfg.GuestMirrorURL == nil { + cfg.GuestMirrorURL = func(_ VM, p string) (string, error) { return p, nil } + } + if cfg.SecretTimeout <= 0 { + cfg.SecretTimeout = 30 * time.Second + } + log := cfg.Logger + if log == nil { + log = slog.Default() + } + ctx, cancel := context.WithCancel(context.Background()) + return &Runner{ + cfg: cfg, log: log, ctx: ctx, cancel: cancel, + jobs: map[string]*runningJob{}, secrets: map[string]chan *opificav1.SecretResponse{}, + } +} + +// Running returns the IDs of jobs this worker holds, for heartbeats. A job +// stays here until its final result has been delivered, so its lease is +// renewed until then. +func (r *Runner) Running() []string { + r.mu.Lock() + defer r.mu.Unlock() + return slices.Sorted(maps.Keys(r.jobs)) +} + +// Close cancels every job and waits for them to finish. +func (r *Runner) Close() { + r.cancel() + r.wg.Wait() +} + +// HandleMessage handles a message from the controller. +func (r *Runner) HandleMessage(_ context.Context, msg *opificav1.ConnectResponse) { + switch m := msg.GetMessage().(type) { + case *opificav1.ConnectResponse_AssignJob: + r.start(m.AssignJob.GetJob()) + case *opificav1.ConnectResponse_CancelJob: + r.mu.Lock() + j := r.jobs[m.CancelJob.GetJobId()] + r.mu.Unlock() + if j != nil { + r.log.Info("job cancelled by the controller", "job", m.CancelJob.GetJobId(), "reason", m.CancelJob.GetReason()) + j.cancel(errCancelledByController) + } + case *opificav1.ConnectResponse_SecretResponse: + r.mu.Lock() + ch := r.secrets[m.SecretResponse.GetRequestId()] + delete(r.secrets, m.SecretResponse.GetRequestId()) + r.mu.Unlock() + if ch != nil { + ch <- m.SecretResponse + } + case *opificav1.ConnectResponse_Drain: + r.mu.Lock() + r.draining = true + r.mu.Unlock() + r.log.Info("draining: finishing running jobs, accepting no new ones", "reason", m.Drain.GetReason()) + } +} + +func (r *Runner) start(spec *opificav1.Job) { + id := spec.GetId() + r.mu.Lock() + if _, dup := r.jobs[id]; dup || r.draining || r.ctx.Err() != nil { + r.mu.Unlock() + if !dup { + // Not running it: let the controller give it to someone else. + r.cfg.Send(update(id, opificav1.JobState_JOB_STATE_ERRORED, nil, "worker is shutting down or draining"), nil) + } + return + } + ctx, cancel := context.WithCancelCause(r.ctx) + j := &runningJob{spec: spec, cancel: cancel} + r.jobs[id] = j + r.mu.Unlock() + + r.wg.Add(1) + go func() { + defer r.wg.Done() + defer cancel(nil) + state, exit, msg := r.run(ctx, j) + r.log.Info("job finished", "job", id, "state", state, "exit", exit, "error", msg) + var exitPtr *int32 + if state == opificav1.JobState_JOB_STATE_SUCCEEDED || state == opificav1.JobState_JOB_STATE_FAILED { + exitPtr = &exit + } + // Forget the job only once the controller has its result. + r.cfg.Send(update(id, state, exitPtr, msg), func() { + r.mu.Lock() + delete(r.jobs, id) + r.mu.Unlock() + }) + }() +} + +func update(job string, state opificav1.JobState, exit *int32, msg string) *opificav1.ConnectRequest { + return &opificav1.ConnectRequest{Message: &opificav1.ConnectRequest_JobUpdate{JobUpdate: &opificav1.JobUpdate{ + JobId: job, State: state, ExitCode: exit, Error: msg, + }}} +} + +// syslog writes an Opifica line into the job's log. +func (r *Runner) syslog(j *runningJob, step uint32, format string, args ...any) { + r.logChunk(j, step, opificav1.LogStream_LOG_STREAM_SYSTEM, []byte("==> "+fmt.Sprintf(format, args...)+"\n")) +} + +func (r *Runner) logChunk(j *runningJob, step uint32, stream opificav1.LogStream, data []byte) { + r.cfg.Send(&opificav1.ConnectRequest{Message: &opificav1.ConnectRequest_LogChunk{LogChunk: &opificav1.LogChunk{ + JobId: j.spec.GetId(), StepIndex: step, Sequence: j.seq.Add(1), Stream: stream, Data: data, + }}}, nil) +} + +// jobError is a failure with the state it should be reported as. +type jobError struct { + state opificav1.JobState + msg string +} + +func failed(format string, args ...any) *jobError { + return &jobError{opificav1.JobState_JOB_STATE_FAILED, fmt.Sprintf(format, args...)} +} + +func errored(format string, args ...any) *jobError { + return &jobError{opificav1.JobState_JOB_STATE_ERRORED, fmt.Sprintf(format, args...)} +} + +// run executes a job and returns how it ended. Errors in the infrastructure +// (VM, agent, mirror) are reported as ERRORED so the controller retries the +// job elsewhere; problems with the job itself are FAILED. +func (r *Runner) run(ctx context.Context, j *runningJob) (opificav1.JobState, int32, string) { + spec := j.spec + if t := spec.GetTimeout().AsDuration(); t > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeoutCause(ctx, t, fmt.Errorf("job timed out after %v", t)) + defer cancel() + } + exit, jerr := r.execute(ctx, j) + switch { + case errors.Is(context.Cause(ctx), errCancelledByController): + return opificav1.JobState_JOB_STATE_CANCELLED, 0, "cancelled" + case r.ctx.Err() != nil: + return opificav1.JobState_JOB_STATE_ERRORED, 0, "worker shut down during the job" + case ctx.Err() != nil: + msg := context.Cause(ctx).Error() + r.syslog(j, 0, "%s", msg) + return opificav1.JobState_JOB_STATE_FAILED, exit, msg + case jerr != nil: + r.syslog(j, 0, "%s", jerr.msg) + return jerr.state, exit, jerr.msg + case exit != 0: + return opificav1.JobState_JOB_STATE_FAILED, exit, "" + } + return opificav1.JobState_JOB_STATE_SUCCEEDED, 0, "" +} + +func (r *Runner) execute(ctx context.Context, j *runningJob) (int32, *jobError) { + spec := j.spec + src := spec.GetSource() + r.cfg.Send(update(spec.GetId(), opificav1.JobState_JOB_STATE_PREPARING, nil, ""), nil) + + r.syslog(j, 0, "Fetching %s at %s", src.GetCloneUrl(), short(src.GetCommitSha())) + mirror, err := r.cfg.Mirrors.Ensure(ctx, src.GetCloneUrl(), src.GetCredential(), src.GetCommitSha()) + switch { + case errors.Is(err, ErrCommitNotFound): + return 0, failed("%v", err) + case err != nil: + return 0, errored("fetching the repository: %v", err) + } + + r.syslog(j, 0, "Starting a fresh VM (image %s)", spec.GetImage()) + lease, err := r.cfg.Pool.Acquire(ctx, spec.GetImage()) + if err != nil { + return 0, errored("starting a VM: %v", err) + } + defer func() { + if err := lease.Release(context.WithoutCancel(ctx)); err != nil { + r.log.Error("destroying job VM", "job", spec.GetId(), "err", err) + } + }() + + client := agent.NewClient(func(ctx context.Context) (net.Conn, error) { return lease.VM.DialAgent(ctx) }) + if _, err := client.Ready(ctx, connect.NewRequest(&opificav1.ReadyRequest{})); err != nil { + return 0, errored("the VM's agent is not responding: %v", err) + } + guestURL, err := r.cfg.GuestMirrorURL(lease.VM, mirror) + if err != nil { + return 0, errored("exposing the repository to the VM: %v", err) + } + if _, err := client.Checkout(ctx, connect.NewRequest(&opificav1.CheckoutRequest{ + MirrorUrl: guestURL, CommitSha: src.GetCommitSha(), Directory: checkoutDir, + })); err != nil { + return 0, errored("checking out %s: %v", short(src.GetCommitSha()), err) + } + + r.cfg.Send(update(spec.GetId(), opificav1.JobState_JOB_STATE_RUNNING, nil, ""), nil) + steps := spec.GetSteps() + var failedExit int32 + for i, step := range steps { + idx := uint32(i) + name := stepName(step) + r.syslog(j, idx, "Step %d/%d: %s", i+1, len(steps), name) + exit, jerr := r.step(ctx, j, client, idx, step) + if jerr != nil { + return exit, jerr + } + if ctx.Err() != nil { + return exit, nil + } + if exit != 0 { + if step.GetContinueOnError() { + r.syslog(j, idx, "Step failed with exit code %d; continuing (continue-on-error)", exit) + continue + } + r.syslog(j, idx, "Step failed with exit code %d", exit) + failedExit = exit + break + } + } + return failedExit, nil +} + +func (r *Runner) step(ctx context.Context, j *runningJob, client agentClient, idx uint32, step *opificav1.Step) (int32, *jobError) { + switch a := step.GetAction().(type) { + case *opificav1.Step_Run: + return r.runStep(ctx, j, client, idx, step, a.Run) + case *opificav1.Step_Report: + r.syslog(j, idx, "Test reports are not stored yet; skipping") + case *opificav1.Step_Upload: + r.syslog(j, idx, "Artifact uploads are not stored yet; skipping") + case *opificav1.Step_Sign: + r.syslog(j, idx, "The signing service is not available yet; skipping") + default: + return 0, failed("step %d has no action", idx+1) + } + return 0, nil +} + +type agentClient interface { + Exec(context.Context, *connect.Request[opificav1.ExecRequest]) (*connect.ServerStreamForClient[opificav1.ExecResponse], error) +} + +func (r *Runner) runStep(ctx context.Context, j *runningJob, client agentClient, idx uint32, step *opificav1.Step, run *opificav1.RunAction) (int32, *jobError) { + spec := j.spec + env := map[string]string{ + "CI": "true", "OPIFICA": "true", + "OPIFICA_JOB_ID": spec.GetId(), "OPIFICA_RUN_ID": spec.GetRunId(), + "OPIFICA_COMMIT_SHA": spec.GetSource().GetCommitSha(), "OPIFICA_REF": spec.GetSource().GetRef(), + } + maps.Copy(env, spec.GetEnv()) + maps.Copy(env, step.GetEnv()) + + var masks []string + for _, name := range step.GetSecrets() { + value, denied, err := r.secret(ctx, spec, name) + if err != nil { + return 0, errored("requesting secret %s: %v", name, err) + } + if denied != "" { + return 0, failed("secret %s was denied: %s", name, denied) + } + env[name] = value + masks = append(masks, value) + } + + timeout := step.GetTimeout().AsDuration() + if dl, ok := ctx.Deadline(); ok && (timeout == 0 || time.Until(dl) < timeout) { + timeout = time.Until(dl) + } + wd := checkoutDir + if d := run.GetWorkingDirectory(); d != "" { + wd = checkoutDir + "/" + strings.TrimPrefix(d, "./") + } + stream, err := client.Exec(ctx, connect.NewRequest(&opificav1.ExecRequest{ + Command: run.GetCommand(), Shell: run.GetShell(), WorkingDirectory: wd, + Env: env, Mask: masks, TimeoutSeconds: uint32(math.Ceil(timeout.Seconds())), + })) + if err != nil { + return 0, errored("starting the step: %v", err) + } + defer stream.Close() + exit := int32(-1) + for stream.Receive() { + switch e := stream.Msg().GetEvent().(type) { + case *opificav1.ExecResponse_Stdout: + r.logChunk(j, idx, opificav1.LogStream_LOG_STREAM_STDOUT, e.Stdout) + case *opificav1.ExecResponse_Stderr: + r.logChunk(j, idx, opificav1.LogStream_LOG_STREAM_STDERR, e.Stderr) + case *opificav1.ExecResponse_ExitCode: + exit = e.ExitCode + } + } + if ctx.Err() != nil { + return 0, nil + } + if err := stream.Err(); err != nil { + var cerr *connect.Error + if errors.As(err, &cerr) && cerr.Code() == connect.CodeInvalidArgument { + return 0, failed("step %d: %s", idx+1, cerr.Message()) + } + return 0, errored("running the step: %v", err) + } + if exit < 0 { + return 0, errored("the agent did not report an exit code") + } + return exit, nil +} + +// secret asks the controller for a secret on the job's behalf. +func (r *Runner) secret(ctx context.Context, spec *opificav1.Job, name string) (value, denied string, err error) { + reqID := uuid.NewString() + ch := make(chan *opificav1.SecretResponse, 1) + r.mu.Lock() + r.secrets[reqID] = ch + r.mu.Unlock() + defer func() { + r.mu.Lock() + delete(r.secrets, reqID) + r.mu.Unlock() + }() + + r.cfg.Send(&opificav1.ConnectRequest{Message: &opificav1.ConnectRequest_SecretRequest{SecretRequest: &opificav1.SecretRequest{ + RequestId: reqID, JobId: spec.GetId(), JobToken: spec.GetJobToken(), Name: name, + }}}, nil) + select { + case resp := <-ch: + if d, ok := resp.GetResult().(*opificav1.SecretResponse_DeniedReason); ok { + return "", d.DeniedReason, nil + } + return string(resp.GetValue()), "", nil + case <-time.After(r.cfg.SecretTimeout): + return "", "", errors.New("no answer from the controller") + case <-ctx.Done(): + return "", "", ctx.Err() + } +} + +func stepName(s *opificav1.Step) string { + if s.GetName() != "" { + return s.GetName() + } + switch a := s.GetAction().(type) { + case *opificav1.Step_Run: + line, _, _ := strings.Cut(strings.TrimSpace(a.Run.GetCommand()), "\n") + return line + case *opificav1.Step_Report: + return "report " + strings.Join(a.Report.GetPaths(), " ") + case *opificav1.Step_Upload: + return "upload " + strings.Join(a.Upload.GetPaths(), " ") + case *opificav1.Step_Sign: + return "sign " + strings.Join(a.Sign.GetPaths(), " ") + } + return "step" +} + +func short(sha string) string { + if len(sha) > 12 { + return sha[:12] + } + return sha +} diff --git a/pkg/worker/worker.go b/pkg/worker/worker.go new file mode 100644 index 0000000..b94a989 --- /dev/null +++ b/pkg/worker/worker.go @@ -0,0 +1,206 @@ +package worker + +import ( + "context" + "errors" + "log/slog" + "maps" + "os" + "path/filepath" + "runtime" + "slices" + "time" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/version" +) + +// Config is everything an OS-specific worker passes to Run. +type Config struct { + ControllerURL string + // StateDir holds the identity from Join and the repository mirrors. + StateDir string + Hypervisor Hypervisor + // Pool sizes the VM pool. Its Capacity is how many jobs run at once. + Pool PoolConfig + // Labels are reported to the controller for scheduling. "pool" is set + // by the join token and can't be chosen here. + Labels map[string]string + // GuestMirrorURL: see RunnerConfig. + GuestMirrorURL func(vm VM, hostPath string) (string, error) + // AllowLocalRepos: see Mirrors.AllowLocal. Development only. + AllowLocalRepos bool + // HeartbeatInterval: see ClientConfig. + HeartbeatInterval time.Duration + Logger *slog.Logger +} + +// Info describes this machine for Join and Hello. +func Info(pool PoolConfig, labels map[string]string) *opificav1.WorkerInfo { + host, _ := os.Hostname() + info := &opificav1.WorkerInfo{ + Hostname: host, + Labels: labels, + MaxConcurrentJobs: uint32(pool.Capacity), + Version: version.Version, + } + switch runtime.GOOS { + case "linux": + info.Os = opificav1.OperatingSystem_OPERATING_SYSTEM_LINUX + case "windows": + info.Os = opificav1.OperatingSystem_OPERATING_SYSTEM_WINDOWS + case "darwin": + info.Os = opificav1.OperatingSystem_OPERATING_SYSTEM_MACOS + } + switch runtime.GOARCH { + case "amd64": + info.Arch = opificav1.Architecture_ARCHITECTURE_AMD64 + case "arm64": + info.Arch = opificav1.Architecture_ARCHITECTURE_ARM64 + } + for _, name := range slices.Sorted(maps.Keys(pool.Images)) { + info.Images = append(info.Images, &opificav1.ImageRef{Name: name, Digest: pool.Images[name].Digest}) + } + return info +} + +// Run connects to the controller and runs jobs until ctx is cancelled or the +// controller rejects the worker for good. The worker must have joined +// (see Join) with the same StateDir. +func Run(ctx context.Context, cfg Config) error { + log := cfg.Logger + if log == nil { + log = slog.Default() + } + id, err := LoadIdentity(cfg.StateDir) + if err != nil { + return err + } + if cfg.Hypervisor == nil { + return errors.New("a hypervisor is required") + } + cfg.Pool.Logger = log + pool, err := NewPool(cfg.Hypervisor, cfg.Pool) + if err != nil { + return err + } + defer pool.Close(context.Background()) + + var outbox *Outbox + runner := NewRunner(RunnerConfig{ + Pool: pool, + Mirrors: &Mirrors{Dir: filepath.Join(cfg.StateDir, "mirrors"), AllowLocal: cfg.AllowLocalRepos}, + Send: func(m *opificav1.ConnectRequest, onSent func()) { outbox.Send(m, onSent) }, + GuestMirrorURL: cfg.GuestMirrorURL, + Logger: log, + }) + + client, err := NewClient(ClientConfig{ + ControllerURL: cfg.ControllerURL, + Identity: id, + Info: Info(cfg.Pool, cfg.Labels), + HeartbeatInterval: cfg.HeartbeatInterval, + Logger: log, + Status: func() (uint32, []string) { + warm := 0 + for _, n := range pool.Stats().Warm { + warm += n + } + return uint32(warm), runner.Running() + }, + OnMessage: runner.HandleMessage, + }) + if err != nil { + return err + } + outbox = NewOutbox(client, 4096) + + outboxCtx, stopOutbox := context.WithCancel(context.Background()) + outboxDone := make(chan struct{}) + go func() { outbox.Run(outboxCtx); close(outboxDone) }() + + err = client.Run(ctx) + + // Stop jobs, then give their final updates a moment to go out (they + // won't if the connection is gone; the controller's leases cover that). + runner.Close() + time.AfterFunc(5*time.Second, stopOutbox) + outbox.Close() + <-outboxDone + stopOutbox() + return err +} + +// Outbox delivers messages to the controller in order, waiting through +// reconnects. Job updates and logs aren't lost when the connection drops +// briefly: they are sent once it is back. +type Outbox struct { + client *Client + ch chan outboxItem + closed chan struct{} +} + +type outboxItem struct { + msg *opificav1.ConnectRequest + onSent func() +} + +// NewOutbox creates an outbox holding up to size messages. When it is full, +// Send blocks, which slows job output down instead of using unbounded memory. +func NewOutbox(c *Client, size int) *Outbox { + return &Outbox{client: c, ch: make(chan outboxItem, size), closed: make(chan struct{})} +} + +// Send queues msg. onSent, if set, runs after it was handed to the stream. +func (o *Outbox) Send(msg *opificav1.ConnectRequest, onSent func()) { + select { + case o.ch <- outboxItem{msg, onSent}: + case <-o.closed: + } +} + +// Close stops accepting messages; Run delivers what is queued, then returns. +func (o *Outbox) Close() { + select { + case <-o.closed: + default: + close(o.closed) + } +} + +// Run sends queued messages until Close was called and the queue is empty, +// or ctx is cancelled. +func (o *Outbox) Run(ctx context.Context) { + for { + var it outboxItem + select { + case it = <-o.ch: + case <-o.closed: + select { + case it = <-o.ch: + default: + return + } + case <-ctx.Done(): + return + } + for { + select { + case <-o.client.Connected(): + case <-ctx.Done(): + return + } + if err := o.client.Send(it.msg); err == nil { + break + } + select { // the stream just ended; wait for the next one + case <-time.After(50 * time.Millisecond): + case <-ctx.Done(): + return + } + } + if it.onSent != nil { + it.onSent() + } + } +} diff --git a/pkg/worker/workertest/hypervisor.go b/pkg/worker/workertest/hypervisor.go index 3a4c2ab..a0f71ea 100644 --- a/pkg/worker/workertest/hypervisor.go +++ b/pkg/worker/workertest/hypervisor.go @@ -6,11 +6,16 @@ import ( "context" "errors" "fmt" + "io" + "log/slog" "net" + "os" "sync" "sync/atomic" "time" + "github.com/opifica/opifica/internal/agent" + "github.com/opifica/opifica/internal/server" "github.com/opifica/opifica/pkg/worker" ) @@ -21,6 +26,10 @@ type Hypervisor struct { BootDelay time.Duration // FailStart, if set, is called on every Start; a non-nil error fails it. FailStart func(spec worker.VMSpec) error + // Agent runs a real agent for every VM, on the host, in a fresh work + // directory that is deleted with the VM. Jobs then really run, but with + // NO isolation: for tests and development only. + Agent bool nextID atomic.Int64 mu sync.Mutex @@ -82,8 +91,16 @@ type VM struct { mu sync.Mutex started bool gone bool + + // With Hypervisor.Agent: + root string + agentAddr string + stopAgent func() } +// Root is the VM's agent work directory (with Hypervisor.Agent). +func (v *VM) Root() string { return v.root } + var _ worker.VM = (*VM)(nil) // ID implements worker.VM. @@ -109,10 +126,41 @@ func (v *VM) Start(ctx context.Context) error { if v.gone { return errors.New("vm destroyed") } + if v.hv.Agent { + if err := v.startAgent(); err != nil { + return err + } + } v.started = true return nil } +func (v *VM) startAgent() error { + root, err := os.MkdirTemp("", "opifica-vm-") + if err != nil { + return err + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + os.RemoveAll(root) + return err + } + ctx, cancel := context.WithCancel(context.Background()) + quiet := slog.New(slog.NewTextHandler(io.Discard, nil)) + done := make(chan struct{}) + go func() { + server.Serve(ctx, ln, agent.Handler(&agent.Server{Root: root, Log: quiet}), quiet) + close(done) + }() + v.root, v.agentAddr = root, ln.Addr().String() + v.stopAgent = func() { + cancel() + <-done + os.RemoveAll(root) + } + return nil +} + // DialAgent implements worker.VM with an in-memory pipe. func (v *VM) DialAgent(ctx context.Context) (net.Conn, error) { v.mu.Lock() @@ -120,8 +168,12 @@ func (v *VM) DialAgent(ctx context.Context) (net.Conn, error) { if !v.started || v.gone { return nil, errors.New("vm not running") } - client, server := net.Pipe() - server.Close() + if v.agentAddr != "" { + var d net.Dialer + return d.DialContext(ctx, "tcp", v.agentAddr) + } + client, srv := net.Pipe() + srv.Close() return client, nil } @@ -130,10 +182,14 @@ func (v *VM) Destroy(context.Context) error { v.mu.Lock() already := v.gone v.gone = true + stop := v.stopAgent v.mu.Unlock() if already { return nil } + if stop != nil { + stop() + } v.hv.mu.Lock() defer v.hv.mu.Unlock() delete(v.hv.alive, v.id)