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.
Run several jobs at once with --jobs
18c559ea0010dc22db13c889afda4fe9af3c15d7 · Keenan Nemetz · 3d ago · parent 91b47a22
Run several jobs at once with --jobs A runner process leased one job and blocked on it, so the only way to get a second job onto a machine was a second process: another host, another key, another allowlist entry, and another --dir that learns the repositories cold. The server was never the limit — a lease is taken per session and claimed by a conditional update — so this is the runner catching up. `--jobs N` is N slots in one process. Each slot owns a session and runs at most one job, and everything that belongs to the process happens once, before the slots start: the directories, the cache bounds, the leftover sweep, and the connections. A slot that refuses a job stops every slot, because reconnecting would ask the same instance the same question. A slot that loses its connection reconnects with a doubling delay instead, so one flaky link does not take the other slots' jobs with it; the job it was running is requeued when its lease expires. The log stream moves from the session to the job, which is what makes two jobs on one session impossible to confuse: a slot has one, and it is the stream for the job in front of it. Deployment: `RUNNER_JOBS` in the first-boot renderer, the packaged argument file, and the docs. The container limits are per job, so the machine has to carry --jobs times what they allow; max_jobs_per_repo is what stops one repository claiming every slot. Refs: 01M6SCWCSX39QXS38QAHQSMVGX (#216)
ci/agent/agent.go +260−63
| @@ -49,6 +49,14 @@ const leaseWait = 25 * time.Second | |||
| 49 | 49 | // a later pass, so a longer gap costs disk, not correctness. | |
| 50 | 50 | const cacheTrimInterval = 5 * time.Minute | |
| 51 | 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 | |||
| 52 | 60 | // errLeaseLost means the queue took the job away while the runner was still | |
| 53 | 61 | // preparing its environment. There is nothing to report in that case. | |
| 54 | 62 | var errLeaseLost = errors.New("lease lost") | |
| @@ -57,6 +65,15 @@ var errLeaseLost = errors.New("lease lost") | |||
| 57 | 65 | // reporting: a canceled run has a story, and the partial log and step timings are it. | |
| 58 | 66 | var errCanceled = errors.New("run canceled") | |
| 59 | 67 | ||
| 68 | // refusedJob says the instance handed this runner a job whose shape this runner does | ||
| 69 | // not accept. Reconnecting would ask the same instance the same question, so the | ||
| 70 | // error stops every slot and the process says why: a runner that kept looping would | ||
| 71 | // lease the same job again and refuse it again. | ||
| 72 | type refusedJob struct{ err error } | ||
| 73 | |||
| 74 | func (r refusedJob) Error() string { return r.err.Error() } | ||
| 75 | func (r refusedJob) Unwrap() error { return r.err } | ||
| 76 | |||
| 60 | 77 | type runnerStep struct { | |
| 61 | 78 | Name string `json:"name"` | |
| 62 | 79 | Run string `json:"run"` | |
| @@ -240,8 +257,13 @@ type Options struct { | |||
| 240 | 257 | // Labels describe the machine. A job is only leased to a runner carrying every | |
| 241 | 258 | // label the job asked for with runs-on. | |
| 242 | 259 | Labels []string | |
| 243 | // Once runs one job and exits, which is what a scheduled container wants. | ||
| 260 | // Once runs one job per slot and exits, which is what a scheduled container wants. | ||
| 244 | 261 | Once bool | |
| 262 | // Jobs is how many jobs one process may run at once, one per session. Zero or | ||
| 263 | // negative means one. Each slot holds its container limits of its own, so the | ||
| 264 | // machine has to carry this many times what --container-memory and the rest | ||
| 265 | // allow. | ||
| 266 | Jobs int | ||
| 245 | 267 | // ContainerRuntime is the CLI used to run containers: docker, podman, or nerdctl. | |
| 246 | 268 | // A runner always requires one; the image itself comes from each job's workflow. | |
| 247 | 269 | ContainerRuntime string | |
| @@ -276,12 +298,23 @@ type Options struct { | |||
| 276 | 298 | GitCacheMaxBytes int64 | |
| 277 | 299 | } | |
| 278 | 300 | ||
| 279 | // Run leases and executes CI jobs until the connection fails, the context ends, | ||
| 280 | // or (with Once) one job has been handled. | ||
| 301 | // Run leases and executes CI jobs until the context ends, a slot is told to stop, | ||
| 302 | // or (with Once) every slot has handled one job. | ||
| 303 | // | ||
| 304 | // Everything that belongs to the process happens here, once: the directories, the | ||
| 305 | // cache bounds, the leftovers of a killed predecessor, and the connections. The | ||
| 306 | // slots then share all of it and differ in nothing but their session, which is what | ||
| 307 | // lets --jobs run several jobs on one machine without each of them repeating the | ||
| 308 | // startup work or racing it. | ||
| 281 | 309 | func Run(ctx context.Context, opts Options) error { | |
| 282 | 310 | if opts.Remote == "" { | |
| 283 | 311 | return errors.New("--remote is required, e.g. --remote ssh://git@example.com:2222") | |
| 284 | 312 | } | |
| 313 | // Everything below is one process's work, so one cancellation covers it: the | ||
| 314 | // cache trimmer and the slots stop together when a slot fails for a reason the | ||
| 315 | // others cannot fix. | ||
| 316 | ctx, cancel := context.WithCancel(ctx) | ||
| 317 | defer cancel() | ||
| 285 | 318 | cache := opts.Dir | |
| 286 | 319 | if cache == "" { | |
| 287 | 320 | base, err := os.UserCacheDir() | |
| @@ -350,55 +383,199 @@ func Run(ctx context.Context, opts Options) error { | |||
| 350 | 383 | } | |
| 351 | 384 | } | |
| 352 | 385 | ||
| 353 | sess, err := dialRunner(opts.Remote) | ||
| 354 | if err != nil { | ||
| 355 | return err | ||
| 386 | jobs := opts.Jobs | ||
| 387 | if jobs < 1 { | ||
| 388 | jobs = 1 | ||
| 356 | 389 | } | |
| 357 | defer sess.close() | ||
| 358 | 390 | ||
| 359 | // Ask who the server thinks this is, so the runner can say whose results it reports | ||
| 360 | // and check its own --repos filter against what the account may run. | ||
| 361 | who, err := sess.call(runnerRequest{Op: "ping"}) | ||
| 362 | if err != nil { | ||
| 363 | return err | ||
| 364 | } | ||
| 365 | if who.Error != "" { | ||
| 366 | return errors.New(who.Error) | ||
| 367 | } | ||
| 368 | // --repos narrows the account's grant. Check it against the grant now: a filter that | ||
| 369 | // matches nothing is indistinguishable from a quiet instance, so it would idle | ||
| 370 | // forever while looking healthy — cheaper to find at startup than after a build that | ||
| 371 | // never arrives. | ||
| 372 | if err := checkRepoFilter(opts.Repos, who); err != nil { | ||
| 373 | return err | ||
| 391 | // One session per slot, all connected before the first lease: a slot that cannot | ||
| 392 | // reach the instance is a startup failure, not a job that fails later and less | ||
| 393 | // clearly. | ||
| 394 | slots := make([]*slot, 0, jobs) | ||
| 395 | for i := 0; i < jobs; i++ { | ||
| 396 | sl := &slot{opts: opts, cache: cache, box: box, cacheRoot: cacheRoot, gitCacheMax: gitCacheMax} | ||
| 397 | if err := sl.connect(); err != nil { | ||
| 398 | closeSlots(slots) | ||
| 399 | return err | ||
| 400 | } | ||
| 401 | // --repos narrows the account's grant. Check it against the grant now: a filter | ||
| 402 | // that matches nothing is indistinguishable from a quiet instance, so it would | ||
| 403 | // idle forever while looking healthy — cheaper to find at startup than after a | ||
| 404 | // build that never arrives. Every slot gets the same answer, so it is asked | ||
| 405 | // once. | ||
| 406 | if i == 0 { | ||
| 407 | if err := checkRepoFilter(opts.Repos, sl.who); err != nil { | ||
| 408 | sl.close() | ||
| 409 | closeSlots(slots) | ||
| 410 | return err | ||
| 411 | } | ||
| 412 | } | ||
| 413 | slots = append(slots, sl) | ||
| 374 | 414 | } | |
| 375 | sess.name = who.Name | ||
| 376 | sess.repos, sess.labels = opts.Repos, opts.Labels | ||
| 377 | sess.network = box.Network | ||
| 378 | sess.cacheRoot, sess.cacheReuse = cacheRoot, !opts.NoCache | ||
| 415 | |||
| 379 | 416 | if opts.NoCache { | |
| 380 | 417 | fmt.Fprintf(os.Stderr, "vole runner: job cache in %s, emptied after every job\n", cacheRoot) | |
| 381 | 418 | } else { | |
| 382 | 419 | fmt.Fprintf(os.Stderr, "vole runner: job cache in %s, capped at %s\n", | |
| 383 | 420 | cacheRoot, format.Bytes(cacheMax)) | |
| 384 | 421 | } | |
| 385 | fmt.Fprintf(os.Stderr, "vole runner: %s\n", sess.describe(who.Name)) | ||
| 422 | fmt.Fprintf(os.Stderr, "vole runner: %s\n", slots[0].describe(jobs)) | ||
| 386 | 423 | fmt.Fprintf(os.Stderr, "vole runner: jobs run in the workflow's image via %s\n", box.Runtime) | |
| 387 | 424 | if container.RunsAsRoot(box.Config) { | |
| 388 | 425 | fmt.Fprintln(os.Stderr, "vole runner: warning: jobs run as root in their container, so a job can "+ | |
| 389 | 426 | "rewrite its own image; run the runner as an unprivileged user (or pass --container-user) "+ | |
| 390 | 427 | "to have the image's permissions keep it immutable") | |
| 391 | 428 | } | |
| 429 | |||
| 430 | // The slots share a process, so a failure one of them cannot fix — a job this | ||
| 431 | // runner refuses to execute — stops them all rather than leaving the process | ||
| 432 | // half-alive with an answer nobody is going to change. | ||
| 433 | var ( | ||
| 434 | wg sync.WaitGroup | ||
| 435 | stopMu sync.Mutex | ||
| 436 | stopErr error | ||
| 437 | ) | ||
| 438 | halt := func(err error) { | ||
| 439 | stopMu.Lock() | ||
| 440 | if stopErr == nil { | ||
| 441 | stopErr = err | ||
| 442 | } | ||
| 443 | stopMu.Unlock() | ||
| 444 | cancel() | ||
| 445 | } | ||
| 446 | for _, sl := range slots { | ||
| 447 | wg.Add(1) | ||
| 448 | go func(sl *slot) { | ||
| 449 | defer wg.Done() | ||
| 450 | if err := sl.run(ctx); err != nil { | ||
| 451 | halt(err) | ||
| 452 | } | ||
| 453 | }(sl) | ||
| 454 | } | ||
| 455 | wg.Wait() | ||
| 456 | return stopErr | ||
| 457 | } | ||
| 458 | |||
| 459 | // slot is one worker: one session, at most one job at a time. --jobs N is N of | ||
| 460 | // these. They share the process, the cache root and the saved checkouts, and | ||
| 461 | // differ only in the session they lease on, because a lease is answered on the | ||
| 462 | // channel that asked for it. | ||
| 463 | type slot struct { | ||
| 464 | opts Options | ||
| 465 | cache string | ||
| 466 | box runnerBox | ||
| 467 | cacheRoot string | ||
| 468 | gitCacheMax int64 | ||
| 469 | // sess is this slot's connection, nil between a failure and the reconnect. | ||
| 470 | sess *runnerSession | ||
| 471 | // who is the account the current session authenticated as. The answer is what | ||
| 472 | // the runner reports its results under, so a session without one is not a | ||
| 473 | // session yet. | ||
| 474 | who runnerReply | ||
| 475 | } | ||
| 476 | |||
| 477 | // connect dials this slot's session and asks who the instance thinks it is. | ||
| 478 | func (sl *slot) connect() error { | ||
| 479 | sess, err := dialRunner(sl.opts.Remote) | ||
| 480 | if err != nil { | ||
| 481 | return err | ||
| 482 | } | ||
| 483 | who, err := sess.call(runnerRequest{Op: "ping"}) | ||
| 484 | if err != nil { | ||
| 485 | sess.close() | ||
| 486 | return err | ||
| 487 | } | ||
| 488 | if who.Error != "" { | ||
| 489 | sess.close() | ||
| 490 | return errors.New(who.Error) | ||
| 491 | } | ||
| 492 | sess.name = who.Name | ||
| 493 | sess.repos, sess.labels = sl.opts.Repos, sl.opts.Labels | ||
| 494 | sess.network = sl.box.Network | ||
| 495 | sess.cacheRoot, sess.cacheReuse = sl.cacheRoot, !sl.opts.NoCache | ||
| 496 | sl.sess, sl.who = sess, who | ||
| 497 | return nil | ||
| 498 | } | ||
| 499 | |||
| 500 | // close ends this slot's session, if it has one. | ||
| 501 | func (sl *slot) close() { | ||
| 502 | if sl.sess != nil { | ||
| 503 | sl.sess.close() | ||
| 504 | sl.sess = nil | ||
| 505 | } | ||
| 506 | } | ||
| 507 | |||
| 508 | // describe is the line the runner prints when it starts work, because "why is | ||
| 509 | // nothing running" is answered by reading what it says it will take. | ||
| 510 | func (sl *slot) describe(jobs int) string { | ||
| 511 | return sl.sess.describe(sl.who.Name, jobs) | ||
| 512 | } | ||
| 513 | |||
| 514 | // run keeps this slot working until the context ends. A lost connection is the | ||
| 515 | // slot's problem rather than the runner's: it reconnects and carries on, so one | ||
| 516 | // flaky link does not take the other slots' jobs with it, and the job that was | ||
| 517 | // running is requeued when its lease expires. A refused job is the exception — see | ||
| 518 | // refusedJob — because reconnecting would ask the same instance the same question. | ||
| 519 | func (sl *slot) run(ctx context.Context) error { | ||
| 520 | defer sl.close() | ||
| 521 | delay := reconnectDelay | ||
| 522 | for { | ||
| 523 | if ctx.Err() != nil { | ||
| 524 | return nil | ||
| 525 | } | ||
| 526 | if sl.sess == nil { | ||
| 527 | if err := sl.connect(); err != nil { | ||
| 528 | if sl.opts.Once { | ||
| 529 | // A one-shot runner is a process per job: the scheduler decides | ||
| 530 | // whether to try again, not a loop here. | ||
| 531 | return err | ||
| 532 | } | ||
| 533 | fmt.Fprintf(os.Stderr, "vole runner: %v; retrying in %s\n", err, delay) | ||
| 534 | if !sleep(ctx, delay) { | ||
| 535 | return nil | ||
| 536 | } | ||
| 537 | delay = nextDelay(delay) | ||
| 538 | continue | ||
| 539 | } | ||
| 540 | delay = reconnectDelay | ||
| 541 | } | ||
| 542 | err := sl.serve(ctx) | ||
| 543 | sl.close() | ||
| 544 | if err == nil { | ||
| 545 | return nil | ||
| 546 | } | ||
| 547 | var refused refusedJob | ||
| 548 | if errors.As(err, &refused) || sl.opts.Once { | ||
| 549 | return err | ||
| 550 | } | ||
| 551 | fmt.Fprintf(os.Stderr, "vole runner: %v; reconnecting in %s\n", err, delay) | ||
| 552 | if !sleep(ctx, delay) { | ||
| 553 | return nil | ||
| 554 | } | ||
| 555 | delay = nextDelay(delay) | ||
| 556 | } | ||
| 557 | } | ||
| 558 | |||
| 559 | // serve leases and runs jobs on this slot's session until something fails. | ||
| 560 | func (sl *slot) serve(ctx context.Context) error { | ||
| 561 | s := sl.sess | ||
| 392 | 562 | for { | |
| 393 | // A resident runner parks on the server's lease: the reply arrives when work is | ||
| 394 | // queued, not on a fixed beat, so an idle pool costs one open channel per | ||
| 395 | // runner instead of a query every couple of seconds. A one-shot runner asks | ||
| 396 | // once and leaves. | ||
| 397 | req := runnerRequest{Op: "lease", Repos: sess.repos, Labels: sess.labels, Network: sess.network} | ||
| 398 | if !opts.Once { | ||
| 563 | // A slot that has been told to stop does not lease again. A job it is | ||
| 564 | // already running finishes and reports, which is what "stop" means here: | ||
| 565 | // throwing away a finished build would only hand the same work to somebody | ||
| 566 | // else. | ||
| 567 | if ctx.Err() != nil { | ||
| 568 | return nil | ||
| 569 | } | ||
| 570 | // A resident runner parks on the server's lease: the reply arrives when work | ||
| 571 | // is queued, not on a fixed beat, so an idle slot costs one open channel | ||
| 572 | // instead of a query every couple of seconds. A one-shot runner asks once and | ||
| 573 | // leaves. | ||
| 574 | req := runnerRequest{Op: "lease", Repos: s.repos, Labels: s.labels, Network: s.network} | ||
| 575 | if !sl.opts.Once { | ||
| 399 | 576 | req.Wait = int(leaseWait.Seconds()) | |
| 400 | 577 | } | |
| 401 | rep, err := sess.call(req) | ||
| 578 | rep, err := s.call(req) | ||
| 402 | 579 | if err != nil { | |
| 403 | 580 | return err | |
| 404 | 581 | } | |
| @@ -406,44 +583,66 @@ func Run(ctx context.Context, opts Options) error { | |||
| 406 | 583 | return errors.New(rep.Error) | |
| 407 | 584 | } | |
| 408 | 585 | if rep.Run == nil { | |
| 409 | if opts.Once { | ||
| 586 | if sl.opts.Once { | ||
| 410 | 587 | return nil | |
| 411 | 588 | } | |
| 412 | 589 | // Either the wait expired with nothing to do, or the server answered at | |
| 413 | 590 | // once because it is older than this client and ignores `wait`. The sleep | |
| 414 | 591 | // keeps the second case at a polite two requests per second instead of a | |
| 415 | 592 | // spin. | |
| 416 | select { | ||
| 417 | case <-ctx.Done(): | ||
| 593 | if !sleep(ctx, 2*time.Second) { | ||
| 418 | 594 | return nil | |
| 419 | case <-time.After(2 * time.Second): | ||
| 420 | 595 | } | |
| 421 | 596 | continue | |
| 422 | 597 | } | |
| 423 | // The job came over the channel, and the runner is the party that acts on | ||
| 424 | // it: every field it builds a path or a command from is checked before the | ||
| 425 | // checkout, the container or the cache sees it. A refusal stops the runner | ||
| 426 | // rather than looping on the same lease, because a job that is malformed in | ||
| 427 | // the payload will not become well-formed by asking again. | ||
| 598 | // The job came over the channel, and the runner is the party that acts on it: | ||
| 599 | // every field it builds a path or a command from is checked before the | ||
| 600 | // checkout, the container or the cache sees it. | ||
| 428 | 601 | if err := validateJob(rep.Run); err != nil { | |
| 429 | return err | ||
| 602 | return refusedJob{err} | ||
| 430 | 603 | } | |
| 431 | 604 | lease := time.Duration(rep.Lease) * time.Second | |
| 432 | 605 | if lease <= 0 { | |
| 433 | 606 | lease = 60 * time.Second | |
| 434 | 607 | } | |
| 435 | sess.run(opts.Remote, cache, rep.Run, lease, box) | ||
| 436 | if opts.Once { | ||
| 608 | s.run(sl.opts.Remote, sl.cache, rep.Run, lease, sl.box) | ||
| 609 | if sl.opts.Once { | ||
| 437 | 610 | // This process is done; the next one bounds the cache at startup instead, | |
| 438 | 611 | // before it leases anything. | |
| 439 | 612 | return nil | |
| 440 | 613 | } | |
| 441 | 614 | // Tidy between jobs, never during one: the checkout this runner just used is | |
| 442 | 615 | // the one the next job is most likely to want, so it is passed over as active. | |
| 443 | maintainTrustedCaches(cache, gitCacheMax, rep.Run.Repo) | ||
| 616 | maintainTrustedCaches(sl.cache, sl.gitCacheMax, rep.Run.Repo) | ||
| 444 | 617 | } | |
| 445 | 618 | } | |
| 446 | 619 | ||
| 620 | func closeSlots(slots []*slot) { | ||
| 621 | for _, sl := range slots { | ||
| 622 | sl.close() | ||
| 623 | } | ||
| 624 | } | ||
| 625 | |||
| 626 | // sleep waits for d or until the context ends, and reports whether the wait | ||
| 627 | // finished. | ||
| 628 | func sleep(ctx context.Context, d time.Duration) bool { | ||
| 629 | select { | ||
| 630 | case <-ctx.Done(): | ||
| 631 | return false | ||
| 632 | case <-time.After(d): | ||
| 633 | return true | ||
| 634 | } | ||
| 635 | } | ||
| 636 | |||
| 637 | // nextDelay doubles the wait between reconnects, up to reconnectMax: a revoked key | ||
| 638 | // should not fill the journal, and a long outage should still be picked up promptly. | ||
| 639 | func nextDelay(d time.Duration) time.Duration { | ||
| 640 | if d *= 2; d > reconnectMax { | ||
| 641 | return reconnectMax | ||
| 642 | } | ||
| 643 | return d | ||
| 644 | } | ||
| 645 | |||
| 447 | 646 | // ---------- connection ---------- | |
| 448 | 647 | ||
| 449 | 648 | type runnerSession struct { | |
| @@ -453,8 +652,6 @@ type runnerSession struct { | |||
| 453 | 652 | enc *json.Encoder | |
| 454 | 653 | dec *json.Decoder | |
| 455 | 654 | mu sync.Mutex | |
| 456 | // stream is the log streamer for the job this session is running. | ||
| 457 | stream *logStream | ||
| 458 | 655 | // name is the account the server resolved from the connection. A record | |
| 459 | 656 | // names it, and the gate refuses a record whose signer is not the pusher. | |
| 460 | 657 | name string | |
| @@ -482,7 +679,7 @@ type runnerSession struct { | |||
| 482 | 679 | // nothing running" is answered by reading what it says it will take. Both filters are | |
| 483 | 680 | // named when they are set and omitted when they are not, so the common case stays one | |
| 484 | 681 | // short line. | |
| 485 | func (s *runnerSession) describe(name string) string { | ||
| 682 | func (s *runnerSession) describe(name string, jobs int) string { | ||
| 486 | 683 | out := "connected as " + name | |
| 487 | 684 | if len(s.labels) > 0 { | |
| 488 | 685 | out += ", taking jobs that ask for " + strings.Join(s.labels, ", ") | |
| @@ -493,6 +690,9 @@ func (s *runnerSession) describe(name string) string { | |||
| 493 | 690 | if s.network == ci.NetworkBridge { | |
| 494 | 691 | out += ", able to give a job the network" | |
| 495 | 692 | } | |
| 693 | if jobs > 1 { | ||
| 694 | out += fmt.Sprintf(", running %d jobs at a time", jobs) | ||
| 695 | } | ||
| 496 | 696 | return out + "; waiting for work" | |
| 497 | 697 | } | |
| 498 | 698 | ||
| @@ -604,11 +804,8 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 604 | 804 | // assumed. | |
| 605 | 805 | mask := redact.New(secretValues(job)) | |
| 606 | 806 | log.SetMask(mask) | |
| 607 | s.stream = s.startLogStream(job, log) | ||
| 608 | defer func() { | ||
| 609 | s.stream.stop() | ||
| 610 | s.stream = nil | ||
| 611 | }() | ||
| 807 | stream := s.startLogStream(job, log) | ||
| 808 | defer stream.stop() | ||
| 612 | 809 | co, err := prepareWorktree(remote, cache, job.Repo, job.SHA, job.ID, job.Attempt) | |
| 613 | 810 | if err != nil { | |
| 614 | 811 | // No checkout means no branch to commit a record to, so there is nothing to | |
| @@ -628,7 +825,7 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 628 | 825 | // A run queued before images were required has no environment to start. | |
| 629 | 826 | // Fail it with a record rather than guessing at the host. | |
| 630 | 827 | fmt.Fprintf(log, "\nthis run has no image; re-queue it from a workflow that declares image or image-file\n") | |
| 631 | s.finish(remote, co, job, log, 1, nil, nil, nil, box) | ||
| 828 | s.finish(remote, co, job, log, stream, 1, nil, nil, nil, box) | ||
| 632 | 829 | return | |
| 633 | 830 | } | |
| 634 | 831 | jobBox := &jobSandbox{} | |
| @@ -638,7 +835,7 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 638 | 835 | msg := mask.String(fmt.Sprintf(format, args...)) | |
| 639 | 836 | fmt.Fprintf(os.Stderr, "vole runner: %s\n", msg) | |
| 640 | 837 | fmt.Fprintf(log, "\n%s\n", msg) | |
| 641 | s.finish(remote, co, job, log, 1, nil, nil, nil, box) | ||
| 838 | s.finish(remote, co, job, log, stream, 1, nil, nil, nil, box) | ||
| 642 | 839 | } | |
| 643 | 840 | // A job that asked to act as an account it cannot act as is broken as written, not | |
| 644 | 841 | // degraded: fail it before the image is pulled and say why. | |
| @@ -677,7 +874,7 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 677 | 874 | if errors.Is(err, errCanceled) { | |
| 678 | 875 | // The build was stopped on a heartbeat, and the log already says why. The | |
| 679 | 876 | // attempt still has a story: report it with the work that did happen. | |
| 680 | s.finish(remote, co, job, log, 1, nil, nil, nil, box) | ||
| 877 | s.finish(remote, co, job, log, stream, 1, nil, nil, nil, box) | ||
| 681 | 878 | return | |
| 682 | 879 | } | |
| 683 | 880 | fail("%s", err) | |
| @@ -861,7 +1058,7 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 861 | 1058 | // the timings are what a reader asks for first after "what had it done". | |
| 862 | 1059 | steps = append(steps, result) | |
| 863 | 1060 | jobBox.StopWithLogs(log) | |
| 864 | s.finish(remote, co, job, log, code, steps, nil, nil, box) | ||
| 1061 | s.finish(remote, co, job, log, stream, code, steps, nil, nil, box) | ||
| 865 | 1062 | return | |
| 866 | 1063 | } | |
| 867 | 1064 | if code != 0 { | |
| @@ -870,7 +1067,7 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 870 | 1067 | result.Finished = time.Now().UTC() | |
| 871 | 1068 | steps = append(steps, result) | |
| 872 | 1069 | jobBox.StopWithLogs(log) | |
| 873 | s.finish(remote, co, job, log, code, steps, nil, nil, box) | ||
| 1070 | s.finish(remote, co, job, log, stream, code, steps, nil, nil, box) | ||
| 874 | 1071 | return | |
| 875 | 1072 | } | |
| 876 | 1073 | steps = append(steps, result) | |
| @@ -896,7 +1093,7 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 896 | 1093 | fail("%s", err) | |
| 897 | 1094 | return | |
| 898 | 1095 | } | |
| 899 | s.finish(remote, co, job, log, 0, steps, arts, releaseArts, box) | ||
| 1096 | s.finish(remote, co, job, log, stream, 0, steps, arts, releaseArts, box) | ||
| 900 | 1097 | } | |
| 901 | 1098 | ||
| 902 | 1099 | // cacheFor returns the mount for one job's cache and the lock that keeps eviction | |
| @@ -972,11 +1169,11 @@ func trimJobCache(ctx context.Context, root string, max int64) { | |||
| 972 | 1169 | // A log that cannot be stored does not lose the result: the record then names no log | |
| 973 | 1170 | // and the failure is reported on stderr rather than swallowed. A record the server | |
| 974 | 1171 | // does not accept is not reported at all. | |
| 975 | func (s *runnerSession) finish(remote string, co checkout, job *runnerJob, log *spool, exit int, steps []runs.StepResult, arts, releaseArts []runnerArtifact, box runnerBox) { | ||
| 1172 | func (s *runnerSession) finish(remote string, co checkout, job *runnerJob, log *spool, stream *logStream, exit int, steps []runs.StepResult, arts, releaseArts []runnerArtifact, box runnerBox) { | ||
| 976 | 1173 | if err := log.seal(); err != nil { | |
| 977 | 1174 | fmt.Fprintf(os.Stderr, "vole runner: could not seal the log for %s: %v\n", job.ID, err) | |
| 978 | 1175 | } | |
| 979 | streamErr := s.stream.drain() | ||
| 1176 | streamErr := stream.drain() | ||
| 980 | 1177 | if streamErr != nil { | |
| 981 | 1178 | fmt.Fprintf(os.Stderr, "vole runner: log stream for %s failed: %v\n", job.ID, streamErr) | |
| 982 | 1179 | // The step ranges describe bytes the server does not have. Keep the timings, | |