package build import ( "context" "encoding/json" "errors" "fmt" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "sort" "strings" "sync" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) // ToolchainEntries are the nodes' toolchain entries of a language (buildrun.ToolchainSet). type ToolchainEntries interface { Entries(language string) []delivery.NodeToolchain } // toolPrefix is the start of each build module's tool names: fetch and build. var toolPrefix = map[string]string{buildrun.GoBuildModule: "build_go_", buildrun.NodeBuildModule: "build_node_", ImageBuildModule: "build_image_"} // Builder builds, dark, every new-form fingerprint the indexer writes (novox/hq ADR 0453 clause 4, ADR 0478): it is // handed each modules value as the watch of the indexer's bucket sees it, and for each module with a fingerprint and an // artifact to build claims the fingerprint's build key, builds it through the build modules on the nodes // the toolchains choose, and writes the key done or failed. It holds no timer; a build starts from a modules value or a // change of the toolchains. type Builder struct { Claimer Claimer Runner buildrun.Runner Toolchains ToolchainEntries // Presence says which nodes are online (the watchman's); a node that is not is not chosen. Presence interface{ Online(node string) bool } Manifests Manifests // Health says whether a node's build module is running and healthy (the node-engine's have); a node whose build module // is not is not chosen. Nil means every node's is. Health interface { Healthy(node, module string) bool } // buildrun.Places holds a step back while its node is at its build capacity; a Builder with none runs one step at a time on each node. Places *buildrun.Places // Nodes is how many build nodes build each fingerprint: two until release R4e (ADR 0478 clause 4). Nodes int // Limit is how many builds run at once; four when 0. Limit int Logf func(format string, args ...any) once sync.Once sem chan struct{} wg sync.WaitGroup mu sync.Mutex known map[string]delivery.Module // by modules key inflight map[string]bool // by build key again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built plans map[string]planned // by manifest digest cands map[string]delivery.Candidate // by candidates key (candidates.go) via map[string]via // by build key: a candidate's fetch (candidates.go) trunkFP map[string]bool // by build key: a trunk module value carries this fingerprint tried map[string]bool // by build key: looked at since this process started // epoch counts the changes of the nodes (online, healthy, a toolchain) seen since this process started; a done key // whose second-node build failed is held back until it moves (issue 1664). epoch uint64 secondAt map[string]uint64 // by build key: the epoch a second-node build of a one-node key failed at } type planned struct { plan Plan err error } func (b *Builder) init() { b.once.Do(func() { limit := b.Limit if limit <= 0 { limit = 4 } b.sem = make(chan struct{}, limit) b.cands, b.via, b.trunkFP = map[string]delivery.Candidate{}, map[string]via{}, map[string]bool{} b.known, b.inflight, b.again, b.plans, b.tried = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}, map[string]bool{} b.secondAt = map[string]uint64{} if b.Places == nil { b.Places = &buildrun.Places{} } }) } func (b *Builder) logf(format string, args ...any) { if b.Logf != nil { b.Logf(format, args...) } } // Wait returns when every build started so far has ended. func (b *Builder) Wait() { b.wg.Wait() } // Module takes one value of the indexer's modules bucket in (the key is the module's name) and builds it if it is // to be built. A value that is not a module is said and left. func (b *Builder) Module(ctx context.Context, key string, value []byte) { b.init() m, err := delivery.DecodeModule(value) if err != nil { b.logf("modules.%s holds what is not a module, left: %v", key, err) return } b.mu.Lock() b.known[key] = m if bk, err := delivery.BuilderBuildKey(key, m.FP2); err == nil && m.FP2 != "" { // A fingerprint the trunk carries is built from the trunk's commit, whoever asked for it first. b.trunkFP[bk] = true delete(b.via, bk) } b.mu.Unlock() b.consider(ctx, key, m) } // NodesChanged is Reconsider after a change of the nodes: one came online or went, a build module became healthy or // stopped being, or a toolchain moved. A done key built on one node whose second-node build failed is tried again only // after such a change, not at every modules value (issue 1664). func (b *Builder) NodesChanged(ctx context.Context) { b.init() b.mu.Lock() b.epoch++ b.mu.Unlock() b.Reconsider(ctx) } // Sweep closes, at start, every key left building for a fingerprint its module no longer has (issue 1660, ADR 0478 // clause 6): the key can never be built, since nothing asks for it, and would stay building for ever. It fails the key // naming who left it. A key of the fingerprint the module has now is not touched: the build takes it over. Call it // once the modules bucket has been read, so every module is known. func (b *Builder) Sweep(ctx context.Context) error { b.init() keys, err := b.Claimer.Store.Keys(ctx) if err != nil { return err } b.mu.Lock() current := map[string]bool{} for name, m := range b.known { if m.FP2 != "" { if k, err := delivery.BuilderBuildKey(name, m.FP2); err == nil { current[k] = true } } } b.mu.Unlock() for _, key := range keys { if current[key] { continue } value, rev, found, err := b.Claimer.Store.Get(ctx, key) if err != nil || !found { continue } cur, err := delivery.DecodeBuilderBuild(value) if err != nil || cur.State != delivery.BuildBuilding { continue } closed := delivery.BuilderBuild{State: delivery.BuildFailed, By: b.Claimer.By, Manifest: cur.Manifest, Step: cur.Step, Error: fmt.Sprintf("closed at start: left building by %s, and the module no longer has this fingerprint, so nothing will build it", cur.By)} data, err := delivery.EncodeBuilderBuild(closed) if err != nil { return err } if _, err := b.Claimer.Store.Update(ctx, key, data, rev); err != nil && !errors.Is(err, ErrConflict) { return fmt.Errorf("%s: %w", key, err) } b.logf("%s: closed, left building by %s for a fingerprint the module no longer has", key, cur.By) } return nil } // Forget takes a module out (its key was deleted). func (b *Builder) Forget(key string) { b.init() b.mu.Lock() delete(b.known, key) b.mu.Unlock() } // Reconsider looks at every module again: the toolchains changed, so a build that failed for them may build now. func (b *Builder) Reconsider(ctx context.Context) { b.init() b.mu.Lock() known := make(map[string]delivery.Module, len(b.known)) for k, m := range b.known { known[k] = m } b.mu.Unlock() for _, k := range sortedKeys(known) { b.consider(ctx, k, known[k]) } b.mu.Lock() cands := make([]delivery.Candidate, 0, len(b.cands)) for _, c := range b.cands { cands = append(cands, c) } b.mu.Unlock() sort.Slice(cands, func(i, j int) bool { return cands[i].Group < cands[j].Group }) for _, c := range cands { b.considerCandidate(ctx, c) } } func sortedKeys(m map[string]delivery.Module) []string { keys := make([]string, 0, len(m)) for k := range m { keys = append(keys, k) } sort.Strings(keys) return keys } // planOf is the plan of the manifest with this digest, read once: a manifest by digest never changes. A manifest that // cannot be read is not kept, so it is read again when the module is. func (b *Builder) planOf(ctx context.Context, digest string) (Plan, error, bool) { b.mu.Lock() p, ok := b.plans[digest] b.mu.Unlock() if ok { return p.plan, p.err, true } m, err := b.Manifests.Manifest(ctx, digest) if err != nil { return Plan{}, fmt.Errorf("its manifest %s cannot be read: %w", digest, err), false } plan, err := PlanOf(m) b.mu.Lock() b.plans[digest] = planned{plan, err} b.mu.Unlock() return plan, err, true } func (b *Builder) consider(ctx context.Context, module string, m delivery.Module) { if m.FP2 == "" || m.Manifest == "" { return } key, err := delivery.BuilderBuildKey(module, m.FP2) if err != nil { b.logf("modules.%s: %v", module, err) return } b.mu.Lock() if b.inflight[key] { b.again[key] = m // looked at once more when the build ends, so a change is never lost b.mu.Unlock() return } b.inflight[key] = true b.mu.Unlock() b.wg.Add(1) go func() { defer b.wg.Done() b.sem <- struct{}{} defer func() { <-b.sem }() for { b.try(ctx, module, m) b.mu.Lock() next, more := b.again[key] delete(b.again, key) if !more || ctx.Err() != nil { delete(b.inflight, key) b.mu.Unlock() return } b.mu.Unlock() m = next } }() } // try claims the fingerprint and builds it, writing the key done or failed. A build that ends because ctx did is left // building by this builder: it takes it again when it starts (ADR 0478 clause 5). func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { plan, planErr, read := b.planOf(ctx, m.Manifest) if read && planErr == nil && len(plan.Tasks) == 0 { return // nothing in the manifest for a builder to build } if planErr == nil { if why := b.stale(m, plan); why != "" { // The fingerprint names a toolchain the build nodes no longer agree on: a new fingerprint is coming, and a build // now would be of this source with another toolchain (ADR 0477 clause 2). Nothing is written. b.logf("%s at %s is not built: %s", module, m.FP2, why) return } } key, _ := delivery.BuilderBuildKey(module, m.FP2) b.mu.Lock() trunk := b.trunkFP[key] first := !b.tried[key] b.tried[key] = true epoch := b.epoch failedAt, failedBefore := b.secondAt[key] b.mu.Unlock() moreNodes := func(modules []string) bool { // A second build that failed is not tried again until the nodes change (issue 1664). return (!failedBefore || epoch > failedAt) && b.moreNodes(modules) } cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, MoreNodes: moreNodes, ImageBuilder: b.imageBuilderRuns, Trunk: trunk}) if err != nil { b.logf("%s: %v", module, err) return } if cl == nil { return } // A done key taken again builds only what is missing: the build modules that ran on one node, and the image when it was // built as a candidate's; the other artifacts stay as the done key has them (ADR 0487 clauses 5 and 8). var prior *delivery.BuilderBuild var only map[string]bool publishing := false if cl.Build.State == delivery.BuildDone { prior, only = &cl.Build, map[string]bool{} for _, bm := range OneNodeModules(cl.Build) { only[bm] = true } if trunk && Unpublished(cl.Build) { only[ImageBuildModule] = true publishing = true } } result := b.build(ctx, module, m, plan, planErr, prior, only) if ctx.Err() != nil { return } if result.State == "" { b.logf("%s: left building, the toolchains moved under it", cl.Key) return } result.By, result.Manifest = b.Claimer.By, m.Manifest if publishing && result.State == delivery.BuildFailed { // The trunk's image was built only as a candidate's and its push failed: the key says so, so the builds view, the // checker and the failed-key retake see it, and a look does not build the image again for nothing. result.Error = "publishing the image of the merged commit failed: " + result.Error } else if cl.Build.State == delivery.BuildDone && result.State == delivery.BuildFailed && !strings.HasPrefix(result.Error, buildrun.DigestsDifferPrefix) { // A done key built again on a second node keeps its done value when the new build fails for another reason than // the digests: the first build stands, and a later change of the nodes looks again. b.logf("%s: built again on a second node and failed, the done build stands: %s", cl.Key, result.Error) b.mu.Lock() b.secondAt[cl.Key] = epoch b.mu.Unlock() return } if err := b.Claimer.Finish(ctx, cl, result); err != nil { b.logf("%s: the build could not be written: %v", cl.Key, err) return } if result.State == delivery.BuildFailed { b.logf("%s: failed: %s", cl.Key, result.Error) } } // stale is why the module's fingerprint is not built now, or empty when it is: the toolchain each language of the plan // has agreed (novox/hq ADR 0465) must be the one the fingerprint was computed with (Module.Toolchains). func (b *Builder) stale(m delivery.Module, plan Plan) string { for _, bm := range plan.Modules() { if !plan.NeedsToolchain(bm) { continue // an archive is packed with no toolchain } language := buildrun.LanguageOf[bm] tc, none := delivery.AgreedToolchain(b.Toolchains.Entries(language)) if none != "" { return fmt.Sprintf("the %s toolchain is not agreed: %s", language, none) } want, ok := m.Toolchains[language] if !ok || want.Version != tc.Version || want.Recipe != tc.Recipe { return fmt.Sprintf("the fingerprint names %s %s (recipe %s) and the nodes agree on %s (recipe %s)", language, want.Version, want.Recipe, tc.Version, tc.Recipe) } } return "" } func failed(format string, args ...any) delivery.BuilderBuild { return delivery.BuilderBuild{State: delivery.BuildFailed, Error: fmt.Sprintf(format, args...)} } // lane is the build of a plan's tasks of one build module on one node. type lane struct { module, node string tasks []Task digests map[string]string // artifact -> digest references map[string]string // artifact -> the reference an image was pushed to (a trunk build only) err error } func (b *Builder) build(ctx context.Context, module string, m delivery.Module, plan Plan, planErr error, prior *delivery.BuilderBuild, only map[string]bool) delivery.BuilderBuild { switch { case planErr != nil: return failed("%s", planErr) case m.Path == "": // Empty is not known (a root module's path is "."): the indexer writes it again (issue 1615). return failed("the module's key names no directory (path), so the build tools cannot be told where %s lives", module) } repository := delivery.RepositoryName(m.Repository) var lanes []*lane used := map[string]delivery.ToolchainUsed{} var steps, oneNode []string // Every node chosen holds a reservation until its lane takes the place; a build that ends before its lanes start gives them up. giveUp := func(nodes ...string) { for _, n := range nodes { b.Places.Cancel(n) } } giveUpLanes := func() { for _, l := range lanes { b.Places.Cancel(l.node) } } var kept []string // build modules whose artifacts stay as the prior done key has them for _, bm := range plan.Modules() { if prior != nil && len(only) > 0 && !only[bm] { kept = append(kept, bm) continue } language := buildrun.LanguageOf[bm] entries := b.Toolchains.Entries(language) qualifies := func(node string) bool { return b.qualifies(node, bm) } var nodes []string if bm == ImageBuildModule { // An image is built on one node, chosen from the nodes whose image build module runs healthy (ADR 0487 clause 8): // the module has no toolchain language, so no toolchain entry names its nodes, and a second build would only // push the image again. ns, err := buildrun.ChooseAnyNodes("image", b.imageEntries(), b.Places, qualifies, 1) if err != nil { giveUpLanes() return failed("%s no node has %s, so the image cannot be built: the provision rootless-image-build is held by the module buildah; assign buildah and %s to a node (%v)", NoImageBuilderPrefix, ImageBuildModule, ImageBuildModule, err) } nodes = ns } else if plan.NeedsToolchain(bm) { ns, tc, err := buildrun.ChooseNodes(language, entries, b.Places, qualifies, b.Nodes) if err != nil { giveUpLanes() return failed("%s", err) } if want := m.Toolchains[language]; want.Version != tc.Version || want.Recipe != tc.Recipe { giveUp(ns...) giveUpLanes() return delivery.BuilderBuild{} // the agreed toolchain moved since try looked: nothing is written } used[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe} nodes = ns } else { // An archive is packed with no toolchain (the fingerprint names none): any node of the build module that qualifies. ns, err := buildrun.ChooseAnyNodes(language, entries, b.Places, qualifies, b.Nodes) if err != nil { giveUpLanes() return failed("%s", err) } nodes = ns } if bm == ImageBuildModule { steps = append(steps, fmt.Sprintf("%s on %s", bm, nodes[0])) } else if len(nodes) == 1 { oneNode = append(oneNode, bm) steps = append(steps, fmt.Sprintf("%s on one node, %s", bm, nodes[0])) } else { sorted := append([]string(nil), nodes...) sort.Strings(sorted) steps = append(steps, fmt.Sprintf("%s on %s", bm, strings.Join(sorted, " and "))) } var tasks []Task for _, t := range plan.Tasks { if t.Module == bm { tasks = append(tasks, t) } } for _, n := range nodes { lanes = append(lanes, &lane{module: bm, node: n, tasks: tasks, digests: map[string]string{}, references: map[string]string{}}) } } var wg sync.WaitGroup for _, l := range lanes { wg.Add(1) go func() { defer wg.Done(); b.runLane(ctx, l, repository, m, module) }() } wg.Wait() var errs []string for _, l := range lanes { if l.err != nil { errs = append(errs, l.err.Error()) } } if len(errs) > 0 { return failed("%s", strings.Join(errs, "; ")) } var artifacts []delivery.BuiltArtifact keptModule := map[string]bool{} for _, bm := range kept { keptModule[bm] = true } for _, t := range plan.Tasks { if keptModule[t.Module] { found := false for _, a := range prior.Artifacts { if a.Name == t.Artifact { artifacts, found = append(artifacts, a), true } } if !found { return failed("the done build holds no artifact %s to keep", t.Artifact) } continue } byNode := map[string]string{} for _, l := range lanes { if d, ok := l.digests[t.Artifact]; ok && l.module == t.Module { byNode[l.node] = d } } d, err := buildrun.Agree(t.Artifact, byNode) if err != nil { return failed("%s", err) } built := delivery.BuiltArtifact{Name: t.Artifact, Kind: t.Kind, Digest: d} for _, l := range lanes { if r := l.references[t.Artifact]; r != "" && l.module == t.Module { built.Reference = r } } artifacts = append(artifacts, built) } for _, bm := range kept { steps = append(steps, bm+" kept") for lang, u := range prior.Toolchains { if _, ok := used[lang]; !ok && buildrun.LanguageOf[bm] == lang { used[lang] = u } } } step := "built " + strings.Join(steps, "; ") if len(oneNode) > 0 { // Marked so that Claim builds the key again once a second node qualifies (ADR 0478 clause 4). step = OneNodePrefix + strings.Join(oneNode, ", ") + "): " + step } if len(used) == 0 { used = nil } return delivery.BuilderBuild{State: delivery.BuildDone, Step: step, Artifacts: artifacts, Toolchains: used} } func (b *Builder) online(node string) bool { return b.Presence == nil || b.Presence.Online(node) } // qualifies is whether the node can build with the build module now: it is online and the build module on it is healthy. func (b *Builder) qualifies(node, bm string) bool { return b.Places.CapacityOf(node) > 0 && b.online(node) && (b.Health == nil || b.Health.Healthy(node, bm)) } // imageEntries are the nodes an image may be built on, as toolchain entries for the chooser: every node the builder has a // capacity for or a build module toolchain from. Whether the image build module runs there is qualifies', from the have. func (b *Builder) imageEntries() []delivery.NodeToolchain { seen := map[string]bool{} var entries []delivery.NodeToolchain add := func(n string) { if !seen[n] { seen[n] = true entries = append(entries, delivery.NodeToolchain{Node: n}) } } for n := range b.Places.Capacity { add(n) } for _, language := range []string{"go", "typescript"} { for _, e := range b.Toolchains.Entries(language) { add(e.Node) } } sort.Slice(entries, func(i, j int) bool { return entries[i].Node < entries[j].Node }) return entries } // imageBuilderRuns is whether a node can build an image now. func (b *Builder) imageBuilderRuns() bool { for _, e := range b.imageEntries() { if b.qualifies(e.Node, ImageBuildModule) { return true } } return false } // moreNodes is whether any of these build modules has two nodes that qualify now. func (b *Builder) moreNodes(modules []string) bool { for _, bm := range modules { seen := map[string]bool{} n := 0 for _, e := range b.Toolchains.Entries(buildrun.LanguageOf[bm]) { if !seen[e.Node] && b.qualifies(e.Node, bm) { seen[e.Node] = true n++ } } if n >= 2 { return true } } return false } // step runs one step of a build module; the lane that runs it holds its node's place. func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) { return b.Runner.Run(ctx, bm, node, toolPrefix[bm]+tool, args) } // fetchResult and buildResult are the results of a build module's fetch and build steps as the builder reads them. type fetchResult struct { Workspace string `json:"workspace"` Merge string `json:"merge"` Conflicts []string `json:"conflicts"` } type buildResult struct { Digest string `json:"digest"` Reference string `json:"reference"` } // runLane takes a place on the node, fetches the commit there and builds each of the lane's artifacts, one step at a time. func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string) { release, err := b.Places.Acquire(ctx, l.node) if err != nil { l.err = err return } defer release() fetchArgs := map[string]any{"repository": repository, "commit": m.Commit} if key, err := delivery.BuilderBuildKey(module, m.FP2); err == nil { b.mu.Lock() v, candidate := b.via[key] b.mu.Unlock() if candidate { fetchArgs["ref"], fetchArgs["base"] = v.Ref, v.Base } } raw, err := b.step(ctx, l.module, l.node, "fetch", fetchArgs) if err != nil { l.err = err return } var fetched fetchResult if err := json.Unmarshal(raw, &fetched); err != nil || fetched.Workspace == "" { l.err = fmt.Errorf("the fetch on %s answered no workspace: %s", l.node, raw) return } // A candidate is built from its head merged onto the trunk. A head that does not merge ends the fetch passed with the tree at // the trunk, and building that would store the trunk's build under the candidate's fingerprint. _, candidate := fetchArgs["base"] if candidate { if len(fetched.Conflicts) > 0 { l.err = fmt.Errorf("the head does not merge into the trunk on %s: files in conflict: %s", l.node, strings.Join(fetched.Conflicts, ", ")) return } if fetched.Merge == "" { l.err = fmt.Errorf("the fetch with a base on %s answered no merge: %s", l.node, raw) return } } for _, t := range l.tasks { args := map[string]any{"workspace": fetched.Workspace, "module": m.Path, "artifact": t.Artifact} if t.Kind == catalogue.ArtifactImage && !candidate { // Only a trunk key pushes the image to the registry; a candidate's build never does (ADR 0487 clause 12). args["publish"] = true } raw, err := b.step(ctx, l.module, l.node, "build", args) if err != nil { l.err = err return } var built buildResult if err := json.Unmarshal(raw, &built); err != nil || built.Digest == "" { l.err = fmt.Errorf("the build of %s of %s on %s answered no digest: %s", t.Artifact, module, l.node, raw) return } l.digests[t.Artifact] = built.Digest if built.Reference != "" && !candidate { l.references[t.Artifact] = built.Reference } } }