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.
7package host
8
9import (
10 "context"
11 "errors"
12 "fmt"
13 "net/http"
14 "net/netip"
15 "net/url"
16 "strings"
17 "time"
18
19 "connectrpc.com/connect"
20
21 ermite "bnema.dev/ermite-api/gen/ermite"
22 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
23)
24
25// bearer adds "Authorization: Bearer <token>" to every request. The token is
26// never logged. It refuses to send the token over a connection that is not
27// protected: plain http is only for loopback, unless AllowInsecure is set.
28type bearer struct {
29 token string
30 base http.RoundTripper
31 allowInsecure bool
32}
33
34// ErrInsecureURL is returned for a server URL that is not https and not
35// loopback: the runner token and the job tokens would travel in clear.
36var ErrInsecureURL = errors.New("the server URL must be https (plain http is only accepted for localhost; ERMITE_RUNNER_INSECURE_HTTP=1 allows it, for tests, and sends every token in clear)")
37
38// CheckURL validates the URL of the server: http or https, a host, no
39// credentials, and https unless the host is loopback or allowInsecure is set.
40func CheckURL(raw string, allowInsecure bool) error {
41 u, err := url.Parse(raw)
42 if err != nil || u.Host == "" {
43 return fmt.Errorf("%q is not a valid URL", raw)
44 }
45 if u.User != nil {
46 return errors.New("the server URL must not contain credentials")
47 }
48 switch u.Scheme {
49 case "https":
50 return nil
51 case "http":
52 if allowInsecure || isLoopback(u.Hostname()) {
53 return nil
54 }
55 return ErrInsecureURL
56 }
57 return fmt.Errorf("%q: the scheme must be https", raw)
58}
59
60func isLoopback(host string) bool {
61 // Host names are not case sensitive (LOCALHOST, Localhost).
62 if strings.EqualFold(host, "localhost") {
63 return true
64 }
65 ip, err := netip.ParseAddr(host)
66 return err == nil && ip.IsLoopback()
67}
68
69func (b bearer) RoundTrip(r *http.Request) (*http.Response, error) {
70 if err := CheckURL(r.URL.String(), b.allowInsecure); err != nil {
71 return nil, err
72 }
73 r = r.Clone(r.Context())
74 if b.token != "" {
75 r.Header.Set("Authorization", "Bearer "+b.token)
76 }
77 base := b.base
78 if base == nil {
79 base = http.DefaultTransport
80 }
81 return base.RoundTrip(r)
82}
83
84// NewClient returns a RunnerService client that authenticates with token. The
85// HTTP client has no overall timeout: calls carry their own context deadline
86// (WatchJob is a long stream). It never follows a redirect: a response from the
87// server that points elsewhere is returned as is, so that the token is not
88// carried to another host (Go would keep the Authorization header on a redirect
89// to the same host and drop it for others, but the client has no business
90// following any).
91func NewClient(baseURL, token string, base http.RoundTripper) ermiteconnect.RunnerServiceClient {
92 return newClient(baseURL, token, base, false)
93}
94
95// NewInsecureClient is NewClient for a server reached over plain http that is
96// not loopback. Only for tests and trusted networks.
97func NewInsecureClient(baseURL, token string, base http.RoundTripper) ermiteconnect.RunnerServiceClient {
98 return newClient(baseURL, token, base, true)
99}
100
101func newClient(baseURL, token string, base http.RoundTripper, insecure bool) ermiteconnect.RunnerServiceClient {
102 hc := &http.Client{
103 Transport: bearer{token: token, base: base, allowInsecure: insecure},
104 CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
105 }
106 return ermiteconnect.NewRunnerServiceClient(hc, strings.TrimRight(baseURL, "/")+APIPath)
107}
108
109// APIPath is where ermite serves its API under the base URL, as in ermite's own
110// client (internal/adapters/client.APIPath there).
111const APIPath = "/-/api"
112
113// ClientFactory builds a client for a token; tests replace it.
114type ClientFactory func(token string) ermiteconnect.RunnerServiceClient
115
116// jobEnded extracts the JobEnded detail of a failed_precondition error.
117func jobEnded(err error) (*ermite.JobEnded, bool) {
118 var ce *connect.Error
119 if !errors.As(err, &ce) || ce.Code() != connect.CodeFailedPrecondition {
120 return nil, false
121 }
122 for _, d := range ce.Details() {
123 m, derr := d.Value()
124 if derr != nil {
125 continue
126 }
127 if je, ok := m.(*ermite.JobEnded); ok {
128 return je, true
129 }
130 }
131 return nil, false
132}
133
134// kind classifies the error of a job call.
135type kind int
136
137const (
138 kindRetry kind = iota // network error, unavailable, internal...: try again
139 kindEnded // failed_precondition with JobEnded: stop, send nothing more
140 kindAbandoned // unauthenticated: the assignment is gone, stop at once
141 kindPermanent // the request itself is wrong: do not repeat it
142)
143
144func classify(err error) kind {
145 if err == nil {
146 return kindRetry
147 }
148 if _, ok := jobEnded(err); ok {
149 return kindEnded
150 }
151 switch connect.CodeOf(err) {
152 case connect.CodeUnauthenticated:
153 return kindAbandoned
154 case connect.CodeInvalidArgument, connect.CodeFailedPrecondition, connect.CodeOutOfRange,
155 connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeUnimplemented, connect.CodeAlreadyExists:
156 return kindPermanent
157 }
158 // Canceled by our own context is handled by the callers through ctx.Err().
159 return kindRetry
160}
161
162// Timing holds the delays of the host loop. Zero fields take the defaults,
163// which are the protocol's.
164type Timing struct {
165 // PollTimeout bounds one FetchJob call (the server holds it up to 50 s).
166 PollTimeout time.Duration
167 // HeartbeatEvery is the heartbeat period (30 s); HeartbeatRetry the delay
168 // after a failed heartbeat (5 s).
169 HeartbeatEvery, HeartbeatRetry time.Duration
170 // CallTimeout bounds one unary job call.
171 CallTimeout time.Duration
172 // BackoffMin and BackoffMax bound the retry delay of calls that fail
173 // without a verdict.
174 BackoffMin, BackoffMax time.Duration
175 // LogFlush is how long log lines are held to fill a batch.
176 LogFlush time.Duration
177 // WatchRetry is the delay before WatchJob is reopened.
178 WatchRetry time.Duration
179 // TimeoutGrace is how long the executor gets after the job timeout to stop
180 // its steps (SIGINT, then SIGTERM after 7.5 s, then SIGKILL 2.5 s later) and
181 // report, before the sandbox is destroyed.
182 TimeoutGrace time.Duration
183 // ShutdownGrace is how long jobs get to stop after a shutdown signal before
184 // their sandbox is destroyed. The executor needs 10 s for its escalation
185 // (SIGINT, SIGTERM after 7.5 s, SIGKILL 2.5 s later) and some for the
186 // post-steps and the report to the host.
187 ShutdownGrace time.Duration
188 // StopDeadline bounds the destruction of a sandbox.
189 StopDeadline time.Duration
190 // ShutdownReport bounds the last flush of the log and CompleteJob of a job
191 // that a shutdown cut short.
192 ShutdownReport time.Duration
193 // CompleteDeadline bounds the flush of logs and CompleteJob at the end of
194 // a job.
195 CompleteDeadline time.Duration
196 // NoJobPause is the pause after a poll without a job and without
197 // retry_after.
198 NoJobPause time.Duration
199 // ReapRetry is the delay between attempts to reap after a leaked sandbox.
200 ReapRetry time.Duration
201}
202
203func (t Timing) withDefaults() Timing {
204 def := func(d *time.Duration, v time.Duration) {
205 if *d == 0 {
206 *d = v
207 }
208 }
209 def(&t.PollTimeout, 60*time.Second)
210 def(&t.HeartbeatEvery, 30*time.Second)
211 def(&t.HeartbeatRetry, 5*time.Second)
212 def(&t.CallTimeout, 30*time.Second)
213 def(&t.BackoffMin, time.Second)
214 def(&t.BackoffMax, 30*time.Second)
215 def(&t.LogFlush, 250*time.Millisecond)
216 def(&t.WatchRetry, time.Second)
217 def(&t.TimeoutGrace, 30*time.Second)
218 def(&t.ShutdownGrace, 30*time.Second)
219 def(&t.StopDeadline, 30*time.Second)
220 def(&t.ShutdownReport, 30*time.Second)
221 def(&t.CompleteDeadline, 5*time.Minute)
222 def(&t.NoJobPause, 0)
223 def(&t.ReapRetry, 5*time.Second)
224 return t
225}
226
227// ShutdownBudget is the longest a graceful shutdown takes with these timings:
228// the grace for the executor, the destruction of the sandbox, and the report.
229// The jobs shut down in parallel, so it does not depend on their number. The
230// service manager must wait at least this long (contrib/ermite-runner.service
231// has TimeoutStopSec=120 for the defaults, 90 s).
232func (t Timing) ShutdownBudget() time.Duration {
233 t = t.withDefaults()
234 return t.ShutdownGrace + t.StopDeadline + t.ShutdownReport
235}
236
237// sleep waits d or until ctx ends; it reports whether the full delay passed.
238func sleep(ctx context.Context, d time.Duration) bool {
239 if d <= 0 {
240 return ctx.Err() == nil
241 }
242 t := time.NewTimer(d)
243 defer t.Stop()
244 select {
245 case <-t.C:
246 return true
247 case <-ctx.Done():
248 return false
249 }
250}
251
252// backoff doubles from min to max.
253type backoff struct{ min, max, cur time.Duration }
254
255func (b *backoff) next() time.Duration {
256 if b.cur == 0 {
257 b.cur = b.min
258 } else if b.cur *= 2; b.cur > b.max {
259 b.cur = b.max
260 }
261 return b.cur
262}