package main import ( "context" "encoding/json" "errors" "flag" "fmt" "io" "os" "regexp" "sort" "strings" "time" "github.com/nats-io/nats.go" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) // The build queue, controlled by hand (novox/hq ADR 0219). // // Seen whole — what waits, what runs where and for how long, what was handed out as often as it // may be and never settled — and changed through the controller: an ask cancelled, the queue // cleared, a module rebuilt, a build replayed. What one machine is running is that machine's // holder's to end or pause, and the controller asks it (`kill`, `pause`, `resume`). // // **Everything that drops an ask leaves a failed outcome for it**, taken in exactly as a build that // failed is (takeIn, then the plan): a plan waiting on an ask a person removed fails, saying so, // rather than waiting for ever on an answer nobody will give. // holderAsks is how long a holder's verb is waited for; killAnswer how long its kill is — longer // than the holder's worst case (its two passes removing containers and its wait between, 50s). const ( holderAsks = 15 * time.Second killAnswer = 75 * time.Second ) // dialTheBus is the controller's connection, for a command that reads or changes the queue: the serving // controller's own, lent, when this process is it (novox/hq issue 327). func dialTheBus() (*broker.JetStream, error) { return aBus() } // queueCommand prints every ask in the build seat's work queue. func queueCommand(ctx context.Context, args []string) error { return listQueue(ctx, args, os.Stdout) } // listQueue is `queue`: the build queue, as a person reads it or as JSON. func listQueue(ctx context.Context, args []string, w io.Writer) error { set := flag.NewFlagSet("queue", flag.ContinueOnError) usageTo(set, w) asJSON := set.Bool("json", false, "the queue as JSON") if _, err := parseAround(set, args); err != nil { return err } js, err := dialTheBus() if err != nil { return err } defer js.Close() seat := buildSeatHeld(ctx) q, err := link.ReadQueue(ctx, js, seat) if err != nil { return err } now := time.Now() if *asJSON { body, err := json.MarshalIndent(queueAnswerOf(q, now), "", " ") if err != nil { return err } fmt.Fprintln(w, string(body)) return nil } fmt.Fprint(w, queueText(q, now)) return nil } // keepingTheQueueInOrder runs the build queue's passes while this controller acts (novox/hq issue 558): // the one process that puts the queue in order, so no two passes pause and resume it over each other. // The build seat is read from the catalogue at most every half minute. func keepingTheQueueInOrder(ctx context.Context, open *stores, js *broker.JetStream) { if js == nil { return } seat, readAt := link.TheBuildMachine, time.Time{} seatNow := func() string { if time.Since(readAt) < watchEvery { return seat } if entries, err := open.inventory.Catalogued(ctx); err == nil { seat, readAt = buildSeatAmong(entries), time.Now() } else { fmt.Printf("the build seat is read as %s: the catalogue cannot be read: %v\n", seat, err) } return seat } link.KeepingInOrder(ctx, js, seatNow, link.Holding, func(format string, args ...any) { fmt.Printf(format+"\n", args...) }) } // queueAnswer is the queue as data (novox/hq issue 455): what a caller reads to say how far back an ask // waits — "waiting in the build queue behind 2" — which the table alone could not tell it. Text is the // table, written from the same read: the verb answers it as its output and the rest as its answer, so the // two never disagree. type queueAnswer struct { Seat string `json:"seat"` Waiting int `json:"waiting"` InFlight int `json:"in-flight"` Dead int `json:"dead"` MaxDeliver int `json:"max-deliver"` Asks []queuedAnswer `json:"asks"` Text string `json:"text"` } // queuedAnswer is one ask as data. Never what the request carries beyond its source — not `held`, every // artifact the mesh has built, nor a check's checkouts beside it. type queuedAnswer struct { ID string `json:"id"` State string `json:"state"` // Kind is "delivery" for a build a walk waits for, "check" for a pull request's check and any ask // whose kind cannot be told (link.KindOf, novox/hq issue 558). Kind string `json:"kind"` Repository string `json:"repository"` Path string `json:"path,omitempty"` Ref string `json:"ref,omitempty"` // Commit is the ref when it names a commit; empty when the ask follows a branch or the default. Commit string `json:"commit,omitempty"` // Machine is the one building an ask in flight; WasOn the one that started it and handed it back. Machine string `json:"machine,omitempty"` WasOn string `json:"was-on,omitempty"` AskedAt time.Time `json:"asked-at,omitzero"` // Since is when the ask came to its state: its start on the machine for one in flight that a // machine said it started, otherwise when it was asked. Since time.Time `json:"since,omitzero"` // Position is its place among the waiting, 1 being handed out next; 0 for an ask not waiting. Position int `json:"position,omitempty"` Seq uint64 `json:"seq"` // Delivery is the how-manyth time an ask in flight was handed out; more than 1 is a restart // (novox/hq issue 594). Delivery int `json:"delivery,omitempty"` } // aCommitRef is a ref that names a commit by its full hash (SHA-1 or SHA-256). var aCommitRef = regexp.MustCompile(`^(?:[0-9a-f]{40}|[0-9a-f]{64})$`) // queueAnswerOf is the queue as data, in the table's order: in flight, the waiting from the next one to // be handed out, then the dead. A pure function of one read (novox/hq issue 455). func queueAnswerOf(q link.Queue, now time.Time) queueAnswer { waiting, running, dead := q.Of(link.AskWaiting), q.Of(link.AskInFlight), q.Of(link.AskDead) out := queueAnswer{Seat: q.Seat, Waiting: len(waiting), InFlight: len(running), Dead: len(dead), MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, Text: queueText(q, now)} one := func(a link.QueuedAsk, position int) queuedAnswer { kind := link.KindOf(a.Request) since := a.AskedAt if a.State == link.AskInFlight && a.On != "" && !a.Started.IsZero() { since = a.Started } commit := "" if aCommitRef.MatchString(a.Ref) { commit = a.Ref } return queuedAnswer{ID: a.ID, State: a.State, Kind: kind, Repository: link.WithoutPassword(a.Repository), Path: a.Path, Ref: a.Ref, Commit: commit, Machine: a.On, WasOn: a.Was, AskedAt: a.AskedAt, Since: since, Position: position, Seq: a.Seq, Delivery: a.Delivery} } for _, a := range running { out.Asks = append(out.Asks, one(a, 0)) } for i, a := range waiting { out.Asks = append(out.Asks, one(a, i+1)) } for _, a := range dead { out.Asks = append(out.Asks, one(a, 0)) } return out } // queueText is the queue as a person reads it: a summary line, then each kind in queue order. // Never what an ask carries as `held` — that is every artifact the mesh has built. func queueText(q link.Queue, now time.Time) string { var b strings.Builder waiting, running, dead := q.Of(link.AskWaiting), q.Of(link.AskInFlight), q.Of(link.AskDead) fmt.Fprintf(&b, "%s: %d waiting, %d in flight, %d dead\n", q.Seat, len(waiting), len(running), len(dead)) ago := func(t time.Time) string { if t.IsZero() { return "asked at an unknown time" } return "asked " + now.Sub(t).Round(time.Second).String() + " ago" } what := func(a link.QueuedAsk) string { s := link.WithoutPassword(a.Repository) if a.Path != "" { s += " at " + a.Path } if a.Ref != "" { s += " on " + a.Ref } return s } if len(running) > 0 { fmt.Fprintln(&b, "\nin flight:") for _, a := range running { where := "taken, not yet said where" switch { case a.On != "": where = fmt.Sprintf("on %s for %s", a.On, now.Sub(a.Started).Round(time.Second)) // Started again after an earlier taker left it (novox/hq issue 594). if a.Delivery > 1 { where += fmt.Sprintf(", handed out %d times", a.Delivery) } case a.Was != "": where = fmt.Sprintf("handed back by %s, to be handed out again", a.Was) } fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), where, ago(a.AskedAt), a.Seq) } } if len(waiting) > 0 { fmt.Fprintln(&b, "\nwaiting, in the order they are handed out:") for _, a := range waiting { fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq) } } if len(dead) > 0 { fmt.Fprintf(&b, "\ndead — handed out as often as the worker allows (%d) and never settled, still held:\n", q.MaxDeliver) for _, a := range dead { fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq) } } switch { case len(q.Asks) == 0: fmt.Fprintln(&b, "nothing is asked of it") default: fmt.Fprintln(&b, "\n`cancel ` drops a waiting or dead ask, `clear` every waiting one, `kill ` ends one in flight") } return b.String() } // cancelCommand drops one waiting or dead ask. func cancelCommand(ctx context.Context, args []string) error { if len(args) != 1 { return errors.New("cancel ") } js, err := dialTheBus() if err != nil { return err } defer js.Close() open, err := openStores(ctx) if err != nil { return err } defer open.Close() seat := buildSeatHeld(ctx) q, err := link.ReadQueue(ctx, js, seat) if err != nil { return err } ask, found := q.Find(args[0]) if !found { return fmt.Errorf("%s is not in the %s queue: it was built, cancelled, or never asked — `builds` says "+ "what came of it", args[0], seat) } said, err := cancelAsk(ctx, js, open, seat, ask) if err != nil { return err } fmt.Println(said) return nil } // cancelAsk drops one ask from the queue and records it failed, cancelled by hand. // // **In this order, and each step for a reason** (novox/hq ADR 0219): // 1. an ask a machine says it is building is refused: deleting its message ends nothing a machine // is running — that is `kill`, on the machine running it; // 2. its id goes into the seat's cancelled set, so a holder that fetches it from now on ends it; // 3. for a waiting ask, the worker is read again: one handed out since the queue was read was // taken in the moment of cancelling, and the cancel is withdrawn and refused — the holder either // read the mark first and ends it as cancelled, or is building it; // 4. for an ask the worker counts handed out that no machine is building — taken and not yet said, // handed back after a restart, or past its deliveries while nobody pulls — the holder's own look // at the set is waited out, and a start heard since withdraws and refuses the cancel: a holder // that took it before the mark is building it, and only `kill` ends that; // 5. the message is deleted by its sequence; // 6. the failed outcome is taken in as any failed build's is, and the plan that asked fails. func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat string, ask link.QueuedAsk) (string, error) { if ask.State == link.AskInFlight && ask.On != "" { return "", fmt.Errorf("%s is in flight on %s: cancelling drops an ask nobody is building. `kill %s` "+ "ends the build where it runs", ask.ID, ask.On, ask.ID) } if err := link.MarkCancelled(ctx, js, seat, ask.ID); err != nil { return "", err } withdraw := func() { if err := link.UnmarkCancelled(ctx, js, seat, ask.ID); err != nil { fmt.Fprintf(os.Stderr, "could not withdraw the cancel of %s: %v — a holder taking it ends it as cancelled\n", ask.ID, err) } } worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}}) // **Where it waits now** (novox/hq issue 558): moved into order since it was read, a waiting ask has a // later sequence, and the worker handing out what was behind its old one is not it being taken. if ask.State == link.AskWaiting { if q, err := link.ReadQueue(ctx, js, seat); err == nil { if now, ok := q.Find(ask.ID); ok && now.State == link.AskWaiting { ask.Seq = now.Seq } } } switch ask.State { case link.AskWaiting: if taken, err := takenSince(js, worker, ask.Seq); err != nil { withdraw() return "", err } else if taken { withdraw() return "", fmt.Errorf("%s was taken by a holder as it was cancelled, so the cancel is withdrawn. If the "+ "holder read it first it ends the ask as %s and its outcome says so; otherwise it is building — "+ "`queue` says which, and `kill %s` ends it", ask.ID, link.CancelledByHand, ask.ID) } case link.AskInFlight: if started, err := startedSince(ctx, js, seat, ask); err != nil { withdraw() return "", err } else if started != "" { withdraw() return "", fmt.Errorf("%s has started on %s since it was read, so the cancel is withdrawn: `kill %s` "+ "ends it there", ask.ID, started, ask.ID) } } // Found by its id when it was moved into order since it was read (novox/hq issue 558). removed, err := link.DeleteAsk(ctx, js, seat, ask) if err != nil { return "", fmt.Errorf("%s is marked cancelled and could not be deleted from the queue (seq %d): %w — a "+ "holder taking it ends it as cancelled", ask.ID, ask.Seq, err) } recordCancelled(ctx, open, ask) return cancelSaid(ask, seat, removed), nil } // cancelSaid is what a cancel says: deleted from the queue only when it was. func cancelSaid(ask link.QueuedAsk, seat string, removed uint64) string { where := fmt.Sprintf("deleted from the %s queue", seat) if removed == 0 { where = fmt.Sprintf("no longer in the %s queue to delete, and marked cancelled, so a holder taking it "+ "ends it", seat) } return fmt.Sprintf("cancelled %s (%s, %s): %s and recorded failed, %s — a plan "+ "that asked for it fails with that", ask.ID, ask.Repository, ask.State, where, link.CancelledByHand) } // holderLooks is how long a cancel waits for a holder that took the ask before the mark to say it // started: longer than a holder's look at the cancelled set (3s) and its start that follows. var holderLooks = 5 * time.Second // startedSince waits out a holder's look at the cancelled set and says the machine that started the // ask since it was read, or nothing. A start it ended as cancelled comes with its outcome and is not // a start of a build. func startedSince(ctx context.Context, js *broker.JetStream, seat string, ask link.QueuedAsk) (string, error) { select { case <-ctx.Done(): return "", ctx.Err() case <-time.After(holderLooks): } since := time.Now().Add(-7 * 24 * time.Hour) if !ask.AskedAt.IsZero() { since = ask.AskedAt.Add(-time.Minute) } heard, err := link.ReadBuildEvents(ctx, js, seat, since) if err != nil { return "", err } s, ok := heard.Started[ask.ID] if !ok || heard.Outcomes[ask.ID] { return "", nil } at, _ := time.Parse(time.RFC3339Nano, s.At) if ask.Started.IsZero() || at.After(ask.Started) { return s.On, nil } return "", nil } // takenSince is whether the worker has handed out the ask at this sequence. func takenSince(js *broker.JetStream, worker broker.Consumer, seq uint64) (bool, error) { info, err := js.Context().ConsumerInfo(worker.Stream, worker.Name) if errors.Is(err, nats.ErrConsumerNotFound) { return false, nil } if err != nil { return false, fmt.Errorf("cannot read the worker of the queue again: %w", err) } return info.Delivered.Stream >= seq, nil } // recordCancelled takes the failed outcome in the way the daemon takes in any build's: recorded, // then the plan that asked for it — by the id it asked with, or by repository and path. func recordCancelled(ctx context.Context, open *stores, ask link.QueuedAsk) { r := ask.Request result := link.BuildResult{ID: ask.ID, Repository: ask.Repository, Path: ask.Path, Ref: ask.Ref, Source: r.Source, DryRun: r.DryRun, Failed: link.CancelledByHand} _ = builds{open.inventory, open}.Built(ctx, result) } // clearCommand cancels every waiting ask, and with --dead every dead one too. func clearCommand(ctx context.Context, args []string) error { set := flag.NewFlagSet("clear", flag.ContinueOnError) dead := set.Bool("dead", false, "the dead asks too") if _, err := parseAround(set, args); err != nil { return err } js, err := dialTheBus() if err != nil { return err } defer js.Close() open, err := openStores(ctx) if err != nil { return err } defer open.Close() seat := buildSeatHeld(ctx) said, err := clearQueue(ctx, js, open, seat, *dead) fmt.Print(said) return err } // clearQueue cancels what clear names, each as `cancel` would, and never anything in flight. func clearQueue(ctx context.Context, js *broker.JetStream, open *stores, seat string, dead bool) (string, error) { q, err := link.ReadQueue(ctx, js, seat) if err != nil { return "", err } var b strings.Builder cancelled, refused := 0, 0 for _, a := range q.Asks { if a.State == link.AskInFlight || (a.State == link.AskDead && !dead) { continue } said, err := cancelAsk(ctx, js, open, seat, a) if err != nil { refused++ fmt.Fprintf(&b, " %s: %v\n", a.ID, err) continue } cancelled++ fmt.Fprintf(&b, " %s\n", said) } what := "waiting" if dead { what = "waiting and dead" } fmt.Fprintf(&b, "%d %s ask(s) cancelled", cancelled, what) if refused > 0 { fmt.Fprintf(&b, ", %d could not be", refused) } fmt.Fprintln(&b) if n := len(q.Of(link.AskInFlight)); n > 0 { fmt.Fprintf(&b, "%d in flight, left running: `kill ` ends one where it runs\n", n) } if n := len(q.Of(link.AskDead)); n > 0 && !dead { fmt.Fprintf(&b, "%d dead, left: `clear --dead` cancels them too\n", n) } if refused > 0 { return b.String(), fmt.Errorf("%d ask(s) could not be cancelled", refused) } return b.String(), nil } // rebuildCommand asks the module's current source again — or, given a build's id, that build's // repository, path and ref — under a new id. func rebuildCommand(ctx context.Context, args []string) error { if len(args) != 1 { return errors.New("rebuild ") } open, err := openStores(ctx) if err != nil { return err } defer open.Close() inv := open.inventory entries, err := inv.Catalogued(ctx) if err != nil { return err } var source buildSource var path, ref, module string if b, found, err := inv.BuildByID(ctx, args[0]); err != nil { return err } else if found { source, module = sourceOfBuild(b, entries) path, ref = b.Path, b.Ref // **A build at a commit is rebuilt at what its module follows now** (novox/hq ADR 0219, issue // 219): asked now, a rebuild of an old commit would be the newest ask of the module and roll // that commit out over everything since. Building that commit again is `replay`, which says // what it would do and refuses to register it while anything newer is asked or registered. if e, known := entryNamed(entries, module); known && ref != "" && followedBranch(ref) == "" { ref = followedBranch(e.Source.Ref) follows := ref if follows == "" { follows = "the repository's default branch" } fmt.Printf("%s was built at commit %s; a rebuild asks what %s follows now, %s — `replay %s` "+ "builds that commit\n", b.ID, short(b.Ref), module, follows, b.ID) } } else if e, known := entryNamed(entries, args[0]); known { // As a plan asks it: the branch it follows, never a commit a build once named (issue 215). source = buildSource{Repository: e.Source.Repository, Seat: e.Source.Seat} path, ref, module = e.Source.Path, followedBranch(e.Source.Ref), e.Manifest.Module } else { return fmt.Errorf("%s is neither a module the catalogue holds nor a build the mesh recorded", args[0]) } // **A module a plan holds unbuilt, or failed, joins that plan** rather than running beside it: the // plan would ask it again, or stay failed on an outcome a rebuild has replaced. if module != "" { joined, said, err := joinAPlan(ctx, open, module) if err != nil { return err } if joined { fmt.Println(said) return nil } } id, err := askABuild(ctx, source, path, ref) if err != nil { return err } recordAsked(ctx, inventory.BuildRequest{ID: id, Repository: source.Repository, Seat: source.Seat, Path: path, Ref: ref, For: "rebuild"}) fmt.Printf("rebuild asked as %s\n", id) return nil } // entryNamed is the catalogue's entry for a module. func entryNamed(entries []inventory.Entry, name string) (inventory.Entry, bool) { for _, e := range entries { if e.Manifest.Module == name { return e, true } } return inventory.Entry{}, false } // sourceOfBuild is where a recorded build's repository is asked from: the catalogued module's own // source when the build is of one — on its seat, so the outcome registers it as the mesh records it // (ADR 0111) — else the repository as it was cloned. func sourceOfBuild(b inventory.Build, entries []inventory.Entry) (buildSource, string) { for _, e := range entries { if (b.Module != "" && e.Manifest.Module == b.Module) || (b.Module == "" && repositoryMatches(e.Source.Repository, b.Repository) && e.Source.Path == b.Path) { return buildSource{Repository: e.Source.Repository, Seat: e.Source.Seat}, e.Manifest.Module } } return buildSource{Repository: b.Repository}, b.Module } // replayCommand asks a recorded build's repository and path again at the commit it built. func replayCommand(ctx context.Context, args []string) error { set := flag.NewFlagSet("replay", flag.ContinueOnError) register := set.Bool("register", false, "register what it builds, as any build is") older := set.Bool("older", false, "with --register: even though a newer build of the module is registered") positionals, err := parseAround(set, args) if err != nil { return err } if len(positionals) != 1 { return errors.New("replay [--register [--older]]") } open, err := openStores(ctx) if err != nil { return err } defer open.Close() inv := open.inventory b, found, err := inv.BuildByID(ctx, positionals[0]) if err != nil { return err } if !found { return fmt.Errorf("no build %s is recorded", positionals[0]) } entries, err := inv.Catalogued(ctx) if err != nil { return err } source, module := sourceOfBuild(b, entries) var history []inventory.Build if module != "" { if history, err = inv.Builds(ctx, module, 100); err != nil { return err } } // What is asked of the module and not yet answered, when the replay would be registered. var outstanding []string if *register { js, err := dialTheBus() if err != nil { return err } q, err := link.ReadQueue(ctx, js, buildSeatHeld(ctx)) js.Close() if err != nil { return err } plans, err := inv.OpenPlans(ctx) if err != nil { return err } outstanding = outstandingFor(module, b.Repository, b.Path, q, plans) } if err := replayRefusal(b, module, history, *register, *older, outstanding); err != nil { return err } asker := "" if *register { asker = "replay" } id, err := buildOneAsked(ctx, source, b.Path, b.Commit, 0, !*register, asker) if err != nil { return err } fmt.Println(replaySaid(b, module, id, *register)) return nil } // replayRefusal is why a replay is not asked, or nothing (novox/hq ADR 0219, issue 207). // // A replay re-asks a recorded build at the commit it built, under a new id — and an id is when it // was asked, which is what orders builds of one module (issue 219). So a registered replay of an // older commit is newer than everything since, and would roll that older commit out as the module's // current version: what issue 207 recorded happening by accident. It is therefore a dry run unless // --register says otherwise, and --register is refused when a newer build of a different commit is // registered, unless --older says that is the point. // // **And refused while anything newer is outstanding**, --older or not: an ask of the module in the // queue, or an open plan holding it unbuilt. Asked now, the replay would be newer than those asks, // and their builds — made from newer source — would be recorded and never registered (issue 219 // orders by ask), the plan waiting on them sending the older commit instead. func replayRefusal(b inventory.Build, module string, history []inventory.Build, register, older bool, outstanding []string) error { if b.Commit == "" { return fmt.Errorf("%s recorded no commit — it failed before it knew what it was building, so there is "+ "nothing to replay; `rebuild %s` asks its repository, path and ref again", b.ID, b.ID) } if older && !register { return errors.New("--older only says what --register may do; a dry run registers nothing") } if register && len(outstanding) > 0 { return fmt.Errorf("%s is asked and not yet answered — %s — and a registered replay asked now would "+ "replace what those build (novox/hq issue 219). Wait for them, or `replay %s` without --register to look", orNone(module), strings.Join(outstanding, "; "), b.ID) } if !register || older { return nil } for _, h := range history { if h.ID == b.ID || !h.Worked() || h.Commit == b.Commit { continue } if h.AskedOrAt().After(b.AskedOrAt()) { return fmt.Errorf("a newer build of %s is registered — %s, from %s — and registering a replay of %s "+ "would roll that older commit out as %s's current version (novox/hq issue 207). `replay %s` "+ "without --register looks at it; --register --older registers it anyway", module, h.ID, short(h.Commit), short(b.Commit), module, b.ID) } } return nil } // replaySaid is what a replay prints once asked: what it builds, and what becomes of the outcome. func replaySaid(b inventory.Build, module, id string, register bool) string { what := b.Repository if module != "" { what = module } said := fmt.Sprintf("replaying %s (%s) at %s as %s", b.ID, what, short(b.Commit), id) if !register { return said + "\n a dry run: the outcome is looked at and not taken in — nothing is recorded or " + "registered, and nothing is sent. `builds --log " + id + "` follows it; `--register` registers it" } return said + "\n registered when it is built, as the module's current version — newer than every build " + "asked before now — and rolled out as its policy says. `builds --log " + id + "` follows it" } // killCommand finds the machine running a build and asks its holder to end it. func killCommand(ctx context.Context, args []string) error { if len(args) != 1 { return errors.New("kill ") } id := args[0] js, err := dialTheBus() if err != nil { return err } defer js.Close() seat := buildSeatHeld(ctx) since := time.Now().Add(-7 * 24 * time.Hour) if at, ok := link.BuildAskedAt(id); ok { since = at.Add(-time.Minute) } heard, err := link.ReadBuildEvents(ctx, js, seat, since) if err != nil { return err } started, ok := heard.Started[id] if !ok { return fmt.Errorf("no machine has said it started %s: if it waits in the queue, `cancel %s` drops it", id, id) } if heard.Outcomes[id] { return fmt.Errorf("%s has already ended on %s — `builds` says how", id, started.On) } answer, err := link.AskSeatTool(ctx, js.Conn(), seat, "kill", started.On, map[string]string{"id": id}, killAnswer) if err != nil { return err } return printHolderAnswer(started.On, answer) } // pauseCommand asks one machine's holder, or every holder's, to pause or resume. func pauseCommand(ctx context.Context, verb string, args []string) error { if len(args) > 1 { return fmt.Errorf("%s [node]", verb) } js, err := dialTheBus() if err != nil { return err } defer js.Close() seat := buildSeatHeld(ctx) nodes := args if len(nodes) == 0 { if nodes, err = buildSeatHolders(ctx, seat); err != nil { return err } if len(nodes) == 0 { return fmt.Errorf("nothing holds %s, so there is nothing to %s", seat, verb) } } failed := 0 for _, node := range nodes { answer, err := link.AskSeatTool(ctx, js.Conn(), seat, verb, node, map[string]string{}, holderAsks) if err != nil { fmt.Printf("%s: %v\n", node, err) failed++ continue } if err := printHolderAnswer(node, answer); err != nil { failed++ } } if failed > 0 { return fmt.Errorf("%d of %d machine(s) did not %s", failed, len(nodes), verb) } return nil } // printHolderAnswer prints what a holder said, or its refusal as the command's failure. func printHolderAnswer(node string, answer link.Answer) error { if answer.Error != "" { fmt.Printf("%s: %s\n", node, answer.Error) return errors.New(answer.Error) } var said struct { Said string `json:"said"` } if json.Unmarshal(answer.Result, &said) == nil && said.Said != "" { fmt.Printf("%s: %s\n", node, said.Said) return nil } fmt.Printf("%s: %s\n", node, string(answer.Result)) return nil } // buildSeatHolders is every machine an assigned module holding the build seat runs on. func buildSeatHolders(ctx context.Context, seat string) ([]string, error) { open, err := openStores(ctx) if err != nil { return nil, err } defer open.Close() entries, err := open.inventory.Catalogued(ctx) if err != nil { return nil, err } return holdersAmong(entries, seat), nil } // holdersAmong is the machines of every catalogued module claiming the seat, sorted, once each. func holdersAmong(entries []inventory.Entry, seat string) []string { seen := map[string]bool{} var out []string for _, e := range entries { if !e.Manifest.ClaimsSeat(seat) { continue } for _, n := range e.On { if !seen[n] { seen[n] = true out = append(out, n) } } } sort.Strings(out) return out } // outstandingFor is everything asked of a module and not yet answered: its asks in the build queue, // by repository and path, and every open plan holding it not yet built. func outstandingFor(module, repository, path string, q link.Queue, plans []inventory.Plan) []string { var out []string for _, a := range q.Asks { if repositoryMatches(a.Repository, repository) && a.Path == path { out = append(out, fmt.Sprintf("%s %s in the queue", a.ID, a.State)) } } if module == "" { return out } for _, p := range plans { if !p.Open() { continue } for _, tier := range p.Tiers { for _, m := range tier { if m == module { if st := p.Modules[m]; st == nil || st.State != "built" { out = append(out, fmt.Sprintf("%s holds it not yet built", p.ID)) } } } } } return out }