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// ServerFeaturesFunc returns the function that reads the features of the
110// server with GetVersion, which needs no authentication. It is called once
111// per job.
112func (c Config) ServerFeaturesFunc(url string) ServerFeatures {
113 hc := &http.Client{
114 Transport: bearer{base: http.DefaultTransport, allowInsecure: c.InsecureHTTP},
115 CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
116 }
117 client := ermiteconnect.NewServerServiceClient(hc, strings.TrimRight(url, "/")+APIPath)
118 return func(ctx context.Context) ([]string, error) {
119 resp, err := client.GetVersion(ctx, connect.NewRequest(&ermite.GetVersionRequest{}))
120 if err != nil {
121 return nil, err
122 }
123 return resp.Msg.GetFeatures(), nil
124 }
125}
126
127// APIPath is where ermite serves its API under the base URL, as in ermite's own
128// client (internal/adapters/client.APIPath there).
129const APIPath = "/-/api"
130
131// ClientFactory builds a client for a token; tests replace it.
132type ClientFactory func(token string) ermiteconnect.RunnerServiceClient
133
134// jobEnded extracts the JobEnded detail of a failed_precondition error.
135func jobEnded(err error) (*ermite.JobEnded, bool) {
136 var ce *connect.Error
137 if !errors.As(err, &ce) || ce.Code() != connect.CodeFailedPrecondition {
138 return nil, false
139 }
140 for _, d := range ce.Details() {
141 m, derr := d.Value()
142 if derr != nil {
143 continue
144 }
145 if je, ok := m.(*ermite.JobEnded); ok {
146 return je, true
147 }
148 }
149 return nil, false
150}
151
152// kind classifies the error of a job call.
153type kind int
154
155const (
156 kindRetry kind = iota // network error, unavailable, internal...: try again
157 kindEnded // failed_precondition with JobEnded: stop, send nothing more
158 kindAbandoned // unauthenticated: the assignment is gone, stop at once
159 kindPermanent // the request itself is wrong: do not repeat it
160)
161
162func classify(err error) kind {
163 if err == nil {
164 return kindRetry
165 }
166 if _, ok := jobEnded(err); ok {
167 return kindEnded
168 }
169 switch connect.CodeOf(err) {
170 case connect.CodeUnauthenticated:
171 return kindAbandoned
172 case connect.CodeInvalidArgument, connect.CodeFailedPrecondition, connect.CodeOutOfRange,
173 connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeUnimplemented, connect.CodeAlreadyExists:
174 return kindPermanent
175 }
176 // Canceled by our own context is handled by the callers through ctx.Err().
177 return kindRetry
178}
179
180// Timing holds the delays of the host loop. Zero fields take the defaults,
181// which are the protocol's.
182type Timing struct {
183 // PollTimeout bounds one FetchJob call (the server holds it up to 50 s).
184 PollTimeout time.Duration
185 // HeartbeatEvery is the heartbeat period (30 s); HeartbeatRetry the delay
186 // after a failed heartbeat (5 s).
187 HeartbeatEvery, HeartbeatRetry time.Duration
188 // CallTimeout bounds one unary job call.
189 CallTimeout time.Duration
190 // BackoffMin and BackoffMax bound the retry delay of calls that fail
191 // without a verdict.
192 BackoffMin, BackoffMax time.Duration
193 // LogFlush is how long log lines are held to fill a batch.
194 LogFlush time.Duration
195 // WatchRetry is the delay before WatchJob is reopened.
196 WatchRetry time.Duration
197 // TimeoutGrace is how long the executor gets after the job timeout to stop
198 // its steps (SIGINT, then SIGTERM after 7.5 s, then SIGKILL 2.5 s later) and
199 // report, before the sandbox is destroyed.
200 TimeoutGrace time.Duration
201 // ShutdownGrace is how long jobs get to stop after a shutdown signal before
202 // their sandbox is destroyed. The executor needs 10 s for its escalation
203 // (SIGINT, SIGTERM after 7.5 s, SIGKILL 2.5 s later) and some for the
204 // post-steps and the report to the host.
205 ShutdownGrace time.Duration
206 // StopDeadline bounds the destruction of a sandbox.
207 StopDeadline time.Duration
208 // ShutdownReport bounds the last flush of the log and CompleteJob of a job
209 // that a shutdown cut short.
210 ShutdownReport time.Duration
211 // CompleteDeadline bounds the flush of logs and CompleteJob at the end of
212 // a job.
213 CompleteDeadline time.Duration
214 // NoJobPause is the pause after a poll without a job and without
215 // retry_after.
216 NoJobPause time.Duration
217 // ReapRetry is the delay between attempts to reap after a leaked sandbox.
218 ReapRetry time.Duration
219}
220
221func (t Timing) withDefaults() Timing {
222 def := func(d *time.Duration, v time.Duration) {
223 if *d == 0 {
224 *d = v
225 }
226 }
227 def(&t.PollTimeout, 60*time.Second)
228 def(&t.HeartbeatEvery, 30*time.Second)
229 def(&t.HeartbeatRetry, 5*time.Second)
230 def(&t.CallTimeout, 30*time.Second)
231 def(&t.BackoffMin, time.Second)
232 def(&t.BackoffMax, 30*time.Second)
233 def(&t.LogFlush, 250*time.Millisecond)
234 def(&t.WatchRetry, time.Second)
235 def(&t.TimeoutGrace, 30*time.Second)
236 def(&t.ShutdownGrace, 30*time.Second)
237 def(&t.StopDeadline, 30*time.Second)
238 def(&t.ShutdownReport, 30*time.Second)
239 def(&t.CompleteDeadline, 5*time.Minute)
240 def(&t.NoJobPause, 0)
241 def(&t.ReapRetry, 5*time.Second)
242 return t
243}
244
245// ShutdownBudget is the longest a graceful shutdown takes with these timings:
246// the grace for the executor, the destruction of the sandbox, and the report.
247// The jobs shut down in parallel, so it does not depend on their number. The
248// service manager must wait at least this long (contrib/ermite-runner.service
249// has TimeoutStopSec=120 for the defaults, 90 s).
250func (t Timing) ShutdownBudget() time.Duration {
251 t = t.withDefaults()
252 return t.ShutdownGrace + t.StopDeadline + t.ShutdownReport
253}
254
255// sleep waits d or until ctx ends; it reports whether the full delay passed.
256func sleep(ctx context.Context, d time.Duration) bool {
257 if d <= 0 {
258 return ctx.Err() == nil
259 }
260 t := time.NewTimer(d)
261 defer t.Stop()
262 select {
263 case <-t.C:
264 return true
265 case <-ctx.Done():
266 return false
267 }
268}
269
270// backoff doubles from min to max.
271type backoff struct{ min, max, cur time.Duration }
272
273func (b *backoff) next() time.Duration {
274 if b.cur == 0 {
275 b.cur = b.min
276 } else if b.cur *= 2; b.cur > b.max {
277 b.cur = b.max
278 }
279 return b.cur
280}