vole
Clone
Read only, no credentials:
https://rodent.software/git/vole.git
Write over SSH with a registered key:
git clone ssh://git@0.0.0.0:2222/vole.git
A free and open-source forge that lives in your Git repo.
Stop a running job at its runner's next heartbeat
0c14f0789af1ab0c43f99043acd7507eec4956e4 · Keenan Nemetz · 6d ago · parent 7e0b5950
Stop a running job at its runner's next heartbeat A maintainer could cancel a queued run and nothing else: CancelRun only retired waiting work, so a build that was clearly wrong or expensive held its runner until its budget ran out, or until somebody killed the container by hand. The queue cannot kill a process it does not own, but it can ask. Canceling a running run now marks the queue row and leaves the lease alone, because the runner still holds it and its partial attempt has to be acceptable. The runner sees the request on the heartbeat it already sends while a step or an image build runs, stops the container, writes one line to the log, and reports what it did. RecordAttempt and Finish then keep the canceled status instead of mapping the nonzero exit code to failed, so the partial log and step timings survive and the run is not a red build. A stop request whose runner dies first becomes canceled on the expiry sweep rather than being leased again, and a rerun clears the request: it is a new decision. The run page says who asked, distinguishes a run stopped while running from one canceled in the queue, and shows the interrupted step as stopped rather than failed. Closes #162.
ci/agent/agent.go +61−24
| @@ -48,6 +48,10 @@ const leaseWait = 25 * time.Second | |||
| 48 | 48 | // preparing its environment. There is nothing to report in that case. | |
| 49 | 49 | var errLeaseLost = errors.New("lease lost") | |
| 50 | 50 | ||
| 51 | // errCanceled means a maintainer asked this run to stop. The attempt is still worth | ||
| 52 | // reporting: a canceled run has a story, and the partial log and step timings are it. | ||
| 53 | var errCanceled = errors.New("run canceled") | ||
| 54 | |||
| 51 | 55 | type runnerStep struct { | |
| 52 | 56 | Name string `json:"name"` | |
| 53 | 57 | Run string `json:"run"` | |
| @@ -202,6 +206,10 @@ type runnerReply struct { | |||
| 202 | 206 | Repos []string `json:"repos,omitempty"` | |
| 203 | 207 | // LogSize is how much of the log the server has after a log chunk. | |
| 204 | 208 | LogSize int64 `json:"log_size,omitempty"` | |
| 209 | // Cancel names the account that asked this run to stop. It rides the heartbeat | ||
| 210 | // reply: empty means keep going, and a server that does not send it is one this | ||
| 211 | // runner keeps working against, exactly as before. | ||
| 212 | Cancel string `json:"cancel,omitempty"` | ||
| 205 | 213 | } | |
| 206 | 214 | ||
| 207 | 215 | // Options configure one runner process. | |
| @@ -614,6 +622,12 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 614 | 622 | fmt.Fprintf(os.Stderr, "vole runner: lease lost while building %s; stopping\n", job.Image) | |
| 615 | 623 | return | |
| 616 | 624 | } | |
| 625 | if errors.Is(err, errCanceled) { | ||
| 626 | // The build was stopped on a heartbeat, and the log already says why. The | ||
| 627 | // attempt still has a story: report it with the work that did happen. | ||
| 628 | s.finish(remote, co, job, log, 1, nil, nil, nil, box) | ||
| 629 | return | ||
| 630 | } | ||
| 617 | 631 | fail("%s", err) | |
| 618 | 632 | return | |
| 619 | 633 | } | |
| @@ -732,14 +746,14 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 732 | 746 | fmt.Fprintf(log, "\n--- %s\n", step.Name) | |
| 733 | 747 | own, _ := time.ParseDuration(step.Timeout) | |
| 734 | 748 | limit := ci.StepLimit(ci.BudgetLeft(deadline, time.Now()), own) | |
| 735 | code, lost := 0, false | ||
| 749 | code, lost, canceled := 0, false, false | ||
| 736 | 750 | if limit <= 0 { | |
| 737 | 751 | // The budget is spent, so the step must not start. Reporting it as a failed | |
| 738 | 752 | // step keeps the record the same shape as any other failure. | |
| 739 | 753 | fmt.Fprintf(log, "\nthe job's %s budget ran out before this step\n", job.Timeout) | |
| 740 | 754 | code = 1 | |
| 741 | 755 | } else { | |
| 742 | code, lost = s.execStep(job.ID, sandbox, step, limit, lease, log) | ||
| 756 | code, lost, canceled = s.execStep(job.ID, sandbox, step, limit, lease, log) | ||
| 743 | 757 | } | |
| 744 | 758 | // The mask holds back a few bytes so a value split across writes is caught; | |
| 745 | 759 | // flushing here keeps this step's byte range its own. | |
| @@ -752,6 +766,16 @@ func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Dur | |||
| 752 | 766 | fmt.Fprintf(os.Stderr, "vole runner: lease lost while running %q; stopping\n", step.Name) | |
| 753 | 767 | return // no finish: the queue already gave it to somebody else | |
| 754 | 768 | } | |
| 769 | if canceled { | ||
| 770 | // The heartbeat said a maintainer asked the run to stop, and runCommand | ||
| 771 | // already wrote the line to the log. Keep the partial attempt: the log and | ||
| 772 | // the timings are what a reader asks for first after "what had it done". | ||
| 773 | steps = append(steps, result) | ||
| 774 | sandbox.Stop() | ||
| 775 | sandbox = nil | ||
| 776 | s.finish(remote, co, job, log, code, steps, nil, nil, box) | ||
| 777 | return | ||
| 778 | } | ||
| 755 | 779 | if code != 0 { | |
| 756 | 780 | fmt.Fprintf(log, "\nstep %q exited %d\n", step.Name, code) | |
| 757 | 781 | result.LogSize = log.Len() - logStart | |
| @@ -1149,37 +1173,42 @@ func collectArtifactFiles(root string, patterns []string) ([]artifactFile, error | |||
| 1149 | 1173 | } | |
| 1150 | 1174 | ||
| 1151 | 1175 | // execStep runs one shell step, heartbeating while it works. It returns the exit | |
| 1152 | // code, the output produced, and whether the lease was lost. | ||
| 1153 | func (s *runnerSession) execStep(runID string, sandbox *container.Container, step runnerStep, limit time.Duration, lease time.Duration, out io.Writer) (int, bool) { | ||
| 1176 | // code, whether the lease was lost, and whether a maintainer asked the run to stop. | ||
| 1177 | func (s *runnerSession) execStep(runID string, sandbox *container.Container, step runnerStep, limit time.Duration, lease time.Duration, out io.Writer) (int, bool, bool) { | ||
| 1154 | 1178 | // limit is already resolved: the step's own timeout, or what is left of the job's | |
| 1155 | 1179 | // budget. The caller never passes zero. | |
| 1156 | 1180 | ctx, cancel := context.WithTimeout(context.Background(), limit) | |
| 1157 | 1181 | defer cancel() | |
| 1158 | 1182 | ||
| 1159 | 1183 | cmd := sandbox.Exec(ctx, step.Run) | |
| 1160 | code, lost := s.runCommand(ctx, cmd, runID, lease, sandbox.Stop, out) | ||
| 1184 | code, lost, canceled := s.runCommand(ctx, cmd, runID, lease, sandbox.Stop, out) | ||
| 1185 | if canceled { | ||
| 1186 | return code, false, true | ||
| 1187 | } | ||
| 1161 | 1188 | if ctx.Err() == context.DeadlineExceeded { | |
| 1162 | 1189 | // The heartbeat will have extended the lease this far; report the timeout as | |
| 1163 | 1190 | // a failure rather than leaving the job hanging. | |
| 1164 | 1191 | // What is left of a budget is rarely a round number; a reader wants the promise, | |
| 1165 | 1192 | // not the microseconds the bookkeeping cost. | |
| 1166 | 1193 | fmt.Fprintf(out, "\nstep timed out after %s\n", limit.Round(time.Millisecond)) | |
| 1167 | return 1, lost | ||
| 1194 | return 1, lost, false | ||
| 1168 | 1195 | } | |
| 1169 | return code, lost | ||
| 1196 | return code, lost, false | ||
| 1170 | 1197 | } | |
| 1171 | 1198 | ||
| 1172 | 1199 | // runCommand starts one host-side command and heartbeats while it runs. It is used | |
| 1173 | 1200 | // for docker exec and for building a custom image; onTimeout is what closes the | |
| 1174 | // sandbox when the command runs out of time. | ||
| 1175 | func (s *runnerSession) runCommand(ctx context.Context, cmd *exec.Cmd, runID string, lease time.Duration, onTimeout func(), out io.Writer) (int, bool) { | ||
| 1201 | // sandbox when the command runs out of time. A stop request from the server rides a | ||
| 1202 | // heartbeat, so the command is killed the same way a timeout kills it; the difference | ||
| 1203 | // is that the caller reports the partial attempt instead of a failure. | ||
| 1204 | 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) { | ||
| 1176 | 1205 | // Own process group: a command that spawns children must not outlive its timeout. | |
| 1177 | 1206 | cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} | |
| 1178 | 1207 | cmd.Stdout, cmd.Stderr = out, out | |
| 1179 | 1208 | ||
| 1180 | 1209 | if err := cmd.Start(); err != nil { | |
| 1181 | 1210 | fmt.Fprintf(out, "cannot start: %v\n", err) | |
| 1182 | return 1, false | ||
| 1211 | return 1, false, false | ||
| 1183 | 1212 | } | |
| 1184 | 1213 | done := make(chan error, 1) | |
| 1185 | 1214 | go func() { done <- cmd.Wait() }() | |
| @@ -1191,33 +1220,41 @@ func (s *runnerSession) runCommand(ctx context.Context, cmd *exec.Cmd, runID str | |||
| 1191 | 1220 | tick := time.NewTicker(interval) | |
| 1192 | 1221 | defer tick.Stop() | |
| 1193 | 1222 | ||
| 1223 | stop := func() { | ||
| 1224 | if onTimeout != nil { | ||
| 1225 | onTimeout() | ||
| 1226 | } | ||
| 1227 | if cmd.Process != nil { | ||
| 1228 | _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) | ||
| 1229 | } | ||
| 1230 | // Wait for the copy goroutines before the caller closes the log. | ||
| 1231 | <-done | ||
| 1232 | } | ||
| 1233 | |||
| 1194 | 1234 | for { | |
| 1195 | 1235 | select { | |
| 1196 | 1236 | case err := <-done: | |
| 1197 | 1237 | if err == nil { | |
| 1198 | return 0, false | ||
| 1238 | return 0, false, false | ||
| 1199 | 1239 | } | |
| 1200 | 1240 | var ee *exec.ExitError | |
| 1201 | 1241 | if errors.As(err, &ee) { | |
| 1202 | return ee.ExitCode(), false | ||
| 1242 | return ee.ExitCode(), false, false | ||
| 1203 | 1243 | } | |
| 1204 | 1244 | fmt.Fprintf(out, "\n%v\n", err) | |
| 1205 | return 1, false | ||
| 1245 | return 1, false, false | ||
| 1206 | 1246 | case <-ctx.Done(): | |
| 1207 | if onTimeout != nil { | ||
| 1208 | onTimeout() | ||
| 1209 | } | ||
| 1210 | if cmd.Process != nil { | ||
| 1211 | _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) | ||
| 1212 | } | ||
| 1213 | // Wait for the copy goroutines before the caller closes the log. | ||
| 1214 | <-done | ||
| 1215 | return 1, false | ||
| 1247 | stop() | ||
| 1248 | return 1, false, false | ||
| 1216 | 1249 | case <-tick.C: | |
| 1217 | // The lease is the only thing a heartbeat carries. | ||
| 1218 | 1250 | rep, err := s.call(runnerRequest{Op: "heartbeat", Run: runID}) | |
| 1219 | 1251 | if err != nil || !rep.OK { | |
| 1220 | return 1, true | ||
| 1252 | return 1, true, false | ||
| 1253 | } | ||
| 1254 | if rep.Cancel != "" { | ||
| 1255 | fmt.Fprintf(out, "\nthe run was canceled by %s; stopping\n", rep.Cancel) | ||
| 1256 | stop() | ||
| 1257 | return 1, false, true | ||
| 1221 | 1258 | } | |
| 1222 | 1259 | } | |
| 1223 | 1260 | } | |