vole
Clone
Read only, no credentials:
https://rodent.software/vole.git
Write over SSH with a registered key:
git clone ssh://git@rodent.software:2222/vole.git
A free and open-source forge that lives in your Git repo.
| 1 | // Package agent implements the Vole CI agent. |
| 2 | // |
| 3 | // A runner connects to the instance over SSH with the key its account pushes |
| 4 | // with, leases one job at a time, and gives the job a disposable checkout. Steps |
| 5 | // run in a container built from the image the workflow names, or from a pinned |
| 6 | // .vole/images definition; the runner's key and trusted Git cache stay outside. |
| 7 | // The runner streams the log while the job runs, uploads artifacts, and sends the |
| 8 | // attempt to the server with its finish message. This package is the agent; |
| 9 | // argument parsing and the CLI's messages stay in cmd/vole. |
| 10 | package agent |
| 11 | |
| 12 | import ( |
| 13 | "bufio" |
| 14 | "context" |
| 15 | "encoding/json" |
| 16 | "errors" |
| 17 | "fmt" |
| 18 | "io" |
| 19 | "net/url" |
| 20 | "os" |
| 21 | "os/exec" |
| 22 | "path" |
| 23 | "path/filepath" |
| 24 | "runtime" |
| 25 | "sort" |
| 26 | "strings" |
| 27 | "sync" |
| 28 | "syscall" |
| 29 | "time" |
| 30 | |
| 31 | "rodent.software/vole/ci" |
| 32 | jobcache "rodent.software/vole/ci/cache" |
| 33 | "rodent.software/vole/ci/container" |
| 34 | "rodent.software/vole/ci/redact" |
| 35 | "rodent.software/vole/ci/runs" |
| 36 | "rodent.software/vole/client" |
| 37 | "rodent.software/vole/forge/access" |
| 38 | "rodent.software/vole/internal/format" |
| 39 | "rodent.software/vole/metadata" |
| 40 | ) |
| 41 | |
| 42 | // leaseWait is how long an idle runner parks on the lease before asking again. It is |
| 43 | // below the 60-second cap the server applies and long enough that a quiet pool is |
| 44 | // invisible in the queue's load. |
| 45 | const leaseWait = 25 * time.Second |
| 46 | |
| 47 | // cacheTrimInterval is how often the runner bounds the job cache between jobs. A |
| 48 | // namespace that is past the cap while a job has it mounted is skipped and taken on |
| 49 | // a later pass, so a longer gap costs disk, not correctness. |
| 50 | const cacheTrimInterval = 5 * time.Minute |
| 51 | |
| 52 | // How long a slot waits before dialling again after its session failed, and the cap |
| 53 | // that keeps a revoked key from filling the journal while still picking a long |
| 54 | // outage up promptly. |
| 55 | const ( |
| 56 | reconnectDelay = 2 * time.Second |
| 57 | reconnectMax = 30 * time.Second |
| 58 | ) |
| 59 | |
| 60 | // errLeaseLost means the queue took the job away while the runner was still |
| 61 | // preparing its environment. There is nothing to report in that case. |
| 62 | var errLeaseLost = errors.New("lease lost") |
| 63 | |
| 64 | // errCanceled means a maintainer asked this run to stop. The attempt is still worth |
| 65 | // reporting: a canceled run has a story, and the partial log and step timings are it. |
| 66 | var errCanceled = errors.New("run canceled") |
| 67 | |
| 68 | // errStopped says this slot was asked to stop while it held no job. The session was |
| 69 | // hung up so a parked lease would answer instead of sitting out its wait, and nothing |
| 70 | // is lost: a slot that holds no job holds no work. |
| 71 | var errStopped = errors.New("the runner is stopping") |
| 72 | |
| 73 | // refusedJob says the instance handed this runner a job whose shape this runner does |
| 74 | // not accept. Reconnecting would ask the same instance the same question, so the |
| 75 | // error stops every slot and the process says why: a runner that kept looping would |
| 76 | // lease the same job again and refuse it again. |
| 77 | type refusedJob struct{ err error } |
| 78 | |
| 79 | func (r refusedJob) Error() string { return r.err.Error() } |
| 80 | func (r refusedJob) Unwrap() error { return r.err } |
| 81 | |
| 82 | type runnerStep struct { |
| 83 | Name string `json:"name"` |
| 84 | Run string `json:"run"` |
| 85 | Timeout string `json:"timeout"` |
| 86 | } |
| 87 | |
| 88 | type runnerJob struct { |
| 89 | ID string `json:"id"` |
| 90 | Repo string `json:"repo"` |
| 91 | SHA string `json:"sha"` |
| 92 | // Attempt is which try at this job this is. The record lives under it, so the |
| 93 | // runner has to know the number it is reporting. |
| 94 | Attempt int `json:"attempt"` |
| 95 | // Ref, WorkflowSHA and StepsSHA are part of what this runner signs when the |
| 96 | // attempt ends: the exact revision, the exact workflow file, and the digest of |
| 97 | // the steps that were pinned when the job was queued. |
| 98 | Ref string `json:"ref"` |
| 99 | WorkflowSHA string `json:"workflow_sha"` |
| 100 | StepsSHA string `json:"steps_sha"` |
| 101 | Branch string `json:"branch"` |
| 102 | Workflow string `json:"workflow"` |
| 103 | Job string `json:"job"` |
| 104 | Steps []runnerStep `json:"steps"` |
| 105 | // Trigger and EventPayload are the pinned event that requested this run. The |
| 106 | // runner writes the payload into a file and sets VOLE_EVENT, VOLE_EVENT_ACTION |
| 107 | // and VOLE_EVENT_PATH; it is data, never an interpolated command. |
| 108 | Trigger string `json:"trigger,omitempty"` |
| 109 | TriggerAction string `json:"trigger_action,omitempty"` |
| 110 | EventPayload string `json:"event_payload,omitempty"` |
| 111 | // Network is the isolation the job asked for with `network:`, or empty when the |
| 112 | // workflow did not say. The runner applies it to the container it starts, so the |
| 113 | // job's request decides and the machine's default is what an unspecified job gets. |
| 114 | Network string `json:"network,omitempty"` |
| 115 | // Timeout is the job's budget as a duration string, or empty when it has none. |
| 116 | // The runner starts the clock with the first step and cuts every later step short |
| 117 | // when the budget runs out. |
| 118 | Timeout string `json:"timeout,omitempty"` |
| 119 | // Bot is the account this job acts as, and Credential is the key it may act with. |
| 120 | // The key is written into the run directory and is valid until the run ends. |
| 121 | Bot string `json:"bot,omitempty"` |
| 122 | Credential *runnerCredential `json:"credential,omitempty"` |
| 123 | // CredentialError means the job asked for an account it cannot act as. The runner |
| 124 | // fails the job before the container starts rather than running it without a key. |
| 125 | CredentialError string `json:"credential_error,omitempty"` |
| 126 | // HostKey is the instance's public host key line, so the job can verify the host |
| 127 | // it pushes to without scanning it. |
| 128 | HostKey string `json:"host_key,omitempty"` |
| 129 | // Image is a registry reference or "file:<name>". For the latter, ImageDefinition |
| 130 | // is the pinned recipe and ImageSHA its digest; the runner builds it locally. |
| 131 | Image string `json:"image"` |
| 132 | ImageFile string `json:"image_file,omitempty"` |
| 133 | ImageDefinition string `json:"image_definition,omitempty"` |
| 134 | ImageSHA string `json:"image_sha,omitempty"` |
| 135 | // Env is the job's declared environment, pinned when the run was queued. |
| 136 | Env map[string]string `json:"env,omitempty"` |
| 137 | // Services are the containers started beside the job, in declaration order. A job |
| 138 | // with services asked for network: bridge, and the runner gives the job and its |
| 139 | // services one network for this attempt. |
| 140 | Services []runnerService `json:"services,omitempty"` |
| 141 | // Secrets are the declared secrets this repository has, and Missing are the |
| 142 | // names it declared that it does not. A missing one leaves its variable unset and |
| 143 | // gets a line in the log; it is not a failure, because a workflow is allowed an |
| 144 | // optional secret. |
| 145 | Secrets map[string]runnerSecret `json:"secrets,omitempty"` |
| 146 | Missing []runnerSecretIssue `json:"missing_secrets,omitempty"` |
| 147 | // Artifacts are path patterns, expanded in the checkout after the steps pass. |
| 148 | Artifacts []string `json:"artifacts,omitempty"` |
| 149 | // Tag is set for a tag-triggered run; TagAnnotated says whether it carries |
| 150 | // notes, which is what makes it a release. ReleaseArtifacts are the patterns |
| 151 | // that also publish to that release. |
| 152 | Tag string `json:"tag,omitempty"` |
| 153 | TagAnnotated bool `json:"tag_annotated,omitempty"` |
| 154 | ReleaseArtifacts []string `json:"release_artifacts,omitempty"` |
| 155 | // Pages is the site deploy this run performs, if any. |
| 156 | Pages *runnerPages `json:"pages,omitempty"` |
| 157 | } |
| 158 | |
| 159 | // runnerCredential is the private half of a run's key, delivered once. |
| 160 | type runnerCredential struct { |
| 161 | PrivateKey string `json:"private_key"` |
| 162 | Fingerprint string `json:"fingerprint"` |
| 163 | } |
| 164 | |
| 165 | // runnerService is one container the job asked to run beside it. |
| 166 | type runnerService struct { |
| 167 | ID string `json:"id"` |
| 168 | Image string `json:"image"` |
| 169 | Env map[string]string `json:"env,omitempty"` |
| 170 | } |
| 171 | |
| 172 | // runnerSecret is one delivered value and the version it came from. The version is |
| 173 | // what the attempt record reports, so a result says which value ran without |
| 174 | // carrying it. |
| 175 | type runnerSecret struct { |
| 176 | Version int `json:"version"` |
| 177 | Value string `json:"value"` |
| 178 | } |
| 179 | |
| 180 | // runnerSecretIssue is a declared secret the instance could not supply. Reason is |
| 181 | // empty when the only reason is that nobody has stored that name. |
| 182 | type runnerSecretIssue struct { |
| 183 | Name string `json:"name"` |
| 184 | Reason string `json:"reason,omitempty"` |
| 185 | } |
| 186 | |
| 187 | type runnerPages struct { |
| 188 | Path string `json:"path"` |
| 189 | Ref string `json:"ref"` |
| 190 | } |
| 191 | |
| 192 | type runnerArtifact struct { |
| 193 | Name string `json:"name"` |
| 194 | OID string `json:"oid"` |
| 195 | Size int64 `json:"size"` |
| 196 | } |
| 197 | |
| 198 | // runnerLogChunk is one piece of an attempt's log. Offset is where it starts in |
| 199 | // the attempt's file, so the server can refuse a gap or a duplicate. |
| 200 | type runnerLogChunk struct { |
| 201 | Attempt int `json:"attempt"` |
| 202 | Offset int64 `json:"offset"` |
| 203 | Data []byte `json:"data"` |
| 204 | } |
| 205 | |
| 206 | // runnerRequest is the wire. A finish carries the attempt's record: the runner |
| 207 | // no longer publishes it to Git, and this channel is what the server checks and |
| 208 | // stores. |
| 209 | type runnerRequest struct { |
| 210 | Op string `json:"op"` |
| 211 | Run string `json:"run"` |
| 212 | // Record is the attempt's result, sent with finish. |
| 213 | Record *runs.Record `json:"record,omitempty"` |
| 214 | // Log is one chunk of the attempt's log, sent while the job runs. |
| 215 | Log *runnerLogChunk `json:"log,omitempty"` |
| 216 | // Repos is the --repos filter, sent with every lease. The server intersects it |
| 217 | // with what the account's allowlists grant, so it can only narrow. |
| 218 | Repos []string `json:"repos,omitempty"` |
| 219 | // Labels are what this machine is, sent with every lease so the server can match |
| 220 | // them against the runs-on a job asked for. |
| 221 | Labels []string `json:"labels,omitempty"` |
| 222 | // Network is the network this machine can give, sent with every lease so the |
| 223 | // server can match it against a job's `network:`. A runner started without |
| 224 | // --container-network bridge says none, which is the truth: it cannot provide one. |
| 225 | Network string `json:"network,omitempty"` |
| 226 | // Wait is how many seconds an idle lease may park before answering "nothing to |
| 227 | // do"; the server caps it. It is what makes an idle runner cost a parked channel |
| 228 | // rather than a query every couple of seconds. |
| 229 | Wait int `json:"wait,omitempty"` |
| 230 | } |
| 231 | |
| 232 | type runnerReply struct { |
| 233 | Op string `json:"op"` |
| 234 | Error string `json:"error,omitempty"` |
| 235 | Run *runnerJob `json:"run,omitempty"` |
| 236 | OK bool `json:"ok"` |
| 237 | Lease int `json:"lease,omitempty"` |
| 238 | // The account name this session authenticated as, so a runner can say whose |
| 239 | // results it reports. |
| 240 | Name string `json:"name,omitempty"` |
| 241 | // Repos are the repositories this account may run jobs for. A runner answers the |
| 242 | // "ping" with these so a --repos filter can be checked against them at startup: |
| 243 | // a filter that matches nothing would otherwise look exactly like a quiet instance. |
| 244 | Repos []string `json:"repos,omitempty"` |
| 245 | // LogSize is how much of the log the server has after a log chunk. |
| 246 | LogSize int64 `json:"log_size,omitempty"` |
| 247 | // Cancel names the account that asked this run to stop. It rides the heartbeat |
| 248 | // reply: empty means keep going, and a server that does not send it is one this |
| 249 | // runner keeps working against, exactly as before. |
| 250 | Cancel string `json:"cancel,omitempty"` |
| 251 | } |
| 252 | |
| 253 | // Options configure one runner process. |
| 254 | type Options struct { |
| 255 | // Remote is the instance's SSH URL, e.g. ssh://git@host:2222. |
| 256 | Remote string |
| 257 | // Dir is where checkouts are cached. Empty means the user cache directory. |
| 258 | Dir string |
| 259 | // Repos narrows this runner to the named repositories. The server intersects |
| 260 | // it with the account's runner allowlist, so it can only subtract. |
| 261 | Repos []string |
| 262 | // Labels describe the machine. A job is only leased to a runner carrying every |
| 263 | // label the job asked for with runs-on. |
| 264 | Labels []string |
| 265 | // Once runs one job per slot and exits, which is what a scheduled container wants. |
| 266 | Once bool |
| 267 | // Jobs is how many jobs one process may run at once, one per session. Zero or |
| 268 | // negative means one. Each slot holds its container limits of its own, so the |
| 269 | // machine has to carry this many times what --container-memory and the rest |
| 270 | // allow. |
| 271 | Jobs int |
| 272 | // ContainerRuntime is the CLI used to run containers: docker, podman, or nerdctl. |
| 273 | // A runner always requires one; the image itself comes from each job's workflow. |
| 274 | ContainerRuntime string |
| 275 | // ContainerNetwork is the network mode: none (the default) or bridge. |
| 276 | ContainerNetwork string |
| 277 | // ContainerPull is the image pull policy: missing (the default), always, or never. |
| 278 | ContainerPull string |
| 279 | // ContainerUser is the user the job runs as inside the container, uid:gid. |
| 280 | // Empty means the runner user's uid:gid. |
| 281 | ContainerUser string |
| 282 | // ContainerUserns is the user namespace mode passed to the runtime. Empty uses |
| 283 | // keep-id for rootless Podman, which is what makes a bind-mounted job tree |
| 284 | // writable by the runner user. |
| 285 | ContainerUserns string |
| 286 | // ContainerMemory and ContainerCPUs are optional resource limits passed to the |
| 287 | // container runtime. Empty leaves the runtime's own behavior. |
| 288 | ContainerMemory string |
| 289 | ContainerCPUs string |
| 290 | // ContainerPIDs bounds one container's processes. Zero means the default in |
| 291 | // ci/container, not "no limit": a job that can fork without a bound is a fork bomb |
| 292 | // away from taking the machine. |
| 293 | ContainerPIDs int |
| 294 | // ContainerTmpfs is the size of the job's /tmp. Empty means the default; it is a |
| 295 | // tmpfs, so this is memory. |
| 296 | ContainerTmpfs string |
| 297 | // DiskReserve is how much free space the runner insists on keeping under Dir. A |
| 298 | // job that takes the filesystem below it is stopped, because a job tree is a host |
| 299 | // bind mount with no quota and the machine is shared. Zero disables the check. |
| 300 | DiskReserve int64 |
| 301 | |
| 302 | // NoCache turns off the per-repository cache that lets a job reuse downloads |
| 303 | // and compiled output between runs. A runner without one is slower and has |
| 304 | // nothing to poison, which is the trade. |
| 305 | NoCache bool |
| 306 | // CacheMaxBytes caps that cache. Zero means the default; see ci/cache. |
| 307 | CacheMaxBytes int64 |
| 308 | |
| 309 | // GitCacheMaxBytes caps the saved checkouts of the repositories this runner has |
| 310 | // built. A job clones from the saved checkout instead of downloading the history |
| 311 | // again, so the cap trades disk for network. Negative means the default, and zero |
| 312 | // keeps every checkout; see DefaultGitCacheBytes and gitcache.go. |
| 313 | GitCacheMaxBytes int64 |
| 314 | } |
| 315 | |
| 316 | // Run leases and executes CI jobs until the context ends, a slot is told to stop, |
| 317 | // or (with Once) every slot has handled one job. |
| 318 | // |
| 319 | // Everything that belongs to the process happens here, once: the directories, the |
| 320 | // cache bounds, the leftovers of a killed predecessor, and the connections. The |
| 321 | // slots then share all of it and differ in nothing but their session, which is what |
| 322 | // lets --jobs run several jobs on one machine without each of them repeating the |
| 323 | // startup work or racing it. |
| 324 | func Run(ctx context.Context, opts Options) error { |
| 325 | if opts.Remote == "" { |
| 326 | return errors.New("--remote is required, e.g. --remote ssh://git@example.com:2222") |
| 327 | } |
| 328 | // Everything below is one process's work, so one cancellation covers it: the |
| 329 | // cache trimmer and the slots stop together when a slot fails for a reason the |
| 330 | // others cannot fix. |
| 331 | ctx, cancel := context.WithCancel(ctx) |
| 332 | defer cancel() |
| 333 | cache := opts.Dir |
| 334 | if cache == "" { |
| 335 | base, err := os.UserCacheDir() |
| 336 | if err != nil { |
| 337 | base = os.TempDir() |
| 338 | } |
| 339 | cache = filepath.Join(base, "vole-runner") |
| 340 | } |
| 341 | cache, err := filepath.Abs(cache) |
| 342 | if err != nil { |
| 343 | return err |
| 344 | } |
| 345 | if err := os.MkdirAll(cache, 0o755); err != nil { |
| 346 | return err |
| 347 | } |
| 348 | |
| 349 | box, err := containerOptions(opts, cache) |
| 350 | if err != nil { |
| 351 | return err |
| 352 | } |
| 353 | // The binary being on PATH is not the runtime being able to create a container. |
| 354 | // Probe before anything is leased: a machine that cannot run a container is not |
| 355 | // a degraded runner, and an attempt spent proving it is a commit blamed for the |
| 356 | // machine's problem. |
| 357 | if err := container.Probe(ctx, box.Config); err != nil { |
| 358 | return err |
| 359 | } |
| 360 | cleanupJobTrees(cache) |
| 361 | cleanupStaleContainers(box) |
| 362 | cleanupStaleNetworks(box) |
| 363 | // Bound the saved checkouts before work starts, the same way the job cache is |
| 364 | // bound below: a runner whose disk filled up while it was not watching should |
| 365 | // not start by failing a job. |
| 366 | gitCacheMax := opts.GitCacheMaxBytes |
| 367 | if gitCacheMax < 0 { |
| 368 | gitCacheMax = DefaultGitCacheBytes |
| 369 | } |
| 370 | maintainTrustedCaches(cache, gitCacheMax, "") |
| 371 | |
| 372 | // The directory is mounted whether or not it is reused: an image that sets |
| 373 | // GOCACHE=/cache/go-build must work under --no-cache too, and a non-root job |
| 374 | // cannot create /cache in the image's own (root-owned) tree. |
| 375 | cacheRoot := filepath.Join(cache, "cache") |
| 376 | if err := os.MkdirAll(cacheRoot, 0o755); err != nil { |
| 377 | return fmt.Errorf("prepare the job cache: %w", err) |
| 378 | } |
| 379 | cacheMax := opts.CacheMaxBytes |
| 380 | if cacheMax <= 0 { |
| 381 | cacheMax = jobcache.DefaultBytes |
| 382 | } |
| 383 | if opts.NoCache { |
| 384 | // Nothing survives, so nothing needs bounding; empty what a killed runner |
| 385 | // left behind. |
| 386 | _ = jobcache.Clear(cacheRoot) |
| 387 | } else { |
| 388 | if removed, err := jobcache.Trim(cacheRoot, cacheMax); err == nil && removed > 0 { |
| 389 | // A killed runner leaves whatever the last job wrote; bound it before work |
| 390 | // starts rather than after the disk is full. |
| 391 | fmt.Fprintf(os.Stderr, "vole runner: trimmed %s from the job cache\n", format.Bytes(removed)) |
| 392 | } |
| 393 | // Bound it between jobs too, but on its own beat rather than at the end of |
| 394 | // every job: eviction is whole-namespace and passes over a namespace a job |
| 395 | // has mounted, so it is a disk-pressure check rather than the last step of a |
| 396 | // build. |
| 397 | go trimJobCache(ctx, cacheRoot, cacheMax) |
| 398 | } |
| 399 | |
| 400 | // A label has to be one the workflow validator would accept, or the machine would |
| 401 | // advertise something no job can ask for — a typo with no symptom at all. |
| 402 | for _, label := range opts.Labels { |
| 403 | if err := ci.ValidLabel(label); err != nil { |
| 404 | return fmt.Errorf("--label %w", err) |
| 405 | } |
| 406 | } |
| 407 | |
| 408 | jobs := opts.Jobs |
| 409 | if jobs < 1 { |
| 410 | jobs = 1 |
| 411 | } |
| 412 | |
| 413 | // One session per slot, all connected before the first lease: a slot that cannot |
| 414 | // reach the instance is a startup failure, not a job that fails later and less |
| 415 | // clearly. |
| 416 | slots := make([]*slot, 0, jobs) |
| 417 | for i := 0; i < jobs; i++ { |
| 418 | sl := &slot{opts: opts, cache: cache, box: box, cacheRoot: cacheRoot, gitCacheMax: gitCacheMax} |
| 419 | if err := sl.connect(); err != nil { |
| 420 | closeSlots(slots) |
| 421 | return err |
| 422 | } |
| 423 | // --repos narrows the account's grant. Check it against the grant now: a filter |
| 424 | // that matches nothing is indistinguishable from a quiet instance, so it would |
| 425 | // idle forever while looking healthy — cheaper to find at startup than after a |
| 426 | // build that never arrives. Every slot gets the same answer, so it is asked |
| 427 | // once. |
| 428 | if i == 0 { |
| 429 | if err := checkRepoFilter(opts.Repos, sl.who); err != nil { |
| 430 | sl.close() |
| 431 | closeSlots(slots) |
| 432 | return err |
| 433 | } |
| 434 | } |
| 435 | slots = append(slots, sl) |
| 436 | } |
| 437 | |
| 438 | if opts.NoCache { |
| 439 | fmt.Fprintf(os.Stderr, "vole runner: job cache in %s, emptied after every job\n", cacheRoot) |
| 440 | } else { |
| 441 | fmt.Fprintf(os.Stderr, "vole runner: job cache in %s, capped at %s\n", |
| 442 | cacheRoot, format.Bytes(cacheMax)) |
| 443 | } |
| 444 | fmt.Fprintf(os.Stderr, "vole runner: %s\n", slots[0].describe(jobs)) |
| 445 | fmt.Fprintf(os.Stderr, "vole runner: jobs run in the workflow's image via %s\n", box.Runtime) |
| 446 | if container.RunsAsRoot(box.Config) { |
| 447 | fmt.Fprintln(os.Stderr, "vole runner: warning: jobs run as root in their container, so a job can "+ |
| 448 | "rewrite its own image; run the runner as an unprivileged user (or pass --container-user) "+ |
| 449 | "to have the image's permissions keep it immutable") |
| 450 | } |
| 451 | |
| 452 | // The slots share a process, so a failure one of them cannot fix — a job this |
| 453 | // runner refuses to execute — stops them all rather than leaving the process |
| 454 | // half-alive with an answer nobody is going to change. |
| 455 | var ( |
| 456 | wg sync.WaitGroup |
| 457 | stopMu sync.Mutex |
| 458 | stopErr error |
| 459 | ) |
| 460 | halt := func(err error) { |
| 461 | stopMu.Lock() |
| 462 | if stopErr == nil { |
| 463 | stopErr = err |
| 464 | } |
| 465 | stopMu.Unlock() |
| 466 | cancel() |
| 467 | } |
| 468 | for _, sl := range slots { |
| 469 | wg.Add(1) |
| 470 | go func(sl *slot) { |
| 471 | defer wg.Done() |
| 472 | if err := sl.run(ctx); err != nil { |
| 473 | halt(err) |
| 474 | } |
| 475 | }(sl) |
| 476 | } |
| 477 | wg.Wait() |
| 478 | return stopErr |
| 479 | } |
| 480 | |
| 481 | // slot is one worker: one session, at most one job at a time. --jobs N is N of |
| 482 | // these. They share the process, the cache root and the saved checkouts, and |
| 483 | // differ only in the session they lease on, because a lease is answered on the |
| 484 | // channel that asked for it. |
| 485 | type slot struct { |
| 486 | opts Options |
| 487 | cache string |
| 488 | box runnerBox |
| 489 | cacheRoot string |
| 490 | gitCacheMax int64 |
| 491 | // sess is this slot's connection, nil between a failure and the reconnect. |
| 492 | sess *runnerSession |
| 493 | // who is the account the current session authenticated as. The answer is what |
| 494 | // the runner reports its results under, so a session without one is not a |
| 495 | // session yet. |
| 496 | who runnerReply |
| 497 | } |
| 498 | |
| 499 | // connect dials this slot's session and asks who the instance thinks it is. |
| 500 | func (sl *slot) connect() error { |
| 501 | sess, err := dialRunner(sl.opts.Remote) |
| 502 | if err != nil { |
| 503 | return err |
| 504 | } |
| 505 | who, err := sess.call(runnerRequest{Op: "ping"}) |
| 506 | if err != nil { |
| 507 | sess.close() |
| 508 | return err |
| 509 | } |
| 510 | if who.Error != "" { |
| 511 | sess.close() |
| 512 | return errors.New(who.Error) |
| 513 | } |
| 514 | sess.name = who.Name |
| 515 | sess.repos, sess.labels = sl.opts.Repos, sl.opts.Labels |
| 516 | sess.network = sl.box.Network |
| 517 | sess.cacheRoot, sess.cacheReuse = sl.cacheRoot, !sl.opts.NoCache |
| 518 | sess.diskRoot, sess.diskReserve = sl.cache, sl.opts.DiskReserve |
| 519 | sl.sess, sl.who = sess, who |
| 520 | return nil |
| 521 | } |
| 522 | |
| 523 | // close ends this slot's session, if it has one. |
| 524 | func (sl *slot) close() { |
| 525 | if sl.sess != nil { |
| 526 | sl.sess.close() |
| 527 | sl.sess = nil |
| 528 | } |
| 529 | } |
| 530 | |
| 531 | // describe is the line the runner prints when it starts work, because "why is |
| 532 | // nothing running" is answered by reading what it says it will take. |
| 533 | func (sl *slot) describe(jobs int) string { |
| 534 | return sl.sess.describe(sl.who.Name, jobs) |
| 535 | } |
| 536 | |
| 537 | // run keeps this slot working until the context ends. A lost connection is the |
| 538 | // slot's problem rather than the runner's: it reconnects and carries on, so one |
| 539 | // flaky link does not take the other slots' jobs with it, and the job that was |
| 540 | // running is requeued when its lease expires. A refused job is the exception — see |
| 541 | // refusedJob — because reconnecting would ask the same instance the same question. |
| 542 | func (sl *slot) run(ctx context.Context) error { |
| 543 | defer sl.close() |
| 544 | delay := reconnectDelay |
| 545 | for { |
| 546 | if ctx.Err() != nil { |
| 547 | return nil |
| 548 | } |
| 549 | if sl.sess == nil { |
| 550 | if err := sl.connect(); err != nil { |
| 551 | if sl.opts.Once { |
| 552 | // A one-shot runner is a process per job: the scheduler decides |
| 553 | // whether to try again, not a loop here. |
| 554 | return err |
| 555 | } |
| 556 | fmt.Fprintf(os.Stderr, "vole runner: %v; retrying in %s\n", err, delay) |
| 557 | if !sleep(ctx, delay) { |
| 558 | return nil |
| 559 | } |
| 560 | delay = nextDelay(delay) |
| 561 | continue |
| 562 | } |
| 563 | delay = reconnectDelay |
| 564 | } |
| 565 | err := sl.serve(ctx) |
| 566 | sl.close() |
| 567 | if err == nil { |
| 568 | return nil |
| 569 | } |
| 570 | if ctx.Err() != nil { |
| 571 | // Stopping: the session went away because it was hung up, or a job |
| 572 | // finished after another slot stopped the runner. Either way this slot is |
| 573 | // done, and there is nothing to reconnect to. |
| 574 | return nil |
| 575 | } |
| 576 | var refused refusedJob |
| 577 | if errors.As(err, &refused) || sl.opts.Once { |
| 578 | return err |
| 579 | } |
| 580 | fmt.Fprintf(os.Stderr, "vole runner: %v; reconnecting in %s\n", err, delay) |
| 581 | if !sleep(ctx, delay) { |
| 582 | return nil |
| 583 | } |
| 584 | delay = nextDelay(delay) |
| 585 | } |
| 586 | } |
| 587 | |
| 588 | // serve leases and runs jobs on this slot's session until something fails. |
| 589 | func (sl *slot) serve(ctx context.Context) error { |
| 590 | s := sl.sess |
| 591 | for { |
| 592 | // A slot that has been told to stop does not lease again. A job it is |
| 593 | // already running finishes and reports, which is what "stop" means here: |
| 594 | // throwing away a finished build would only hand the same work to somebody |
| 595 | // else. |
| 596 | if ctx.Err() != nil { |
| 597 | return nil |
| 598 | } |
| 599 | // A resident runner parks on the server's lease: the reply arrives when work |
| 600 | // is queued, not on a fixed beat, so an idle slot costs one open channel |
| 601 | // instead of a query every couple of seconds. A one-shot runner asks once and |
| 602 | // leaves. |
| 603 | req := runnerRequest{Op: "lease", Repos: s.repos, Labels: s.labels, Network: s.network} |
| 604 | if !sl.opts.Once { |
| 605 | req.Wait = int(leaseWait.Seconds()) |
| 606 | } |
| 607 | rep, err := s.lease(ctx, req) |
| 608 | if err != nil { |
| 609 | return err |
| 610 | } |
| 611 | if rep.Error != "" { |
| 612 | return errors.New(rep.Error) |
| 613 | } |
| 614 | if rep.Run == nil { |
| 615 | if sl.opts.Once { |
| 616 | return nil |
| 617 | } |
| 618 | // Either the wait expired with nothing to do, or the server answered at |
| 619 | // once because it is older than this client and ignores `wait`. The sleep |
| 620 | // keeps the second case at a polite two requests per second instead of a |
| 621 | // spin. |
| 622 | if !sleep(ctx, 2*time.Second) { |
| 623 | return nil |
| 624 | } |
| 625 | continue |
| 626 | } |
| 627 | // The job came over the channel, and the runner is the party that acts on it: |
| 628 | // every field it builds a path or a command from is checked before the |
| 629 | // checkout, the container or the cache sees it. |
| 630 | if err := validateJob(rep.Run); err != nil { |
| 631 | return refusedJob{err} |
| 632 | } |
| 633 | lease := time.Duration(rep.Lease) * time.Second |
| 634 | if lease <= 0 { |
| 635 | lease = 60 * time.Second |
| 636 | } |
| 637 | s.run(sl.opts.Remote, sl.cache, rep.Run, lease, sl.box) |
| 638 | if sl.opts.Once { |
| 639 | // This process is done; the next one bounds the cache at startup instead, |
| 640 | // before it leases anything. |
| 641 | return nil |
| 642 | } |
| 643 | // Tidy between jobs, never during one: the checkout this runner just used is |
| 644 | // the one the next job is most likely to want, so it is passed over as active. |
| 645 | maintainTrustedCaches(sl.cache, sl.gitCacheMax, rep.Run.Repo) |
| 646 | } |
| 647 | } |
| 648 | |
| 649 | // lease asks for a job and returns when the server answers or the runner is stopping. |
| 650 | // A resident runner parks on this call for up to leaseWait, and a shutdown that had |
| 651 | // to sit out the park would be half a minute of nothing happening. |
| 652 | // |
| 653 | // The reply is read by the session's decoder, which cannot be interrupted from |
| 654 | // inside it; ending the connection is what unblocks it, and that costs nothing here |
| 655 | // because a slot parked on a lease is holding no job. |
| 656 | func (s *runnerSession) lease(ctx context.Context, req runnerRequest) (runnerReply, error) { |
| 657 | type answer struct { |
| 658 | rep runnerReply |
| 659 | err error |
| 660 | } |
| 661 | done := make(chan answer, 1) |
| 662 | go func() { |
| 663 | rep, err := s.call(req) |
| 664 | done <- answer{rep, err} |
| 665 | }() |
| 666 | select { |
| 667 | case a := <-done: |
| 668 | return a.rep, a.err |
| 669 | case <-ctx.Done(): |
| 670 | s.hangup() |
| 671 | return runnerReply{}, errStopped |
| 672 | } |
| 673 | } |
| 674 | |
| 675 | // hangup ends the connection from outside the session's lock: taking the lock would |
| 676 | // wait for the goroutine that holds it, which is the one parked on the read. |
| 677 | func (s *runnerSession) hangup() { |
| 678 | if s.cmd != nil && s.cmd.Process != nil { |
| 679 | _ = s.cmd.Process.Signal(syscall.SIGTERM) |
| 680 | } |
| 681 | } |
| 682 | |
| 683 | func closeSlots(slots []*slot) { |
| 684 | for _, sl := range slots { |
| 685 | sl.close() |
| 686 | } |
| 687 | } |
| 688 | |
| 689 | // sleep waits for d or until the context ends, and reports whether the wait |
| 690 | // finished. |
| 691 | func sleep(ctx context.Context, d time.Duration) bool { |
| 692 | select { |
| 693 | case <-ctx.Done(): |
| 694 | return false |
| 695 | case <-time.After(d): |
| 696 | return true |
| 697 | } |
| 698 | } |
| 699 | |
| 700 | // nextDelay doubles the wait between reconnects, up to reconnectMax: a revoked key |
| 701 | // should not fill the journal, and a long outage should still be picked up promptly. |
| 702 | func nextDelay(d time.Duration) time.Duration { |
| 703 | if d *= 2; d > reconnectMax { |
| 704 | return reconnectMax |
| 705 | } |
| 706 | return d |
| 707 | } |
| 708 | |
| 709 | // ---------- connection ---------- |
| 710 | |
| 711 | type runnerSession struct { |
| 712 | cmd *exec.Cmd |
| 713 | in *bufio.Writer |
| 714 | out *bufio.Reader |
| 715 | enc *json.Encoder |
| 716 | dec *json.Decoder |
| 717 | mu sync.Mutex |
| 718 | // name is the account the server resolved from the connection. A record |
| 719 | // names it, and the gate refuses a record whose signer is not the pusher. |
| 720 | name string |
| 721 | // repos is the --repos filter, sent with every lease. It is what the operator |
| 722 | // asked for, not the effective list: the server intersects it with the grant on |
| 723 | // each request, so an allowlist edit takes effect without a restart. |
| 724 | repos []string |
| 725 | // cacheRoot is where this runner keeps the caches it mounts into jobs. cacheReuse |
| 726 | // says whether a job keeps what the last one wrote; with --no-cache the directory |
| 727 | // is still mounted and emptied instead, so an image that points at /cache works |
| 728 | // either way. The cap and the trimmer live in Run: a cache is bounded on the |
| 729 | // runner's own beat, not per job. |
| 730 | cacheRoot string |
| 731 | cacheReuse bool |
| 732 | // labels is what this machine is (--label). A job is only given to a runner that |
| 733 | // carries every label the job asked for, so an unlabelled runner only sees jobs |
| 734 | // that ask for nothing. |
| 735 | labels []string |
| 736 | // network is the network this runner can give, sent with every lease so the server |
| 737 | // only hands it a job whose `network:` it can satisfy. |
| 738 | network string |
| 739 | // diskRoot is the filesystem a job's tree is written to, and diskReserve is how |
| 740 | // much of it must stay free. Together they are the only bound on a job tree: it is |
| 741 | // a host bind mount, so the runtime cannot quota it. |
| 742 | diskRoot string |
| 743 | diskReserve int64 |
| 744 | } |
| 745 | |
| 746 | // describe is the one line a runner prints when it starts work, because "why is |
| 747 | // nothing running" is answered by reading what it says it will take. Both filters are |
| 748 | // named when they are set and omitted when they are not, so the common case stays one |
| 749 | // short line. |
| 750 | func (s *runnerSession) describe(name string, jobs int) string { |
| 751 | out := "connected as " + name |
| 752 | if len(s.labels) > 0 { |
| 753 | out += ", taking jobs that ask for " + strings.Join(s.labels, ", ") |
| 754 | } |
| 755 | if len(s.repos) > 0 { |
| 756 | out += ", only in " + strings.Join(s.repos, ", ") |
| 757 | } |
| 758 | if s.network == ci.NetworkBridge { |
| 759 | out += ", able to give a job the network" |
| 760 | } |
| 761 | if jobs > 1 { |
| 762 | out += fmt.Sprintf(", running %d jobs at a time", jobs) |
| 763 | } |
| 764 | return out + "; waiting for work" |
| 765 | } |
| 766 | |
| 767 | // checkRepoFilter refuses a --repos filter that can never match. A name the server |
| 768 | // did not grant is dropped by the intersection rather than refused — an operator may |
| 769 | // reasonably name a repository before it exists — but a filter whose names are *all* |
| 770 | // outside the grant is a typo or a permissions problem, and it would otherwise present |
| 771 | // as a runner that connected happily and never did anything. |
| 772 | func checkRepoFilter(asked []string, who runnerReply) error { |
| 773 | if len(asked) == 0 { |
| 774 | if len(who.Repos) == 0 { |
| 775 | // Not an error: an allowlist nobody has written yet is a normal state, and it |
| 776 | // takes effect without a restart. Say so instead of idling silently. |
| 777 | fmt.Fprintf(os.Stderr, "vole runner: %s may not run jobs for any repository yet; "+ |
| 778 | "waiting (an admin allows it with `vole admin acl runners add <repo> %s`)\n", |
| 779 | who.Name, who.Name) |
| 780 | } |
| 781 | return nil |
| 782 | } |
| 783 | granted := make(map[string]bool, len(who.Repos)) |
| 784 | for _, r := range who.Repos { |
| 785 | granted[r] = true |
| 786 | } |
| 787 | for _, r := range asked { |
| 788 | if granted[r] { |
| 789 | return nil |
| 790 | } |
| 791 | } |
| 792 | allowed := "none" |
| 793 | if len(who.Repos) > 0 { |
| 794 | allowed = strings.Join(who.Repos, ", ") |
| 795 | } |
| 796 | return fmt.Errorf("--repos %s matches nothing %s may run (allowed: %s); "+ |
| 797 | "a runner can only narrow what its account is allowed", |
| 798 | strings.Join(asked, ", "), who.Name, allowed) |
| 799 | } |
| 800 | |
| 801 | func dialRunner(remote string) (*runnerSession, error) { |
| 802 | u, err := url.Parse(remote) |
| 803 | if err != nil { |
| 804 | return nil, fmt.Errorf("%q is not an ssh URL: %w", remote, err) |
| 805 | } |
| 806 | if u.Scheme != "ssh" || u.Hostname() == "" { |
| 807 | return nil, fmt.Errorf("%q is not an ssh URL (expected ssh://user@host:port)", remote) |
| 808 | } |
| 809 | user := u.User.Username() |
| 810 | if user == "" { |
| 811 | user = "git" |
| 812 | } |
| 813 | cmd := client.SSHCommand(user+"@"+u.Hostname(), u.Port(), "vole-runner") |
| 814 | cmd.Stderr = os.Stderr |
| 815 | stdin, err := cmd.StdinPipe() |
| 816 | if err != nil { |
| 817 | return nil, err |
| 818 | } |
| 819 | stdout, err := cmd.StdoutPipe() |
| 820 | if err != nil { |
| 821 | return nil, err |
| 822 | } |
| 823 | if err := cmd.Start(); err != nil { |
| 824 | return nil, fmt.Errorf("start ssh: %w", err) |
| 825 | } |
| 826 | s := &runnerSession{cmd: cmd, in: bufio.NewWriter(stdin), out: bufio.NewReader(stdout)} |
| 827 | s.enc = json.NewEncoder(s.in) |
| 828 | s.dec = json.NewDecoder(s.out) |
| 829 | return s, nil |
| 830 | } |
| 831 | |
| 832 | func (s *runnerSession) call(req runnerRequest) (runnerReply, error) { |
| 833 | var rep runnerReply |
| 834 | s.mu.Lock() |
| 835 | defer s.mu.Unlock() |
| 836 | if err := s.enc.Encode(req); err != nil { |
| 837 | return rep, fmt.Errorf("connection lost: %w", err) |
| 838 | } |
| 839 | if err := s.in.Flush(); err != nil { |
| 840 | return rep, fmt.Errorf("connection lost: %w", err) |
| 841 | } |
| 842 | if err := s.dec.Decode(&rep); err != nil { |
| 843 | return rep, fmt.Errorf("connection lost: %w", err) |
| 844 | } |
| 845 | return rep, nil |
| 846 | } |
| 847 | |
| 848 | // heartbeat renews the lease and asks whether a maintainer has stopped the run. It |
| 849 | // is for the places that are not a running command: a command heartbeats on its own |
| 850 | // ticker while it runs, and this is what keeps the lease across the steps that are |
| 851 | // too short to have one. A broken connection answers false, so an attempt that can |
| 852 | // no longer reach the instance stops instead of working without a lease. |
| 853 | func (s *runnerSession) heartbeat(runID string) bool { |
| 854 | rep, err := s.call(runnerRequest{Op: "heartbeat", Run: runID}) |
| 855 | if err != nil { |
| 856 | return false |
| 857 | } |
| 858 | return rep.OK |
| 859 | } |
| 860 | |
| 861 | func (s *runnerSession) close() { |
| 862 | s.mu.Lock() |
| 863 | _ = s.enc.Encode(runnerRequest{Op: "bye"}) |
| 864 | _ = s.in.Flush() |
| 865 | s.mu.Unlock() |
| 866 | _ = s.cmd.Process.Signal(syscall.SIGTERM) |
| 867 | _ = s.cmd.Wait() |
| 868 | } |
| 869 | |
| 870 | // ---------- execution ---------- |
| 871 | |
| 872 | // run executes a leased job and reports the outcome. A lost lease ends the job |
| 873 | // without reporting a result: another runner owns it now. |
| 874 | func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Duration, box runnerBox) { |
| 875 | log, err := newSpool(MaxLogBytes) |
| 876 | if err != nil { |
| 877 | fmt.Fprintf(os.Stderr, "vole runner: %v\n", err) |
| 878 | return |
| 879 | } |
| 880 | defer log.Close() |
| 881 | // The values are masked as the log is written, so neither the file the streamer |
| 882 | // sends nor the instance's copy ever holds one. Errors the runner itself prints |
| 883 | // go through the same mask: an instance error or a runtime message is not |
| 884 | // supposed to contain a value, and this is what makes that true rather than |
| 885 | // assumed. |
| 886 | mask := redact.New(secretValues(job)) |
| 887 | log.SetMask(mask) |
| 888 | stream := s.startLogStream(job, log) |
| 889 | defer stream.stop() |
| 890 | co, err := prepareWorktree(remote, cache, job.Repo, job.SHA, job.ID, job.Attempt) |
| 891 | if err != nil { |
| 892 | // No checkout means no branch to commit a record to, so there is nothing to |
| 893 | // report: the lease expires and the queue gives the job to a runner that can. |
| 894 | fmt.Fprintf(os.Stderr, "vole runner: cannot prepare a checkout of %s at %s: %v\n", |
| 895 | job.Repo, format.SHA(job.SHA), err) |
| 896 | return |
| 897 | } |
| 898 | defer co.cleanup() |
| 899 | // The cache is held from here until after the container is gone, so eviction |
| 900 | // cannot take a namespace this job has mounted. The lock is taken before the |
| 901 | // container starts and released after it stops: the other order would leave a |
| 902 | // window where the tree is up for eviction but the job can still write to it. |
| 903 | cacheMounts, releaseCache := s.cacheFor(job) |
| 904 | defer releaseCache() |
| 905 | if job.Image == "" { |
| 906 | // A run queued before images were required has no environment to start. |
| 907 | // Fail it with a record rather than guessing at the host. |
| 908 | const noImage = "this run has no image; re-queue it from a workflow that declares image or image-file" |
| 909 | fmt.Fprintf(log, "\n%s\n", noImage) |
| 910 | s.finish(remote, co, job, log, stream, attemptOutcome{ |
| 911 | Exit: 1, Failure: runs.FailureSetup, Message: noImage, |
| 912 | }, box) |
| 913 | return |
| 914 | } |
| 915 | jobBox := &jobSandbox{} |
| 916 | defer jobBox.Stop() |
| 917 | // The steps that ran, so a failure after them keeps what did run and a failure |
| 918 | // before them can say that nothing did. |
| 919 | var steps []runs.StepResult |
| 920 | fail := func(format string, args ...any) { |
| 921 | jobBox.StopWithLogs(log) |
| 922 | msg := mask.String(fmt.Sprintf(format, args...)) |
| 923 | fmt.Fprintf(os.Stderr, "vole runner: %s\n", msg) |
| 924 | fmt.Fprintf(log, "\n%s\n", msg) |
| 925 | out := attemptOutcome{Exit: 1, Steps: steps} |
| 926 | if len(steps) == 0 { |
| 927 | // Nothing was tested. The attempt stopped in the runner's own setup (the |
| 928 | // image build, the container, the network) or in a definition the workflow |
| 929 | // got wrong; either way the commit is not what failed, so the record says |
| 930 | // where it stopped and why. |
| 931 | out.Failure = runs.FailureSetup |
| 932 | out.Message = failureLine(msg) |
| 933 | } |
| 934 | s.finish(remote, co, job, log, stream, out, box) |
| 935 | } |
| 936 | // A job that asked to act as an account it cannot act as is broken as written, not |
| 937 | // degraded: fail it before the image is pulled and say why. |
| 938 | if job.CredentialError != "" { |
| 939 | fail("%s", job.CredentialError) |
| 940 | return |
| 941 | } |
| 942 | fmt.Fprintf(log, "=== %s · %s · %s @ %s\n", job.Repo, job.Job, job.Branch, format.SHA(job.SHA)) |
| 943 | // The job's network request is resolved before anything runs, so a runner that was |
| 944 | // handed a job it cannot isolate refuses it instead of leaking the network into a |
| 945 | // job that asked for none, or starving a job that asked for one. |
| 946 | boxConfig, err := box.networkFor(job) |
| 947 | if err != nil { |
| 948 | fail("%s", err) |
| 949 | return |
| 950 | } |
| 951 | fmt.Fprintf(log, "environment: %s via %s, network %s\n", job.Image, box.Runtime, boxConfig.Network) |
| 952 | // A declared secret the repository does not have is not a failure: the variable |
| 953 | // is left unset and the log says which one, because a workflow is allowed an |
| 954 | // optional secret and a required one is ${NAME:?} in the step. |
| 955 | for _, issue := range job.Missing { |
| 956 | if issue.Reason != "" { |
| 957 | fmt.Fprintf(log, "vole: %s cannot be read (%s), so this job runs without it.\n", |
| 958 | issue.Name, issue.Reason) |
| 959 | continue |
| 960 | } |
| 961 | fmt.Fprintf(log, "vole: %s is not set for this repository, so this job runs without it.\n", issue.Name) |
| 962 | } |
| 963 | |
| 964 | image, err := s.resolveImage(job, box, log, lease) |
| 965 | if err != nil { |
| 966 | if errors.Is(err, errLeaseLost) { |
| 967 | fmt.Fprintf(os.Stderr, "vole runner: lease lost while building %s; stopping\n", job.Image) |
| 968 | return |
| 969 | } |
| 970 | if errors.Is(err, errCanceled) { |
| 971 | // The build was stopped on a heartbeat, and the log already says why. The |
| 972 | // attempt still has a story: report it with the work that did happen. |
| 973 | s.finish(remote, co, job, log, stream, attemptOutcome{Exit: 1}, box) |
| 974 | return |
| 975 | } |
| 976 | fail("%s", err) |
| 977 | return |
| 978 | } |
| 979 | // Services are containers beside the job on a network of their own. They start |
| 980 | // after the image is resolved so a slow build does not leave a database running, |
| 981 | // and before the job so their names resolve for the first step. |
| 982 | if len(job.Services) > 0 { |
| 983 | netName := serviceNetworkName(box, job) |
| 984 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 985 | net, err := container.CreateNetwork(ctx, box.Config, netName, box.labels(job)) |
| 986 | cancel() |
| 987 | if err != nil { |
| 988 | fail("%s", err) |
| 989 | return |
| 990 | } |
| 991 | jobBox.network = net |
| 992 | serviceConfig := boxConfig |
| 993 | serviceConfig.Network = netName |
| 994 | boxConfig.Network = netName |
| 995 | ids := make([]string, 0, len(job.Services)) |
| 996 | for _, svc := range job.Services { |
| 997 | c, err := container.StartService(serviceConfig, container.ServiceSpec{ |
| 998 | Name: serviceContainerName(box, job, svc.ID), |
| 999 | Alias: svc.ID, |
| 1000 | Image: svc.Image, |
| 1001 | Env: svc.Env, |
| 1002 | Labels: box.labels(job), |
| 1003 | }) |
| 1004 | if err != nil { |
| 1005 | fail("%s", err) |
| 1006 | return |
| 1007 | } |
| 1008 | jobBox.services = append(jobBox.services, c) |
| 1009 | jobBox.serviceIDs = append(jobBox.serviceIDs, svc.ID) |
| 1010 | ids = append(ids, svc.ID+" ("+svc.Image+")") |
| 1011 | } |
| 1012 | fmt.Fprintf(log, "services: %s\n", strings.Join(ids, ", ")) |
| 1013 | for i, c := range jobBox.services { |
| 1014 | running, err := c.Running(context.Background()) |
| 1015 | if err != nil { |
| 1016 | fail("service %s: %s", job.Services[i].ID, err) |
| 1017 | return |
| 1018 | } |
| 1019 | if !running { |
| 1020 | fail("service %s exited before the steps started", job.Services[i].ID) |
| 1021 | return |
| 1022 | } |
| 1023 | } |
| 1024 | } |
| 1025 | // The run directory is mounted at /run/vole for the pinned event payload and for |
| 1026 | // the job's secrets. Both arrive as files rather than as interpolated commands: a |
| 1027 | // large issue body cannot make an argument oversized, and a secret never becomes |
| 1028 | // an argument at all. |
| 1029 | mounts := cacheMounts |
| 1030 | env := job.Env |
| 1031 | runDir := "" |
| 1032 | if job.EventPayload != "" || len(job.Secrets) > 0 || job.Credential != nil { |
| 1033 | runDir = filepath.Join(filepath.Dir(co.Work), "run-"+job.ID+"-"+fmt.Sprint(job.Attempt)) |
| 1034 | if err := os.MkdirAll(runDir, 0o700); err != nil { |
| 1035 | fail("cannot prepare the run directory: %s", err) |
| 1036 | return |
| 1037 | } |
| 1038 | defer os.RemoveAll(runDir) |
| 1039 | mounts = append(mounts, container.Mount{Host: runDir, Path: "/run/vole"}) |
| 1040 | } |
| 1041 | if job.EventPayload != "" { |
| 1042 | if err := os.WriteFile(filepath.Join(runDir, "event.json"), []byte(job.EventPayload), 0o600); err != nil { |
| 1043 | fail("cannot write the event payload: %s", err) |
| 1044 | return |
| 1045 | } |
| 1046 | env = make(map[string]string, len(job.Env)+3) |
| 1047 | for name, value := range job.Env { |
| 1048 | env[name] = value |
| 1049 | } |
| 1050 | env["VOLE_EVENT"] = job.Trigger |
| 1051 | env["VOLE_EVENT_ACTION"] = job.TriggerAction |
| 1052 | env["VOLE_EVENT_PATH"] = "/run/vole/event.json" |
| 1053 | } |
| 1054 | secretsFile := "" |
| 1055 | if len(job.Secrets) > 0 { |
| 1056 | script, err := secretsScript(job.Secrets) |
| 1057 | if err != nil { |
| 1058 | fail("%s", err) |
| 1059 | return |
| 1060 | } |
| 1061 | if err := os.WriteFile(filepath.Join(runDir, "secrets.env"), script, 0o600); err != nil { |
| 1062 | fail("cannot prepare the secrets: %s", err) |
| 1063 | return |
| 1064 | } |
| 1065 | secretsFile = "/run/vole/secrets.env" |
| 1066 | } |
| 1067 | if job.Credential != nil { |
| 1068 | // The key is a file, not a value in the environment, so `git push` finds it by |
| 1069 | // path and no step has to interpolate it. The known_hosts entry comes from the |
| 1070 | // lease: the job cannot scan the host without trusting whatever answers. |
| 1071 | if env, err = withCredentialEnv(env, remote, job); err != nil { |
| 1072 | fail("%s", err) |
| 1073 | return |
| 1074 | } |
| 1075 | if err := os.WriteFile(filepath.Join(runDir, "id_bot"), []byte(job.Credential.PrivateKey), 0o600); err != nil { |
| 1076 | fail("cannot write the run credential: %s", err) |
| 1077 | return |
| 1078 | } |
| 1079 | known, err := knownHostsLine(remote, job.HostKey) |
| 1080 | if err != nil { |
| 1081 | fail("%s", err) |
| 1082 | return |
| 1083 | } |
| 1084 | if err := os.WriteFile(filepath.Join(runDir, "known_hosts"), []byte(known), 0o644); err != nil { |
| 1085 | fail("cannot write the known_hosts entry: %s", err) |
| 1086 | return |
| 1087 | } |
| 1088 | fmt.Fprintf(log, "acting as %s for this run\n", job.Bot) |
| 1089 | } |
| 1090 | if s.cacheRoot != "" && !s.cacheReuse { |
| 1091 | defer func() { |
| 1092 | // The namespace is held exclusively for this job and this defer runs before |
| 1093 | // the lock is released, so nothing else is inside it while it is emptied. |
| 1094 | // Only the cache mount: the run directory is the job's own and is removed |
| 1095 | // by the defer above. |
| 1096 | for _, m := range cacheMounts { |
| 1097 | _ = os.RemoveAll(m.Host) |
| 1098 | } |
| 1099 | }() |
| 1100 | } |
| 1101 | sandbox, err := container.Start(boxConfig, container.Spec{ |
| 1102 | Image: image, |
| 1103 | Name: containerName(box, job), |
| 1104 | Work: co.Work, |
| 1105 | Labels: box.labels(job), |
| 1106 | Mounts: mounts, |
| 1107 | Env: env, |
| 1108 | SecretsFile: secretsFile, |
| 1109 | }) |
| 1110 | if err != nil { |
| 1111 | fail("%s", err) |
| 1112 | return |
| 1113 | } |
| 1114 | jobBox.job = sandbox |
| 1115 | |
| 1116 | steps = make([]runs.StepResult, 0, len(job.Steps)) |
| 1117 | // The job's budget starts with its first step. An image build has its own cap, so a |
| 1118 | // slow pull must not eat the time the steps were given. |
| 1119 | var deadline time.Time |
| 1120 | if d, err := time.ParseDuration(job.Timeout); err == nil && d > 0 { |
| 1121 | deadline = time.Now().Add(d) |
| 1122 | } |
| 1123 | for _, step := range job.Steps { |
| 1124 | // The step's section of the log starts at its header, so a reader can open |
| 1125 | // one step without holding the image build that ran before it. |
| 1126 | logStart := log.Len() |
| 1127 | started := time.Now().UTC() |
| 1128 | fmt.Fprintf(log, "\n--- %s\n", step.Name) |
| 1129 | own, _ := time.ParseDuration(step.Timeout) |
| 1130 | limit := ci.StepLimit(ci.BudgetLeft(deadline, time.Now()), own) |
| 1131 | code, lost, canceled := 0, false, false |
| 1132 | if limit <= 0 { |
| 1133 | // The budget is spent, so the step must not start. Reporting it as a failed |
| 1134 | // step keeps the record the same shape as any other failure. |
| 1135 | fmt.Fprintf(log, "\nthe job's %s budget ran out before this step\n", job.Timeout) |
| 1136 | code = 1 |
| 1137 | } else { |
| 1138 | // Renew before the step, not only during it: a step heartbeats while it runs, |
| 1139 | // but its ticker starts with the step. Steps that each finish inside one |
| 1140 | // interval never heartbeat at all, so an attempt whose steps add up to more |
| 1141 | // than the lease expires between two of them and the next heartbeat arrives |
| 1142 | // to find the queue has already given the job away. |
| 1143 | if !s.heartbeat(job.ID) { |
| 1144 | fmt.Fprintf(os.Stderr, "vole runner: lease lost before %q; stopping\n", step.Name) |
| 1145 | return |
| 1146 | } |
| 1147 | code, lost, canceled = s.execStep(job.ID, sandbox, step, limit, lease, log) |
| 1148 | } |
| 1149 | // The mask holds back a few bytes so a value split across writes is caught; |
| 1150 | // flushing here keeps this step's byte range its own. |
| 1151 | log.Flush() |
| 1152 | result := runs.StepResult{ |
| 1153 | Name: step.Name, Started: started, Finished: time.Now().UTC(), Exit: code, |
| 1154 | LogOffset: logStart, LogSize: log.Len() - logStart, |
| 1155 | } |
| 1156 | if lost { |
| 1157 | fmt.Fprintf(os.Stderr, "vole runner: lease lost while running %q; stopping\n", step.Name) |
| 1158 | return // no finish: the queue already gave it to somebody else |
| 1159 | } |
| 1160 | if canceled { |
| 1161 | // The heartbeat said a maintainer asked the run to stop, and runCommand |
| 1162 | // already wrote the line to the log. Keep the partial attempt: the log and |
| 1163 | // the timings are what a reader asks for first after "what had it done". |
| 1164 | steps = append(steps, result) |
| 1165 | jobBox.StopWithLogs(log) |
| 1166 | s.finish(remote, co, job, log, stream, attemptOutcome{Exit: code, Steps: steps}, box) |
| 1167 | return |
| 1168 | } |
| 1169 | if code != 0 { |
| 1170 | fmt.Fprintf(log, "\nstep %q exited %d\n", step.Name, code) |
| 1171 | result.LogSize = log.Len() - logStart |
| 1172 | result.Finished = time.Now().UTC() |
| 1173 | steps = append(steps, result) |
| 1174 | jobBox.StopWithLogs(log) |
| 1175 | s.finish(remote, co, job, log, stream, attemptOutcome{Exit: code, Steps: steps}, box) |
| 1176 | return |
| 1177 | } |
| 1178 | steps = append(steps, result) |
| 1179 | } |
| 1180 | // The steps are done. Stop the container before reading artifacts or deploying a |
| 1181 | // site, so a process the workflow left in the background cannot change the files |
| 1182 | // while the runner is publishing them. |
| 1183 | jobBox.StopJob() |
| 1184 | arts, releaseArts, err := publishArtifacts(remote, co.Work, job, log) |
| 1185 | if err != nil { |
| 1186 | fail("%s", err) |
| 1187 | return |
| 1188 | } |
| 1189 | if len(releaseArts) > 0 { |
| 1190 | // The pointers ride the record's push, but a declaration that cannot name a |
| 1191 | // release is a mistake worth failing on before the push rather than after it. |
| 1192 | if err := checkReleaseJob(job); err != nil { |
| 1193 | fail("%s", err) |
| 1194 | return |
| 1195 | } |
| 1196 | } |
| 1197 | if err := publishSite(remote, co, job, log); err != nil { |
| 1198 | fail("%s", err) |
| 1199 | return |
| 1200 | } |
| 1201 | s.finish(remote, co, job, log, stream, attemptOutcome{ |
| 1202 | Steps: steps, Artifacts: arts, ReleaseArtifacts: releaseArts, |
| 1203 | }, box) |
| 1204 | } |
| 1205 | |
| 1206 | // cacheFor returns the mount for one job's cache and the lock that keeps eviction |
| 1207 | // off the namespace until release is called. The release function is always |
| 1208 | // non-nil, so a caller can defer it without asking whether there is a cache: a |
| 1209 | // runner with none returns no mounts and a no-op release. |
| 1210 | // |
| 1211 | // It sets no environment: a job opts in by pointing its own variables at Mount. The |
| 1212 | // namespace separates repositories from each other, and contributor branches from |
| 1213 | // everything else: a cache is shared mutable state, and a branch nobody has vouched |
| 1214 | // for must not be able to seed the one a protected branch builds into. |
| 1215 | func (s *runnerSession) cacheFor(job *runnerJob) ([]container.Mount, func()) { |
| 1216 | if s.cacheRoot == "" { |
| 1217 | return nil, func() {} |
| 1218 | } |
| 1219 | // A contributor branch is untrusted code, and a cache is shared mutable state: |
| 1220 | // one contributor's branch must not be able to seed another's cache or the one a |
| 1221 | // protected branch builds into. Namespacing by account means a contributor's |
| 1222 | // cache follows them across their branches and nobody else can write it. |
| 1223 | trust := access.TrustNamespace(job.Ref) |
| 1224 | ns := jobcache.Namespace(s.cacheRoot, job.Repo, trust, jobcache.ImageKey(job.ImageSHA, job.Image)) |
| 1225 | // Take the lock before creating the directory, so an eviction between the two |
| 1226 | // cannot remove what this job has just made. The shared lock is what lets two |
| 1227 | // jobs of the same repository run at once while eviction passes over both; |
| 1228 | // --no-cache takes it exclusively, because emptying the namespace afterwards must |
| 1229 | // not land on somebody else's build. |
| 1230 | var release func() |
| 1231 | var err error |
| 1232 | if s.cacheReuse { |
| 1233 | release, err = jobcache.Lock(s.cacheRoot, ns) |
| 1234 | } else { |
| 1235 | release, err = jobcache.LockExclusive(s.cacheRoot, ns) |
| 1236 | } |
| 1237 | if err != nil { |
| 1238 | fmt.Fprintf(os.Stderr, "vole runner: cannot lock a cache for %s: %v\n", job.Repo, err) |
| 1239 | return nil, func() {} |
| 1240 | } |
| 1241 | if err := os.MkdirAll(ns, 0o755); err != nil { |
| 1242 | release() |
| 1243 | fmt.Fprintf(os.Stderr, "vole runner: cannot prepare a cache for %s: %v\n", job.Repo, err) |
| 1244 | return nil, func() {} |
| 1245 | } |
| 1246 | return []container.Mount{{Host: ns, Path: jobcache.Mount}}, release |
| 1247 | } |
| 1248 | |
| 1249 | // trimJobCache bounds the job cache until the runner stops. Each pass skips a |
| 1250 | // namespace a job has mounted; the next pass tries again, so the cap is a bound on |
| 1251 | // what the runner keeps rather than a promise about the moment a job ends. |
| 1252 | func trimJobCache(ctx context.Context, root string, max int64) { |
| 1253 | t := time.NewTicker(cacheTrimInterval) |
| 1254 | defer t.Stop() |
| 1255 | for { |
| 1256 | select { |
| 1257 | case <-ctx.Done(): |
| 1258 | return |
| 1259 | case <-t.C: |
| 1260 | if removed, err := jobcache.Trim(root, max); err == nil && removed > 0 { |
| 1261 | fmt.Fprintf(os.Stderr, "vole runner: trimmed %s from the job cache\n", format.Bytes(removed)) |
| 1262 | } |
| 1263 | } |
| 1264 | } |
| 1265 | } |
| 1266 | |
| 1267 | // attemptOutcome is what the runner has to say about a finished attempt: the exit |
| 1268 | // code, the steps that ran, what it published, and — when the attempt stopped before |
| 1269 | // its first step — where it stopped and the runner's own reason. A setup failure is |
| 1270 | // not a verdict about the commit, so it is carried apart from the exit code rather |
| 1271 | // than folded into it. |
| 1272 | type attemptOutcome struct { |
| 1273 | Exit int |
| 1274 | Steps []runs.StepResult |
| 1275 | Artifacts []runnerArtifact |
| 1276 | ReleaseArtifacts []runnerArtifact |
| 1277 | // Failure is empty when a step decided the result, or runs.FailureSetup when no |
| 1278 | // step ever started. Message is the one-line reason that goes with it, already |
| 1279 | // masked. |
| 1280 | Failure string |
| 1281 | Message string |
| 1282 | } |
| 1283 | |
| 1284 | // failureLine is the one line a record keeps for a failure that never reached a |
| 1285 | // step. The message is already masked, so this only makes it display text: one line, |
| 1286 | // bounded to what a record accepts. |
| 1287 | func failureLine(msg string) string { |
| 1288 | line := strings.Join(strings.Fields(msg), " ") |
| 1289 | if len(line) > runs.MaxFailureMessage { |
| 1290 | line = strings.ToValidUTF8(line[:runs.MaxFailureMessage], "") |
| 1291 | } |
| 1292 | return line |
| 1293 | } |
| 1294 | |
| 1295 | // finish uploads this attempt's log, sends its record to the server with the |
| 1296 | // finish message, and then attaches any release pointers the build published. |
| 1297 | // |
| 1298 | // The bytes go in first, the same rule artifacts follow: a record is a set of |
| 1299 | // pointers, and the server refuses one whose log or artifact is not in the store. |
| 1300 | // The record is not a commit; the server stores it with the run's result in one |
| 1301 | // transaction. A tagged build's release pointers are product metadata and are pushed |
| 1302 | // afterwards, so a release cannot exist without the build that produced it. |
| 1303 | // |
| 1304 | // A log that cannot be stored does not lose the result: the record then names no log |
| 1305 | // and the failure is reported on stderr rather than swallowed. A record the server |
| 1306 | // does not accept is not reported at all. |
| 1307 | func (s *runnerSession) finish(remote string, co checkout, job *runnerJob, log *spool, stream *logStream, out attemptOutcome, box runnerBox) { |
| 1308 | if err := log.seal(); err != nil { |
| 1309 | fmt.Fprintf(os.Stderr, "vole runner: could not seal the log for %s: %v\n", job.ID, err) |
| 1310 | } |
| 1311 | streamErr := stream.drain() |
| 1312 | if streamErr != nil { |
| 1313 | fmt.Fprintf(os.Stderr, "vole runner: log stream for %s failed: %v\n", job.ID, streamErr) |
| 1314 | // The step ranges describe bytes the server does not have. Keep the timings, |
| 1315 | // which are still true, and drop the claims about the log. |
| 1316 | for i := range out.Steps { |
| 1317 | out.Steps[i].LogOffset, out.Steps[i].LogSize = 0, 0 |
| 1318 | } |
| 1319 | } |
| 1320 | artifacts := make([]runs.Artifact, 0, len(out.Artifacts)) |
| 1321 | for _, a := range out.Artifacts { |
| 1322 | artifacts = append(artifacts, runs.Artifact{Name: a.Name, OID: a.OID, Size: a.Size}) |
| 1323 | } |
| 1324 | rec := runs.Record{ |
| 1325 | RunID: job.ID, Attempt: job.Attempt, Repo: job.Repo, Ref: job.Ref, SHA: job.SHA, |
| 1326 | Workflow: job.Workflow, WorkflowSHA: job.WorkflowSHA, Job: job.Job, |
| 1327 | StepsSHA: job.StepsSHA, Exit: out.Exit, Finished: time.Now().UTC(), |
| 1328 | Signer: s.name, Steps: out.Steps, Artifacts: artifacts, |
| 1329 | Failure: out.Failure, FailureMessage: out.Message, |
| 1330 | // The machine this attempt ran on. The runner binary's own build target, |
| 1331 | // which is the honest thing it can say: a live probe would add a subprocess |
| 1332 | // and a platform quirk for a fact the binary already knows. |
| 1333 | OS: runtime.GOOS, Arch: runtime.GOARCH, |
| 1334 | } |
| 1335 | rec.Image, rec.ImageFile, rec.ImageSHA = job.Image, job.ImageFile, job.ImageSHA |
| 1336 | rec.Runtime = box.Runtime |
| 1337 | // The record says which version of each secret the job ran with, never the value. |
| 1338 | // The names are sorted so the same attempt reported twice produces the same bytes. |
| 1339 | for name, delivered := range job.Secrets { |
| 1340 | rec.Secrets = append(rec.Secrets, runs.SecretRef{Name: name, Version: delivered.Version}) |
| 1341 | } |
| 1342 | sort.Slice(rec.Secrets, func(i, j int) bool { return rec.Secrets[i].Name < rec.Secrets[j].Name }) |
| 1343 | if streamErr == nil { |
| 1344 | rec.LogSize = log.Len() |
| 1345 | } |
| 1346 | |
| 1347 | // Keep the lease through the report: the record is checked against it, and a |
| 1348 | // slow upload must not turn a finished job into one that expired mid-report. |
| 1349 | if rep, err := s.call(runnerRequest{Op: "heartbeat", Run: job.ID}); err != nil { |
| 1350 | // The channel is gone, so there is no way to send the record. The queue |
| 1351 | // will hand the job to another runner after the lease expires. |
| 1352 | fmt.Fprintf(os.Stderr, "vole runner: lost the connection while finishing %s: %v\n", job.ID, err) |
| 1353 | return |
| 1354 | } else if rep.Error != "" { |
| 1355 | // The server could not answer for the lease, so there is no evidence that |
| 1356 | // it is still ours. A record sent now would be a guess. |
| 1357 | fmt.Fprintf(os.Stderr, "vole runner: cannot renew the lease for %s: %s\n", job.ID, rep.Error) |
| 1358 | return |
| 1359 | } else if !rep.OK { |
| 1360 | // The lease expired while the last step was running. The queue may not |
| 1361 | // have given the job to anyone else yet, and this attempt has a real |
| 1362 | // result. Send it: the server accepts it only while this is still the |
| 1363 | // attempt the run is on, and refuses it once another runner leases the job. |
| 1364 | fmt.Fprintf(os.Stderr, "vole runner: the lease for %s expired before the result was reported; sending it anyway\n", job.ID) |
| 1365 | } |
| 1366 | |
| 1367 | // The record goes to the server on the same channel that holds the lease. If |
| 1368 | // this call is lost after the server stored the record, the retry sends the |
| 1369 | // same bytes and the server answers ok again; a run record is append-only. |
| 1370 | rep, err := s.call(runnerRequest{Op: "finish", Run: job.ID, Record: &rec}) |
| 1371 | switch { |
| 1372 | case err != nil: |
| 1373 | fmt.Fprintf(os.Stderr, "vole runner: connection lost while reporting %s; the result was not stored\n", job.ID) |
| 1374 | return |
| 1375 | case rep.Error != "": |
| 1376 | fmt.Fprintf(os.Stderr, "vole runner: %s\n", rep.Error) |
| 1377 | return |
| 1378 | case !rep.OK: |
| 1379 | fmt.Fprintf(os.Stderr, "vole runner: the lease for %s moved; the result was not stored\n", job.ID) |
| 1380 | return |
| 1381 | } |
| 1382 | |
| 1383 | // The result is stored. A tagged build's release pointers are product metadata |
| 1384 | // and still live on the branch; attach them only after the record, so a release |
| 1385 | // cannot exist without the build that produced it. |
| 1386 | if len(out.ReleaseArtifacts) > 0 { |
| 1387 | if err := publishReleases(remote, co, job, out.ReleaseArtifacts, log); err != nil { |
| 1388 | fmt.Fprintf(os.Stderr, "vole runner: could not attach %d artifact(s) to %s: %v\n", |
| 1389 | len(out.ReleaseArtifacts), job.Tag, err) |
| 1390 | fmt.Fprintf(log, "\nrelease: could not attach %d artifact(s) to %s: %v\n", |
| 1391 | len(out.ReleaseArtifacts), job.Tag, err) |
| 1392 | } |
| 1393 | } |
| 1394 | } |
| 1395 | |
| 1396 | // publishReleases attaches a tagged build's release pointers to the metadata |
| 1397 | // branch. The run record is already in the database, so a failure here loses the |
| 1398 | // release attachment, not the result; the operator can attach it again with |
| 1399 | // `vole release attach`. |
| 1400 | // |
| 1401 | // The branch is pushed once, and a lost race against somebody else's issue is |
| 1402 | // re-applied on the new tip the same way every other metadata write is. The |
| 1403 | // runner's trusted cache is used, never the job tree: the workflow may have edited |
| 1404 | // the job tree's .git, and a push from there could run a hook the workflow wrote. |
| 1405 | func publishReleases(remote string, co checkout, job *runnerJob, rel []runnerArtifact, log io.Writer) error { |
| 1406 | url, err := client.RepoURL(remote, job.Repo) |
| 1407 | if err != nil { |
| 1408 | return err |
| 1409 | } |
| 1410 | // Opening the branch fetches into the trusted cache and the write adds objects to |
| 1411 | // it. Hold the cache's lock so a repack or a deletion in another runner process |
| 1412 | // cannot move the object store while this reads and writes it. |
| 1413 | unlock, err := co.lock() |
| 1414 | if err != nil { |
| 1415 | return err |
| 1416 | } |
| 1417 | defer unlock() |
| 1418 | c, err := client.OpenBranch(co.GitDir, url, strings.TrimPrefix(metadata.Ref, "refs/heads/"), "vole-runner") |
| 1419 | if err != nil { |
| 1420 | return err |
| 1421 | } |
| 1422 | arts := make([]metadata.Artifact, 0, len(rel)) |
| 1423 | for _, a := range rel { |
| 1424 | arts = append(arts, metadata.Artifact{Name: a.Name, OID: a.OID, Size: a.Size}) |
| 1425 | } |
| 1426 | message := fmt.Sprintf("release: attach %d artifact(s) to %s from run %s", len(arts), job.Tag, job.ID) |
| 1427 | write := func() error { |
| 1428 | return metadata.WriteReleasePointers(c.Repo, metadata.Ref, job.Tag, arts, message, c.Author) |
| 1429 | } |
| 1430 | if err := write(); err != nil { |
| 1431 | return err |
| 1432 | } |
| 1433 | if err := c.CommitAndPush(false, write); err != nil { |
| 1434 | return err |
| 1435 | } |
| 1436 | fmt.Fprintf(log, "\nrelease: attached %d artifact(s) to %s\n", len(arts), job.Tag) |
| 1437 | return nil |
| 1438 | } |
| 1439 | |
| 1440 | // checkReleaseJob refuses a workflow that publishes to a release it cannot name. It |
| 1441 | // runs before the push so the refusal lands in the job's log, where somebody will see |
| 1442 | // it, rather than as a push error after a successful build. |
| 1443 | func checkReleaseJob(job *runnerJob) error { |
| 1444 | if job.Tag == "" { |
| 1445 | return fmt.Errorf("this job declares release artifacts but ran on a branch; a release " + |
| 1446 | "is built by a tag push (add on.push.tags and push a tag)") |
| 1447 | } |
| 1448 | if !job.TagAnnotated { |
| 1449 | return fmt.Errorf("%s is a lightweight tag; a release needs an annotated tag with notes "+ |
| 1450 | "(git tag -a %s -F notes.md)", job.Tag, job.Tag) |
| 1451 | } |
| 1452 | return nil |
| 1453 | } |
| 1454 | |
| 1455 | // publishArtifacts expands the job's declared patterns in the checkout and uploads |
| 1456 | // what they match through the same SSH object transfer git-lfs uses. The bytes go |
| 1457 | // in before the finish, so the server can refuse a claim it cannot serve. |
| 1458 | // publishArtifacts uploads every declared output and returns them twice: once as |
| 1459 | // the run's outputs, once as the subset that belongs to the tag's release. |
| 1460 | func publishArtifacts(remote, work string, job *runnerJob, log io.Writer) (all, release []runnerArtifact, err error) { |
| 1461 | if len(job.Artifacts) == 0 { |
| 1462 | return nil, nil, nil |
| 1463 | } |
| 1464 | files, err := collectArtifactFiles(work, job.Artifacts) |
| 1465 | if err != nil { |
| 1466 | return nil, nil, err |
| 1467 | } |
| 1468 | if err := checkArtifactLimits(work, files); err != nil { |
| 1469 | return nil, nil, err |
| 1470 | } |
| 1471 | isRelease := map[string]bool{} |
| 1472 | for _, pat := range job.ReleaseArtifacts { |
| 1473 | isRelease[pat] = true |
| 1474 | } |
| 1475 | url, err := client.RepoURL(remote, job.Repo) |
| 1476 | if err != nil { |
| 1477 | return nil, nil, err |
| 1478 | } |
| 1479 | tx, err := client.DialTransfer(url, "upload") |
| 1480 | if err != nil { |
| 1481 | return nil, nil, fmt.Errorf("connect to publish %d artifact(s): %w", len(files), err) |
| 1482 | } |
| 1483 | defer tx.Close() |
| 1484 | |
| 1485 | all = make([]runnerArtifact, 0, len(files)) |
| 1486 | for _, f := range files { |
| 1487 | ptr, err := tx.Upload(filepath.Join(work, filepath.FromSlash(f.Path))) |
| 1488 | if err != nil { |
| 1489 | return nil, nil, fmt.Errorf("upload %s: %w", f.Path, err) |
| 1490 | } |
| 1491 | art := runnerArtifact{Name: f.Name, OID: ptr.OID, Size: ptr.Size} |
| 1492 | fmt.Fprintf(log, "\nartifact %s (%s) sha256:%s\n", f.Name, format.Bytes(ptr.Size), ptr.OID) |
| 1493 | all = append(all, art) |
| 1494 | if isRelease[f.Pattern] { |
| 1495 | release = append(release, art) |
| 1496 | } |
| 1497 | } |
| 1498 | return all, release, nil |
| 1499 | } |
| 1500 | |
| 1501 | // publishSite deploys the job's declared directory as the repository's site, using |
| 1502 | // the same snapshot code the pages CLI uses. |
| 1503 | // |
| 1504 | // The staleness check first: a site is a snapshot, so a run that finishes after a |
| 1505 | // newer commit has landed would quietly revert the site to an older build. Refusing |
| 1506 | // is the honest outcome — the log says why, and the job fails instead of publishing |
| 1507 | // a stale tree that nobody notices for a week. |
| 1508 | func publishSite(remote string, co checkout, job *runnerJob, log io.Writer) error { |
| 1509 | if job.Pages == nil || job.Pages.Path == "" { |
| 1510 | return nil |
| 1511 | } |
| 1512 | url, err := client.RepoURL(remote, job.Repo) |
| 1513 | if err != nil { |
| 1514 | return err |
| 1515 | } |
| 1516 | // The snapshot is committed and pushed from the trusted cache, which this runner |
| 1517 | // shares with any other process given the same --dir. Take its lock, as the |
| 1518 | // release push and the maintenance pass do. |
| 1519 | unlock, err := co.lock() |
| 1520 | if err != nil { |
| 1521 | return err |
| 1522 | } |
| 1523 | defer unlock() |
| 1524 | tip, err := client.LSRemote(co.GitDir, url, "refs/heads/"+job.Branch) |
| 1525 | if err != nil { |
| 1526 | return fmt.Errorf("check %s before deploying the site: %w", job.Branch, err) |
| 1527 | } |
| 1528 | switch { |
| 1529 | case tip == "": |
| 1530 | return fmt.Errorf("%s no longer exists, so there is no site to deploy to", job.Branch) |
| 1531 | case tip != job.SHA: |
| 1532 | return fmt.Errorf("this run built %s but %s is at %s now; the site was not deployed", |
| 1533 | format.SHA(job.SHA), job.Branch, format.SHA(tip)) |
| 1534 | } |
| 1535 | |
| 1536 | c, err := client.OpenBranch(co.GitDir, url, job.Pages.Ref, "vole-runner") |
| 1537 | if err != nil { |
| 1538 | return err |
| 1539 | } |
| 1540 | dir := filepath.Join(co.Work, filepath.FromSlash(job.Pages.Path)) |
| 1541 | msg := fmt.Sprintf("site: %s from run %s (%s)", job.Branch, format.SHA(job.SHA), job.ID) |
| 1542 | if len(msg) > 200 { |
| 1543 | msg = fmt.Sprintf("site: %s from run %s", job.Branch, job.ID) |
| 1544 | } |
| 1545 | diff, files, deployed, err := client.DeploySite(c, dir, msg, false) |
| 1546 | if err != nil { |
| 1547 | return err |
| 1548 | } |
| 1549 | if deployed { |
| 1550 | fmt.Fprintf(log, "\nsite: published %d file(s) to %s (%s)\n", files, job.Pages.Ref, diff) |
| 1551 | } else { |
| 1552 | fmt.Fprintf(log, "\nsite: %s already matches %s\n", job.Pages.Ref, job.Pages.Path) |
| 1553 | } |
| 1554 | return nil |
| 1555 | } |
| 1556 | |
| 1557 | // artifactFile is one matched output. Pattern is the declaration that matched it, |
| 1558 | // because a pattern can also mean "publish this to the release". |
| 1559 | type artifactFile struct { |
| 1560 | Path string // slash-separated, relative to the checkout |
| 1561 | Name string // the download name: the file's base name |
| 1562 | Pattern string // the declaration that matched |
| 1563 | } |
| 1564 | |
| 1565 | // collectArtifactFiles expands patterns against a checkout. A pattern that matches |
| 1566 | // nothing is an error rather than an empty artifact list: a build that stopped |
| 1567 | // producing the file is the failure a declaration is supposed to catch, and a |
| 1568 | // silent publish of nothing is how that goes unnoticed for a week. |
| 1569 | // checkArtifactLimits bounds what one run may publish. The workflow caps the number |
| 1570 | // of declarations; a declaration is a pattern, and a pattern can match a directory |
| 1571 | // tree, so the bound on what they expand to has to be here. An artifact set is |
| 1572 | // uploaded into the instance's object store, and a job choosing its size is a job |
| 1573 | // choosing the instance's disk. |
| 1574 | func checkArtifactLimits(work string, files []artifactFile) error { |
| 1575 | if len(files) > MaxRunArtifactFiles { |
| 1576 | return fmt.Errorf("this run publishes %d artifact files, limit %d; narrow the patterns", |
| 1577 | len(files), MaxRunArtifactFiles) |
| 1578 | } |
| 1579 | var total int64 |
| 1580 | for _, f := range files { |
| 1581 | info, err := os.Stat(filepath.Join(work, filepath.FromSlash(f.Path))) |
| 1582 | if err != nil { |
| 1583 | continue |
| 1584 | } |
| 1585 | total += info.Size() |
| 1586 | } |
| 1587 | if total > MaxRunArtifactBytes { |
| 1588 | return fmt.Errorf("this run publishes %s of artifacts, limit %s; narrow the patterns", |
| 1589 | format.Bytes(total), format.Bytes(MaxRunArtifactBytes)) |
| 1590 | } |
| 1591 | return nil |
| 1592 | } |
| 1593 | |
| 1594 | func collectArtifactFiles(root string, patterns []string) ([]artifactFile, error) { |
| 1595 | matched := make([]int, len(patterns)) |
| 1596 | var files []artifactFile |
| 1597 | byName := map[string]string{} |
| 1598 | err := filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error { |
| 1599 | if err != nil { |
| 1600 | return err |
| 1601 | } |
| 1602 | if d.IsDir() { |
| 1603 | if d.Name() == ".git" { |
| 1604 | return filepath.SkipDir |
| 1605 | } |
| 1606 | return nil |
| 1607 | } |
| 1608 | rel, err := filepath.Rel(root, p) |
| 1609 | if err != nil { |
| 1610 | return err |
| 1611 | } |
| 1612 | rel = filepath.ToSlash(rel) |
| 1613 | for i, pat := range patterns { |
| 1614 | if ok, err := path.Match(pat, rel); err != nil || !ok { |
| 1615 | continue |
| 1616 | } |
| 1617 | matched[i]++ |
| 1618 | info, err := d.Info() |
| 1619 | if err != nil { |
| 1620 | return err |
| 1621 | } |
| 1622 | if !info.Mode().IsRegular() { |
| 1623 | return nil // a symlink or device: not an artifact |
| 1624 | } |
| 1625 | name := path.Base(rel) |
| 1626 | if prev, dup := byName[name]; dup { |
| 1627 | return fmt.Errorf("two files would be published as %q (%s and %s); rename one "+ |
| 1628 | "or narrow the patterns", name, prev, rel) |
| 1629 | } |
| 1630 | byName[name] = rel |
| 1631 | files = append(files, artifactFile{Path: rel, Name: name, Pattern: pat}) |
| 1632 | return nil |
| 1633 | } |
| 1634 | return nil |
| 1635 | }) |
| 1636 | if err != nil { |
| 1637 | return nil, err |
| 1638 | } |
| 1639 | for i, n := range matched { |
| 1640 | if n == 0 { |
| 1641 | return nil, fmt.Errorf("artifact pattern %q matched nothing in the checkout", patterns[i]) |
| 1642 | } |
| 1643 | } |
| 1644 | sort.Slice(files, func(i, j int) bool { return files[i].Name < files[j].Name }) |
| 1645 | return files, nil |
| 1646 | } |
| 1647 | |
| 1648 | // execStep runs one shell step, heartbeating while it works. It returns the exit |
| 1649 | // code, whether the lease was lost, and whether a maintainer asked the run to stop. |
| 1650 | func (s *runnerSession) execStep(runID string, sandbox *container.Container, step runnerStep, limit time.Duration, lease time.Duration, out io.Writer) (int, bool, bool) { |
| 1651 | // limit is already resolved: the step's own timeout, or what is left of the job's |
| 1652 | // budget. The caller never passes zero. |
| 1653 | ctx, cancel := context.WithTimeout(context.Background(), limit) |
| 1654 | defer cancel() |
| 1655 | |
| 1656 | cmd := sandbox.Exec(ctx, step.Run) |
| 1657 | code, lost, canceled := s.runCommand(ctx, cmd, runID, lease, sandbox.Stop, out) |
| 1658 | if canceled { |
| 1659 | return code, false, true |
| 1660 | } |
| 1661 | if ctx.Err() == context.DeadlineExceeded { |
| 1662 | // The heartbeat will have extended the lease this far; report the timeout as |
| 1663 | // a failure rather than leaving the job hanging. |
| 1664 | // What is left of a budget is rarely a round number; a reader wants the promise, |
| 1665 | // not the microseconds the bookkeeping cost. |
| 1666 | fmt.Fprintf(out, "\nstep timed out after %s\n", limit.Round(time.Millisecond)) |
| 1667 | return 1, lost, false |
| 1668 | } |
| 1669 | return code, lost, false |
| 1670 | } |
| 1671 | |
| 1672 | // runCommand starts one host-side command and heartbeats while it runs. It is used |
| 1673 | // for docker exec and for building a custom image; onTimeout is what closes the |
| 1674 | // sandbox when the command runs out of time. A stop request from the server rides a |
| 1675 | // heartbeat, so the command is killed the same way a timeout kills it; the difference |
| 1676 | // is that the caller reports the partial attempt instead of a failure. |
| 1677 | func (s *runnerSession) runCommand(ctx context.Context, cmd *exec.Cmd, runID string, lease time.Duration, onTimeout func(), out io.Writer) (code int, lost bool, canceled bool) { |
| 1678 | // Own process group: a command that spawns children must not outlive its timeout. |
| 1679 | cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} |
| 1680 | cmd.Stdout, cmd.Stderr = out, out |
| 1681 | |
| 1682 | if err := cmd.Start(); err != nil { |
| 1683 | fmt.Fprintf(out, "cannot start: %v\n", err) |
| 1684 | return 1, false, false |
| 1685 | } |
| 1686 | done := make(chan error, 1) |
| 1687 | go func() { done <- cmd.Wait() }() |
| 1688 | |
| 1689 | interval := lease / 3 |
| 1690 | if interval < time.Second { |
| 1691 | interval = time.Second |
| 1692 | } |
| 1693 | tick := time.NewTicker(interval) |
| 1694 | defer tick.Stop() |
| 1695 | // The disk is checked on its own beat, not the heartbeat's: a build that writes |
| 1696 | // gigabytes a second must not get a whole lease interval to do it. |
| 1697 | disk := time.NewTicker(time.Second) |
| 1698 | defer disk.Stop() |
| 1699 | |
| 1700 | stop := func() { |
| 1701 | if onTimeout != nil { |
| 1702 | onTimeout() |
| 1703 | } |
| 1704 | if cmd.Process != nil { |
| 1705 | _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) |
| 1706 | } |
| 1707 | // Wait for the copy goroutines before the caller closes the log. |
| 1708 | <-done |
| 1709 | } |
| 1710 | |
| 1711 | for { |
| 1712 | select { |
| 1713 | case err := <-done: |
| 1714 | if err == nil { |
| 1715 | return 0, false, false |
| 1716 | } |
| 1717 | var ee *exec.ExitError |
| 1718 | if errors.As(err, &ee) { |
| 1719 | return ee.ExitCode(), false, false |
| 1720 | } |
| 1721 | fmt.Fprintf(out, "\n%v\n", err) |
| 1722 | return 1, false, false |
| 1723 | case <-ctx.Done(): |
| 1724 | stop() |
| 1725 | return 1, false, false |
| 1726 | case <-disk.C: |
| 1727 | free, low, err := diskLow(s.diskRoot, s.diskReserve) |
| 1728 | if err != nil { |
| 1729 | // One failed check is worth saying once; a stat on a directory that |
| 1730 | // vanished would otherwise fill the log with the same line. |
| 1731 | fmt.Fprintf(out, "\ncannot check the runner's free space: %v\n", err) |
| 1732 | disk.Stop() |
| 1733 | continue |
| 1734 | } |
| 1735 | if low { |
| 1736 | fmt.Fprintf(out, "\nthe runner has %s free, below its %s reserve; stopping the job "+ |
| 1737 | "so the machine does not fill (--disk-reserve)\n", |
| 1738 | format.Bytes(free), format.Bytes(s.diskReserve)) |
| 1739 | stop() |
| 1740 | return 1, false, false |
| 1741 | } |
| 1742 | case <-tick.C: |
| 1743 | rep, err := s.call(runnerRequest{Op: "heartbeat", Run: runID}) |
| 1744 | if err != nil || !rep.OK { |
| 1745 | return 1, true, false |
| 1746 | } |
| 1747 | if rep.Cancel != "" { |
| 1748 | fmt.Fprintf(out, "\nthe run was canceled by %s; stopping\n", rep.Cancel) |
| 1749 | stop() |
| 1750 | return 1, false, true |
| 1751 | } |
| 1752 | } |
| 1753 | } |
| 1754 | } |
| 1755 | |
| 1756 | // secretValues is what the log has to redact: the values this job was actually |
| 1757 | // given, not the names it asked for. |
| 1758 | func secretValues(job *runnerJob) []string { |
| 1759 | values := make([]string, 0, len(job.Secrets)+1) |
| 1760 | for _, delivered := range job.Secrets { |
| 1761 | values = append(values, delivered.Value) |
| 1762 | } |
| 1763 | // The run credential is a secret in the same sense: short-lived and scoped, but a |
| 1764 | // step that cats the key should not leave it in the log the instance keeps. |
| 1765 | if job.Credential != nil && job.Credential.PrivateKey != "" { |
| 1766 | values = append(values, job.Credential.PrivateKey) |
| 1767 | } |
| 1768 | return values |
| 1769 | } |
| 1770 | |
| 1771 | // secretsScript renders a job's delivered values as the file every step sources. |
| 1772 | // The rendering lives in ci/container because that is the package that sources it, |
| 1773 | // so a local run writes exactly the same file a runner does. |
| 1774 | func secretsScript(secrets map[string]runnerSecret) ([]byte, error) { |
| 1775 | values := make(map[string]string, len(secrets)) |
| 1776 | for name, delivered := range secrets { |
| 1777 | values[name] = delivered.Value |
| 1778 | } |
| 1779 | return container.SecretsEnv(values) |
| 1780 | } |
| 1781 |