fix(host): a shutdown ends the wait for the acknowledgement and the report fits a bounded budget (M5); private state directory (m7); goroutines end with their job (m8)
- commit
- 320711b37b84ccc1b343b4355b8eaacb062b6eec
- author
- brice · · 2026-10-12 19:37 UTC
- parent
- 495cf95
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +150 -12
M internal/guest/jobexec.go +6 -1
| @@ -150,7 +150,11 @@ func runJob(ctx context.Context, conn *execstream.Conn, cfg ExecConfig, spec *ex | |||
| 150 | 150 | rep := &frameReporter{conn: conn, failed: make(chan struct{})} | |
| 151 | 151 | ||
| 152 | 152 | // Frames from the host while the job runs: a Cancel stops the job; a lost | |
| 153 | - | // host does too, like a runner shutdown. | |
| 153 | + | // host does too, like a runner shutdown. The reader ends with the stream: the | |
| 154 | + | // caller closes it after runJob, and the goroutine that reports a failed | |
| 155 | + | // send ends with the job (done). | |
| 156 | + | done := make(chan struct{}) | |
| 157 | + | defer close(done) | |
| 154 | 158 | go func() { | |
| 155 | 159 | for { | |
| 156 | 160 | f, err := conn.Recv() | |
| @@ -172,6 +176,7 @@ func runJob(ctx context.Context, conn *execstream.Conn, cfg ExecConfig, spec *ex | |||
| 172 | 176 | case <-rep.failed: | |
| 173 | 177 | canceller.Cancel(executor.CancelShutdown) | |
| 174 | 178 | case <-canceller.Done(): | |
| 179 | + | case <-done: | |
| 175 | 180 | } | |
| 176 | 181 | }() | |
| 177 | 182 | ||
| @@ -150,7 +150,11 @@ func runJob(ctx context.Context, conn *execstream.Conn, cfg ExecConfig, spec *ex | |||||
| 150 | rep := &frameReporter{conn: conn, failed: make(chan struct{})} | 150 | rep := &frameReporter{conn: conn, failed: make(chan struct{})} | ||
| 151 | 151 | ||||
| 152 | // Frames from the host while the job runs: a Cancel stops the job; a lost | 152 | // Frames from the host while the job runs: a Cancel stops the job; a lost | ||
| 153 | - | // host does too, like a runner shutdown. | 153 | + | // host does too, like a runner shutdown. The reader ends with the stream: the |
| 154 | + | // caller closes it after runJob, and the goroutine that reports a failed | |||
| 155 | + | // send ends with the job (done). | |||
| 156 | + | done := make(chan struct{}) | |||
| 157 | + | defer close(done) | |||
| 154 | go func() { | 158 | go func() { | ||
| 155 | for { | 159 | for { | ||
| 156 | f, err := conn.Recv() | 160 | f, err := conn.Recv() | ||
| @@ -172,6 +176,7 @@ func runJob(ctx context.Context, conn *execstream.Conn, cfg ExecConfig, spec *ex | |||||
| 172 | case <-rep.failed: | 176 | case <-rep.failed: | ||
| 173 | canceller.Cancel(executor.CancelShutdown) | 177 | canceller.Cancel(executor.CancelShutdown) | ||
| 174 | case <-canceller.Done(): | 178 | case <-canceller.Done(): | ||
| 179 | + | case <-done: | |||
| 175 | } | 180 | } | ||
| 176 | }() | 181 | }() | ||
| 177 | 182 | ||||
M internal/host/config.go +6 -0
| @@ -133,6 +133,12 @@ func SaveState(dir string, s State) error { | |||
| 133 | 133 | if err := os.MkdirAll(dir, 0o700); err != nil { | |
| 134 | 134 | return err | |
| 135 | 135 | } | |
| 136 | + | // MkdirAll leaves an existing directory as it is; the directory holds the | |
| 137 | + | // runner token, so it is made private (the VM images live in it too, and | |
| 138 | + | // the runner runs as root: nobody else needs to enter). | |
| 139 | + | if err := os.Chmod(dir, 0o700); err != nil { | |
| 140 | + | return fmt.Errorf("make the state directory private: %w", err) | |
| 141 | + | } | |
| 136 | 142 | b, err := json.MarshalIndent(s, "", " ") | |
| 137 | 143 | if err != nil { | |
| 138 | 144 | return err | |
| @@ -133,6 +133,12 @@ func SaveState(dir string, s State) error { | |||||
| 133 | if err := os.MkdirAll(dir, 0o700); err != nil { | 133 | if err := os.MkdirAll(dir, 0o700); err != nil { | ||
| 134 | return err | 134 | return err | ||
| 135 | } | 135 | } | ||
| 136 | + | // MkdirAll leaves an existing directory as it is; the directory holds the | |||
| 137 | + | // runner token, so it is made private (the VM images live in it too, and | |||
| 138 | + | // the runner runs as root: nobody else needs to enter). | |||
| 139 | + | if err := os.Chmod(dir, 0o700); err != nil { | |||
| 140 | + | return fmt.Errorf("make the state directory private: %w", err) | |||
| 141 | + | } | |||
| 136 | b, err := json.MarshalIndent(s, "", " ") | 142 | b, err := json.MarshalIndent(s, "", " ") | ||
| 137 | if err != nil { | 143 | if err != nil { | ||
| 138 | return err | 144 | return err | ||
M internal/host/job.go +45 -10
| @@ -422,9 +422,27 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) { | |||
| 422 | 422 | defer j.cancel(nil) | |
| 423 | 423 | ctx := j.ctx | |
| 424 | 424 | ||
| 425 | - | // 1. The acknowledgement comes first, before anything else. | |
| 426 | - | if !j.ack(ctx) { | |
| 427 | - | j.log.Warn().Msg("the assignment was lost before the job started") | |
| 425 | + | // 1. The acknowledgement comes first, before anything else. A shutdown ends | |
| 426 | + | // the wait: the job is not started and not acknowledged, and the server | |
| 427 | + | // gives it to another runner when no heartbeat comes within 30 s (runner.proto, | |
| 428 | + | // FetchJob: this is not a failure of the job). | |
| 429 | + | ackCtx, stopAck := context.WithCancel(ctx) | |
| 430 | + | go func() { | |
| 431 | + | select { | |
| 432 | + | case <-shutdown: | |
| 433 | + | stopAck() | |
| 434 | + | case <-ackCtx.Done(): | |
| 435 | + | } | |
| 436 | + | }() | |
| 437 | + | acked := j.ack(ackCtx) | |
| 438 | + | stopAck() | |
| 439 | + | if !acked { | |
| 440 | + | select { | |
| 441 | + | case <-shutdown: | |
| 442 | + | j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner") | |
| 443 | + | default: | |
| 444 | + | j.log.Warn().Msg("the assignment was lost before the job started") | |
| 445 | + | } | |
| 428 | 446 | return | |
| 429 | 447 | } | |
| 430 | 448 | var bg sync.WaitGroup | |
| @@ -456,22 +474,28 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) { | |||
| 456 | 474 | j.stopping = true | |
| 457 | 475 | j.cancelMu.Unlock() | |
| 458 | 476 | j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN) | |
| 477 | + | timer := time.NewTimer(j.t.ShutdownGrace) | |
| 478 | + | defer timer.Stop() | |
| 459 | 479 | select { | |
| 460 | - | case <-time.After(j.t.ShutdownGrace): | |
| 480 | + | case <-timer.C: | |
| 461 | 481 | hard() | |
| 462 | 482 | case <-ctx.Done(): | |
| 463 | 483 | } | |
| 464 | 484 | }() | |
| 465 | 485 | go func() { | |
| 486 | + | wait := time.NewTimer(time.Until(j.deadline())) | |
| 487 | + | defer wait.Stop() | |
| 466 | 488 | select { | |
| 467 | - | case <-time.After(time.Until(j.deadline())): | |
| 489 | + | case <-wait.C: | |
| 468 | 490 | case <-ctx.Done(): | |
| 469 | 491 | return | |
| 470 | 492 | } | |
| 471 | 493 | j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed") | |
| 472 | 494 | j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT) | |
| 495 | + | timer := time.NewTimer(j.t.TimeoutGrace) | |
| 496 | + | defer timer.Stop() | |
| 473 | 497 | select { | |
| 474 | - | case <-time.After(j.t.TimeoutGrace): | |
| 498 | + | case <-timer.C: | |
| 475 | 499 | hard() | |
| 476 | 500 | case <-ctx.Done(): | |
| 477 | 501 | } | |
| @@ -486,7 +510,13 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) { | |||
| 486 | 510 | } | |
| 487 | 511 | // The report is made even when the runner is shutting down: it uses its own | |
| 488 | 512 | // deadline. | |
| 489 | - | rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.CompleteDeadline) | |
| 513 | + | reportBudget := j.t.CompleteDeadline | |
| 514 | + | if j.isStopping() { | |
| 515 | + | // The service manager waits only so long (see Timing): the report of a job | |
| 516 | + | // cut short by a shutdown gets what is left of the budget. | |
| 517 | + | reportBudget = min(reportBudget, j.t.ShutdownReport) | |
| 518 | + | } | |
| 519 | + | rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), reportBudget) | |
| 490 | 520 | defer cancel() | |
| 491 | 521 | // Output limits are checked before the log is flushed: a job whose outputs | |
| 492 | 522 | // the server would refuse fails without outputs, with a line in the log | |
| @@ -515,7 +545,7 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct | |||
| 515 | 545 | } | |
| 516 | 546 | // Stop runs to completion even when the job context is done. | |
| 517 | 547 | defer func() { | |
| 518 | - | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute) | |
| 548 | + | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline) | |
| 519 | 549 | defer cancel() | |
| 520 | 550 | if err := sb.Stop(sctx); err != nil { | |
| 521 | 551 | j.log.Error().Err(err).Msg("stopping the sandbox") | |
| @@ -546,13 +576,18 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct | |||
| 546 | 576 | } | |
| 547 | 577 | defer func() { _ = conn.Close() }() | |
| 548 | 578 | ||
| 549 | - | // Hard stops (grace elapsed, or the job was abandoned) close the stream. | |
| 579 | + | // Hard stops (grace elapsed, or the job was abandoned) close the stream. The | |
| 580 | + | // goroutine ends when this function returns, whatever happens to ctx. | |
| 581 | + | sessionDone := make(chan struct{}) | |
| 582 | + | defer close(sessionDone) | |
| 550 | 583 | go func() { | |
| 551 | 584 | select { | |
| 552 | 585 | case <-stopHard: | |
| 586 | + | _ = conn.Close() | |
| 553 | 587 | case <-ctx.Done(): | |
| 588 | + | _ = conn.Close() | |
| 589 | + | case <-sessionDone: | |
| 554 | 590 | } | |
| 555 | - | _ = conn.Close() | |
| 556 | 591 | }() | |
| 557 | 592 | ||
| 558 | 593 | run := &execpb.Run{Job: &execpb.JobSpec{ | |
| @@ -422,9 +422,27 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) { | |||||
| 422 | defer j.cancel(nil) | 422 | defer j.cancel(nil) | ||
| 423 | ctx := j.ctx | 423 | ctx := j.ctx | ||
| 424 | 424 | ||||
| 425 | - | // 1. The acknowledgement comes first, before anything else. | 425 | + | // 1. The acknowledgement comes first, before anything else. A shutdown ends |
| 426 | - | if !j.ack(ctx) { | 426 | + | // the wait: the job is not started and not acknowledged, and the server |
| 427 | - | j.log.Warn().Msg("the assignment was lost before the job started") | 427 | + | // gives it to another runner when no heartbeat comes within 30 s (runner.proto, |
| 428 | + | // FetchJob: this is not a failure of the job). | |||
| 429 | + | ackCtx, stopAck := context.WithCancel(ctx) | |||
| 430 | + | go func() { | |||
| 431 | + | select { | |||
| 432 | + | case <-shutdown: | |||
| 433 | + | stopAck() | |||
| 434 | + | case <-ackCtx.Done(): | |||
| 435 | + | } | |||
| 436 | + | }() | |||
| 437 | + | acked := j.ack(ackCtx) | |||
| 438 | + | stopAck() | |||
| 439 | + | if !acked { | |||
| 440 | + | select { | |||
| 441 | + | case <-shutdown: | |||
| 442 | + | j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner") | |||
| 443 | + | default: | |||
| 444 | + | j.log.Warn().Msg("the assignment was lost before the job started") | |||
| 445 | + | } | |||
| 428 | return | 446 | return | ||
| 429 | } | 447 | } | ||
| 430 | var bg sync.WaitGroup | 448 | var bg sync.WaitGroup | ||
| @@ -456,22 +474,28 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) { | |||||
| 456 | j.stopping = true | 474 | j.stopping = true | ||
| 457 | j.cancelMu.Unlock() | 475 | j.cancelMu.Unlock() | ||
| 458 | j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN) | 476 | j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN) | ||
| 477 | + | timer := time.NewTimer(j.t.ShutdownGrace) | |||
| 478 | + | defer timer.Stop() | |||
| 459 | select { | 479 | select { | ||
| 460 | - | case <-time.After(j.t.ShutdownGrace): | 480 | + | case <-timer.C: |
| 461 | hard() | 481 | hard() | ||
| 462 | case <-ctx.Done(): | 482 | case <-ctx.Done(): | ||
| 463 | } | 483 | } | ||
| 464 | }() | 484 | }() | ||
| 465 | go func() { | 485 | go func() { | ||
| 486 | + | wait := time.NewTimer(time.Until(j.deadline())) | |||
| 487 | + | defer wait.Stop() | |||
| 466 | select { | 488 | select { | ||
| 467 | - | case <-time.After(time.Until(j.deadline())): | 489 | + | case <-wait.C: |
| 468 | case <-ctx.Done(): | 490 | case <-ctx.Done(): | ||
| 469 | return | 491 | return | ||
| 470 | } | 492 | } | ||
| 471 | j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed") | 493 | j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed") | ||
| 472 | j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT) | 494 | j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT) | ||
| 495 | + | timer := time.NewTimer(j.t.TimeoutGrace) | |||
| 496 | + | defer timer.Stop() | |||
| 473 | select { | 497 | select { | ||
| 474 | - | case <-time.After(j.t.TimeoutGrace): | 498 | + | case <-timer.C: |
| 475 | hard() | 499 | hard() | ||
| 476 | case <-ctx.Done(): | 500 | case <-ctx.Done(): | ||
| 477 | } | 501 | } | ||
| @@ -486,7 +510,13 @@ func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) { | |||||
| 486 | } | 510 | } | ||
| 487 | // The report is made even when the runner is shutting down: it uses its own | 511 | // The report is made even when the runner is shutting down: it uses its own | ||
| 488 | // deadline. | 512 | // deadline. | ||
| 489 | - | rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.CompleteDeadline) | 513 | + | reportBudget := j.t.CompleteDeadline |
| 514 | + | if j.isStopping() { | |||
| 515 | + | // The service manager waits only so long (see Timing): the report of a job | |||
| 516 | + | // cut short by a shutdown gets what is left of the budget. | |||
| 517 | + | reportBudget = min(reportBudget, j.t.ShutdownReport) | |||
| 518 | + | } | |||
| 519 | + | rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), reportBudget) | |||
| 490 | defer cancel() | 520 | defer cancel() | ||
| 491 | // Output limits are checked before the log is flushed: a job whose outputs | 521 | // Output limits are checked before the log is flushed: a job whose outputs | ||
| 492 | // the server would refuse fails without outputs, with a line in the log | 522 | // the server would refuse fails without outputs, with a line in the log | ||
| @@ -515,7 +545,7 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct | |||||
| 515 | } | 545 | } | ||
| 516 | // Stop runs to completion even when the job context is done. | 546 | // Stop runs to completion even when the job context is done. | ||
| 517 | defer func() { | 547 | defer func() { | ||
| 518 | - | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute) | 548 | + | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline) |
| 519 | defer cancel() | 549 | defer cancel() | ||
| 520 | if err := sb.Stop(sctx); err != nil { | 550 | if err := sb.Stop(sctx); err != nil { | ||
| 521 | j.log.Error().Err(err).Msg("stopping the sandbox") | 551 | j.log.Error().Err(err).Msg("stopping the sandbox") | ||
| @@ -546,13 +576,18 @@ func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct | |||||
| 546 | } | 576 | } | ||
| 547 | defer func() { _ = conn.Close() }() | 577 | defer func() { _ = conn.Close() }() | ||
| 548 | 578 | ||||
| 549 | - | // Hard stops (grace elapsed, or the job was abandoned) close the stream. | 579 | + | // Hard stops (grace elapsed, or the job was abandoned) close the stream. The |
| 580 | + | // goroutine ends when this function returns, whatever happens to ctx. | |||
| 581 | + | sessionDone := make(chan struct{}) | |||
| 582 | + | defer close(sessionDone) | |||
| 550 | go func() { | 583 | go func() { | ||
| 551 | select { | 584 | select { | ||
| 552 | case <-stopHard: | 585 | case <-stopHard: | ||
| 586 | + | _ = conn.Close() | |||
| 553 | case <-ctx.Done(): | 587 | case <-ctx.Done(): | ||
| 588 | + | _ = conn.Close() | |||
| 589 | + | case <-sessionDone: | |||
| 554 | } | 590 | } | ||
| 555 | - | _ = conn.Close() | |||
| 556 | }() | 591 | }() | ||
| 557 | 592 | ||||
| 558 | run := &execpb.Run{Job: &execpb.JobSpec{ | 593 | run := &execpb.Run{Job: &execpb.JobSpec{ | ||
M internal/host/rpc.go +20 -1
| @@ -176,8 +176,15 @@ type Timing struct { | |||
| 176 | 176 | // report, before the sandbox is destroyed. | |
| 177 | 177 | TimeoutGrace time.Duration | |
| 178 | 178 | // ShutdownGrace is how long jobs get to stop after a shutdown signal before | |
| 179 | - | // their sandbox is destroyed. | |
| 179 | + | // their sandbox is destroyed. The executor needs 10 s for its escalation | |
| 180 | + | // (SIGINT, SIGTERM after 7.5 s, SIGKILL 2.5 s later) and some for the | |
| 181 | + | // post-steps and the report to the host. | |
| 180 | 182 | ShutdownGrace time.Duration | |
| 183 | + | // StopDeadline bounds the destruction of a sandbox. | |
| 184 | + | StopDeadline time.Duration | |
| 185 | + | // ShutdownReport bounds the last flush of the log and CompleteJob of a job | |
| 186 | + | // that a shutdown cut short. | |
| 187 | + | ShutdownReport time.Duration | |
| 181 | 188 | // CompleteDeadline bounds the flush of logs and CompleteJob at the end of | |
| 182 | 189 | // a job. | |
| 183 | 190 | CompleteDeadline time.Duration | |
| @@ -204,12 +211,24 @@ func (t Timing) withDefaults() Timing { | |||
| 204 | 211 | def(&t.WatchRetry, time.Second) | |
| 205 | 212 | def(&t.TimeoutGrace, 30*time.Second) | |
| 206 | 213 | def(&t.ShutdownGrace, 30*time.Second) | |
| 214 | + | def(&t.StopDeadline, 30*time.Second) | |
| 215 | + | def(&t.ShutdownReport, 30*time.Second) | |
| 207 | 216 | def(&t.CompleteDeadline, 5*time.Minute) | |
| 208 | 217 | def(&t.NoJobPause, 0) | |
| 209 | 218 | def(&t.ReapRetry, 5*time.Second) | |
| 210 | 219 | return t | |
| 211 | 220 | } | |
| 212 | 221 | ||
| 222 | + | // ShutdownBudget is the longest a graceful shutdown takes with these timings: | |
| 223 | + | // the grace for the executor, the destruction of the sandbox, and the report. | |
| 224 | + | // The jobs shut down in parallel, so it does not depend on their number. The | |
| 225 | + | // service manager must wait at least this long (contrib/ermite-runner.service | |
| 226 | + | // has TimeoutStopSec=120 for the defaults, 90 s). | |
| 227 | + | func (t Timing) ShutdownBudget() time.Duration { | |
| 228 | + | t = t.withDefaults() | |
| 229 | + | return t.ShutdownGrace + t.StopDeadline + t.ShutdownReport | |
| 230 | + | } | |
| 231 | + | ||
| 213 | 232 | // sleep waits d or until ctx ends; it reports whether the full delay passed. | |
| 214 | 233 | func sleep(ctx context.Context, d time.Duration) bool { | |
| 215 | 234 | if d <= 0 { | |
| @@ -176,8 +176,15 @@ type Timing struct { | |||||
| 176 | // report, before the sandbox is destroyed. | 176 | // report, before the sandbox is destroyed. | ||
| 177 | TimeoutGrace time.Duration | 177 | TimeoutGrace time.Duration | ||
| 178 | // ShutdownGrace is how long jobs get to stop after a shutdown signal before | 178 | // ShutdownGrace is how long jobs get to stop after a shutdown signal before | ||
| 179 | - | // their sandbox is destroyed. | 179 | + | // their sandbox is destroyed. The executor needs 10 s for its escalation |
| 180 | + | // (SIGINT, SIGTERM after 7.5 s, SIGKILL 2.5 s later) and some for the | |||
| 181 | + | // post-steps and the report to the host. | |||
| 180 | ShutdownGrace time.Duration | 182 | ShutdownGrace time.Duration | ||
| 183 | + | // StopDeadline bounds the destruction of a sandbox. | |||
| 184 | + | StopDeadline time.Duration | |||
| 185 | + | // ShutdownReport bounds the last flush of the log and CompleteJob of a job | |||
| 186 | + | // that a shutdown cut short. | |||
| 187 | + | ShutdownReport time.Duration | |||
| 181 | // CompleteDeadline bounds the flush of logs and CompleteJob at the end of | 188 | // CompleteDeadline bounds the flush of logs and CompleteJob at the end of | ||
| 182 | // a job. | 189 | // a job. | ||
| 183 | CompleteDeadline time.Duration | 190 | CompleteDeadline time.Duration | ||
| @@ -204,12 +211,24 @@ func (t Timing) withDefaults() Timing { | |||||
| 204 | def(&t.WatchRetry, time.Second) | 211 | def(&t.WatchRetry, time.Second) | ||
| 205 | def(&t.TimeoutGrace, 30*time.Second) | 212 | def(&t.TimeoutGrace, 30*time.Second) | ||
| 206 | def(&t.ShutdownGrace, 30*time.Second) | 213 | def(&t.ShutdownGrace, 30*time.Second) | ||
| 214 | + | def(&t.StopDeadline, 30*time.Second) | |||
| 215 | + | def(&t.ShutdownReport, 30*time.Second) | |||
| 207 | def(&t.CompleteDeadline, 5*time.Minute) | 216 | def(&t.CompleteDeadline, 5*time.Minute) | ||
| 208 | def(&t.NoJobPause, 0) | 217 | def(&t.NoJobPause, 0) | ||
| 209 | def(&t.ReapRetry, 5*time.Second) | 218 | def(&t.ReapRetry, 5*time.Second) | ||
| 210 | return t | 219 | return t | ||
| 211 | } | 220 | } | ||
| 212 | 221 | ||||
| 222 | + | // ShutdownBudget is the longest a graceful shutdown takes with these timings: | |||
| 223 | + | // the grace for the executor, the destruction of the sandbox, and the report. | |||
| 224 | + | // The jobs shut down in parallel, so it does not depend on their number. The | |||
| 225 | + | // service manager must wait at least this long (contrib/ermite-runner.service | |||
| 226 | + | // has TimeoutStopSec=120 for the defaults, 90 s). | |||
| 227 | + | func (t Timing) ShutdownBudget() time.Duration { | |||
| 228 | + | t = t.withDefaults() | |||
| 229 | + | return t.ShutdownGrace + t.StopDeadline + t.ShutdownReport | |||
| 230 | + | } | |||
| 231 | + | ||||
| 213 | // sleep waits d or until ctx ends; it reports whether the full delay passed. | 232 | // sleep waits d or until ctx ends; it reports whether the full delay passed. | ||
| 214 | func sleep(ctx context.Context, d time.Duration) bool { | 233 | func sleep(ctx context.Context, d time.Duration) bool { | ||
| 215 | if d <= 0 { | 234 | if d <= 0 { | ||
M internal/host/runner_test.go +73 -0
| @@ -9,6 +9,9 @@ import ( | |||
| 9 | 9 | "io" | |
| 10 | 10 | "net/http" | |
| 11 | 11 | "os" | |
| 12 | + | "path/filepath" | |
| 13 | + | "regexp" | |
| 14 | + | "strconv" | |
| 12 | 15 | "strings" | |
| 13 | 16 | "testing" | |
| 14 | 17 | "time" | |
| @@ -1173,3 +1176,73 @@ func TestSandboxSizesAreLeftToTheDriversConfiguration(t *testing.T) { | |||
| 1173 | 1176 | t.Fatalf("%+v", got) | |
| 1174 | 1177 | } | |
| 1175 | 1178 | } | |
| 1179 | + | ||
| 1180 | + | // ---- M5: shutdown | |
| 1181 | + | ||
| 1182 | + | // A job whose acknowledgement is pending when the runner shuts down is left to | |
| 1183 | + | // the server (it requeues it after 30 s): the sandbox is never started, no | |
| 1184 | + | // result is reported, and Run returns promptly. | |
| 1185 | + | func TestShutdownBeforeTheAcknowledgementLeavesTheJobToTheServer(t *testing.T) { | |
| 1186 | + | h := newHarness(t, simpleScript("x")) | |
| 1187 | + | h.srv.jobs = []*ermite.Job{testJob("j1")} | |
| 1188 | + | h.srv.beatHook = func(int) error { return connect.NewError(connect.CodeUnavailable, io.EOF) } // never acknowledged | |
| 1189 | + | stop := h.run(t) | |
| 1190 | + | eventually(t, "a failed acknowledgement", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return len(h.srv.beats) >= 2 }) | |
| 1191 | + | start := time.Now() | |
| 1192 | + | if err := stop(); err != nil { | |
| 1193 | + | t.Fatal(err) | |
| 1194 | + | } | |
| 1195 | + | if time.Since(start) > 3*time.Second { | |
| 1196 | + | t.Fatalf("Run took %v to return", time.Since(start)) | |
| 1197 | + | } | |
| 1198 | + | h.driver.mu.Lock() | |
| 1199 | + | defer h.driver.mu.Unlock() | |
| 1200 | + | if h.driver.news != 0 || h.driver.started != 0 { | |
| 1201 | + | t.Fatalf("a sandbox was created for a job that was never acknowledged: news=%d started=%d", h.driver.news, h.driver.started) | |
| 1202 | + | } | |
| 1203 | + | if n := h.srv.completeCount(); n != 0 { | |
| 1204 | + | t.Fatalf("CompleteJob was called %d times", n) | |
| 1205 | + | } | |
| 1206 | + | if !strings.Contains(h.logs.String(), "before the job was acknowledged") { | |
| 1207 | + | t.Fatalf("no explanation in the log:\n%s", h.logs.String()) | |
| 1208 | + | } | |
| 1209 | + | } | |
| 1210 | + | ||
| 1211 | + | // The graceful shutdown of the defaults fits the time the unit gives. | |
| 1212 | + | func TestShutdownBudgetFitsTheUnit(t *testing.T) { | |
| 1213 | + | budget := Timing{}.ShutdownBudget() | |
| 1214 | + | if budget != 90*time.Second { | |
| 1215 | + | t.Fatalf("the budget is %v, README and contrib/ermite-runner.service say 90 s", budget) | |
| 1216 | + | } | |
| 1217 | + | unit, err := os.ReadFile("../../contrib/ermite-runner.service") | |
| 1218 | + | if err != nil { | |
| 1219 | + | t.Skip(err) | |
| 1220 | + | } | |
| 1221 | + | m := regexp.MustCompile(`(?m)^TimeoutStopSec=(\d+)$`).FindStringSubmatch(string(unit)) | |
| 1222 | + | if m == nil { | |
| 1223 | + | t.Fatal("no TimeoutStopSec in the unit") | |
| 1224 | + | } | |
| 1225 | + | secs, _ := strconv.Atoi(m[1]) | |
| 1226 | + | if time.Duration(secs)*time.Second < budget+15*time.Second { | |
| 1227 | + | t.Fatalf("TimeoutStopSec=%d does not leave 15 s of margin over the %v budget", secs, budget) | |
| 1228 | + | } | |
| 1229 | + | } | |
| 1230 | + | ||
| 1231 | + | // ---- m7: the state directory is private | |
| 1232 | + | ||
| 1233 | + | func TestSaveStateMakesAnExistingDirectoryPrivate(t *testing.T) { | |
| 1234 | + | dir := filepath.Join(t.TempDir(), "state") | |
| 1235 | + | if err := os.Mkdir(dir, 0o755); err != nil { | |
| 1236 | + | t.Fatal(err) | |
| 1237 | + | } | |
| 1238 | + | if err := SaveState(dir, State{URL: "https://x", RunnerID: "r", Name: "n", RunnerToken: "ermiterun_t"}); err != nil { | |
| 1239 | + | t.Fatal(err) | |
| 1240 | + | } | |
| 1241 | + | fi, err := os.Stat(dir) | |
| 1242 | + | if err != nil || fi.Mode().Perm() != 0o700 { | |
| 1243 | + | t.Fatalf("%v %v", fi, err) | |
| 1244 | + | } | |
| 1245 | + | if fi, _ := os.Stat(StateFile(dir)); fi.Mode().Perm() != 0o600 { | |
| 1246 | + | t.Fatalf("%v", fi.Mode()) | |
| 1247 | + | } | |
| 1248 | + | } | |
| @@ -9,6 +9,9 @@ import ( | |||||
| 9 | "io" | 9 | "io" | ||
| 10 | "net/http" | 10 | "net/http" | ||
| 11 | "os" | 11 | "os" | ||
| 12 | + | "path/filepath" | |||
| 13 | + | "regexp" | |||
| 14 | + | "strconv" | |||
| 12 | "strings" | 15 | "strings" | ||
| 13 | "testing" | 16 | "testing" | ||
| 14 | "time" | 17 | "time" | ||
| @@ -1173,3 +1176,73 @@ func TestSandboxSizesAreLeftToTheDriversConfiguration(t *testing.T) { | |||||
| 1173 | t.Fatalf("%+v", got) | 1176 | t.Fatalf("%+v", got) | ||
| 1174 | } | 1177 | } | ||
| 1175 | } | 1178 | } | ||
| 1179 | + | ||||
| 1180 | + | // ---- M5: shutdown | |||
| 1181 | + | ||||
| 1182 | + | // A job whose acknowledgement is pending when the runner shuts down is left to | |||
| 1183 | + | // the server (it requeues it after 30 s): the sandbox is never started, no | |||
| 1184 | + | // result is reported, and Run returns promptly. | |||
| 1185 | + | func TestShutdownBeforeTheAcknowledgementLeavesTheJobToTheServer(t *testing.T) { | |||
| 1186 | + | h := newHarness(t, simpleScript("x")) | |||
| 1187 | + | h.srv.jobs = []*ermite.Job{testJob("j1")} | |||
| 1188 | + | h.srv.beatHook = func(int) error { return connect.NewError(connect.CodeUnavailable, io.EOF) } // never acknowledged | |||
| 1189 | + | stop := h.run(t) | |||
| 1190 | + | eventually(t, "a failed acknowledgement", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return len(h.srv.beats) >= 2 }) | |||
| 1191 | + | start := time.Now() | |||
| 1192 | + | if err := stop(); err != nil { | |||
| 1193 | + | t.Fatal(err) | |||
| 1194 | + | } | |||
| 1195 | + | if time.Since(start) > 3*time.Second { | |||
| 1196 | + | t.Fatalf("Run took %v to return", time.Since(start)) | |||
| 1197 | + | } | |||
| 1198 | + | h.driver.mu.Lock() | |||
| 1199 | + | defer h.driver.mu.Unlock() | |||
| 1200 | + | if h.driver.news != 0 || h.driver.started != 0 { | |||
| 1201 | + | t.Fatalf("a sandbox was created for a job that was never acknowledged: news=%d started=%d", h.driver.news, h.driver.started) | |||
| 1202 | + | } | |||
| 1203 | + | if n := h.srv.completeCount(); n != 0 { | |||
| 1204 | + | t.Fatalf("CompleteJob was called %d times", n) | |||
| 1205 | + | } | |||
| 1206 | + | if !strings.Contains(h.logs.String(), "before the job was acknowledged") { | |||
| 1207 | + | t.Fatalf("no explanation in the log:\n%s", h.logs.String()) | |||
| 1208 | + | } | |||
| 1209 | + | } | |||
| 1210 | + | ||||
| 1211 | + | // The graceful shutdown of the defaults fits the time the unit gives. | |||
| 1212 | + | func TestShutdownBudgetFitsTheUnit(t *testing.T) { | |||
| 1213 | + | budget := Timing{}.ShutdownBudget() | |||
| 1214 | + | if budget != 90*time.Second { | |||
| 1215 | + | t.Fatalf("the budget is %v, README and contrib/ermite-runner.service say 90 s", budget) | |||
| 1216 | + | } | |||
| 1217 | + | unit, err := os.ReadFile("../../contrib/ermite-runner.service") | |||
| 1218 | + | if err != nil { | |||
| 1219 | + | t.Skip(err) | |||
| 1220 | + | } | |||
| 1221 | + | m := regexp.MustCompile(`(?m)^TimeoutStopSec=(\d+)$`).FindStringSubmatch(string(unit)) | |||
| 1222 | + | if m == nil { | |||
| 1223 | + | t.Fatal("no TimeoutStopSec in the unit") | |||
| 1224 | + | } | |||
| 1225 | + | secs, _ := strconv.Atoi(m[1]) | |||
| 1226 | + | if time.Duration(secs)*time.Second < budget+15*time.Second { | |||
| 1227 | + | t.Fatalf("TimeoutStopSec=%d does not leave 15 s of margin over the %v budget", secs, budget) | |||
| 1228 | + | } | |||
| 1229 | + | } | |||
| 1230 | + | ||||
| 1231 | + | // ---- m7: the state directory is private | |||
| 1232 | + | ||||
| 1233 | + | func TestSaveStateMakesAnExistingDirectoryPrivate(t *testing.T) { | |||
| 1234 | + | dir := filepath.Join(t.TempDir(), "state") | |||
| 1235 | + | if err := os.Mkdir(dir, 0o755); err != nil { | |||
| 1236 | + | t.Fatal(err) | |||
| 1237 | + | } | |||
| 1238 | + | if err := SaveState(dir, State{URL: "https://x", RunnerID: "r", Name: "n", RunnerToken: "ermiterun_t"}); err != nil { | |||
| 1239 | + | t.Fatal(err) | |||
| 1240 | + | } | |||
| 1241 | + | fi, err := os.Stat(dir) | |||
| 1242 | + | if err != nil || fi.Mode().Perm() != 0o700 { | |||
| 1243 | + | t.Fatalf("%v %v", fi, err) | |||
| 1244 | + | } | |||
| 1245 | + | if fi, _ := os.Stat(StateFile(dir)); fi.Mode().Perm() != 0o600 { | |||
| 1246 | + | t.Fatalf("%v", fi.Mode()) | |||
| 1247 | + | } | |||
| 1248 | + | } | |||