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