jochen
9873b3bf13
A registered replay asked now outranked newer asks of its module, and a plan took any later outcome as its answer. A replay is now refused while the module is asked anywhere, a plan module asked under an id is answered by that id alone, and a rebuild of a commit asks what the module follows. An ask handed back after a restart no longer reads as dead; a cancel that meets a start is withdrawn; pause holds for an ask fetched as it lands; kill removes containers before and after the build ends and says whether its outcome went out; a holder may say only its own machine is paused.
387 lines
13 KiB
Go
387 lines
13 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/micro"
|
|
|
|
"github.com/novox/mesh-controller/internal/builder"
|
|
"github.com/novox/mesh-controller/internal/catalogue"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// What a holder of the build seat answers for, on its own machine (novox/hq ADR 0219).
|
|
//
|
|
// **The queue is the controller's; the build running here is this machine's.** The controller can
|
|
// see and change what waits in the seat's queue, but an ask a holder already took is a process tree
|
|
// on this machine and containers in this machine's runtime, and only this machine can end them. So
|
|
// the holder serves four verbs on the seat's subjects for this machine: what it is building, kill
|
|
// it, pause, resume.
|
|
//
|
|
// **One build at a time** (ADR 0190), which is also what builder.Said assumes — a package global
|
|
// set per build — so "the build running here" is one or none, and kill names it by id so a call
|
|
// that arrives as one build ends and the next begins cannot end the wrong one.
|
|
|
|
// holder is this machine's state as a holder of the build seat.
|
|
type holder struct {
|
|
on, seat string
|
|
// workspace is where the paused flag is kept, so a holder restarted while paused stays paused
|
|
// rather than silently taking work again.
|
|
workspace string
|
|
// say publishes this machine's state: whether it takes work (link.HolderState).
|
|
say func(link.HolderState) error
|
|
// remove runs what a kill needs outside the build: the containers left behind.
|
|
remove builder.Runner
|
|
|
|
mu sync.Mutex
|
|
paused bool
|
|
running *running
|
|
}
|
|
|
|
// running is the build this machine is doing.
|
|
type running struct {
|
|
request link.BuildRequest
|
|
step string
|
|
started time.Time
|
|
cancel context.CancelFunc
|
|
killed bool
|
|
// returned is the build's own work having ended, before a kill or not; outcome is what was then
|
|
// announced, and announced whether it went out.
|
|
returned bool
|
|
outcome string
|
|
announced bool
|
|
done chan struct{}
|
|
}
|
|
|
|
// pausedFile is where the flag lives in the workspace.
|
|
func pausedFile(workspace string) string { return filepath.Join(workspace, ".mesh-builder-paused") }
|
|
|
|
// newHolder reads the paused flag the workspace keeps.
|
|
func newHolder(on, seat, workspace string, say func(link.HolderState) error) *holder {
|
|
h := &holder{on: on, seat: seat, workspace: workspace, say: say, remove: plainRun}
|
|
if _, err := os.Stat(pausedFile(workspace)); err == nil {
|
|
h.paused = true
|
|
}
|
|
return h
|
|
}
|
|
|
|
// Paused is asked by the taking loop before every fetch.
|
|
func (h *holder) Paused() bool {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
return h.paused
|
|
}
|
|
|
|
// setPaused records the flag in the workspace first and then in memory, so what this holder says
|
|
// it is and what it would be after a restart never differ.
|
|
func (h *holder) setPaused(paused bool) error {
|
|
path := pausedFile(h.workspace)
|
|
if paused {
|
|
if err := os.MkdirAll(h.workspace, 0o755); err != nil {
|
|
return err
|
|
}
|
|
if err := os.WriteFile(path, []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil {
|
|
return fmt.Errorf("cannot keep the paused flag in %s: %w", path, err)
|
|
}
|
|
} else if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
|
|
return fmt.Errorf("cannot remove the paused flag %s: %w", path, err)
|
|
}
|
|
h.mu.Lock()
|
|
h.paused = paused
|
|
h.mu.Unlock()
|
|
h.announce()
|
|
return nil
|
|
}
|
|
|
|
// announce says whether this machine takes work. Never fatal: the flag is kept either way, and the
|
|
// controller reading an older state is a plan read as late rather than a build lost.
|
|
func (h *holder) announce() {
|
|
if h.say == nil {
|
|
return
|
|
}
|
|
if err := h.say(link.HolderState{On: h.on, Paused: h.Paused(), At: time.Now().UTC().Format(time.RFC3339Nano)}); err != nil {
|
|
fmt.Fprintf(os.Stderr, "cannot say whether this machine takes builds: %v\n", err)
|
|
}
|
|
}
|
|
|
|
// stateWhilePaused says this machine's state at start, and again every while it stays paused, so
|
|
// a pause outlives the events stream's retention.
|
|
func (h *holder) stateWhilePaused(ctx context.Context, every time.Duration) {
|
|
h.announce()
|
|
tick := time.NewTicker(every)
|
|
defer tick.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-tick.C:
|
|
if h.Paused() {
|
|
h.announce()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// begin records the build this machine took, with the cancel that ends it.
|
|
func (h *holder) begin(request link.BuildRequest, cancel context.CancelFunc) *running {
|
|
r := &running{request: request, started: time.Now(), cancel: cancel, done: make(chan struct{})}
|
|
h.mu.Lock()
|
|
h.running = r
|
|
h.mu.Unlock()
|
|
return r
|
|
}
|
|
|
|
// end clears it, and lets a kill waiting on it know it is over.
|
|
func (h *holder) end(r *running) {
|
|
h.mu.Lock()
|
|
if h.running == r {
|
|
h.running = nil
|
|
}
|
|
h.mu.Unlock()
|
|
close(r.done)
|
|
}
|
|
|
|
// stepped records the step a build is at, for `current`.
|
|
func (h *holder) stepped(r *running, step string) {
|
|
h.mu.Lock()
|
|
r.step = step
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
// currentBuild is what `current` answers.
|
|
type currentBuild struct {
|
|
On string `json:"on"`
|
|
Paused bool `json:"paused"`
|
|
Running *struct {
|
|
ID string `json:"id"`
|
|
Repository string `json:"repository"`
|
|
Path string `json:"path,omitempty"`
|
|
Ref string `json:"ref,omitempty"`
|
|
Step string `json:"step,omitempty"`
|
|
Started string `json:"started"`
|
|
Elapsed string `json:"elapsed"`
|
|
} `json:"running,omitempty"`
|
|
Said string `json:"said"`
|
|
}
|
|
|
|
func (h *holder) current() currentBuild {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
out := currentBuild{On: h.on, Paused: h.paused}
|
|
taking := "taking builds"
|
|
if h.paused {
|
|
taking = "paused, taking no new build"
|
|
}
|
|
if h.running == nil {
|
|
out.Said = fmt.Sprintf("%s is building nothing; %s", h.on, taking)
|
|
return out
|
|
}
|
|
r := h.running
|
|
elapsed := time.Since(r.started).Round(time.Second)
|
|
out.Running = &struct {
|
|
ID string `json:"id"`
|
|
Repository string `json:"repository"`
|
|
Path string `json:"path,omitempty"`
|
|
Ref string `json:"ref,omitempty"`
|
|
Step string `json:"step,omitempty"`
|
|
Started string `json:"started"`
|
|
Elapsed string `json:"elapsed"`
|
|
}{r.request.ID, r.request.Repository, r.request.Path, r.request.Ref, r.step,
|
|
r.started.UTC().Format(time.RFC3339), elapsed.String()}
|
|
out.Said = fmt.Sprintf("%s is building %s (%s) at %s for %s; %s",
|
|
h.on, r.request.Repository, r.request.ID, orNothing(r.step), elapsed, taking)
|
|
return out
|
|
}
|
|
|
|
// The bounds a kill keeps: each pass removing containers, and the wait for the build to end between
|
|
// them. The worst case — 15s, 20s, 15s — is inside what the controller waits for the answer
|
|
// (killAnswer in the controller's queue.go, 75s).
|
|
var (
|
|
killRemoves = 15 * time.Second
|
|
killWaits = 20 * time.Second
|
|
)
|
|
|
|
// kill ends the build with this id, if it is the one running here and still working: its context
|
|
// cancelled — which kills each command's process group — and the containers it started removed by
|
|
// their label, once at once and again after the build has ended, so one created while it was being
|
|
// killed is not left. The build's own goroutine announces it failed, killed by hand, and settles the
|
|
// ask so it is not redelivered; the answer says whether that happened.
|
|
func (h *holder) kill(id string) (string, error) {
|
|
h.mu.Lock()
|
|
r := h.running
|
|
if r == nil || r.request.ID != id {
|
|
doing := "nothing"
|
|
if r != nil {
|
|
doing = r.request.ID
|
|
}
|
|
h.mu.Unlock()
|
|
return "", fmt.Errorf("%s is not building %s; it is building %s. `queue` says where an ask is", h.on, id, doing)
|
|
}
|
|
if r.returned {
|
|
h.mu.Unlock()
|
|
return "", fmt.Errorf("%s on %s has already ended on its own and is saying how; `builds` shows it", id, h.on)
|
|
}
|
|
r.killed = true
|
|
cancel, done := r.cancel, r.done
|
|
h.mu.Unlock()
|
|
|
|
cancel()
|
|
removed, removeErr := h.removeContainers(id)
|
|
ended := false
|
|
select {
|
|
case <-done:
|
|
ended = true
|
|
case <-time.After(killWaits):
|
|
}
|
|
again, againErr := h.removeContainers(id)
|
|
removed += again
|
|
if removeErr == nil {
|
|
removeErr = againErr
|
|
}
|
|
containers := fmt.Sprintf("%d container(s) it started removed", removed)
|
|
if removeErr != nil {
|
|
containers = "its containers could not all be listed or removed: " + removeErr.Error()
|
|
}
|
|
if !ended {
|
|
return fmt.Sprintf("killed %s on %s: %s; the build has not finished ending yet — `builds` says when "+
|
|
"its outcome is in", id, h.on, containers), nil
|
|
}
|
|
h.mu.Lock()
|
|
outcome, announced := r.outcome, r.announced
|
|
h.mu.Unlock()
|
|
if !announced {
|
|
return fmt.Sprintf("killed %s (%s) on %s: its commands ended and %s, and its outcome could not be "+
|
|
"announced — the ask is not settled and will be handed out again", id, r.request.Repository, h.on,
|
|
containers), nil
|
|
}
|
|
return fmt.Sprintf("killed %s (%s) on %s: its commands ended, %s, and its outcome announced as failed, %s — "+
|
|
"settled, so it is not handed to another machine", id, r.request.Repository, h.on, containers, outcome), nil
|
|
}
|
|
|
|
// removeContainers is one pass of removing what the build left, bounded.
|
|
func (h *holder) removeContainers(id string) (int, error) {
|
|
cleanup, stop := context.WithTimeout(context.Background(), killRemoves)
|
|
defer stop()
|
|
return builder.RemoveContainersOf(cleanup, h.remove, id)
|
|
}
|
|
|
|
// returned records that the build's work ended, and says whether a kill came first — only then is
|
|
// the build killed; an error it ended with on its own is its own outcome.
|
|
func (h *holder) returned(r *running) bool {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
r.returned = true
|
|
return r.killed
|
|
}
|
|
|
|
// said records the outcome announced, and whether it went out, for a kill to answer with.
|
|
func (h *holder) said(r *running, outcome string, announced bool) {
|
|
h.mu.Lock()
|
|
r.outcome, r.announced = outcome, announced
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
// handlers are the seat's verbs, as this machine answers them.
|
|
func (h *holder) handlers() map[string]link.ToolHandler {
|
|
return map[string]link.ToolHandler{
|
|
"current": func(context.Context, json.RawMessage) (any, error) { return h.current(), nil },
|
|
"kill": func(_ context.Context, raw json.RawMessage) (any, error) {
|
|
var args struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(raw, &args); err != nil || strings.TrimSpace(args.ID) == "" {
|
|
return nil, errors.New("kill needs the build's id")
|
|
}
|
|
said, err := h.kill(strings.TrimSpace(args.ID))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{"said": said}, nil
|
|
},
|
|
"pause": func(context.Context, json.RawMessage) (any, error) {
|
|
if err := h.setPaused(true); err != nil {
|
|
return nil, err
|
|
}
|
|
said := h.on + " is paused: it takes no new build until resumed"
|
|
if c := h.current(); c.Running != nil {
|
|
said += "; " + c.Running.ID + " runs on and finishes"
|
|
}
|
|
return map[string]any{"said": said, "paused": true}, nil
|
|
},
|
|
"resume": func(context.Context, json.RawMessage) (any, error) {
|
|
if err := h.setPaused(false); err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{"said": h.on + " takes builds again", "paused": false}, nil
|
|
},
|
|
}
|
|
}
|
|
|
|
func orNothing(s string) string {
|
|
if s == "" {
|
|
return "its start"
|
|
}
|
|
return s
|
|
}
|
|
|
|
// plainRun runs a command outside any build: no line reaches a build's log, because the build whose
|
|
// log it would be is the one being ended.
|
|
func plainRun(ctx context.Context, dir, name string, args ...string) (string, error) {
|
|
cmd := exec.CommandContext(ctx, name, args...)
|
|
cmd.Dir = dir
|
|
out, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
return string(out), fmt.Errorf("%s %s: %w: %s", name, strings.Join(args, " "), err, strings.TrimSpace(string(out)))
|
|
}
|
|
return string(out), nil
|
|
}
|
|
|
|
// announcement is what this holder answers discovery with (novox/hq ADR 0195, ADR 0197): this
|
|
// machine's verbs of the seat, in the shape every tool runtime announces a seat's verb — kind seat,
|
|
// the module answering, the seat, scope node, the machine, its description and argument schema — so
|
|
// the console finds `<node>/node-build-agent.kill` by searching, as it finds any seat's verb.
|
|
//
|
|
// One service per machine, named for the seat and identified by the machine, so the answer is this
|
|
// machine's four verbs and nothing more. The verbs are the compiled seat row's: a build machine has no
|
|
// store, and what it serves is what this binary was built to serve.
|
|
func announcement(seat, module, node string) micro.Info {
|
|
s, _ := catalogue.SeatNamed(seat)
|
|
var endpoints []micro.EndpointInfo
|
|
for _, v := range s.Serves {
|
|
schema, _ := json.Marshal(v.Input)
|
|
endpoints = append(endpoints, micro.EndpointInfo{
|
|
Name: seat + "__" + v.Name,
|
|
Subject: link.NodeSeatToolSubject(seat, v.Name, node),
|
|
// The queue group the verbs are served in, as every runtime announces its own.
|
|
QueueGroup: "seat." + seat,
|
|
Metadata: map[string]string{
|
|
"kind": "seat", "module": module, "tool": v.Name, "seat": seat, "scope": "node",
|
|
"node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema),
|
|
},
|
|
})
|
|
}
|
|
return micro.Info{
|
|
ServiceIdentity: micro.ServiceIdentity{Name: seat, ID: node, Version: "0.1.0",
|
|
Metadata: map[string]string{"seat": seat, "scope": "node", "node": node, "module": module}},
|
|
Description: "what the build running on " + node + " is, and ending, pausing and resuming it (novox/hq ADR 0219)",
|
|
Endpoints: endpoints,
|
|
}
|
|
}
|
|
|
|
// moduleOf is the module a credential was issued for: its user is `<node>.<module>`.
|
|
func moduleOf(user string) string {
|
|
if _, module, ok := strings.Cut(user, "."); ok && module != "" {
|
|
return module
|
|
}
|
|
return "build-agent"
|
|
}
|