amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package main
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "flag"
10 "fmt"
11 "io"
12 "log"
13 "mime/multipart"
14 "net/http"
15 "net/url"
16 "os"
17 "os/signal"
18 "path/filepath"
19 "slices"
20 "strconv"
21 "strings"
22 "sync"
23 "syscall"
24 "time"
25
26 "github.com/carlosframework/platform/internal/provision"
27 "github.com/carlosframework/platform/internal/runs"
28)
29
30const runUsageRun = "usage: carlos run --app <app> --instance <host> [--file <path>]... [--timeout 30m] -- <arg>..."
31
32const runUsageFollow = "carlos run --app <app> --instance <host> --follow <id>"
33
34const runUsageKey = "add --idempotency-key <key> to start a run you can safely run again: the same key and request follow the run already started"
35
36func runUsageText() string {
37 return runUsageRun + "\n " + runUsageFollow + "\n " + runUsageKey + "\n"
38}
39
40const runDirectModeMsg = "carlos run works through a console, and this shell is in direct bucket mode. unset CARLOS_DEPLOYMENT_BUCKET and CARLOS_DEPLOYMENT_DIR, then run carlos auth login --console <url>"
41
42// errRunUsage is a command line that does not parse: the usage text says
43// what was wrong.
44var errRunUsage = errors.New("carlos run: usage")
45
46// runOpts is `carlos run` parsed.
47type runOpts struct {
48 app, instance, follow, console string
49 idempotencyKey string
50 acctRaw, acctIDRaw *string
51 files []string
52 timeout time.Duration
53 argv []string
54}
55
56// parseRunArgs splits at the first "--": carlos's flags before it, the
57// command after it. Nothing may sit between the flags and the "--", so a
58// forgotten "--" is a usage error rather than a command whose flags carlos
59// swallowed.
60func parseRunArgs(args []string, stderr io.Writer) (runOpts, error) {
61 flagArgs, argv := args, []string(nil)
62 if i := slices.Index(args, "--"); i >= 0 {
63 flagArgs, argv = args[:i], args[i+1:]
64 }
65 fs := flag.NewFlagSet("run", flag.ContinueOnError)
66 fs.SetOutput(stderr)
67 app := appFlag(fs, "the app whose command to run")
68 instance := fs.String("instance", "", "the instance to run it against")
69 var files multiFlag
70 fs.Var(&files, "file", "a file to upload with the run (repeatable; 64 MB in total)")
71 timeout := fs.Duration("timeout", runs.DefaultTimeout, "how long the command may run (1s to 6h)")
72 follow := fs.String("follow", "", "attach to a run already started, by its id")
73 key := fs.String("idempotency-key", "", "a key of your choosing: run again with the same key and the same request, and you follow the run already started instead of starting a second")
74 acctRaw, acctIDRaw, _ := accountFlags(fs)
75 console := consoleOption(fs, "console to run through (default: the only logged-in one)")
76 fs.Usage = flagUsage(fs, runUsageText())
77 if err := fs.Parse(flagArgs); err != nil {
78 return runOpts{}, err
79 }
80 if fs.NArg() > 0 {
81 return runOpts{}, errRunUsage
82 }
83 o := runOpts{app: strings.TrimSpace(*app), instance: strings.TrimSpace(*instance), follow: strings.TrimSpace(*follow),
84 console: *console, acctRaw: acctRaw, acctIDRaw: acctIDRaw, files: files, timeout: *timeout, argv: argv,
85 idempotencyKey: *key}
86 switch {
87 case o.instance == "":
88 return o, errors.New("--app and --instance name the instance to run against")
89 case o.follow != "" && (len(o.argv) > 0 || len(o.files) > 0):
90 return o, errors.New("--follow takes a run id and no command after --")
91 case runs.CheckTimeout(o.timeout) != nil:
92 return o, errors.New("--timeout must be between 1s and 6h")
93 case o.idempotencyKey != "" && !runs.ValidKey(o.idempotencyKey):
94 return o, errors.New(runKeyShapeMsg)
95 case o.idempotencyKey != "" && o.follow != "":
96 return o, errors.New("--idempotency-key is for starting a run. --follow starts none")
97 }
98 return o, nil
99}
100
101// runKeyShapeMsg is what the CLI says of a key that the console would refuse.
102const runKeyShapeMsg = "--idempotency-key is 1 to 128 characters: letters, digits, dots, underscores, colons and hyphens"
103
104// errRunNotRegular is a --file that is a directory or a device: only a
105// regular file has a size to promise the console.
106var errRunNotRegular = errors.New("not a regular file")
107
108// errRunFileChanged is a --file whose size is no longer what meta promised.
109var errRunFileChanged = errors.New("the file changed while carlos was reading it")
110
111// errRunLocalFile is what errors.Is finds in a --file that cannot be
112// uploaded: a failure on this machine (exit 2), not the console's answer.
113var errRunLocalFile = errors.New("a --file could not be read")
114
115// runFileError is a --file failure carrying the RUN-C53 text.
116type runFileError struct{ err error }
117
118func (e *runFileError) Error() string { return e.err.Error() }
119func (e *runFileError) Unwrap() error { return e.err }
120func (e *runFileError) Is(target error) bool { return target == errRunLocalFile }
121
122// runReadError is a --file that cannot be uploaded (RUN-C53). It is a local
123// failure, and either nothing has been sent or the upload is aborted before
124// its last byte.
125func runReadError(path string, err error) error {
126 return &runFileError{fmt.Errorf("could not read %s: %w", path, err)}
127}
128
129// runMetaFor is the create request's meta: argv, the timeout, and each
130// file's base name and size. A run with an idempotency key also declares
131// each file's sha256: the console needs it to tell a retry of the same
132// request from a different one. The console checks all of it again.
133func runMetaFor(o runOpts) (runs.Meta, error) {
134 m := runs.Meta{Argv: o.argv, TimeoutS: int64(o.timeout / time.Second)}
135 for _, p := range o.files {
136 st, err := os.Stat(p)
137 if err != nil {
138 return m, runReadError(p, err)
139 }
140 if !st.Mode().IsRegular() {
141 return m, runReadError(p, errRunNotRegular)
142 }
143 f := runs.MetaFile{Name: filepath.Base(p), Size: st.Size()}
144 if o.idempotencyKey != "" {
145 if f.SHA256, err = runFileSum(p); err != nil {
146 return m, runReadError(p, err)
147 }
148 }
149 m.Files = append(m.Files, f)
150 }
151 return m, nil
152}
153
154// runFileSum is the hex sha256 of the file at path.
155func runFileSum(path string) (string, error) {
156 f, err := os.Open(path)
157 if err != nil {
158 return "", err
159 }
160 defer f.Close()
161 h := sha256.New()
162 if _, err := io.Copy(h, f); err != nil {
163 return "", err
164 }
165 return hex.EncodeToString(h.Sum(nil)), nil
166}
167
168// runClient is the console's five run routes (run spec §5.1, §5.6, §5.7).
169type runClient interface {
170 create(ctx context.Context, meta runs.Meta, paths []string) (runs.Created, error)
171 page(ctx context.Context, id string, after int64) (runs.Page, error)
172 cancel(ctx context.Context, id string) (string, error)
173 expire(ctx context.Context, id string) (string, error)
174 ack(ctx context.Context, id string, lastSeq int64) error
175}
176
177// runConsoleError is a 4xx the console answered, with its message: a
178// refusal by the console, exit 2.
179type runConsoleError struct {
180 status int
181 msg string
182 retryAfter time.Duration // a 429's Retry-After, when the console sent one
183}
184
185func (e *runConsoleError) Error() string { return e.msg }
186
187// runAnswer decodes a want-status answer into out, or turns any other
188// answer into an error: a 4xx into a runConsoleError.
189func runAnswer(status int, body []byte, err error, want int, out any) error {
190 if err != nil {
191 return err
192 }
193 if status != want {
194 e := apiErrorFromBody(status, body)
195 if status >= 400 && status < 500 {
196 return &runConsoleError{status: status, msg: e.Error()}
197 }
198 return e
199 }
200 if out == nil {
201 return nil
202 }
203 return json.Unmarshal(body, out)
204}
205
206// runAPI is runClient over the signed-in console.
207type runAPI struct {
208 s *apiSession
209 a apiAcct
210 app string
211 host string
212 key string // the Idempotency-Key a create carries; "" sends none
213}
214
215func (c runAPI) path(tail string) string {
216 return apiPath(c.a, c.app, "/instances/"+url.PathEscape(c.host)+"/runs"+tail)
217}
218
219// create sends meta, then one file part per meta file in meta's order, each
220// named as meta names it and exactly as long as meta says. A file whose size
221// is no longer meta's fails the call before its part is written, or aborts
222// the upload part-way: the console never gets a whole request whose parts
223// disagree with its meta.
224func (c runAPI) create(ctx context.Context, meta runs.Meta, paths []string) (runs.Created, error) {
225 if len(paths) != len(meta.Files) {
226 return runs.Created{}, errors.New("carlos run: the files to upload and the files in meta differ")
227 }
228 var mu sync.Mutex
229 var writeErr error
230 var hdr http.Header
231 if c.key != "" {
232 hdr = http.Header{"Idempotency-Key": {c.key}}
233 }
234 status, body, err := c.s.c.doMultipartHeaders(ctx, http.MethodPost, c.path(""), hdr, func(mw *multipart.Writer) (err error) {
235 defer func() {
236 mu.Lock()
237 writeErr = err
238 mu.Unlock()
239 }()
240 mp, err := mw.CreateFormField("meta")
241 if err != nil {
242 return err
243 }
244 if err := json.NewEncoder(mp).Encode(meta); err != nil {
245 return err
246 }
247 for i, p := range paths {
248 if err := writeRunFilePart(mw, p, meta.Files[i]); err != nil {
249 return err
250 }
251 }
252 return nil
253 })
254 if err != nil {
255 // The http client wraps a failed body in a *url.Error. When the
256 // failure was a file the writer could not read, the user is told that,
257 // not which URL the request was for.
258 mu.Lock()
259 we := writeErr
260 mu.Unlock()
261 if errors.Is(we, errRunLocalFile) {
262 return runs.Created{}, we
263 }
264 }
265 var out runs.Created
266 return out, runAnswer(status, body, err, http.StatusAccepted, &out)
267}
268
269// runFileHandle is the open file writeRunFilePart reads: the reader and the
270// file's current size.
271type runFileHandle interface {
272 io.Reader
273 Stat() (os.FileInfo, error)
274}
275
276// writeRunFilePart writes one file part: exactly want.Size bytes of the file
277// at path, under want.Name. The size is checked on the open file, before and
278// after the copy, so a file that grew or shrank since the stat is an error
279// and not a part that disagrees with meta.
280func writeRunFilePart(mw *multipart.Writer, path string, want runs.MetaFile) error {
281 f, err := os.Open(path)
282 if err != nil {
283 return runReadError(path, err)
284 }
285 defer f.Close()
286 return copyRunFilePart(mw, f, path, want)
287}
288
289// copyRunFilePart is writeRunFilePart over an open file.
290func copyRunFilePart(mw *multipart.Writer, f runFileHandle, path string, want runs.MetaFile) error {
291 st, err := f.Stat()
292 if err != nil {
293 return runReadError(path, err)
294 }
295 if !st.Mode().IsRegular() {
296 return runReadError(path, errRunNotRegular)
297 }
298 if st.Size() != want.Size {
299 return runReadError(path, errRunFileChanged)
300 }
301 fp, err := mw.CreateFormFile("file", want.Name)
302 if err != nil {
303 return err
304 }
305 if n, err := io.CopyN(fp, f, want.Size); err != nil {
306 if n < want.Size && errors.Is(err, io.EOF) {
307 err = errRunFileChanged
308 }
309 return runReadError(path, err)
310 }
311 if st, err := f.Stat(); err != nil || st.Size() != want.Size {
312 return runReadError(path, errRunFileChanged)
313 }
314 return nil
315}
316
317func (c runAPI) page(ctx context.Context, id string, after int64) (runs.Page, error) {
318 status, h, body, err := c.s.getJSONHeader(ctx, c.path("/"+url.PathEscape(id)+fmt.Sprintf("?after=%d", after)))
319 var out runs.Page
320 err = runAnswer(status, body, err, http.StatusOK, &out)
321 var ce *runConsoleError
322 if errors.As(err, &ce) && status == http.StatusTooManyRequests {
323 if n, perr := strconv.Atoi(h.Get("Retry-After")); perr == nil && n > 0 {
324 ce.retryAfter = time.Duration(n) * time.Second
325 }
326 }
327 return out, err
328}
329
330func (c runAPI) cancel(ctx context.Context, id string) (string, error) {
331 status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/cancel"), struct{}{})
332 var out runs.StateAnswer
333 return out.State, runAnswer(status, body, err, http.StatusOK, &out)
334}
335
336func (c runAPI) expire(ctx context.Context, id string) (string, error) {
337 status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/expire"), struct{}{})
338 var out runs.StateAnswer
339 return out.State, runAnswer(status, body, err, http.StatusOK, &out)
340}
341
342func (c runAPI) ack(ctx context.Context, id string, lastSeq int64) error {
343 status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/ack"), runs.Ack{LastSeq: lastSeq})
344 return runAnswer(status, body, err, http.StatusOK, nil)
345}
346
347// runTerm is how the follow loop meets the terminal and the clock; tests
348// replace all of it.
349type runTerm struct {
350 stdout, stderr io.Writer
351 interrupts <-chan os.Signal
352 after func(time.Duration) <-chan time.Time
353 now func() time.Time
354}
355
356func (t runTerm) say(s string) { fmt.Fprintln(t.stderr, s) }
357
358// runAckTries is how many times an ack that the console cannot take yet, or
359// that did not arrive, is sent; runAckWait is the pause between them.
360const (
361 runAckTries = 10
362 runAckWait = time.Second
363)
364
365// How the follow loop treats a console that is slow to answer. A page that
366// fails for a reason that may pass (a 5xx, a lost connection, a 429) is asked
367// for again; after runPageGiveUp of failing in a row the follow is given up
368// (the run carries on, and --follow picks it up again). A 429 waits longer,
369// or as long as the console's Retry-After says, at most runRateWaitMax. A
370// page cut short is followed after a short pause, so that catching up on a
371// long run cannot spend the console's per-token budget in a burst.
372const (
373 runPageGiveUp = 2 * time.Minute
374 runRateWait = 5 * time.Second
375 runRateWaitMax = time.Minute
376 runMorePause = 200 * time.Millisecond
377)
378
379// runNotPickedUp is RUN-C55.
380func runNotPickedUp(host string) string {
381 return fmt.Sprintf("%s: not picked up within 60 seconds. the run did not start. try again", host)
382}
383
384// runLetGo is RUN-C86: the record no longer names the run and no result
385// came back for it within runPageGiveUp.
386func runLetGo(host, id string) string {
387 return fmt.Sprintf("%s: the instance no longer holds run %s, and no result came back. it may have done part of its work", host, id)
388}
389
390// runConflict reports whether err is the console's 409 with exactly msg.
391func runConflict(err error, msg string) bool {
392 var ce *runConsoleError
393 return errors.As(err, &ce) && ce.status == http.StatusConflict && ce.msg == msg
394}
395
396// ackRun tells the console the output has been read, so it can delete the
397// run's objects. The output is already on the terminal and the exit code is
398// the command's, so an ack that cannot be made is said and nothing more.
399// An ack retried after a lost 200 finds the run gone (404), which is done.
400// An ack that beats the box's last write (409, not finished) or does not
401// arrive is sent again a few times.
402func ackRun(ctx context.Context, c runClient, t runTerm, id string, lastSeq int64) {
403 var err error
404 for try := 0; try < runAckTries && ctx.Err() == nil; try++ {
405 if try > 0 {
406 select {
407 case <-t.after(runAckWait):
408 case <-ctx.Done():
409 return
410 }
411 }
412 err = c.ack(ctx, id, lastSeq)
413 var ce *runConsoleError
414 switch {
415 case err == nil:
416 return
417 case errors.As(err, &ce) && ce.status == http.StatusNotFound:
418 return
419 case errors.As(err, &ce) && !runConflict(err, fmt.Sprintf("run %s has not finished yet", id)):
420 t.say(fmt.Sprintf("carlos run: %s", err))
421 return
422 }
423 }
424 if ctx.Err() != nil {
425 return // cut off by Ctrl-C: nothing hangs on the ack
426 }
427 t.say(fmt.Sprintf("carlos run: %s", err))
428}
429
430// followRun prints a run's output until its result, acknowledges it, and
431// returns the exit code (run spec §5.6). It polls once a second; a page
432// cut short is followed at once; the result is in hand only when a page
433// carries it and is not cut short, because the console reads the result
434// before it lists chunks. A run still queued at expireAt is expired (§5.2);
435// zero never expires. A page with no state and no result is a run the
436// record no longer names (the instance was deleted, or the run was dropped):
437// a result may still arrive, so it is asked for again for runPageGiveUp, and
438// then the follow ends with RUN-C86 and 125. The first Ctrl-C cancels the
439// run on the box and keeps following; the second stops waiting (§5.7).
440func followRun(ctx context.Context, c runClient, t runTerm, host, id string, expireAt time.Time) int {
441 var after int64
442 watch := &runWatch{ch: t.interrupts}
443 cancel := runCancel{w: watch}
444 var failingSince, letGoSince time.Time
445 for {
446 var p runs.Page
447 var err error
448 watch.run(ctx, func(ctx context.Context) { p, err = c.page(ctx, id, after) })
449 if watch.stopped() {
450 return stopWaiting(ctx, t, c, id, &cancel)
451 }
452 if err != nil {
453 var ce *runConsoleError
454 switch {
455 case errors.As(err, &ce) && ce.status == http.StatusNotFound:
456 t.say(fmt.Sprintf("%s has no run %s, or this console has no carlos run yet", host, id))
457 return 2
458 case errors.As(err, &ce) && ce.status != http.StatusTooManyRequests:
459 t.say(fmt.Sprintf("carlos run: %s", err))
460 return 2
461 }
462 // A 5xx, a 429 or a connection that failed may pass: ask again,
463 // for a while.
464 if failingSince.IsZero() {
465 failingSince = t.now()
466 }
467 if t.now().Sub(failingSince) >= runPageGiveUp {
468 t.say(fmt.Sprintf("carlos run: %s", err))
469 return 1
470 }
471 wait := time.Second
472 if ce != nil {
473 wait = min(max(runRateWait, ce.retryAfter), runRateWaitMax)
474 }
475 if interrupted(ctx, t, c, id, &cancel, wait) {
476 return 130
477 }
478 continue
479 }
480 failingSince = time.Time{}
481 for _, ch := range p.Chunks {
482 if ch.Seq <= after {
483 continue
484 }
485 w := t.stdout
486 if ch.Stream == runs.StreamErr {
487 w = t.stderr
488 }
489 w.Write(ch.Data)
490 after = ch.Seq
491 }
492 if p.More {
493 if interrupted(ctx, t, c, id, &cancel, runMorePause) {
494 return 130
495 }
496 continue
497 }
498 if p.Result != nil {
499 // The result is in hand and the ack is a courtesy: Ctrl-C during
500 // it ends the ack, not the exit code, and cancels nothing.
501 watch.run(ctx, func(ctx context.Context) { ackRun(ctx, c, t, id, p.Result.LastSeq) })
502 code, lines := runExit(host, *p.Result)
503 switch {
504 case p.Result.Cancelled && p.Result.Started.IsZero():
505 // Cancelled before any box launched it: there was no command.
506 lines = []string{fmt.Sprintf("%s: run %s was cancelled before it started", host, id)}
507 case p.Result.TimedOut && p.Result.Started.IsZero():
508 // Its deadline passed before any box launched it: there was
509 // no command, and so no duration to report (re-review 8).
510 lines = []string{fmt.Sprintf("%s: run %s timed out before it started", host, id)}
511 }
512 for _, l := range lines {
513 t.say(l)
514 }
515 return code
516 }
517 if p.State != "" {
518 letGoSince = time.Time{}
519 } else {
520 if letGoSince.IsZero() {
521 letGoSince = t.now()
522 }
523 if t.now().Sub(letGoSince) >= runPageGiveUp {
524 t.say(runLetGo(host, id))
525 return 125
526 }
527 }
528 switch p.State {
529 case provision.RunExpired:
530 t.say(runNotPickedUp(host))
531 return 3
532 case provision.RunCancelled:
533 t.say(fmt.Sprintf("%s: run %s was cancelled before it started", host, id))
534 return 130
535 case provision.RunQueued:
536 if !expireAt.IsZero() && !t.now().Before(expireAt) {
537 var state string
538 var err error
539 watch.run(ctx, func(ctx context.Context) { state, err = c.expire(ctx, id) })
540 if watch.stopped() {
541 return stopWaiting(ctx, t, c, id, &cancel)
542 }
543 var ce *runConsoleError
544 switch {
545 case err == nil && state == provision.RunExpired:
546 // The run's uploaded files go now, not with the sweep.
547 // The ack is a courtesy: nothing hangs on it.
548 watch.run(ctx, func(ctx context.Context) { _ = c.ack(ctx, id, 0) })
549 t.say(runNotPickedUp(host))
550 return 3
551 case err == nil || errors.As(err, &ce):
552 // A box picked the run first (409), or the console will
553 // not expire it: follow it. A call that did not arrive is
554 // asked again on the next poll.
555 expireAt = time.Time{}
556 }
557 }
558 }
559 if interrupted(ctx, t, c, id, &cancel, time.Second) {
560 return 130
561 }
562 }
563}
564
565// runWatch is the follow loop's view of Ctrl-C. Between requests the loop
566// takes a Ctrl-C itself; while a request is in flight run takes it, so the
567// second one can end that request at once (run spec §5.7). n counts every
568// Ctrl-C taken, and handled counts those the loop has acted on.
569type runWatch struct {
570 ch <-chan os.Signal
571 n, handled int
572}
573
574// run calls f with a context that the second Ctrl-C cancels, and reports
575// whether it did. Ctrl-Cs that arrive meanwhile are counted for the loop. If
576// the second has already been taken, the context starts cancelled: no request
577// gets going after the user said stop.
578func (w *runWatch) run(ctx context.Context, f func(context.Context)) bool {
579 return w.do(ctx, 0, f)
580}
581
582// do is run with a deadline when limit is set. A request made after the
583// second Ctrl-C is bounded this way instead of started cancelled: it is still
584// attempted, and a further Ctrl-C ends it.
585func (w *runWatch) do(ctx context.Context, limit time.Duration, f func(context.Context)) (aborted bool) {
586 rctx, cancel := context.WithCancel(ctx)
587 if limit > 0 {
588 var stopTimer context.CancelFunc
589 rctx, stopTimer = context.WithTimeout(rctx, limit)
590 defer stopTimer()
591 } else if w.n >= 2 {
592 aborted = true
593 cancel()
594 }
595 stop, fin := make(chan struct{}), make(chan struct{})
596 go func() {
597 defer close(fin)
598 for {
599 select {
600 case <-w.ch:
601 w.n++
602 if w.n >= 2 && !aborted {
603 aborted = true
604 cancel()
605 }
606 case <-stop:
607 return
608 }
609 }
610 }()
611 f(rctx)
612 close(stop)
613 <-fin
614 cancel()
615 return aborted
616}
617
618// stopped reports whether the second Ctrl-C has been taken.
619func (w *runWatch) stopped() bool { return w.n >= 2 }
620
621// take claims a Ctrl-C that is waiting, taken by run or still in the channel.
622func (w *runWatch) take() bool {
623 if w.n > w.handled {
624 w.handled++
625 return true
626 }
627 select {
628 case <-w.ch:
629 w.n++
630 w.handled++
631 return true
632 default:
633 return false
634 }
635}
636
637// runLateCancelWait is how long a cancel sent after the second Ctrl-C may
638// take. A variable so a test can shorten it.
639var runLateCancelWait = 3 * time.Second
640
641// runCancel is the state of the cancel a Ctrl-C asks for.
642type runCancel struct {
643 w *runWatch
644 asked, done, said bool
645 abandoned bool // the second Ctrl-C cut the request off: do not send it again
646}
647
648// send makes the cancel, if one was asked for and has not been answered. A
649// cancel retried after a lost 200 finds the run already ended, which is what
650// was asked for. One that did not arrive is sent again on the next call, so
651// the cancel still happens (run spec §5.7); it is said once. A cancel the
652// second Ctrl-C cut off is not sent again: that Ctrl-C said stop waiting.
653func (k *runCancel) send(ctx context.Context, t runTerm, c runClient, id string) {
654 if !k.asked || k.done || k.abandoned {
655 return
656 }
657 var err error
658 var aborted bool
659 // A cancel sent after the second Ctrl-C is still attempted, but only for
660 // a moment: the user has said stop waiting.
661 late := k.w.stopped()
662 if late {
663 aborted = k.w.do(ctx, runLateCancelWait, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
664 } else {
665 aborted = k.w.run(ctx, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
666 }
667 if (aborted && errors.Is(err, context.Canceled)) || (late && errors.Is(err, context.DeadlineExceeded)) {
668 k.abandoned = true
669 return
670 }
671 var ce *runConsoleError
672 switch {
673 case err == nil, runConflict(err, fmt.Sprintf("run %s has already ended", id)):
674 k.done = true
675 case errors.As(err, &ce):
676 k.done = true
677 t.say(fmt.Sprintf("carlos run: %s", err))
678 case !k.said:
679 k.said = true
680 t.say(fmt.Sprintf("carlos run: %s", err))
681 }
682}
683
684// press acts on one Ctrl-C that has been taken. The first asks for the cancel
685// and returns false so following goes on; the second tries the cancel once
686// more if it is unanswered and returns true: stop waiting.
687func (k *runCancel) press(ctx context.Context, t runTerm, c runClient, id string) bool {
688 if k.asked {
689 k.send(ctx, t, c, id)
690 return true
691 }
692 k.asked = true
693 t.say(fmt.Sprintf("cancelling run %s. press Ctrl-C again to stop waiting", id))
694 k.send(ctx, t, c, id)
695 return false
696}
697
698// stopWaiting is the exit once the second Ctrl-C has been taken, possibly
699// while a request was in flight and the first not yet acted on.
700func stopWaiting(ctx context.Context, t runTerm, c runClient, id string, k *runCancel) int {
701 for k.w.take() {
702 if k.press(ctx, t, c, id) {
703 break
704 }
705 }
706 return 130
707}
708
709// interrupted waits out d, the pause before the next request, and handles a
710// Ctrl-C that arrives first or was already waiting: a pending Ctrl-C is
711// taken before the next tick, never after it. The first asks for the cancel
712// and returns false so following goes on; the second stops waiting and
713// returns true. A cancel still unanswered when the wait ends is sent again.
714func interrupted(ctx context.Context, t runTerm, c runClient, id string, k *runCancel, d time.Duration) bool {
715 if !k.w.take() {
716 select {
717 case <-t.after(d):
718 k.send(ctx, t, c, id)
719 // A second Ctrl-C during the resend ends the wait here, not
720 // after another request.
721 return k.w.stopped()
722 case <-k.w.ch:
723 k.w.n++
724 k.w.handled++
725 }
726 }
727 if k.press(ctx, t, c, id) {
728 return true
729 }
730 // A second Ctrl-C already waiting is not left for the next poll.
731 if k.w.take() {
732 k.press(ctx, t, c, id)
733 return true
734 }
735 return false
736}
737
738// runExit is a result's exit code and what stderr says about it (run spec
739// §5.6, §8; index item 6).
740func runExit(host string, r runs.Result) (int, []string) {
741 var lines []string
742 if r.Truncated {
743 lines = append(lines, fmt.Sprintf("%s: the output passed 64 MB and was cut off there. the command ran on to its end", host))
744 }
745 switch {
746 case r.Refused == runs.RefusedExpired:
747 return 3, append(lines, runNotPickedUp(host))
748 case r.Refused == runs.RefusedFlat:
749 return 3, append(lines, fmt.Sprintf("%s: the box is not on the nested layout yet, so carlos run cannot run there", host))
750 case r.Refused == runs.RefusedClaim:
751 return 3, append(lines, fmt.Sprintf("%s: the box has host claims armed. carlos run cannot run there yet", host))
752 case r.Refused == runs.RefusedWake:
753 return 3, append(lines, fmt.Sprintf("%s: the instance did not wake. %s", host, r.Detail))
754 case r.Refused == runs.RefusedIncarnation:
755 return 3, append(lines, fmt.Sprintf("%s was deleted and created again after this run was requested. the run did not start", host))
756 case r.Refused == runs.RefusedImport:
757 return 3, append(lines, fmt.Sprintf("%s is an import that is not committed yet. the run did not start", host))
758 case r.Refused != "":
759 return 3, append(lines, fmt.Sprintf("carlos run: %s", r.Refused))
760 case r.Cancelled:
761 return 130, append(lines, fmt.Sprintf("%s: the command was cancelled", host))
762 case r.TimedOut:
763 return 124, append(lines, fmt.Sprintf("%s: the command timed out after %s", host, r.Ended.Sub(r.Started).Round(time.Second)))
764 case r.Interrupted != "":
765 return 125, append(lines, fmt.Sprintf("%s: the run was interrupted: %s. it may have done part of its work", host, r.Interrupted))
766 case r.Signal != "":
767 n, _ := strconv.Atoi(r.Signal)
768 return 128 + n, lines
769 }
770 return r.Exit, lines
771}
772
773// startRun sends the run and follows it. A console refusal, or a --file that
774// cannot be read, is exit 2.
775func startRun(ctx context.Context, c runClient, t runTerm, o runOpts, app string) int {
776 meta, err := runMetaFor(o)
777 if err != nil {
778 t.say(err.Error())
779 return 2
780 }
781 created, err := c.create(ctx, meta, o.files)
782 if err != nil {
783 var ce *runConsoleError
784 switch {
785 case errors.Is(err, errRunLocalFile):
786 t.say(err.Error())
787 return 2
788 case errors.As(err, &ce):
789 t.say(fmt.Sprintf("carlos run: %s", err))
790 return 2
791 }
792 t.say(fmt.Sprintf("carlos run: %s", err))
793 return 1
794 }
795 if created.Replayed {
796 // An earlier attempt with this key started the run: this one starts
797 // nothing, and follows it.
798 if created.State == "" && runGone(ctx, c, created.ID) {
799 // The record no longer names the run and its objects are gone: it
800 // was acknowledged or swept. The key proves it started.
801 t.say(runReplayGone(created.ID))
802 return 2
803 }
804 t.say(fmt.Sprintf("already started: run %s on %s. to follow it again: carlos run --app %s --instance %s --follow %s", created.ID, o.instance, app, o.instance, created.ID))
805 } else {
806 t.say(fmt.Sprintf("run %s on %s. to follow it again: carlos run --app %s --instance %s --follow %s", created.ID, o.instance, app, o.instance, created.ID))
807 }
808 return followRun(ctx, c, t, o.instance, created.ID, t.now().Add(runs.PickupWindow))
809}
810
811// runReplayGone is what a retry says when its key started a run that is no
812// longer there to follow.
813func runReplayGone(id string) string {
814 return fmt.Sprintf("run %s was already started with this key, and its output is no longer kept", id)
815}
816
817// runGone reports whether the console says it has no run id (a 404 on the
818// first page). Any other answer, an error included, is not "gone": the
819// follow that comes next meets it again and deals with it.
820func runGone(ctx context.Context, c runClient, id string) bool {
821 _, err := c.page(ctx, id, 0)
822 var ce *runConsoleError
823 return errors.As(err, &ce) && ce.status == http.StatusNotFound
824}
825
826// runRun is `carlos run` (run spec §3).
827func runRun(args []string) {
828 o, err := parseRunArgs(args, os.Stderr)
829 if err != nil {
830 if errors.Is(err, flag.ErrHelp) {
831 os.Exit(0)
832 }
833 if !errors.Is(err, errRunUsage) {
834 fmt.Fprintln(os.Stderr, err.Error())
835 }
836 fmt.Fprint(os.Stderr, runUsageText())
837 os.Exit(2)
838 }
839 sess, err := resolveMode(o.console)
840 if err != nil {
841 log.Fatal(err)
842 }
843 if sess == nil {
844 log.Fatal(runDirectModeMsg)
845 }
846 printUsing(os.Stderr, sess)
847 acct := resolvedAccount(sess, o.acctRaw, o.acctIDRaw)
848 app := resolvedApp(sess, acct, o.app, runUsageText())
849 interrupts := make(chan os.Signal, 2)
850 signal.Notify(interrupts, os.Interrupt, syscall.SIGTERM)
851 t := runTerm{stdout: os.Stdout, stderr: os.Stderr, interrupts: interrupts, after: time.After, now: time.Now}
852 c := runAPI{s: sess, a: acct, app: app, host: o.instance, key: o.idempotencyKey}
853 ctx := context.Background()
854 if o.follow != "" {
855 os.Exit(followRun(ctx, c, t, o.instance, o.follow, time.Time{}))
856 }
857 os.Exit(startRun(ctx, c, t, o, app))
858}
859