detached 7d17a05

feat(host): the host loop: register, fetch jobs, run them in sandboxes, report logs, steps and results

commit
7d17a05bc2dbadc03a704cb22f74d734cf9a3413
author
brice · · 2026-10-12 18:30 UTC
parent
90f106a
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
11 files · +1494 -66
Split
UnifiedSplit
M go.mod +1 -0
@@ -14,6 +14,7 @@ require (
1414 )
1515
1616 require (
17+ bnema.dev/ermite-api v0.0.0-20261012155218-f8f64413e63c // indirect
1718 buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect
1819 buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect
1920 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect
@@ -14,6 +14,7 @@ require (
14 )14 )
15 15
16 require (16 require (
17+ bnema.dev/ermite-api v0.0.0-20261012155218-f8f64413e63c // indirect
17 buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect18 buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect
18 buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect19 buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect
19 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect20 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect
M go.sum +2 -0 generated file: collapsed
load diff
A internal/execstream/split.go +38 -0
@@ -0,0 +1,38 @@
1+// SPDX-License-Identifier: MIT
2+
3+package execstream
4+
5+import (
6+ "strings"
7+ "unicode/utf8"
8+)
9+
10+// MaxLogLine is the largest piece of a log line that fits an AppendLog line
11+// (64 KiB); longer lines are cut by SplitLine.
12+const MaxLogLine = 64 * 1024
13+
14+// SplitLine cuts text into pieces of at most max bytes at UTF-8 character
15+// boundaries. The result has one piece for a short line, and always at least
16+// one (an empty line is one empty piece). Invalid UTF-8 in text is replaced
17+// with U+FFFD first, because the protocol only carries valid text.
18+func SplitLine(text string, max int) []string {
19+ if !utf8.ValidString(text) {
20+ text = strings.ToValidUTF8(text, "\ufffd")
21+ }
22+ if len(text) <= max {
23+ return []string{text}
24+ }
25+ var out []string
26+ for len(text) > max {
27+ cut := max
28+ for cut > 0 && !utf8.RuneStart(text[cut]) {
29+ cut--
30+ }
31+ if cut == 0 { // cannot happen with max >= utf8.UTFMax
32+ cut = max
33+ }
34+ out = append(out, text[:cut])
35+ text = text[cut:]
36+ }
37+ return append(out, text)
38+}
@@ -0,0 +1,38 @@
1+// SPDX-License-Identifier: MIT
2+
3+package execstream
4+
5+import (
6+ "strings"
7+ "unicode/utf8"
8+)
9+
10+// MaxLogLine is the largest piece of a log line that fits an AppendLog line
11+// (64 KiB); longer lines are cut by SplitLine.
12+const MaxLogLine = 64 * 1024
13+
14+// SplitLine cuts text into pieces of at most max bytes at UTF-8 character
15+// boundaries. The result has one piece for a short line, and always at least
16+// one (an empty line is one empty piece). Invalid UTF-8 in text is replaced
17+// with U+FFFD first, because the protocol only carries valid text.
18+func SplitLine(text string, max int) []string {
19+ if !utf8.ValidString(text) {
20+ text = strings.ToValidUTF8(text, "\ufffd")
21+ }
22+ if len(text) <= max {
23+ return []string{text}
24+ }
25+ var out []string
26+ for len(text) > max {
27+ cut := max
28+ for cut > 0 && !utf8.RuneStart(text[cut]) {
29+ cut--
30+ }
31+ if cut == 0 { // cannot happen with max >= utf8.UTFMax
32+ cut = max
33+ }
34+ out = append(out, text[:cut])
35+ text = text[cut:]
36+ }
37+ return append(out, text)
38+}
A internal/execstream/split_test.go +41 -0
@@ -0,0 +1,41 @@
1+// SPDX-License-Identifier: MIT
2+
3+package execstream
4+
5+import (
6+ "strings"
7+ "testing"
8+ "unicode/utf8"
9+)
10+
11+func TestSplitLine(t *testing.T) {
12+ if got := SplitLine("", 10); len(got) != 1 || got[0] != "" {
13+ t.Fatalf("empty: %q", got)
14+ }
15+ if got := SplitLine("short", 10); len(got) != 1 || got[0] != "short" {
16+ t.Fatalf("%q", got)
17+ }
18+ // Cuts at character boundaries and joins back exactly.
19+ in := strings.Repeat("é€😀a", 40) // 2, 3, 4, 1 bytes
20+ for _, max := range []int{4, 5, 7, 10, 33} {
21+ parts := SplitLine(in, max)
22+ var b strings.Builder
23+ for _, p := range parts {
24+ if len(p) > max || !utf8.ValidString(p) || p == "" {
25+ t.Fatalf("max %d: bad piece %q", max, p)
26+ }
27+ b.WriteString(p)
28+ }
29+ if b.String() != in {
30+ t.Fatalf("max %d: pieces do not rebuild the line", max)
31+ }
32+ }
33+ // Invalid UTF-8 is replaced before cutting.
34+ if got := SplitLine("a\xffb", 10); got[0] != "a\ufffdb" {
35+ t.Fatalf("%q", got)
36+ }
37+ big := SplitLine(strings.Repeat("x", 3*MaxLogLine+5), MaxLogLine)
38+ if len(big) != 4 || len(big[0]) != MaxLogLine || len(big[3]) != 5 {
39+ t.Fatalf("%d pieces", len(big))
40+ }
41+}
@@ -0,0 +1,41 @@
1+// SPDX-License-Identifier: MIT
2+
3+package execstream
4+
5+import (
6+ "strings"
7+ "testing"
8+ "unicode/utf8"
9+)
10+
11+func TestSplitLine(t *testing.T) {
12+ if got := SplitLine("", 10); len(got) != 1 || got[0] != "" {
13+ t.Fatalf("empty: %q", got)
14+ }
15+ if got := SplitLine("short", 10); len(got) != 1 || got[0] != "short" {
16+ t.Fatalf("%q", got)
17+ }
18+ // Cuts at character boundaries and joins back exactly.
19+ in := strings.Repeat("é€😀a", 40) // 2, 3, 4, 1 bytes
20+ for _, max := range []int{4, 5, 7, 10, 33} {
21+ parts := SplitLine(in, max)
22+ var b strings.Builder
23+ for _, p := range parts {
24+ if len(p) > max || !utf8.ValidString(p) || p == "" {
25+ t.Fatalf("max %d: bad piece %q", max, p)
26+ }
27+ b.WriteString(p)
28+ }
29+ if b.String() != in {
30+ t.Fatalf("max %d: pieces do not rebuild the line", max)
31+ }
32+ }
33+ // Invalid UTF-8 is replaced before cutting.
34+ if got := SplitLine("a\xffb", 10); got[0] != "a\ufffdb" {
35+ t.Fatalf("%q", got)
36+ }
37+ big := SplitLine(strings.Repeat("x", 3*MaxLogLine+5), MaxLogLine)
38+ if len(big) != 4 || len(big[0]) != MaxLogLine || len(big[3]) != 5 {
39+ t.Fatalf("%d pieces", len(big))
40+ }
41+}
M internal/guest/jobexec.go +1 -32
@@ -10,43 +10,12 @@ import (
1010 "strings"
1111 "syscall"
1212 "time"
13- "unicode/utf8"
1413
1514 "bnema.dev/ermite-runner/internal/execstream"
1615 "bnema.dev/ermite-runner/internal/execstream/execpb"
1716 "bnema.dev/ermite-runner/internal/executor"
1817 )
1918
20-// MaxLogLine is the largest piece of a log line that fits an AppendLog line
21-// (64 KiB); longer lines are cut by SplitLine.
22-const MaxLogLine = 64 * 1024
23-
24-// SplitLine cuts text into pieces of at most max bytes at UTF-8 character
25-// boundaries. The result has one piece for a short line, and always at least
26-// one (an empty line is one empty piece). Invalid UTF-8 in text is replaced
27-// with U+FFFD first, because the protocol only carries valid text.
28-func SplitLine(text string, max int) []string {
29- if !utf8.ValidString(text) {
30- text = strings.ToValidUTF8(text, "\ufffd")
31- }
32- if len(text) <= max {
33- return []string{text}
34- }
35- var out []string
36- for len(text) > max {
37- cut := max
38- for cut > 0 && !utf8.RuneStart(text[cut]) {
39- cut--
40- }
41- if cut == 0 { // cannot happen with max >= utf8.UTFMax
42- cut = max
43- }
44- out = append(out, text[:cut])
45- text = text[cut:]
46- }
47- return append(out, text)
48-}
49-
5019 // frameReporter is the executor.Reporter that writes frames on the stream.
5120 type frameReporter struct {
5221 conn *execstream.Conn
@@ -87,7 +56,7 @@ func (r *frameReporter) StepStarted(n int, name string, at time.Time) {
8756 }
8857
8958 func (r *frameReporter) Log(n int, at time.Time, line string) {
90- for i, piece := range SplitLine(line, MaxLogLine) {
59+ for i, piece := range execstream.SplitLine(line, execstream.MaxLogLine) {
9160 r.send(&execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: ts(at), Text: piece, Continued: i > 0}}})
9261 }
9362 }
@@ -10,43 +10,12 @@ import (
10 "strings"10 "strings"
11 "syscall"11 "syscall"
12 "time"12 "time"
13- "unicode/utf8"
14 13
15 "bnema.dev/ermite-runner/internal/execstream"14 "bnema.dev/ermite-runner/internal/execstream"
16 "bnema.dev/ermite-runner/internal/execstream/execpb"15 "bnema.dev/ermite-runner/internal/execstream/execpb"
17 "bnema.dev/ermite-runner/internal/executor"16 "bnema.dev/ermite-runner/internal/executor"
18 )17 )
19 18
20-// MaxLogLine is the largest piece of a log line that fits an AppendLog line
21-// (64 KiB); longer lines are cut by SplitLine.
22-const MaxLogLine = 64 * 1024
23-
24-// SplitLine cuts text into pieces of at most max bytes at UTF-8 character
25-// boundaries. The result has one piece for a short line, and always at least
26-// one (an empty line is one empty piece). Invalid UTF-8 in text is replaced
27-// with U+FFFD first, because the protocol only carries valid text.
28-func SplitLine(text string, max int) []string {
29- if !utf8.ValidString(text) {
30- text = strings.ToValidUTF8(text, "\ufffd")
31- }
32- if len(text) <= max {
33- return []string{text}
34- }
35- var out []string
36- for len(text) > max {
37- cut := max
38- for cut > 0 && !utf8.RuneStart(text[cut]) {
39- cut--
40- }
41- if cut == 0 { // cannot happen with max >= utf8.UTFMax
42- cut = max
43- }
44- out = append(out, text[:cut])
45- text = text[cut:]
46- }
47- return append(out, text)
48-}
49-
50 // frameReporter is the executor.Reporter that writes frames on the stream.19 // frameReporter is the executor.Reporter that writes frames on the stream.
51 type frameReporter struct {20 type frameReporter struct {
52 conn *execstream.Conn21 conn *execstream.Conn
@@ -87,7 +56,7 @@ func (r *frameReporter) StepStarted(n int, name string, at time.Time) {
87 }56 }
88 57
89 func (r *frameReporter) Log(n int, at time.Time, line string) {58 func (r *frameReporter) Log(n int, at time.Time, line string) {
90- for i, piece := range SplitLine(line, MaxLogLine) {59+ for i, piece := range execstream.SplitLine(line, execstream.MaxLogLine) {
91 r.send(&execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: ts(at), Text: piece, Continued: i > 0}}})60 r.send(&execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: ts(at), Text: piece, Continued: i > 0}}})
92 }61 }
93 }62 }
M internal/guest/jobexec_test.go +1 -34
@@ -11,44 +11,11 @@ import (
1111 "sync"
1212 "testing"
1313 "time"
14- "unicode/utf8"
1514
1615 "bnema.dev/ermite-runner/internal/execstream"
1716 "bnema.dev/ermite-runner/internal/execstream/execpb"
1817 )
1918
20-func TestSplitLine(t *testing.T) {
21- if got := SplitLine("", 10); len(got) != 1 || got[0] != "" {
22- t.Fatalf("empty: %q", got)
23- }
24- if got := SplitLine("short", 10); len(got) != 1 || got[0] != "short" {
25- t.Fatalf("%q", got)
26- }
27- // Cuts at character boundaries and joins back exactly.
28- in := strings.Repeat("é€😀a", 40) // 2, 3, 4, 1 bytes
29- for _, max := range []int{4, 5, 7, 10, 33} {
30- parts := SplitLine(in, max)
31- var b strings.Builder
32- for _, p := range parts {
33- if len(p) > max || !utf8.ValidString(p) || p == "" {
34- t.Fatalf("max %d: bad piece %q", max, p)
35- }
36- b.WriteString(p)
37- }
38- if b.String() != in {
39- t.Fatalf("max %d: pieces do not rebuild the line", max)
40- }
41- }
42- // Invalid UTF-8 is replaced before cutting.
43- if got := SplitLine("a\xffb", 10); got[0] != "a\ufffdb" {
44- t.Fatalf("%q", got)
45- }
46- big := SplitLine(strings.Repeat("x", 3*MaxLogLine+5), MaxLogLine)
47- if len(big) != 4 || len(big[0]) != MaxLogLine || len(big[3]) != 5 {
48- t.Fatalf("%d pieces", len(big))
49- }
50-}
51-
5219 // runJobSession starts Exec on a unix socket and drives it with a job.
5320 func runJobSession(t *testing.T, yaml string, cancelAfter time.Duration, tweak func(*ExecConfig)) (steps []string, logs []*execpb.Log, res *execpb.JobResult, exit *execpb.Exit) {
5421 t.Helper()
@@ -149,7 +116,7 @@ func TestExecJobLongLinesAreContinued(t *testing.T) {
149116 if strings.HasPrefix(l.Text, "yyyy") {
150117 pieces++
151118 total += len(l.Text)
152- if len(l.Text) > MaxLogLine || (pieces > 1) != l.Continued {
119+ if len(l.Text) > execstream.MaxLogLine || (pieces > 1) != l.Continued {
153120 t.Fatalf("piece %d: len %d continued %v", pieces, len(l.Text), l.Continued)
154121 }
155122 }
@@ -11,44 +11,11 @@ import (
11 "sync"11 "sync"
12 "testing"12 "testing"
13 "time"13 "time"
14- "unicode/utf8"
15 14
16 "bnema.dev/ermite-runner/internal/execstream"15 "bnema.dev/ermite-runner/internal/execstream"
17 "bnema.dev/ermite-runner/internal/execstream/execpb"16 "bnema.dev/ermite-runner/internal/execstream/execpb"
18 )17 )
19 18
20-func TestSplitLine(t *testing.T) {
21- if got := SplitLine("", 10); len(got) != 1 || got[0] != "" {
22- t.Fatalf("empty: %q", got)
23- }
24- if got := SplitLine("short", 10); len(got) != 1 || got[0] != "short" {
25- t.Fatalf("%q", got)
26- }
27- // Cuts at character boundaries and joins back exactly.
28- in := strings.Repeat("é€😀a", 40) // 2, 3, 4, 1 bytes
29- for _, max := range []int{4, 5, 7, 10, 33} {
30- parts := SplitLine(in, max)
31- var b strings.Builder
32- for _, p := range parts {
33- if len(p) > max || !utf8.ValidString(p) || p == "" {
34- t.Fatalf("max %d: bad piece %q", max, p)
35- }
36- b.WriteString(p)
37- }
38- if b.String() != in {
39- t.Fatalf("max %d: pieces do not rebuild the line", max)
40- }
41- }
42- // Invalid UTF-8 is replaced before cutting.
43- if got := SplitLine("a\xffb", 10); got[0] != "a\ufffdb" {
44- t.Fatalf("%q", got)
45- }
46- big := SplitLine(strings.Repeat("x", 3*MaxLogLine+5), MaxLogLine)
47- if len(big) != 4 || len(big[0]) != MaxLogLine || len(big[3]) != 5 {
48- t.Fatalf("%d pieces", len(big))
49- }
50-}
51-
52 // runJobSession starts Exec on a unix socket and drives it with a job.19 // runJobSession starts Exec on a unix socket and drives it with a job.
53 func runJobSession(t *testing.T, yaml string, cancelAfter time.Duration, tweak func(*ExecConfig)) (steps []string, logs []*execpb.Log, res *execpb.JobResult, exit *execpb.Exit) {20 func runJobSession(t *testing.T, yaml string, cancelAfter time.Duration, tweak func(*ExecConfig)) (steps []string, logs []*execpb.Log, res *execpb.JobResult, exit *execpb.Exit) {
54 t.Helper()21 t.Helper()
@@ -149,7 +116,7 @@ func TestExecJobLongLinesAreContinued(t *testing.T) {
149 if strings.HasPrefix(l.Text, "yyyy") {116 if strings.HasPrefix(l.Text, "yyyy") {
150 pieces++117 pieces++
151 total += len(l.Text)118 total += len(l.Text)
152- if len(l.Text) > MaxLogLine || (pieces > 1) != l.Continued {119+ if len(l.Text) > execstream.MaxLogLine || (pieces > 1) != l.Continued {
153 t.Fatalf("piece %d: len %d continued %v", pieces, len(l.Text), l.Continued)120 t.Fatalf("piece %d: len %d continued %v", pieces, len(l.Text), l.Continued)
154 }121 }
155 }122 }
A internal/host/config.go +160 -0
@@ -0,0 +1,160 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "encoding/json"
7+ "errors"
8+ "fmt"
9+ "os"
10+ "path/filepath"
11+ "regexp"
12+ "strings"
13+)
14+
15+// Environment variables of the host process. The sandbox settings
16+// (ERMITE_RUNNER_ISOLATION, _CAPACITY, _IMAGE, _JOB_*) are read by package
17+// sandbox.
18+const (
19+ EnvURL = "ERMITE_RUNNER_URL"
20+ EnvName = "ERMITE_RUNNER_NAME"
21+ EnvLabels = "ERMITE_RUNNER_LABELS"
22+ EnvRegisterToken = "ERMITE_RUNNER_REGISTER_TOKEN"
23+)
24+
25+// Config is the host part of the runner configuration.
26+type Config struct {
27+ // URL is the base URL of the ermite server.
28+ URL string
29+ // Name is the runner name: 1 to 64 letters, digits, ".", "_" and "-".
30+ Name string
31+ // Labels are the extra labels of the runner.
32+ Labels []string
33+ // StateDir holds the runner token (mode 0600).
34+ StateDir string
35+ // Capacity is the number of jobs in parallel.
36+ Capacity int
37+ // Version is the version of the runner binary.
38+ Version string
39+}
40+
41+var nameRe = regexp.MustCompile(`^[A-Za-z0-9._-]{1,64}$`)
42+
43+// LoadConfig reads the host settings. hostname is the default runner name.
44+// stateDir and capacity come from the sandbox configuration, so that both
45+// agree.
46+func LoadConfig(getenv func(string) string, hostname, stateDir string, capacity int, version string) (Config, error) {
47+ c := Config{URL: strings.TrimSpace(getenv(EnvURL)), Name: strings.TrimSpace(getenv(EnvName)), StateDir: stateDir, Capacity: capacity, Version: version}
48+ if c.Name == "" {
49+ c.Name = sanitizeName(hostname)
50+ }
51+ if !nameRe.MatchString(c.Name) {
52+ return c, fmt.Errorf("%s: %q is not a valid runner name (1 to 64 letters, digits, '.', '_' and '-')", EnvName, c.Name)
53+ }
54+ seen := map[string]bool{}
55+ for _, l := range strings.Split(getenv(EnvLabels), ",") {
56+ l = strings.TrimSpace(l)
57+ if l == "" {
58+ continue
59+ }
60+ if !nameRe.MatchString(l) {
61+ return c, fmt.Errorf("%s: %q is not a valid label (1 to 64 letters, digits, '.', '_' and '-')", EnvLabels, l)
62+ }
63+ if k := strings.ToLower(l); !seen[k] {
64+ seen[k] = true
65+ c.Labels = append(c.Labels, l)
66+ }
67+ }
68+ return c, nil
69+}
70+
71+func sanitizeName(s string) string {
72+ b := []byte(s)
73+ for i, ch := range b {
74+ if !(ch >= 'a' && ch <= 'z' || ch >= 'A' && ch <= 'Z' || ch >= '0' && ch <= '9' || ch == '.' || ch == '_' || ch == '-') {
75+ b[i] = '-'
76+ }
77+ }
78+ if len(b) > 64 {
79+ b = b[:64]
80+ }
81+ if len(b) == 0 {
82+ return "runner"
83+ }
84+ return string(b)
85+}
86+
87+// State is what register stores: the runner token and who it belongs to. The
88+// file is mode 0600 and the token is never logged.
89+type State struct {
90+ URL string `json:"url"`
91+ RunnerID string `json:"runner_id"`
92+ Name string `json:"name"`
93+ RunnerToken string `json:"runner_token"`
94+}
95+
96+// StateFile is the path of the state file in dir.
97+func StateFile(dir string) string { return filepath.Join(dir, "runner.json") }
98+
99+// ErrNotRegistered is returned by LoadState when no runner is registered.
100+var ErrNotRegistered = errors.New("this runner is not registered: run `ermite-runner register` first")
101+
102+// SaveState writes the state atomically with mode 0600.
103+func SaveState(dir string, s State) error {
104+ if err := os.MkdirAll(dir, 0o700); err != nil {
105+ return err
106+ }
107+ b, err := json.MarshalIndent(s, "", " ")
108+ if err != nil {
109+ return err
110+ }
111+ tmp, err := os.CreateTemp(dir, ".runner-*.json")
112+ if err != nil {
113+ return err
114+ }
115+ defer func() { _ = os.Remove(tmp.Name()) }()
116+ if err := tmp.Chmod(0o600); err != nil {
117+ _ = tmp.Close()
118+ return err
119+ }
120+ if _, err := tmp.Write(append(b, '\n')); err != nil {
121+ _ = tmp.Close()
122+ return err
123+ }
124+ if err := tmp.Sync(); err != nil {
125+ _ = tmp.Close()
126+ return err
127+ }
128+ if err := tmp.Close(); err != nil {
129+ return err
130+ }
131+ return os.Rename(tmp.Name(), StateFile(dir))
132+}
133+
134+// LoadState reads the state; it refuses a file readable by others, because the
135+// runner token is a secret.
136+func LoadState(dir string) (State, error) {
137+ p := StateFile(dir)
138+ st, err := os.Stat(p)
139+ if errors.Is(err, os.ErrNotExist) {
140+ return State{}, ErrNotRegistered
141+ }
142+ if err != nil {
143+ return State{}, err
144+ }
145+ if st.Mode().Perm()&0o077 != 0 {
146+ return State{}, fmt.Errorf("%s is accessible by other users (mode %o): chmod 600 it", p, st.Mode().Perm())
147+ }
148+ b, err := os.ReadFile(p)
149+ if err != nil {
150+ return State{}, err
151+ }
152+ var s State
153+ if err := json.Unmarshal(b, &s); err != nil {
154+ return State{}, fmt.Errorf("%s: %w", p, err)
155+ }
156+ if s.RunnerToken == "" {
157+ return State{}, fmt.Errorf("%s has no runner token", p)
158+ }
159+ return s, nil
160+}
@@ -0,0 +1,160 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "encoding/json"
7+ "errors"
8+ "fmt"
9+ "os"
10+ "path/filepath"
11+ "regexp"
12+ "strings"
13+)
14+
15+// Environment variables of the host process. The sandbox settings
16+// (ERMITE_RUNNER_ISOLATION, _CAPACITY, _IMAGE, _JOB_*) are read by package
17+// sandbox.
18+const (
19+ EnvURL = "ERMITE_RUNNER_URL"
20+ EnvName = "ERMITE_RUNNER_NAME"
21+ EnvLabels = "ERMITE_RUNNER_LABELS"
22+ EnvRegisterToken = "ERMITE_RUNNER_REGISTER_TOKEN"
23+)
24+
25+// Config is the host part of the runner configuration.
26+type Config struct {
27+ // URL is the base URL of the ermite server.
28+ URL string
29+ // Name is the runner name: 1 to 64 letters, digits, ".", "_" and "-".
30+ Name string
31+ // Labels are the extra labels of the runner.
32+ Labels []string
33+ // StateDir holds the runner token (mode 0600).
34+ StateDir string
35+ // Capacity is the number of jobs in parallel.
36+ Capacity int
37+ // Version is the version of the runner binary.
38+ Version string
39+}
40+
41+var nameRe = regexp.MustCompile(`^[A-Za-z0-9._-]{1,64}$`)
42+
43+// LoadConfig reads the host settings. hostname is the default runner name.
44+// stateDir and capacity come from the sandbox configuration, so that both
45+// agree.
46+func LoadConfig(getenv func(string) string, hostname, stateDir string, capacity int, version string) (Config, error) {
47+ c := Config{URL: strings.TrimSpace(getenv(EnvURL)), Name: strings.TrimSpace(getenv(EnvName)), StateDir: stateDir, Capacity: capacity, Version: version}
48+ if c.Name == "" {
49+ c.Name = sanitizeName(hostname)
50+ }
51+ if !nameRe.MatchString(c.Name) {
52+ return c, fmt.Errorf("%s: %q is not a valid runner name (1 to 64 letters, digits, '.', '_' and '-')", EnvName, c.Name)
53+ }
54+ seen := map[string]bool{}
55+ for _, l := range strings.Split(getenv(EnvLabels), ",") {
56+ l = strings.TrimSpace(l)
57+ if l == "" {
58+ continue
59+ }
60+ if !nameRe.MatchString(l) {
61+ return c, fmt.Errorf("%s: %q is not a valid label (1 to 64 letters, digits, '.', '_' and '-')", EnvLabels, l)
62+ }
63+ if k := strings.ToLower(l); !seen[k] {
64+ seen[k] = true
65+ c.Labels = append(c.Labels, l)
66+ }
67+ }
68+ return c, nil
69+}
70+
71+func sanitizeName(s string) string {
72+ b := []byte(s)
73+ for i, ch := range b {
74+ if !(ch >= 'a' && ch <= 'z' || ch >= 'A' && ch <= 'Z' || ch >= '0' && ch <= '9' || ch == '.' || ch == '_' || ch == '-') {
75+ b[i] = '-'
76+ }
77+ }
78+ if len(b) > 64 {
79+ b = b[:64]
80+ }
81+ if len(b) == 0 {
82+ return "runner"
83+ }
84+ return string(b)
85+}
86+
87+// State is what register stores: the runner token and who it belongs to. The
88+// file is mode 0600 and the token is never logged.
89+type State struct {
90+ URL string `json:"url"`
91+ RunnerID string `json:"runner_id"`
92+ Name string `json:"name"`
93+ RunnerToken string `json:"runner_token"`
94+}
95+
96+// StateFile is the path of the state file in dir.
97+func StateFile(dir string) string { return filepath.Join(dir, "runner.json") }
98+
99+// ErrNotRegistered is returned by LoadState when no runner is registered.
100+var ErrNotRegistered = errors.New("this runner is not registered: run `ermite-runner register` first")
101+
102+// SaveState writes the state atomically with mode 0600.
103+func SaveState(dir string, s State) error {
104+ if err := os.MkdirAll(dir, 0o700); err != nil {
105+ return err
106+ }
107+ b, err := json.MarshalIndent(s, "", " ")
108+ if err != nil {
109+ return err
110+ }
111+ tmp, err := os.CreateTemp(dir, ".runner-*.json")
112+ if err != nil {
113+ return err
114+ }
115+ defer func() { _ = os.Remove(tmp.Name()) }()
116+ if err := tmp.Chmod(0o600); err != nil {
117+ _ = tmp.Close()
118+ return err
119+ }
120+ if _, err := tmp.Write(append(b, '\n')); err != nil {
121+ _ = tmp.Close()
122+ return err
123+ }
124+ if err := tmp.Sync(); err != nil {
125+ _ = tmp.Close()
126+ return err
127+ }
128+ if err := tmp.Close(); err != nil {
129+ return err
130+ }
131+ return os.Rename(tmp.Name(), StateFile(dir))
132+}
133+
134+// LoadState reads the state; it refuses a file readable by others, because the
135+// runner token is a secret.
136+func LoadState(dir string) (State, error) {
137+ p := StateFile(dir)
138+ st, err := os.Stat(p)
139+ if errors.Is(err, os.ErrNotExist) {
140+ return State{}, ErrNotRegistered
141+ }
142+ if err != nil {
143+ return State{}, err
144+ }
145+ if st.Mode().Perm()&0o077 != 0 {
146+ return State{}, fmt.Errorf("%s is accessible by other users (mode %o): chmod 600 it", p, st.Mode().Perm())
147+ }
148+ b, err := os.ReadFile(p)
149+ if err != nil {
150+ return State{}, err
151+ }
152+ var s State
153+ if err := json.Unmarshal(b, &s); err != nil {
154+ return State{}, fmt.Errorf("%s: %w", p, err)
155+ }
156+ if s.RunnerToken == "" {
157+ return State{}, fmt.Errorf("%s has no runner token", p)
158+ }
159+ return s, nil
160+}
A internal/host/job.go +558 -0
@@ -0,0 +1,558 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "sort"
10+ "strings"
11+ "sync"
12+ "time"
13+
14+ "connectrpc.com/connect"
15+ "google.golang.org/protobuf/proto"
16+ "google.golang.org/protobuf/types/known/timestamppb"
17+
18+ ermite "bnema.dev/ermite-api/gen/ermite"
19+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20+
21+ "bnema.dev/ermite-runner/internal/execstream"
22+ "bnema.dev/ermite-runner/internal/execstream/execpb"
23+ "bnema.dev/ermite-runner/internal/sandbox"
24+)
25+
26+// jobClient is what a job needs from the RunnerService client (job token).
27+type jobClient interface {
28+ callClient
29+ Heartbeat(context.Context, *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error)
30+ WatchJob(context.Context, *connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error)
31+ CompleteJob(context.Context, *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error)
32+}
33+
34+var _ jobClient = ermiteconnect.RunnerServiceClient(nil)
35+
36+// jobRun is the host's side of one job: the assignment, the sandbox, and the
37+// calls to ermite with the job token.
38+type jobRun struct {
39+ r *Runner
40+ job *ermite.Job
41+ client jobClient
42+ t Timing
43+ logf func(string, ...any)
44+
45+ // ctx lives until the job is finished or abandoned. Shutdown does not end
46+ // it: a shutdown is a cancellation that the job reports.
47+ ctx context.Context
48+ cancel context.CancelCauseFunc
49+
50+ cancelCh chan *execpb.Cancel // the first value is sent to the executor
51+ cancelMu sync.Mutex
52+ cancelAt string // reason of the first cancel request, "" if none
53+ stopping bool // the runner is shutting down
54+
55+ mu sync.Mutex
56+ steps map[int32]*ermite.StepUpdate
57+ result *execpb.JobResult
58+ outbox *outbox
59+ endedBy error // the server ended the job, or the assignment is gone
60+}
61+
62+var errAbandoned = errors.New("the server ended the job or revoked the assignment")
63+
64+func (j *jobRun) abandon(err error) {
65+ j.mu.Lock()
66+ if j.endedBy == nil {
67+ j.endedBy = err
68+ }
69+ j.mu.Unlock()
70+ j.cancel(errAbandoned)
71+}
72+
73+func (j *jobRun) abandoned() error {
74+ j.mu.Lock()
75+ defer j.mu.Unlock()
76+ return j.endedBy
77+}
78+
79+// requestCancel asks the executor to stop; only the first request counts.
80+func (j *jobRun) requestCancel(reason string, kind execpb.CancelKind) {
81+ j.cancelMu.Lock()
82+ defer j.cancelMu.Unlock()
83+ if j.cancelAt != "" {
84+ return
85+ }
86+ j.cancelAt = reason
87+ j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}
88+}
89+
90+func (j *jobRun) cancelRequested() bool {
91+ j.cancelMu.Lock()
92+ defer j.cancelMu.Unlock()
93+ return j.cancelAt != ""
94+}
95+
96+// heartbeat sends the acknowledgement. It retries every HeartbeatRetry until
97+// it succeeds, so that the 30 s of the server are used well; it returns false
98+// when the assignment is gone.
99+func (j *jobRun) ack(ctx context.Context) bool {
100+ for {
101+ err := j.beat(ctx)
102+ if err == nil {
103+ return true
104+ }
105+ if ctx.Err() != nil {
106+ return false
107+ }
108+ switch classify(err) {
109+ case kindEnded, kindAbandoned:
110+ j.abandon(err)
111+ return false
112+ }
113+ j.logf("first heartbeat failed, retrying: %v", err)
114+ if !sleep(ctx, j.t.HeartbeatRetry) {
115+ return false
116+ }
117+ }
118+}
119+
120+func (j *jobRun) beat(ctx context.Context) error {
121+ cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
122+ defer cancel()
123+ _, err := j.client.Heartbeat(cctx, connect.NewRequest(&ermite.HeartbeatRequest{JobId: j.job.JobId}))
124+ return err
125+}
126+
127+// heartbeats keeps the job alive: every HeartbeatEvery, and after a failure
128+// every HeartbeatRetry until one succeeds (runner.proto: Heartbeat).
129+func (j *jobRun) heartbeats(ctx context.Context) {
130+ wait := j.t.HeartbeatEvery
131+ for {
132+ if !sleep(ctx, wait) {
133+ return
134+ }
135+ if err := j.beat(ctx); err != nil {
136+ if ctx.Err() != nil {
137+ return
138+ }
139+ switch classify(err) {
140+ case kindEnded, kindAbandoned:
141+ j.abandon(err)
142+ return
143+ }
144+ j.logf("heartbeat failed, retrying in %s: %v", j.t.HeartbeatRetry, err)
145+ wait = j.t.HeartbeatRetry
146+ continue
147+ }
148+ wait = j.t.HeartbeatEvery
149+ }
150+}
151+
152+// watch follows WatchJob: a cancel request stops the job, an Ended message or
153+// the end-of-job error abandons it. The stream is reopened after any other
154+// error and after a clean close without Ended.
155+func (j *jobRun) watch(ctx context.Context) {
156+ for ctx.Err() == nil {
157+ stream, err := j.client.WatchJob(ctx, connect.NewRequest(&ermite.WatchJobRequest{JobId: j.job.JobId}))
158+ if err == nil {
159+ for stream.Receive() {
160+ switch m := stream.Msg().Msg.(type) {
161+ case *ermite.WatchJobResponse_Cancel:
162+ j.requestCancel(cancelText(m.Cancel), kindOf(m.Cancel.GetReason()))
163+ case *ermite.WatchJobResponse_Ended:
164+ _ = stream.Close()
165+ j.abandon(fmt.Errorf("job ended: %s", m.Ended.GetConclusion()))
166+ return
167+ }
168+ }
169+ err = stream.Err()
170+ _ = stream.Close()
171+ }
172+ if ctx.Err() != nil {
173+ return
174+ }
175+ if err != nil {
176+ switch classify(err) {
177+ case kindEnded, kindAbandoned:
178+ j.abandon(err)
179+ return
180+ }
181+ j.logf("WatchJob: %v", err)
182+ }
183+ if !sleep(ctx, j.t.WatchRetry) {
184+ return
185+ }
186+ }
187+}
188+
189+func cancelText(c *ermite.CancelRequest) string {
190+ if c.GetMessage() != "" {
191+ return c.GetMessage()
192+ }
193+ return c.GetReason().String()
194+}
195+
196+func kindOf(r ermite.CancelReason) execpb.CancelKind {
197+ if r == ermite.CancelReason_CANCEL_REASON_TIMEOUT {
198+ return execpb.CancelKind_CANCEL_KIND_TIMEOUT
199+ }
200+ // User, fail-fast, and any reason this runner does not know: cancelled.
201+ return execpb.CancelKind_CANCEL_KIND_USER
202+}
203+
204+// ---- steps
205+
206+func (j *jobRun) stepStarted(s *execpb.StepStarted) {
207+ u := &ermite.StepUpdate{Number: s.Number, Name: s.Name, State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
208+ j.mu.Lock()
209+ j.steps[s.Number] = u
210+ j.mu.Unlock()
211+ j.outbox.Step(u)
212+}
213+
214+func (j *jobRun) stepCompleted(s *execpb.StepCompleted) {
215+ j.mu.Lock()
216+ u := j.steps[s.Number]
217+ if u == nil {
218+ u = &ermite.StepUpdate{Number: s.Number, StartedAt: tsOf(s.UnixNano)}
219+ } else {
220+ u = cloneStep(u)
221+ }
222+ if s.Name != "" {
223+ u.Name = s.Name
224+ }
225+ u.State = ermite.StepState_STEP_STATE_COMPLETED
226+ u.Conclusion = conclusionOf(s.Conclusion)
227+ if u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
228+ u.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
229+ }
230+ u.CompletedAt = tsOf(s.UnixNano)
231+ j.steps[s.Number] = u
232+ j.mu.Unlock()
233+ j.outbox.Step(u)
234+}
235+
236+func cloneStep(u *ermite.StepUpdate) *ermite.StepUpdate {
237+ return proto.Clone(u).(*ermite.StepUpdate)
238+}
239+
240+func tsOf(unixNano int64) *timestamppb.Timestamp {
241+ if unixNano == 0 {
242+ return timestamppb.Now()
243+ }
244+ return timestamppb.New(time.Unix(0, unixNano))
245+}
246+
247+func conclusionOf(c execpb.StepConclusion) ermite.CiConclusion {
248+ switch c {
249+ case execpb.StepConclusion_STEP_CONCLUSION_SUCCESS:
250+ return ermite.CiConclusion_CI_CONCLUSION_SUCCESS
251+ case execpb.StepConclusion_STEP_CONCLUSION_FAILURE:
252+ return ermite.CiConclusion_CI_CONCLUSION_FAILURE
253+ case execpb.StepConclusion_STEP_CONCLUSION_CANCELLED:
254+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED
255+ case execpb.StepConclusion_STEP_CONCLUSION_SKIPPED:
256+ return ermite.CiConclusion_CI_CONCLUSION_SKIPPED
257+ case execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT:
258+ return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT
259+ }
260+ return ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED
261+}
262+
263+// finalSteps closes the steps that never completed (the executor died or was
264+// stopped) with the conclusion of the job, and returns all snapshots in order.
265+func (j *jobRun) finalSteps(jobConclusion ermite.CiConclusion) []*ermite.StepUpdate {
266+ j.mu.Lock()
267+ defer j.mu.Unlock()
268+ end := ermite.CiConclusion_CI_CONCLUSION_FAILURE
269+ if jobConclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED || jobConclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
270+ end = ermite.CiConclusion_CI_CONCLUSION_CANCELLED
271+ }
272+ var out []*ermite.StepUpdate
273+ for _, u := range j.steps {
274+ if u.State != ermite.StepState_STEP_STATE_COMPLETED {
275+ u = cloneStep(u)
276+ u.State, u.Conclusion, u.CompletedAt = ermite.StepState_STEP_STATE_COMPLETED, end, timestamppb.Now()
277+ }
278+ out = append(out, u)
279+ }
280+ sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
281+ return out
282+}
283+
284+// ---- the job
285+
286+// run runs the whole job: acknowledgement, sandbox, executor session, final
287+// report. It returns when everything is over and the sandbox is gone.
288+func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
289+ defer j.cancel(nil)
290+ ctx := j.ctx
291+
292+ // 1. The acknowledgement comes first, before anything else.
293+ if !j.ack(ctx) {
294+ j.logf("job %s: assignment lost before it started", j.job.JobId)
295+ return
296+ }
297+ var bg sync.WaitGroup
298+ bg.Add(2)
299+ go func() { defer bg.Done(); j.heartbeats(ctx) }()
300+ go func() { defer bg.Done(); j.watch(ctx) }()
301+ defer bg.Wait()
302+
303+ j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
304+ j.outbox.start(ctx)
305+
306+ // A shutdown is a cancellation of kind shutdown, then a hard stop.
307+ stopHard := make(chan struct{})
308+ go func() {
309+ select {
310+ case <-shutdown:
311+ case <-ctx.Done():
312+ return
313+ }
314+ j.cancelMu.Lock()
315+ j.stopping = true
316+ j.cancelMu.Unlock()
317+ j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
318+ select {
319+ case <-time.After(j.t.ShutdownGrace):
320+ close(stopHard)
321+ case <-ctx.Done():
322+ }
323+ }()
324+
325+ conclusion, outputs, shutdownFlag := j.execute(ctx, rt, stopHard)
326+
327+ // 2. Nothing more is sent to a job that is over.
328+ if err := j.abandoned(); err != nil {
329+ j.logf("job %s: not reporting a result: %v", j.job.JobId, err)
330+ return
331+ }
332+ // The report is made even when the runner is shutting down: it uses its own
333+ // deadline.
334+ rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.CompleteDeadline)
335+ defer cancel()
336+ // Output limits are checked before the log is flushed: a job whose outputs
337+ // the server would refuse fails without outputs, with a line in the log
338+ // (runner.proto: CompleteJobRequest.outputs), and lines sent after
339+ // CompleteJob are rejected.
340+ if msg := checkOutputs(outputs); msg != "" {
341+ j.sandboxError(msg)
342+ outputs, conclusion, shutdownFlag = nil, ermite.CiConclusion_CI_CONCLUSION_FAILURE, false
343+ }
344+ j.outbox.Close(rctx)
345+ if err := j.abandoned(); err != nil {
346+ return
347+ }
348+ j.complete(rctx, conclusion, outputs, shutdownFlag)
349+}
350+
351+// execute starts the sandbox, runs the job in it and returns the conclusion.
352+func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) (ermite.CiConclusion, map[string]string, bool) {
353+ fail := func(format string, a ...any) (ermite.CiConclusion, map[string]string, bool) {
354+ j.failEarly(fmt.Sprintf(format, a...))
355+ return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
356+ }
357+ sb, err := rt.Driver.New()
358+ if err != nil {
359+ return fail("The runner has no free sandbox: %v", err)
360+ }
361+ // Stop runs to completion even when the job context is done.
362+ defer func() {
363+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute)
364+ defer cancel()
365+ if err := sb.Stop(sctx); err != nil {
366+ j.logf("job %s: stopping the sandbox: %v", j.job.JobId, err)
367+ if errors.Is(err, sandbox.ErrLeaked) {
368+ rt.markLeaked()
369+ }
370+ }
371+ }()
372+
373+ // Starting the sandbox is bounded by the job context and a hard stop.
374+ startCtx, cancelStart := context.WithCancel(ctx)
375+ stopWatch := make(chan struct{})
376+ go func() {
377+ select {
378+ case <-stopHard:
379+ cancelStart()
380+ case <-stopWatch:
381+ }
382+ }()
383+ conn, err := sb.Start(startCtx, rt.spec(j.job))
384+ close(stopWatch)
385+ cancelStart()
386+ if err != nil {
387+ if ctx.Err() != nil || j.cancelRequested() {
388+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, j.isStopping()
389+ }
390+ return fail("The sandbox did not start: %v", err)
391+ }
392+ defer func() { _ = conn.Close() }()
393+
394+ // Hard stops (grace elapsed, or the job was abandoned) close the stream.
395+ go func() {
396+ select {
397+ case <-stopHard:
398+ case <-ctx.Done():
399+ }
400+ _ = conn.Close()
401+ }()
402+
403+ run := &execpb.Run{Job: &execpb.JobSpec{
404+ WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
405+ Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: j.job.TimeoutMinutes,
406+ RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: "self-hosted",
407+ }}
408+ _, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
409+ StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
410+ Log: func(l *execpb.Log) {
411+ j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: l.Text, continued: l.Continued})
412+ },
413+ JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
414+ })
415+ j.mu.Lock()
416+ res := j.result
417+ j.mu.Unlock()
418+
419+ switch {
420+ case res != nil:
421+ c := conclusionOf(res.Conclusion)
422+ if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {
423+ c = ermite.CiConclusion_CI_CONCLUSION_FAILURE
424+ }
425+ if res.Error != "" {
426+ j.logf("job %s: %s", j.job.JobId, res.Error)
427+ }
428+ return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
429+ case j.cancelRequested() || ctx.Err() != nil:
430+ // The executor did not report: it was stopped.
431+ if j.isStopping() {
432+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true
433+ }
434+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false
435+ }
436+ why := "the executor ended without a result"
437+ if err != nil {
438+ why = err.Error()
439+ } else if exit != nil && exit.Error != "" {
440+ why = exit.Error
441+ }
442+ j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))
443+ return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
444+}
445+
446+func (j *jobRun) isStopping() bool {
447+ j.cancelMu.Lock()
448+ defer j.cancelMu.Unlock()
449+ return j.stopping
450+}
451+
452+// failEarly reports a job that could not start: step 1 fails with the reason
453+// in its log.
454+func (j *jobRun) failEarly(msg string) {
455+ now := time.Now()
456+ j.stepStarted(&execpb.StepStarted{Number: 1, Name: "Set up job", UnixNano: now.UnixNano()})
457+ j.outbox.Log(logLine{step: 1, unixNano: now.UnixNano(), text: "##[error]" + msg})
458+ j.stepCompleted(&execpb.StepCompleted{Number: 1, Name: "Set up job", Conclusion: execpb.StepConclusion_STEP_CONCLUSION_FAILURE, UnixNano: time.Now().UnixNano()})
459+}
460+
461+// sandboxError writes the reason into the log of the step that was running.
462+func (j *jobRun) sandboxError(msg string) {
463+ j.mu.Lock()
464+ var last int32 = 1
465+ for n := range j.steps {
466+ last = max(last, n)
467+ }
468+ j.mu.Unlock()
469+ j.outbox.Log(logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
470+}
471+
472+// complete reports the end of the job. CompleteJob is idempotent, so a lost
473+// response is retried until it is answered (applied or not).
474+func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {
475+ req := &ermite.CompleteJobRequest{
476+ JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
477+ RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
478+ Steps: j.finalSteps(conclusion),
479+ }
480+ b := backoff{min: j.t.BackoffMin, max: j.t.BackoffMax}
481+ for {
482+ cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
483+ resp, err := j.client.CompleteJob(cctx, connect.NewRequest(req))
484+ cancel()
485+ if err == nil {
486+ if resp.Msg.Conclusion != conclusion {
487+ j.logf("job %s: ended as %s, the server holds %s", j.job.JobId, conclusion, resp.Msg.Conclusion)
488+ }
489+ return
490+ }
491+ if ctx.Err() != nil {
492+ j.logf("job %s: could not report the result: %v", j.job.JobId, ctx.Err())
493+ return
494+ }
495+ switch classify(err) {
496+ case kindEnded, kindAbandoned:
497+ return
498+ case kindPermanent:
499+ j.logf("job %s: CompleteJob refused: %v", j.job.JobId, err)
500+ return
501+ }
502+ j.logf("CompleteJob failed, retrying: %v", err)
503+ if !sleep(ctx, b.next()) {
504+ return
505+ }
506+ }
507+}
508+
509+// Limits of CompleteJobRequest.outputs.
510+const (
511+ maxOutputBytes = 1 << 20
512+ maxOutputName = 255
513+)
514+
515+// checkOutputs returns a message when the server would refuse the outputs: a
516+// name of 1 to 255 letters, digits, "_" and "-" starting with a letter or "_",
517+// no two names equal ignoring case, each value and the sum of names and values
518+// at most 1 MiB.
519+func checkOutputs(outputs map[string]string) string {
520+ total := 0
521+ seen := map[string]string{}
522+ names := make([]string, 0, len(outputs))
523+ for n := range outputs {
524+ names = append(names, n)
525+ }
526+ sort.Strings(names)
527+ for _, n := range names {
528+ v := outputs[n]
529+ if !validOutputName(n) {
530+ return fmt.Sprintf("Job output name %q is not valid: use 1 to 255 letters, digits, '_' and '-', starting with a letter or '_'.", n)
531+ }
532+ if prev, dup := seen[strings.ToLower(n)]; dup {
533+ return fmt.Sprintf("Job outputs %q and %q differ only by case.", prev, n)
534+ }
535+ seen[strings.ToLower(n)] = n
536+ if len(v) > maxOutputBytes {
537+ return fmt.Sprintf("Job output %q is larger than 1 MiB.", n)
538+ }
539+ total += len(n) + len(v)
540+ }
541+ if total > maxOutputBytes {
542+ return "The job outputs are larger than 1 MiB in total."
543+ }
544+ return ""
545+}
546+
547+func validOutputName(n string) bool {
548+ if n == "" || len(n) > maxOutputName {
549+ return false
550+ }
551+ for i, r := range n {
552+ letter := r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r == '_'
553+ if !(letter || (i > 0 && (r >= '0' && r <= '9' || r == '-'))) {
554+ return false
555+ }
556+ }
557+ return true
558+}
@@ -0,0 +1,558 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "sort"
10+ "strings"
11+ "sync"
12+ "time"
13+
14+ "connectrpc.com/connect"
15+ "google.golang.org/protobuf/proto"
16+ "google.golang.org/protobuf/types/known/timestamppb"
17+
18+ ermite "bnema.dev/ermite-api/gen/ermite"
19+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20+
21+ "bnema.dev/ermite-runner/internal/execstream"
22+ "bnema.dev/ermite-runner/internal/execstream/execpb"
23+ "bnema.dev/ermite-runner/internal/sandbox"
24+)
25+
26+// jobClient is what a job needs from the RunnerService client (job token).
27+type jobClient interface {
28+ callClient
29+ Heartbeat(context.Context, *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error)
30+ WatchJob(context.Context, *connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error)
31+ CompleteJob(context.Context, *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error)
32+}
33+
34+var _ jobClient = ermiteconnect.RunnerServiceClient(nil)
35+
36+// jobRun is the host's side of one job: the assignment, the sandbox, and the
37+// calls to ermite with the job token.
38+type jobRun struct {
39+ r *Runner
40+ job *ermite.Job
41+ client jobClient
42+ t Timing
43+ logf func(string, ...any)
44+
45+ // ctx lives until the job is finished or abandoned. Shutdown does not end
46+ // it: a shutdown is a cancellation that the job reports.
47+ ctx context.Context
48+ cancel context.CancelCauseFunc
49+
50+ cancelCh chan *execpb.Cancel // the first value is sent to the executor
51+ cancelMu sync.Mutex
52+ cancelAt string // reason of the first cancel request, "" if none
53+ stopping bool // the runner is shutting down
54+
55+ mu sync.Mutex
56+ steps map[int32]*ermite.StepUpdate
57+ result *execpb.JobResult
58+ outbox *outbox
59+ endedBy error // the server ended the job, or the assignment is gone
60+}
61+
62+var errAbandoned = errors.New("the server ended the job or revoked the assignment")
63+
64+func (j *jobRun) abandon(err error) {
65+ j.mu.Lock()
66+ if j.endedBy == nil {
67+ j.endedBy = err
68+ }
69+ j.mu.Unlock()
70+ j.cancel(errAbandoned)
71+}
72+
73+func (j *jobRun) abandoned() error {
74+ j.mu.Lock()
75+ defer j.mu.Unlock()
76+ return j.endedBy
77+}
78+
79+// requestCancel asks the executor to stop; only the first request counts.
80+func (j *jobRun) requestCancel(reason string, kind execpb.CancelKind) {
81+ j.cancelMu.Lock()
82+ defer j.cancelMu.Unlock()
83+ if j.cancelAt != "" {
84+ return
85+ }
86+ j.cancelAt = reason
87+ j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}
88+}
89+
90+func (j *jobRun) cancelRequested() bool {
91+ j.cancelMu.Lock()
92+ defer j.cancelMu.Unlock()
93+ return j.cancelAt != ""
94+}
95+
96+// heartbeat sends the acknowledgement. It retries every HeartbeatRetry until
97+// it succeeds, so that the 30 s of the server are used well; it returns false
98+// when the assignment is gone.
99+func (j *jobRun) ack(ctx context.Context) bool {
100+ for {
101+ err := j.beat(ctx)
102+ if err == nil {
103+ return true
104+ }
105+ if ctx.Err() != nil {
106+ return false
107+ }
108+ switch classify(err) {
109+ case kindEnded, kindAbandoned:
110+ j.abandon(err)
111+ return false
112+ }
113+ j.logf("first heartbeat failed, retrying: %v", err)
114+ if !sleep(ctx, j.t.HeartbeatRetry) {
115+ return false
116+ }
117+ }
118+}
119+
120+func (j *jobRun) beat(ctx context.Context) error {
121+ cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
122+ defer cancel()
123+ _, err := j.client.Heartbeat(cctx, connect.NewRequest(&ermite.HeartbeatRequest{JobId: j.job.JobId}))
124+ return err
125+}
126+
127+// heartbeats keeps the job alive: every HeartbeatEvery, and after a failure
128+// every HeartbeatRetry until one succeeds (runner.proto: Heartbeat).
129+func (j *jobRun) heartbeats(ctx context.Context) {
130+ wait := j.t.HeartbeatEvery
131+ for {
132+ if !sleep(ctx, wait) {
133+ return
134+ }
135+ if err := j.beat(ctx); err != nil {
136+ if ctx.Err() != nil {
137+ return
138+ }
139+ switch classify(err) {
140+ case kindEnded, kindAbandoned:
141+ j.abandon(err)
142+ return
143+ }
144+ j.logf("heartbeat failed, retrying in %s: %v", j.t.HeartbeatRetry, err)
145+ wait = j.t.HeartbeatRetry
146+ continue
147+ }
148+ wait = j.t.HeartbeatEvery
149+ }
150+}
151+
152+// watch follows WatchJob: a cancel request stops the job, an Ended message or
153+// the end-of-job error abandons it. The stream is reopened after any other
154+// error and after a clean close without Ended.
155+func (j *jobRun) watch(ctx context.Context) {
156+ for ctx.Err() == nil {
157+ stream, err := j.client.WatchJob(ctx, connect.NewRequest(&ermite.WatchJobRequest{JobId: j.job.JobId}))
158+ if err == nil {
159+ for stream.Receive() {
160+ switch m := stream.Msg().Msg.(type) {
161+ case *ermite.WatchJobResponse_Cancel:
162+ j.requestCancel(cancelText(m.Cancel), kindOf(m.Cancel.GetReason()))
163+ case *ermite.WatchJobResponse_Ended:
164+ _ = stream.Close()
165+ j.abandon(fmt.Errorf("job ended: %s", m.Ended.GetConclusion()))
166+ return
167+ }
168+ }
169+ err = stream.Err()
170+ _ = stream.Close()
171+ }
172+ if ctx.Err() != nil {
173+ return
174+ }
175+ if err != nil {
176+ switch classify(err) {
177+ case kindEnded, kindAbandoned:
178+ j.abandon(err)
179+ return
180+ }
181+ j.logf("WatchJob: %v", err)
182+ }
183+ if !sleep(ctx, j.t.WatchRetry) {
184+ return
185+ }
186+ }
187+}
188+
189+func cancelText(c *ermite.CancelRequest) string {
190+ if c.GetMessage() != "" {
191+ return c.GetMessage()
192+ }
193+ return c.GetReason().String()
194+}
195+
196+func kindOf(r ermite.CancelReason) execpb.CancelKind {
197+ if r == ermite.CancelReason_CANCEL_REASON_TIMEOUT {
198+ return execpb.CancelKind_CANCEL_KIND_TIMEOUT
199+ }
200+ // User, fail-fast, and any reason this runner does not know: cancelled.
201+ return execpb.CancelKind_CANCEL_KIND_USER
202+}
203+
204+// ---- steps
205+
206+func (j *jobRun) stepStarted(s *execpb.StepStarted) {
207+ u := &ermite.StepUpdate{Number: s.Number, Name: s.Name, State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
208+ j.mu.Lock()
209+ j.steps[s.Number] = u
210+ j.mu.Unlock()
211+ j.outbox.Step(u)
212+}
213+
214+func (j *jobRun) stepCompleted(s *execpb.StepCompleted) {
215+ j.mu.Lock()
216+ u := j.steps[s.Number]
217+ if u == nil {
218+ u = &ermite.StepUpdate{Number: s.Number, StartedAt: tsOf(s.UnixNano)}
219+ } else {
220+ u = cloneStep(u)
221+ }
222+ if s.Name != "" {
223+ u.Name = s.Name
224+ }
225+ u.State = ermite.StepState_STEP_STATE_COMPLETED
226+ u.Conclusion = conclusionOf(s.Conclusion)
227+ if u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
228+ u.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
229+ }
230+ u.CompletedAt = tsOf(s.UnixNano)
231+ j.steps[s.Number] = u
232+ j.mu.Unlock()
233+ j.outbox.Step(u)
234+}
235+
236+func cloneStep(u *ermite.StepUpdate) *ermite.StepUpdate {
237+ return proto.Clone(u).(*ermite.StepUpdate)
238+}
239+
240+func tsOf(unixNano int64) *timestamppb.Timestamp {
241+ if unixNano == 0 {
242+ return timestamppb.Now()
243+ }
244+ return timestamppb.New(time.Unix(0, unixNano))
245+}
246+
247+func conclusionOf(c execpb.StepConclusion) ermite.CiConclusion {
248+ switch c {
249+ case execpb.StepConclusion_STEP_CONCLUSION_SUCCESS:
250+ return ermite.CiConclusion_CI_CONCLUSION_SUCCESS
251+ case execpb.StepConclusion_STEP_CONCLUSION_FAILURE:
252+ return ermite.CiConclusion_CI_CONCLUSION_FAILURE
253+ case execpb.StepConclusion_STEP_CONCLUSION_CANCELLED:
254+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED
255+ case execpb.StepConclusion_STEP_CONCLUSION_SKIPPED:
256+ return ermite.CiConclusion_CI_CONCLUSION_SKIPPED
257+ case execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT:
258+ return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT
259+ }
260+ return ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED
261+}
262+
263+// finalSteps closes the steps that never completed (the executor died or was
264+// stopped) with the conclusion of the job, and returns all snapshots in order.
265+func (j *jobRun) finalSteps(jobConclusion ermite.CiConclusion) []*ermite.StepUpdate {
266+ j.mu.Lock()
267+ defer j.mu.Unlock()
268+ end := ermite.CiConclusion_CI_CONCLUSION_FAILURE
269+ if jobConclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED || jobConclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
270+ end = ermite.CiConclusion_CI_CONCLUSION_CANCELLED
271+ }
272+ var out []*ermite.StepUpdate
273+ for _, u := range j.steps {
274+ if u.State != ermite.StepState_STEP_STATE_COMPLETED {
275+ u = cloneStep(u)
276+ u.State, u.Conclusion, u.CompletedAt = ermite.StepState_STEP_STATE_COMPLETED, end, timestamppb.Now()
277+ }
278+ out = append(out, u)
279+ }
280+ sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
281+ return out
282+}
283+
284+// ---- the job
285+
286+// run runs the whole job: acknowledgement, sandbox, executor session, final
287+// report. It returns when everything is over and the sandbox is gone.
288+func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
289+ defer j.cancel(nil)
290+ ctx := j.ctx
291+
292+ // 1. The acknowledgement comes first, before anything else.
293+ if !j.ack(ctx) {
294+ j.logf("job %s: assignment lost before it started", j.job.JobId)
295+ return
296+ }
297+ var bg sync.WaitGroup
298+ bg.Add(2)
299+ go func() { defer bg.Done(); j.heartbeats(ctx) }()
300+ go func() { defer bg.Done(); j.watch(ctx) }()
301+ defer bg.Wait()
302+
303+ j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.logf)
304+ j.outbox.start(ctx)
305+
306+ // A shutdown is a cancellation of kind shutdown, then a hard stop.
307+ stopHard := make(chan struct{})
308+ go func() {
309+ select {
310+ case <-shutdown:
311+ case <-ctx.Done():
312+ return
313+ }
314+ j.cancelMu.Lock()
315+ j.stopping = true
316+ j.cancelMu.Unlock()
317+ j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
318+ select {
319+ case <-time.After(j.t.ShutdownGrace):
320+ close(stopHard)
321+ case <-ctx.Done():
322+ }
323+ }()
324+
325+ conclusion, outputs, shutdownFlag := j.execute(ctx, rt, stopHard)
326+
327+ // 2. Nothing more is sent to a job that is over.
328+ if err := j.abandoned(); err != nil {
329+ j.logf("job %s: not reporting a result: %v", j.job.JobId, err)
330+ return
331+ }
332+ // The report is made even when the runner is shutting down: it uses its own
333+ // deadline.
334+ rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.CompleteDeadline)
335+ defer cancel()
336+ // Output limits are checked before the log is flushed: a job whose outputs
337+ // the server would refuse fails without outputs, with a line in the log
338+ // (runner.proto: CompleteJobRequest.outputs), and lines sent after
339+ // CompleteJob are rejected.
340+ if msg := checkOutputs(outputs); msg != "" {
341+ j.sandboxError(msg)
342+ outputs, conclusion, shutdownFlag = nil, ermite.CiConclusion_CI_CONCLUSION_FAILURE, false
343+ }
344+ j.outbox.Close(rctx)
345+ if err := j.abandoned(); err != nil {
346+ return
347+ }
348+ j.complete(rctx, conclusion, outputs, shutdownFlag)
349+}
350+
351+// execute starts the sandbox, runs the job in it and returns the conclusion.
352+func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) (ermite.CiConclusion, map[string]string, bool) {
353+ fail := func(format string, a ...any) (ermite.CiConclusion, map[string]string, bool) {
354+ j.failEarly(fmt.Sprintf(format, a...))
355+ return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
356+ }
357+ sb, err := rt.Driver.New()
358+ if err != nil {
359+ return fail("The runner has no free sandbox: %v", err)
360+ }
361+ // Stop runs to completion even when the job context is done.
362+ defer func() {
363+ sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute)
364+ defer cancel()
365+ if err := sb.Stop(sctx); err != nil {
366+ j.logf("job %s: stopping the sandbox: %v", j.job.JobId, err)
367+ if errors.Is(err, sandbox.ErrLeaked) {
368+ rt.markLeaked()
369+ }
370+ }
371+ }()
372+
373+ // Starting the sandbox is bounded by the job context and a hard stop.
374+ startCtx, cancelStart := context.WithCancel(ctx)
375+ stopWatch := make(chan struct{})
376+ go func() {
377+ select {
378+ case <-stopHard:
379+ cancelStart()
380+ case <-stopWatch:
381+ }
382+ }()
383+ conn, err := sb.Start(startCtx, rt.spec(j.job))
384+ close(stopWatch)
385+ cancelStart()
386+ if err != nil {
387+ if ctx.Err() != nil || j.cancelRequested() {
388+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, j.isStopping()
389+ }
390+ return fail("The sandbox did not start: %v", err)
391+ }
392+ defer func() { _ = conn.Close() }()
393+
394+ // Hard stops (grace elapsed, or the job was abandoned) close the stream.
395+ go func() {
396+ select {
397+ case <-stopHard:
398+ case <-ctx.Done():
399+ }
400+ _ = conn.Close()
401+ }()
402+
403+ run := &execpb.Run{Job: &execpb.JobSpec{
404+ WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
405+ Matrix: j.job.Matrix, Contexts: j.job.Contexts, Repo: j.job.Repo, TimeoutMinutes: j.job.TimeoutMinutes,
406+ RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: "self-hosted",
407+ }}
408+ _, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
409+ StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
410+ Log: func(l *execpb.Log) {
411+ j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: l.Text, continued: l.Continued})
412+ },
413+ JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
414+ })
415+ j.mu.Lock()
416+ res := j.result
417+ j.mu.Unlock()
418+
419+ switch {
420+ case res != nil:
421+ c := conclusionOf(res.Conclusion)
422+ if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {
423+ c = ermite.CiConclusion_CI_CONCLUSION_FAILURE
424+ }
425+ if res.Error != "" {
426+ j.logf("job %s: %s", j.job.JobId, res.Error)
427+ }
428+ return c, res.Outputs, res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED
429+ case j.cancelRequested() || ctx.Err() != nil:
430+ // The executor did not report: it was stopped.
431+ if j.isStopping() {
432+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, true
433+ }
434+ return ermite.CiConclusion_CI_CONCLUSION_CANCELLED, nil, false
435+ }
436+ why := "the executor ended without a result"
437+ if err != nil {
438+ why = err.Error()
439+ } else if exit != nil && exit.Error != "" {
440+ why = exit.Error
441+ }
442+ j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))
443+ return ermite.CiConclusion_CI_CONCLUSION_FAILURE, nil, false
444+}
445+
446+func (j *jobRun) isStopping() bool {
447+ j.cancelMu.Lock()
448+ defer j.cancelMu.Unlock()
449+ return j.stopping
450+}
451+
452+// failEarly reports a job that could not start: step 1 fails with the reason
453+// in its log.
454+func (j *jobRun) failEarly(msg string) {
455+ now := time.Now()
456+ j.stepStarted(&execpb.StepStarted{Number: 1, Name: "Set up job", UnixNano: now.UnixNano()})
457+ j.outbox.Log(logLine{step: 1, unixNano: now.UnixNano(), text: "##[error]" + msg})
458+ j.stepCompleted(&execpb.StepCompleted{Number: 1, Name: "Set up job", Conclusion: execpb.StepConclusion_STEP_CONCLUSION_FAILURE, UnixNano: time.Now().UnixNano()})
459+}
460+
461+// sandboxError writes the reason into the log of the step that was running.
462+func (j *jobRun) sandboxError(msg string) {
463+ j.mu.Lock()
464+ var last int32 = 1
465+ for n := range j.steps {
466+ last = max(last, n)
467+ }
468+ j.mu.Unlock()
469+ j.outbox.Log(logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
470+}
471+
472+// complete reports the end of the job. CompleteJob is idempotent, so a lost
473+// response is retried until it is answered (applied or not).
474+func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {
475+ req := &ermite.CompleteJobRequest{
476+ JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
477+ RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
478+ Steps: j.finalSteps(conclusion),
479+ }
480+ b := backoff{min: j.t.BackoffMin, max: j.t.BackoffMax}
481+ for {
482+ cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
483+ resp, err := j.client.CompleteJob(cctx, connect.NewRequest(req))
484+ cancel()
485+ if err == nil {
486+ if resp.Msg.Conclusion != conclusion {
487+ j.logf("job %s: ended as %s, the server holds %s", j.job.JobId, conclusion, resp.Msg.Conclusion)
488+ }
489+ return
490+ }
491+ if ctx.Err() != nil {
492+ j.logf("job %s: could not report the result: %v", j.job.JobId, ctx.Err())
493+ return
494+ }
495+ switch classify(err) {
496+ case kindEnded, kindAbandoned:
497+ return
498+ case kindPermanent:
499+ j.logf("job %s: CompleteJob refused: %v", j.job.JobId, err)
500+ return
501+ }
502+ j.logf("CompleteJob failed, retrying: %v", err)
503+ if !sleep(ctx, b.next()) {
504+ return
505+ }
506+ }
507+}
508+
509+// Limits of CompleteJobRequest.outputs.
510+const (
511+ maxOutputBytes = 1 << 20
512+ maxOutputName = 255
513+)
514+
515+// checkOutputs returns a message when the server would refuse the outputs: a
516+// name of 1 to 255 letters, digits, "_" and "-" starting with a letter or "_",
517+// no two names equal ignoring case, each value and the sum of names and values
518+// at most 1 MiB.
519+func checkOutputs(outputs map[string]string) string {
520+ total := 0
521+ seen := map[string]string{}
522+ names := make([]string, 0, len(outputs))
523+ for n := range outputs {
524+ names = append(names, n)
525+ }
526+ sort.Strings(names)
527+ for _, n := range names {
528+ v := outputs[n]
529+ if !validOutputName(n) {
530+ return fmt.Sprintf("Job output name %q is not valid: use 1 to 255 letters, digits, '_' and '-', starting with a letter or '_'.", n)
531+ }
532+ if prev, dup := seen[strings.ToLower(n)]; dup {
533+ return fmt.Sprintf("Job outputs %q and %q differ only by case.", prev, n)
534+ }
535+ seen[strings.ToLower(n)] = n
536+ if len(v) > maxOutputBytes {
537+ return fmt.Sprintf("Job output %q is larger than 1 MiB.", n)
538+ }
539+ total += len(n) + len(v)
540+ }
541+ if total > maxOutputBytes {
542+ return "The job outputs are larger than 1 MiB in total."
543+ }
544+ return ""
545+}
546+
547+func validOutputName(n string) bool {
548+ if n == "" || len(n) > maxOutputName {
549+ return false
550+ }
551+ for i, r := range n {
552+ letter := r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r == '_'
553+ if !(letter || (i > 0 && (r >= '0' && r <= '9' || r == '-'))) {
554+ return false
555+ }
556+ }
557+ return true
558+}
A internal/host/outbox.go +287 -0
@@ -0,0 +1,287 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "sync"
8+ "time"
9+
10+ "connectrpc.com/connect"
11+ "google.golang.org/protobuf/types/known/timestamppb"
12+
13+ ermite "bnema.dev/ermite-api/gen/ermite"
14+)
15+
16+// Limits of AppendLog (runner.proto): at most 1,000 lines and 64 KiB of text
17+// per request, all in the same step.
18+const (
19+ maxBatchLines = 1000
20+ maxBatchBytes = 64 * 1024
21+ // maxQueuedBytes is the log kept in memory while the server cannot be
22+ // reached; beyond it the executor is slowed down (the stream is not read)
23+ // rather than the host's memory filled.
24+ maxQueuedBytes = 16 << 20
25+)
26+
27+type logLine struct {
28+ step int32
29+ unixNano int64
30+ text string
31+ continued bool
32+}
33+
34+// event is something to send, in the order it happened: a log line, or a
35+// snapshot of a step (UpdateJob).
36+type event struct {
37+ line *logLine
38+ step *ermite.StepUpdate
39+}
40+
41+// outbox sends logs and step updates of one job to ermite, in order, one call
42+// at a time. Log lines are numbered from 0 across the whole job; the number of
43+// the next line is the offset of the next AppendLog, so that a retry after a
44+// lost response sends the same lines at the same offset (the server skips what
45+// it already holds).
46+type outbox struct {
47+ client callClient
48+ jobID string
49+ t Timing
50+ // onEnded is called once when the server says the job is over (JobEnded)
51+ // or the assignment is gone; the outbox then drops everything.
52+ onEnded func(error)
53+ logf func(format string, args ...any)
54+
55+ mu sync.Mutex
56+ cond *sync.Cond
57+ events []event
58+ queued int // bytes of log text in events
59+ closing bool
60+ dead bool
61+ done chan struct{}
62+
63+ offset int64 // lines the server holds, as far as we know
64+}
65+
66+// callClient is the part of the RunnerService client the job uses.
67+type callClient interface {
68+ UpdateJob(context.Context, *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error)
69+ AppendLog(context.Context, *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error)
70+}
71+
72+func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), logf func(string, ...any)) *outbox {
73+ o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, logf: logf, done: make(chan struct{})}
74+ o.cond = sync.NewCond(&o.mu)
75+ return o
76+}
77+
78+// start runs the sender until Close or until ctx ends.
79+func (o *outbox) start(ctx context.Context) {
80+ go func() {
81+ defer close(o.done)
82+ o.loop(ctx)
83+ }()
84+ // A cancelled context must wake the loop and the producers.
85+ go func() {
86+ select {
87+ case <-ctx.Done():
88+ o.kill()
89+ case <-o.done:
90+ }
91+ }()
92+}
93+
94+func (o *outbox) kill() {
95+ o.mu.Lock()
96+ o.dead = true
97+ o.events = nil
98+ o.queued = 0
99+ o.mu.Unlock()
100+ o.cond.Broadcast()
101+}
102+
103+// Log queues a line of the job log. It blocks while too much is queued.
104+func (o *outbox) Log(l logLine) {
105+ o.mu.Lock()
106+ for o.queued > maxQueuedBytes && !o.dead && !o.closing {
107+ o.cond.Wait()
108+ }
109+ if o.dead {
110+ o.mu.Unlock()
111+ return
112+ }
113+ o.events = append(o.events, event{line: &l})
114+ o.queued += len(l.text) + 32
115+ o.mu.Unlock()
116+ o.cond.Broadcast()
117+}
118+
119+// Step queues a step snapshot.
120+func (o *outbox) Step(s *ermite.StepUpdate) {
121+ o.mu.Lock()
122+ if o.dead {
123+ o.mu.Unlock()
124+ return
125+ }
126+ o.events = append(o.events, event{step: s})
127+ o.mu.Unlock()
128+ o.cond.Broadcast()
129+}
130+
131+// Close sends what is queued, then stops the sender. It waits at most until
132+// ctx ends.
133+func (o *outbox) Close(ctx context.Context) {
134+ o.mu.Lock()
135+ o.closing = true
136+ o.mu.Unlock()
137+ o.cond.Broadcast()
138+ select {
139+ case <-o.done:
140+ case <-ctx.Done():
141+ o.kill()
142+ <-o.done
143+ }
144+}
145+
146+// Lines returns how many lines the server holds (as far as we know).
147+func (o *outbox) Lines() int64 {
148+ o.mu.Lock()
149+ defer o.mu.Unlock()
150+ return o.offset
151+}
152+
153+// next waits for a batch of events. It returns nil when the outbox is done.
154+func (o *outbox) next(ctx context.Context) []event {
155+ o.mu.Lock()
156+ defer o.mu.Unlock()
157+ for len(o.events) == 0 {
158+ if o.dead || o.closing {
159+ return nil
160+ }
161+ o.cond.Wait()
162+ }
163+ // Let a burst of lines fill the batch, unless it is full, a step event is
164+ // at the front, or we are closing.
165+ if o.events[0].line != nil && !o.closing && o.queued < maxBatchBytes && len(o.events) < maxBatchLines {
166+ o.mu.Unlock()
167+ sleep(ctx, o.t.LogFlush)
168+ o.mu.Lock()
169+ if o.dead {
170+ return nil
171+ }
172+ }
173+ if len(o.events) == 0 {
174+ return nil
175+ }
176+ first := o.events[0]
177+ if first.step != nil {
178+ o.events = o.events[1:]
179+ return []event{first}
180+ }
181+ // A run of lines of the same step, within the limits.
182+ n, bytes := 0, 0
183+ step := first.line.step
184+ for _, e := range o.events {
185+ if e.line == nil || e.line.step != step || n == maxBatchLines || (n > 0 && bytes+len(e.line.text) > maxBatchBytes) {
186+ break
187+ }
188+ n++
189+ bytes += len(e.line.text)
190+ }
191+ batch := append([]event(nil), o.events[:n]...)
192+ o.events = o.events[n:]
193+ for _, e := range batch {
194+ o.queued -= len(e.line.text) + 32
195+ }
196+ o.cond.Broadcast() // producers waiting for room
197+ return batch
198+}
199+
200+func (o *outbox) loop(ctx context.Context) {
201+ for {
202+ batch := o.next(ctx)
203+ if batch == nil {
204+ return
205+ }
206+ if batch[0].step != nil {
207+ o.sendStep(ctx, batch[0].step)
208+ } else {
209+ o.sendLines(ctx, batch)
210+ }
211+ o.mu.Lock()
212+ dead := o.dead
213+ o.mu.Unlock()
214+ if dead {
215+ return
216+ }
217+ }
218+}
219+
220+// retry runs call until it succeeds, the server gives a verdict, or ctx ends.
221+// It returns nil on success.
222+func (o *outbox) retry(ctx context.Context, what string, call func(context.Context) error) error {
223+ b := backoff{min: o.t.BackoffMin, max: o.t.BackoffMax}
224+ for {
225+ cctx, cancel := context.WithTimeout(ctx, o.t.CallTimeout)
226+ err := call(cctx)
227+ cancel()
228+ if err == nil {
229+ return nil
230+ }
231+ if ctx.Err() != nil {
232+ return ctx.Err()
233+ }
234+ switch classify(err) {
235+ case kindEnded, kindAbandoned:
236+ o.onEnded(err)
237+ o.kill()
238+ return err
239+ case kindPermanent:
240+ return err
241+ }
242+ o.logf("%s failed, retrying: %v", what, err)
243+ if !sleep(ctx, b.next()) {
244+ return ctx.Err()
245+ }
246+ }
247+}
248+
249+func (o *outbox) sendStep(ctx context.Context, s *ermite.StepUpdate) {
250+ err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
251+ _, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
252+ return err
253+ })
254+ if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
255+ o.logf("UpdateJob refused step %d: %v", s.Number, err)
256+ }
257+}
258+
259+func (o *outbox) sendLines(ctx context.Context, batch []event) {
260+ req := &ermite.AppendLogRequest{JobId: o.jobID, Step: batch[0].line.step}
261+ for _, e := range batch {
262+ req.Lines = append(req.Lines, &ermite.LogLine{Time: timestamppb.New(time.Unix(0, e.line.unixNano)), Text: e.line.text, Continued: e.line.continued})
263+ }
264+ // The offset is fixed before the first attempt: every retry sends the same
265+ // lines at the same offset.
266+ req.Offset = o.offset
267+ err := o.retry(ctx, "AppendLog", func(c context.Context) error {
268+ resp, err := o.client.AppendLog(c, connect.NewRequest(req))
269+ if err != nil {
270+ return err
271+ }
272+ if want := req.Offset + int64(len(req.Lines)); resp.Msg.NextOffset != want {
273+ o.logf("AppendLog: the server holds %d lines, expected %d", resp.Msg.NextOffset, want)
274+ }
275+ return nil
276+ })
277+ switch {
278+ case err == nil:
279+ o.mu.Lock()
280+ o.offset += int64(len(req.Lines))
281+ o.mu.Unlock()
282+ case ctx.Err() == nil && classify(err) == kindPermanent:
283+ // The server refused these lines; they are dropped and the offset stays
284+ // where the server is.
285+ o.logf("AppendLog refused %d lines of step %d: %v", len(req.Lines), req.Step, err)
286+ }
287+}
@@ -0,0 +1,287 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "sync"
8+ "time"
9+
10+ "connectrpc.com/connect"
11+ "google.golang.org/protobuf/types/known/timestamppb"
12+
13+ ermite "bnema.dev/ermite-api/gen/ermite"
14+)
15+
16+// Limits of AppendLog (runner.proto): at most 1,000 lines and 64 KiB of text
17+// per request, all in the same step.
18+const (
19+ maxBatchLines = 1000
20+ maxBatchBytes = 64 * 1024
21+ // maxQueuedBytes is the log kept in memory while the server cannot be
22+ // reached; beyond it the executor is slowed down (the stream is not read)
23+ // rather than the host's memory filled.
24+ maxQueuedBytes = 16 << 20
25+)
26+
27+type logLine struct {
28+ step int32
29+ unixNano int64
30+ text string
31+ continued bool
32+}
33+
34+// event is something to send, in the order it happened: a log line, or a
35+// snapshot of a step (UpdateJob).
36+type event struct {
37+ line *logLine
38+ step *ermite.StepUpdate
39+}
40+
41+// outbox sends logs and step updates of one job to ermite, in order, one call
42+// at a time. Log lines are numbered from 0 across the whole job; the number of
43+// the next line is the offset of the next AppendLog, so that a retry after a
44+// lost response sends the same lines at the same offset (the server skips what
45+// it already holds).
46+type outbox struct {
47+ client callClient
48+ jobID string
49+ t Timing
50+ // onEnded is called once when the server says the job is over (JobEnded)
51+ // or the assignment is gone; the outbox then drops everything.
52+ onEnded func(error)
53+ logf func(format string, args ...any)
54+
55+ mu sync.Mutex
56+ cond *sync.Cond
57+ events []event
58+ queued int // bytes of log text in events
59+ closing bool
60+ dead bool
61+ done chan struct{}
62+
63+ offset int64 // lines the server holds, as far as we know
64+}
65+
66+// callClient is the part of the RunnerService client the job uses.
67+type callClient interface {
68+ UpdateJob(context.Context, *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error)
69+ AppendLog(context.Context, *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error)
70+}
71+
72+func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), logf func(string, ...any)) *outbox {
73+ o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, logf: logf, done: make(chan struct{})}
74+ o.cond = sync.NewCond(&o.mu)
75+ return o
76+}
77+
78+// start runs the sender until Close or until ctx ends.
79+func (o *outbox) start(ctx context.Context) {
80+ go func() {
81+ defer close(o.done)
82+ o.loop(ctx)
83+ }()
84+ // A cancelled context must wake the loop and the producers.
85+ go func() {
86+ select {
87+ case <-ctx.Done():
88+ o.kill()
89+ case <-o.done:
90+ }
91+ }()
92+}
93+
94+func (o *outbox) kill() {
95+ o.mu.Lock()
96+ o.dead = true
97+ o.events = nil
98+ o.queued = 0
99+ o.mu.Unlock()
100+ o.cond.Broadcast()
101+}
102+
103+// Log queues a line of the job log. It blocks while too much is queued.
104+func (o *outbox) Log(l logLine) {
105+ o.mu.Lock()
106+ for o.queued > maxQueuedBytes && !o.dead && !o.closing {
107+ o.cond.Wait()
108+ }
109+ if o.dead {
110+ o.mu.Unlock()
111+ return
112+ }
113+ o.events = append(o.events, event{line: &l})
114+ o.queued += len(l.text) + 32
115+ o.mu.Unlock()
116+ o.cond.Broadcast()
117+}
118+
119+// Step queues a step snapshot.
120+func (o *outbox) Step(s *ermite.StepUpdate) {
121+ o.mu.Lock()
122+ if o.dead {
123+ o.mu.Unlock()
124+ return
125+ }
126+ o.events = append(o.events, event{step: s})
127+ o.mu.Unlock()
128+ o.cond.Broadcast()
129+}
130+
131+// Close sends what is queued, then stops the sender. It waits at most until
132+// ctx ends.
133+func (o *outbox) Close(ctx context.Context) {
134+ o.mu.Lock()
135+ o.closing = true
136+ o.mu.Unlock()
137+ o.cond.Broadcast()
138+ select {
139+ case <-o.done:
140+ case <-ctx.Done():
141+ o.kill()
142+ <-o.done
143+ }
144+}
145+
146+// Lines returns how many lines the server holds (as far as we know).
147+func (o *outbox) Lines() int64 {
148+ o.mu.Lock()
149+ defer o.mu.Unlock()
150+ return o.offset
151+}
152+
153+// next waits for a batch of events. It returns nil when the outbox is done.
154+func (o *outbox) next(ctx context.Context) []event {
155+ o.mu.Lock()
156+ defer o.mu.Unlock()
157+ for len(o.events) == 0 {
158+ if o.dead || o.closing {
159+ return nil
160+ }
161+ o.cond.Wait()
162+ }
163+ // Let a burst of lines fill the batch, unless it is full, a step event is
164+ // at the front, or we are closing.
165+ if o.events[0].line != nil && !o.closing && o.queued < maxBatchBytes && len(o.events) < maxBatchLines {
166+ o.mu.Unlock()
167+ sleep(ctx, o.t.LogFlush)
168+ o.mu.Lock()
169+ if o.dead {
170+ return nil
171+ }
172+ }
173+ if len(o.events) == 0 {
174+ return nil
175+ }
176+ first := o.events[0]
177+ if first.step != nil {
178+ o.events = o.events[1:]
179+ return []event{first}
180+ }
181+ // A run of lines of the same step, within the limits.
182+ n, bytes := 0, 0
183+ step := first.line.step
184+ for _, e := range o.events {
185+ if e.line == nil || e.line.step != step || n == maxBatchLines || (n > 0 && bytes+len(e.line.text) > maxBatchBytes) {
186+ break
187+ }
188+ n++
189+ bytes += len(e.line.text)
190+ }
191+ batch := append([]event(nil), o.events[:n]...)
192+ o.events = o.events[n:]
193+ for _, e := range batch {
194+ o.queued -= len(e.line.text) + 32
195+ }
196+ o.cond.Broadcast() // producers waiting for room
197+ return batch
198+}
199+
200+func (o *outbox) loop(ctx context.Context) {
201+ for {
202+ batch := o.next(ctx)
203+ if batch == nil {
204+ return
205+ }
206+ if batch[0].step != nil {
207+ o.sendStep(ctx, batch[0].step)
208+ } else {
209+ o.sendLines(ctx, batch)
210+ }
211+ o.mu.Lock()
212+ dead := o.dead
213+ o.mu.Unlock()
214+ if dead {
215+ return
216+ }
217+ }
218+}
219+
220+// retry runs call until it succeeds, the server gives a verdict, or ctx ends.
221+// It returns nil on success.
222+func (o *outbox) retry(ctx context.Context, what string, call func(context.Context) error) error {
223+ b := backoff{min: o.t.BackoffMin, max: o.t.BackoffMax}
224+ for {
225+ cctx, cancel := context.WithTimeout(ctx, o.t.CallTimeout)
226+ err := call(cctx)
227+ cancel()
228+ if err == nil {
229+ return nil
230+ }
231+ if ctx.Err() != nil {
232+ return ctx.Err()
233+ }
234+ switch classify(err) {
235+ case kindEnded, kindAbandoned:
236+ o.onEnded(err)
237+ o.kill()
238+ return err
239+ case kindPermanent:
240+ return err
241+ }
242+ o.logf("%s failed, retrying: %v", what, err)
243+ if !sleep(ctx, b.next()) {
244+ return ctx.Err()
245+ }
246+ }
247+}
248+
249+func (o *outbox) sendStep(ctx context.Context, s *ermite.StepUpdate) {
250+ err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
251+ _, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
252+ return err
253+ })
254+ if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
255+ o.logf("UpdateJob refused step %d: %v", s.Number, err)
256+ }
257+}
258+
259+func (o *outbox) sendLines(ctx context.Context, batch []event) {
260+ req := &ermite.AppendLogRequest{JobId: o.jobID, Step: batch[0].line.step}
261+ for _, e := range batch {
262+ req.Lines = append(req.Lines, &ermite.LogLine{Time: timestamppb.New(time.Unix(0, e.line.unixNano)), Text: e.line.text, Continued: e.line.continued})
263+ }
264+ // The offset is fixed before the first attempt: every retry sends the same
265+ // lines at the same offset.
266+ req.Offset = o.offset
267+ err := o.retry(ctx, "AppendLog", func(c context.Context) error {
268+ resp, err := o.client.AppendLog(c, connect.NewRequest(req))
269+ if err != nil {
270+ return err
271+ }
272+ if want := req.Offset + int64(len(req.Lines)); resp.Msg.NextOffset != want {
273+ o.logf("AppendLog: the server holds %d lines, expected %d", resp.Msg.NextOffset, want)
274+ }
275+ return nil
276+ })
277+ switch {
278+ case err == nil:
279+ o.mu.Lock()
280+ o.offset += int64(len(req.Lines))
281+ o.mu.Unlock()
282+ case ctx.Err() == nil && classify(err) == kindPermanent:
283+ // The server refused these lines; they are dropped and the offset stays
284+ // where the server is.
285+ o.logf("AppendLog refused %d lines of step %d: %v", len(req.Lines), req.Step, err)
286+ }
287+}
A internal/host/rpc.go +170 -0
@@ -0,0 +1,170 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package host is the host side of ermite-runner: it registers the runner with
4+// ermite, fetches jobs, runs each one in a sandbox through the executor, and
5+// reports logs, steps and the result with the job token. The sandbox never sees
6+// a token.
7+package host
8+
9+import (
10+ "context"
11+ "errors"
12+ "net/http"
13+ "strings"
14+ "time"
15+
16+ "connectrpc.com/connect"
17+
18+ ermite "bnema.dev/ermite-api/gen/ermite"
19+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20+)
21+
22+// bearer adds "Authorization: Bearer <token>" to every request. The token is
23+// never logged.
24+type bearer struct {
25+ token string
26+ base http.RoundTripper
27+}
28+
29+func (b bearer) RoundTrip(r *http.Request) (*http.Response, error) {
30+ r = r.Clone(r.Context())
31+ if b.token != "" {
32+ r.Header.Set("Authorization", "Bearer "+b.token)
33+ }
34+ base := b.base
35+ if base == nil {
36+ base = http.DefaultTransport
37+ }
38+ return base.RoundTrip(r)
39+}
40+
41+// NewClient returns a RunnerService client that authenticates with token. The
42+// HTTP client has no overall timeout: calls carry their own context deadline
43+// (WatchJob is a long stream).
44+func NewClient(baseURL, token string, base http.RoundTripper) ermiteconnect.RunnerServiceClient {
45+ return ermiteconnect.NewRunnerServiceClient(&http.Client{Transport: bearer{token, base}}, strings.TrimRight(baseURL, "/"))
46+}
47+
48+// ClientFactory builds a client for a token; tests replace it.
49+type ClientFactory func(token string) ermiteconnect.RunnerServiceClient
50+
51+// jobEnded extracts the JobEnded detail of a failed_precondition error.
52+func jobEnded(err error) (*ermite.JobEnded, bool) {
53+ var ce *connect.Error
54+ if !errors.As(err, &ce) || ce.Code() != connect.CodeFailedPrecondition {
55+ return nil, false
56+ }
57+ for _, d := range ce.Details() {
58+ m, derr := d.Value()
59+ if derr != nil {
60+ continue
61+ }
62+ if je, ok := m.(*ermite.JobEnded); ok {
63+ return je, true
64+ }
65+ }
66+ return nil, false
67+}
68+
69+// kind classifies the error of a job call.
70+type kind int
71+
72+const (
73+ kindRetry kind = iota // network error, unavailable, internal...: try again
74+ kindEnded // failed_precondition with JobEnded: stop, send nothing more
75+ kindAbandoned // unauthenticated: the assignment is gone, stop at once
76+ kindPermanent // the request itself is wrong: do not repeat it
77+)
78+
79+func classify(err error) kind {
80+ if err == nil {
81+ return kindRetry
82+ }
83+ if _, ok := jobEnded(err); ok {
84+ return kindEnded
85+ }
86+ switch connect.CodeOf(err) {
87+ case connect.CodeUnauthenticated:
88+ return kindAbandoned
89+ case connect.CodeInvalidArgument, connect.CodeFailedPrecondition, connect.CodeOutOfRange,
90+ connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeUnimplemented, connect.CodeAlreadyExists:
91+ return kindPermanent
92+ }
93+ // Canceled by our own context is handled by the callers through ctx.Err().
94+ return kindRetry
95+}
96+
97+// Timing holds the delays of the host loop. Zero fields take the defaults,
98+// which are the protocol's.
99+type Timing struct {
100+ // PollTimeout bounds one FetchJob call (the server holds it up to 50 s).
101+ PollTimeout time.Duration
102+ // HeartbeatEvery is the heartbeat period (30 s); HeartbeatRetry the delay
103+ // after a failed heartbeat (5 s).
104+ HeartbeatEvery, HeartbeatRetry time.Duration
105+ // CallTimeout bounds one unary job call.
106+ CallTimeout time.Duration
107+ // BackoffMin and BackoffMax bound the retry delay of calls that fail
108+ // without a verdict.
109+ BackoffMin, BackoffMax time.Duration
110+ // LogFlush is how long log lines are held to fill a batch.
111+ LogFlush time.Duration
112+ // WatchRetry is the delay before WatchJob is reopened.
113+ WatchRetry time.Duration
114+ // ShutdownGrace is how long jobs get to stop after a shutdown signal before
115+ // their sandbox is destroyed.
116+ ShutdownGrace time.Duration
117+ // CompleteDeadline bounds the flush of logs and CompleteJob at the end of
118+ // a job.
119+ CompleteDeadline time.Duration
120+ // NoJobPause is the pause after a poll without a job and without
121+ // retry_after.
122+ NoJobPause time.Duration
123+}
124+
125+func (t Timing) withDefaults() Timing {
126+ def := func(d *time.Duration, v time.Duration) {
127+ if *d == 0 {
128+ *d = v
129+ }
130+ }
131+ def(&t.PollTimeout, 60*time.Second)
132+ def(&t.HeartbeatEvery, 30*time.Second)
133+ def(&t.HeartbeatRetry, 5*time.Second)
134+ def(&t.CallTimeout, 30*time.Second)
135+ def(&t.BackoffMin, time.Second)
136+ def(&t.BackoffMax, 30*time.Second)
137+ def(&t.LogFlush, 250*time.Millisecond)
138+ def(&t.WatchRetry, time.Second)
139+ def(&t.ShutdownGrace, 30*time.Second)
140+ def(&t.CompleteDeadline, 5*time.Minute)
141+ def(&t.NoJobPause, 0)
142+ return t
143+}
144+
145+// sleep waits d or until ctx ends; it reports whether the full delay passed.
146+func sleep(ctx context.Context, d time.Duration) bool {
147+ if d <= 0 {
148+ return ctx.Err() == nil
149+ }
150+ t := time.NewTimer(d)
151+ defer t.Stop()
152+ select {
153+ case <-t.C:
154+ return true
155+ case <-ctx.Done():
156+ return false
157+ }
158+}
159+
160+// backoff doubles from min to max.
161+type backoff struct{ min, max, cur time.Duration }
162+
163+func (b *backoff) next() time.Duration {
164+ if b.cur == 0 {
165+ b.cur = b.min
166+ } else if b.cur *= 2; b.cur > b.max {
167+ b.cur = b.max
168+ }
169+ return b.cur
170+}
@@ -0,0 +1,170 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package host is the host side of ermite-runner: it registers the runner with
4+// ermite, fetches jobs, runs each one in a sandbox through the executor, and
5+// reports logs, steps and the result with the job token. The sandbox never sees
6+// a token.
7+package host
8+
9+import (
10+ "context"
11+ "errors"
12+ "net/http"
13+ "strings"
14+ "time"
15+
16+ "connectrpc.com/connect"
17+
18+ ermite "bnema.dev/ermite-api/gen/ermite"
19+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20+)
21+
22+// bearer adds "Authorization: Bearer <token>" to every request. The token is
23+// never logged.
24+type bearer struct {
25+ token string
26+ base http.RoundTripper
27+}
28+
29+func (b bearer) RoundTrip(r *http.Request) (*http.Response, error) {
30+ r = r.Clone(r.Context())
31+ if b.token != "" {
32+ r.Header.Set("Authorization", "Bearer "+b.token)
33+ }
34+ base := b.base
35+ if base == nil {
36+ base = http.DefaultTransport
37+ }
38+ return base.RoundTrip(r)
39+}
40+
41+// NewClient returns a RunnerService client that authenticates with token. The
42+// HTTP client has no overall timeout: calls carry their own context deadline
43+// (WatchJob is a long stream).
44+func NewClient(baseURL, token string, base http.RoundTripper) ermiteconnect.RunnerServiceClient {
45+ return ermiteconnect.NewRunnerServiceClient(&http.Client{Transport: bearer{token, base}}, strings.TrimRight(baseURL, "/"))
46+}
47+
48+// ClientFactory builds a client for a token; tests replace it.
49+type ClientFactory func(token string) ermiteconnect.RunnerServiceClient
50+
51+// jobEnded extracts the JobEnded detail of a failed_precondition error.
52+func jobEnded(err error) (*ermite.JobEnded, bool) {
53+ var ce *connect.Error
54+ if !errors.As(err, &ce) || ce.Code() != connect.CodeFailedPrecondition {
55+ return nil, false
56+ }
57+ for _, d := range ce.Details() {
58+ m, derr := d.Value()
59+ if derr != nil {
60+ continue
61+ }
62+ if je, ok := m.(*ermite.JobEnded); ok {
63+ return je, true
64+ }
65+ }
66+ return nil, false
67+}
68+
69+// kind classifies the error of a job call.
70+type kind int
71+
72+const (
73+ kindRetry kind = iota // network error, unavailable, internal...: try again
74+ kindEnded // failed_precondition with JobEnded: stop, send nothing more
75+ kindAbandoned // unauthenticated: the assignment is gone, stop at once
76+ kindPermanent // the request itself is wrong: do not repeat it
77+)
78+
79+func classify(err error) kind {
80+ if err == nil {
81+ return kindRetry
82+ }
83+ if _, ok := jobEnded(err); ok {
84+ return kindEnded
85+ }
86+ switch connect.CodeOf(err) {
87+ case connect.CodeUnauthenticated:
88+ return kindAbandoned
89+ case connect.CodeInvalidArgument, connect.CodeFailedPrecondition, connect.CodeOutOfRange,
90+ connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeUnimplemented, connect.CodeAlreadyExists:
91+ return kindPermanent
92+ }
93+ // Canceled by our own context is handled by the callers through ctx.Err().
94+ return kindRetry
95+}
96+
97+// Timing holds the delays of the host loop. Zero fields take the defaults,
98+// which are the protocol's.
99+type Timing struct {
100+ // PollTimeout bounds one FetchJob call (the server holds it up to 50 s).
101+ PollTimeout time.Duration
102+ // HeartbeatEvery is the heartbeat period (30 s); HeartbeatRetry the delay
103+ // after a failed heartbeat (5 s).
104+ HeartbeatEvery, HeartbeatRetry time.Duration
105+ // CallTimeout bounds one unary job call.
106+ CallTimeout time.Duration
107+ // BackoffMin and BackoffMax bound the retry delay of calls that fail
108+ // without a verdict.
109+ BackoffMin, BackoffMax time.Duration
110+ // LogFlush is how long log lines are held to fill a batch.
111+ LogFlush time.Duration
112+ // WatchRetry is the delay before WatchJob is reopened.
113+ WatchRetry time.Duration
114+ // ShutdownGrace is how long jobs get to stop after a shutdown signal before
115+ // their sandbox is destroyed.
116+ ShutdownGrace time.Duration
117+ // CompleteDeadline bounds the flush of logs and CompleteJob at the end of
118+ // a job.
119+ CompleteDeadline time.Duration
120+ // NoJobPause is the pause after a poll without a job and without
121+ // retry_after.
122+ NoJobPause time.Duration
123+}
124+
125+func (t Timing) withDefaults() Timing {
126+ def := func(d *time.Duration, v time.Duration) {
127+ if *d == 0 {
128+ *d = v
129+ }
130+ }
131+ def(&t.PollTimeout, 60*time.Second)
132+ def(&t.HeartbeatEvery, 30*time.Second)
133+ def(&t.HeartbeatRetry, 5*time.Second)
134+ def(&t.CallTimeout, 30*time.Second)
135+ def(&t.BackoffMin, time.Second)
136+ def(&t.BackoffMax, 30*time.Second)
137+ def(&t.LogFlush, 250*time.Millisecond)
138+ def(&t.WatchRetry, time.Second)
139+ def(&t.ShutdownGrace, 30*time.Second)
140+ def(&t.CompleteDeadline, 5*time.Minute)
141+ def(&t.NoJobPause, 0)
142+ return t
143+}
144+
145+// sleep waits d or until ctx ends; it reports whether the full delay passed.
146+func sleep(ctx context.Context, d time.Duration) bool {
147+ if d <= 0 {
148+ return ctx.Err() == nil
149+ }
150+ t := time.NewTimer(d)
151+ defer t.Stop()
152+ select {
153+ case <-t.C:
154+ return true
155+ case <-ctx.Done():
156+ return false
157+ }
158+}
159+
160+// backoff doubles from min to max.
161+type backoff struct{ min, max, cur time.Duration }
162+
163+func (b *backoff) next() time.Duration {
164+ if b.cur == 0 {
165+ b.cur = b.min
166+ } else if b.cur *= 2; b.cur > b.max {
167+ b.cur = b.max
168+ }
169+ return b.cur
170+}
A internal/host/runner.go +235 -0
@@ -0,0 +1,235 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "crypto/rand"
8+ "encoding/base64"
9+ "errors"
10+ "fmt"
11+ "runtime"
12+ "sync"
13+ "sync/atomic"
14+ "time"
15+
16+ "connectrpc.com/connect"
17+
18+ ermite "bnema.dev/ermite-api/gen/ermite"
19+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20+
21+ "bnema.dev/ermite-runner/internal/execstream/execpb"
22+ "bnema.dev/ermite-runner/internal/sandbox"
23+)
24+
25+// Runner fetches jobs from ermite and runs each one in its own sandbox.
26+type Runner struct {
27+ Cfg Config
28+ Driver sandbox.Driver
29+ // Poll is the client that carries the runner token (FetchJob).
30+ Poll ermiteconnect.RunnerServiceClient
31+ // Job builds the client of a job from its job token.
32+ Job func(jobToken string) ermiteconnect.RunnerServiceClient
33+ Timing Timing
34+ Logf func(format string, args ...any)
35+ // JobSpec sizes the sandbox of a job; the zero value takes the defaults.
36+ JobSpec func(*ermite.Job) sandbox.JobSpec
37+
38+ leaked atomic.Bool
39+ active sync.WaitGroup
40+}
41+
42+func (r *Runner) logf(format string, a ...any) {
43+ if r.Logf != nil {
44+ r.Logf(format, a...)
45+ }
46+}
47+
48+func (r *Runner) spec(j *ermite.Job) sandbox.JobSpec {
49+ if r.JobSpec != nil {
50+ return r.JobSpec(j)
51+ }
52+ return sandbox.JobSpec{JobID: j.JobId}
53+}
54+
55+// markLeaked records that a sandbox could not be cleaned up: no new job is
56+// taken until the running ones are over and Driver.Reap has succeeded.
57+func (r *Runner) markLeaked() { r.leaked.Store(true) }
58+
59+// ErrRemoved is returned by Run when ermite no longer knows the runner token:
60+// the runner was removed or replaced.
61+var ErrRemoved = errors.New("ermite rejected the runner token: the runner was removed or replaced; register it again")
62+
63+// newPollID is 16 random bytes in base64url: 22 characters of the set the
64+// protocol allows.
65+func newPollID() string {
66+ var b [16]byte
67+ _, _ = rand.Read(b[:])
68+ return base64.RawURLEncoding.EncodeToString(b[:])
69+}
70+
71+// Run is the main loop. It returns nil after a graceful shutdown (ctx ended and
72+// every running job reported), or an error when the runner cannot go on.
73+//
74+// Capacity: a poll is made only when a slot is free. A lost response is retried
75+// with the same poll_id, so that the server hands back the same assignment.
76+func (r *Runner) Run(ctx context.Context) error {
77+ t := r.Timing.withDefaults()
78+ shutdown := make(chan struct{})
79+ slots := make(chan struct{}, max(1, r.Cfg.Capacity))
80+ defer func() {
81+ // Graceful shutdown: running jobs are cancelled with runner_shutdown and
82+ // get ShutdownGrace to stop; Run returns once they have reported.
83+ close(shutdown)
84+ r.active.Wait()
85+ }()
86+
87+ if err := r.Driver.Reap(ctx); err != nil && !errors.Is(err, sandbox.ErrBusy) {
88+ r.logf("reaping what a previous run left: %v", err)
89+ }
90+
91+ var pollID string
92+ bo := backoff{min: t.BackoffMin, max: t.BackoffMax}
93+ for ctx.Err() == nil {
94+ // Wait for a free slot, and for the end of a leak.
95+ select {
96+ case slots <- struct{}{}:
97+ case <-ctx.Done():
98+ return nil
99+ }
100+ if r.leaked.Load() {
101+ <-slots
102+ if !r.recover(ctx, slots) {
103+ return nil
104+ }
105+ continue
106+ }
107+ if pollID == "" {
108+ pollID = newPollID()
109+ }
110+ pctx, cancel := context.WithTimeout(ctx, t.PollTimeout)
111+ resp, err := r.Poll.FetchJob(pctx, connect.NewRequest(&ermite.FetchJobRequest{PollId: pollID}))
112+ cancel()
113+ if err != nil {
114+ <-slots
115+ if ctx.Err() != nil {
116+ return nil
117+ }
118+ if connect.CodeOf(err) == connect.CodeUnauthenticated {
119+ return ErrRemoved
120+ }
121+ if connect.CodeOf(err) == connect.CodeInvalidArgument {
122+ pollID = "" // our id was refused: a new one
123+ }
124+ d := bo.next()
125+ r.logf("FetchJob failed, retrying in %s: %v", d, err)
126+ if !sleep(ctx, d) {
127+ return nil
128+ }
129+ continue
130+ }
131+ bo.cur = 0
132+ pollID = "" // answered: the next poll is a new one
133+ if resp.Msg.Job == nil {
134+ <-slots
135+ d := t.NoJobPause
136+ if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > d {
137+ d = ra.AsDuration()
138+ }
139+ if !sleep(ctx, d) {
140+ return nil
141+ }
142+ continue
143+ }
144+ r.startJob(ctx, resp.Msg.Job, t, shutdown, slots)
145+ if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > 0 {
146+ if !sleep(ctx, ra.AsDuration()) {
147+ return nil
148+ }
149+ }
150+ }
151+ return nil
152+}
153+
154+// recover waits until no job runs, then reaps. It returns false when ctx ends.
155+func (r *Runner) recover(ctx context.Context, slots chan struct{}) bool {
156+ r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")
157+ for ctx.Err() == nil {
158+ if len(slots) == 0 {
159+ if err := r.Driver.Reap(ctx); err == nil {
160+ r.leaked.Store(false)
161+ r.logf("reaped: taking jobs again")
162+ return true
163+ } else if !errors.Is(err, sandbox.ErrBusy) {
164+ r.logf("reap failed, retrying: %v", err)
165+ }
166+ }
167+ if !sleep(ctx, 5*time.Second) {
168+ return false
169+ }
170+ }
171+ return false
172+}
173+
174+func (r *Runner) startJob(ctx context.Context, job *ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
175+ jctx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
176+ jr := &jobRun{
177+ r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
178+ logf: func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
179+ cancelCh: make(chan *execpb.Cancel, 1), steps: map[int32]*ermite.StepUpdate{},
180+ }
181+ r.logf("job %s (run %d, attempt %d, %s) assigned", job.JobId, job.RunId, job.Attempt, job.JobKey)
182+ r.active.Add(1)
183+ go func() {
184+ defer r.active.Done()
185+ defer func() { <-slots }()
186+ jr.run(r, shutdown)
187+ r.logf("job %s finished", job.JobId)
188+ }()
189+}
190+
191+// ---- registration
192+
193+// Isolation maps the driver's isolation to the protocol's.
194+func Isolation(i sandbox.Isolation) ermite.RunnerIsolation {
195+ switch i {
196+ case sandbox.IsolationVM:
197+ return ermite.RunnerIsolation_RUNNER_ISOLATION_VM
198+ case sandbox.IsolationContainerSysbox:
199+ return ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX
200+ case sandbox.IsolationContainerPrivileged:
201+ return ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED
202+ }
203+ return ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED
204+}
205+
206+// GoArch is the architecture in GitHub's names.
207+func GoArch() string {
208+ if runtime.GOARCH == "arm64" {
209+ return "arm64"
210+ }
211+ return "x64"
212+}
213+
214+// Register registers the runner with ermite using a token that has the
215+// runner:register scope, and stores the runner token (mode 0600) in the state
216+// directory. replace removes a runner of the same name first.
217+func Register(ctx context.Context, cfg Config, client ermiteconnect.RunnerServiceClient, isolation sandbox.Isolation, replace bool) (State, error) {
218+ iso := Isolation(isolation)
219+ if iso == ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED {
220+ return State{}, fmt.Errorf("unknown isolation %q", isolation)
221+ }
222+ rctx, cancel := context.WithTimeout(ctx, 30*time.Second)
223+ defer cancel()
224+ resp, err := client.RegisterRunner(rctx, connect.NewRequest(&ermite.RegisterRunnerRequest{
225+ Name: cfg.Name, Labels: cfg.Labels, Version: cfg.Version, Isolation: iso, Arch: GoArch(), Os: "linux", Replace: replace,
226+ }))
227+ if err != nil {
228+ return State{}, fmt.Errorf("register: %w", err)
229+ }
230+ st := State{URL: cfg.URL, RunnerID: resp.Msg.RunnerId, Name: cfg.Name, RunnerToken: resp.Msg.RunnerToken}
231+ if err := SaveState(cfg.StateDir, st); err != nil {
232+ return State{}, fmt.Errorf("the runner is registered but its token could not be stored: %w", err)
233+ }
234+ return st, nil
235+}
@@ -0,0 +1,235 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "crypto/rand"
8+ "encoding/base64"
9+ "errors"
10+ "fmt"
11+ "runtime"
12+ "sync"
13+ "sync/atomic"
14+ "time"
15+
16+ "connectrpc.com/connect"
17+
18+ ermite "bnema.dev/ermite-api/gen/ermite"
19+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
20+
21+ "bnema.dev/ermite-runner/internal/execstream/execpb"
22+ "bnema.dev/ermite-runner/internal/sandbox"
23+)
24+
25+// Runner fetches jobs from ermite and runs each one in its own sandbox.
26+type Runner struct {
27+ Cfg Config
28+ Driver sandbox.Driver
29+ // Poll is the client that carries the runner token (FetchJob).
30+ Poll ermiteconnect.RunnerServiceClient
31+ // Job builds the client of a job from its job token.
32+ Job func(jobToken string) ermiteconnect.RunnerServiceClient
33+ Timing Timing
34+ Logf func(format string, args ...any)
35+ // JobSpec sizes the sandbox of a job; the zero value takes the defaults.
36+ JobSpec func(*ermite.Job) sandbox.JobSpec
37+
38+ leaked atomic.Bool
39+ active sync.WaitGroup
40+}
41+
42+func (r *Runner) logf(format string, a ...any) {
43+ if r.Logf != nil {
44+ r.Logf(format, a...)
45+ }
46+}
47+
48+func (r *Runner) spec(j *ermite.Job) sandbox.JobSpec {
49+ if r.JobSpec != nil {
50+ return r.JobSpec(j)
51+ }
52+ return sandbox.JobSpec{JobID: j.JobId}
53+}
54+
55+// markLeaked records that a sandbox could not be cleaned up: no new job is
56+// taken until the running ones are over and Driver.Reap has succeeded.
57+func (r *Runner) markLeaked() { r.leaked.Store(true) }
58+
59+// ErrRemoved is returned by Run when ermite no longer knows the runner token:
60+// the runner was removed or replaced.
61+var ErrRemoved = errors.New("ermite rejected the runner token: the runner was removed or replaced; register it again")
62+
63+// newPollID is 16 random bytes in base64url: 22 characters of the set the
64+// protocol allows.
65+func newPollID() string {
66+ var b [16]byte
67+ _, _ = rand.Read(b[:])
68+ return base64.RawURLEncoding.EncodeToString(b[:])
69+}
70+
71+// Run is the main loop. It returns nil after a graceful shutdown (ctx ended and
72+// every running job reported), or an error when the runner cannot go on.
73+//
74+// Capacity: a poll is made only when a slot is free. A lost response is retried
75+// with the same poll_id, so that the server hands back the same assignment.
76+func (r *Runner) Run(ctx context.Context) error {
77+ t := r.Timing.withDefaults()
78+ shutdown := make(chan struct{})
79+ slots := make(chan struct{}, max(1, r.Cfg.Capacity))
80+ defer func() {
81+ // Graceful shutdown: running jobs are cancelled with runner_shutdown and
82+ // get ShutdownGrace to stop; Run returns once they have reported.
83+ close(shutdown)
84+ r.active.Wait()
85+ }()
86+
87+ if err := r.Driver.Reap(ctx); err != nil && !errors.Is(err, sandbox.ErrBusy) {
88+ r.logf("reaping what a previous run left: %v", err)
89+ }
90+
91+ var pollID string
92+ bo := backoff{min: t.BackoffMin, max: t.BackoffMax}
93+ for ctx.Err() == nil {
94+ // Wait for a free slot, and for the end of a leak.
95+ select {
96+ case slots <- struct{}{}:
97+ case <-ctx.Done():
98+ return nil
99+ }
100+ if r.leaked.Load() {
101+ <-slots
102+ if !r.recover(ctx, slots) {
103+ return nil
104+ }
105+ continue
106+ }
107+ if pollID == "" {
108+ pollID = newPollID()
109+ }
110+ pctx, cancel := context.WithTimeout(ctx, t.PollTimeout)
111+ resp, err := r.Poll.FetchJob(pctx, connect.NewRequest(&ermite.FetchJobRequest{PollId: pollID}))
112+ cancel()
113+ if err != nil {
114+ <-slots
115+ if ctx.Err() != nil {
116+ return nil
117+ }
118+ if connect.CodeOf(err) == connect.CodeUnauthenticated {
119+ return ErrRemoved
120+ }
121+ if connect.CodeOf(err) == connect.CodeInvalidArgument {
122+ pollID = "" // our id was refused: a new one
123+ }
124+ d := bo.next()
125+ r.logf("FetchJob failed, retrying in %s: %v", d, err)
126+ if !sleep(ctx, d) {
127+ return nil
128+ }
129+ continue
130+ }
131+ bo.cur = 0
132+ pollID = "" // answered: the next poll is a new one
133+ if resp.Msg.Job == nil {
134+ <-slots
135+ d := t.NoJobPause
136+ if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > d {
137+ d = ra.AsDuration()
138+ }
139+ if !sleep(ctx, d) {
140+ return nil
141+ }
142+ continue
143+ }
144+ r.startJob(ctx, resp.Msg.Job, t, shutdown, slots)
145+ if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > 0 {
146+ if !sleep(ctx, ra.AsDuration()) {
147+ return nil
148+ }
149+ }
150+ }
151+ return nil
152+}
153+
154+// recover waits until no job runs, then reaps. It returns false when ctx ends.
155+func (r *Runner) recover(ctx context.Context, slots chan struct{}) bool {
156+ r.logf("a sandbox could not be cleaned up: waiting for the running jobs, then reaping")
157+ for ctx.Err() == nil {
158+ if len(slots) == 0 {
159+ if err := r.Driver.Reap(ctx); err == nil {
160+ r.leaked.Store(false)
161+ r.logf("reaped: taking jobs again")
162+ return true
163+ } else if !errors.Is(err, sandbox.ErrBusy) {
164+ r.logf("reap failed, retrying: %v", err)
165+ }
166+ }
167+ if !sleep(ctx, 5*time.Second) {
168+ return false
169+ }
170+ }
171+ return false
172+}
173+
174+func (r *Runner) startJob(ctx context.Context, job *ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
175+ jctx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
176+ jr := &jobRun{
177+ r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
178+ logf: func(f string, a ...any) { r.logf("job "+job.JobId+": "+f, a...) },
179+ cancelCh: make(chan *execpb.Cancel, 1), steps: map[int32]*ermite.StepUpdate{},
180+ }
181+ r.logf("job %s (run %d, attempt %d, %s) assigned", job.JobId, job.RunId, job.Attempt, job.JobKey)
182+ r.active.Add(1)
183+ go func() {
184+ defer r.active.Done()
185+ defer func() { <-slots }()
186+ jr.run(r, shutdown)
187+ r.logf("job %s finished", job.JobId)
188+ }()
189+}
190+
191+// ---- registration
192+
193+// Isolation maps the driver's isolation to the protocol's.
194+func Isolation(i sandbox.Isolation) ermite.RunnerIsolation {
195+ switch i {
196+ case sandbox.IsolationVM:
197+ return ermite.RunnerIsolation_RUNNER_ISOLATION_VM
198+ case sandbox.IsolationContainerSysbox:
199+ return ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX
200+ case sandbox.IsolationContainerPrivileged:
201+ return ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED
202+ }
203+ return ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED
204+}
205+
206+// GoArch is the architecture in GitHub's names.
207+func GoArch() string {
208+ if runtime.GOARCH == "arm64" {
209+ return "arm64"
210+ }
211+ return "x64"
212+}
213+
214+// Register registers the runner with ermite using a token that has the
215+// runner:register scope, and stores the runner token (mode 0600) in the state
216+// directory. replace removes a runner of the same name first.
217+func Register(ctx context.Context, cfg Config, client ermiteconnect.RunnerServiceClient, isolation sandbox.Isolation, replace bool) (State, error) {
218+ iso := Isolation(isolation)
219+ if iso == ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED {
220+ return State{}, fmt.Errorf("unknown isolation %q", isolation)
221+ }
222+ rctx, cancel := context.WithTimeout(ctx, 30*time.Second)
223+ defer cancel()
224+ resp, err := client.RegisterRunner(rctx, connect.NewRequest(&ermite.RegisterRunnerRequest{
225+ Name: cfg.Name, Labels: cfg.Labels, Version: cfg.Version, Isolation: iso, Arch: GoArch(), Os: "linux", Replace: replace,
226+ }))
227+ if err != nil {
228+ return State{}, fmt.Errorf("register: %w", err)
229+ }
230+ st := State{URL: cfg.URL, RunnerID: resp.Msg.RunnerId, Name: cfg.Name, RunnerToken: resp.Msg.RunnerToken}
231+ if err := SaveState(cfg.StateDir, st); err != nil {
232+ return State{}, fmt.Errorf("the runner is registered but its token could not be stored: %w", err)
233+ }
234+ return st, nil
235+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close