Files
mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go
T

525 lines
21 KiB
Go

//go:build integration
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
// controller's user and the builder's own principals <node>.<module> (two instances, for the test of one build).
const busConf = `
listen: 127.0.0.1:-1
jetstream { store_dir: %q }
accounts {
MESH {
jetstream: enabled
users = [
{ user: "ctl", password: "pw" }
{ user: "n1.mesh-delivery-builder", password: "pw" }
{ user: "n2.mesh-delivery-builder", password: "pw" }
]
}
}
`
const (
commit = "7a74cee83f88e6585a5511c3cc53bdf0f5303d20"
fp = "src2:3333333333333333333333333333333333333333333333333333333333333333"
)
// fakeBuildModules serve the tools of mesh-build-go and mesh-build-node as the real ones do (ADR 0469 clause 2): a call on
// mesh.mod.<module>.call.<tool>.<node>.<caller> answers at once with the step and its key, and the step's end is written to
// the key <node>.<id> of the module's steps bucket, some milliseconds later.
type fakeBuildModules struct {
js jetstream.JetStream
ctx context.Context
n atomic.Int64
failBuild atomic.Bool
mu sync.Mutex
calls map[string]int // by `<module>.<tool>`
}
func (f *fakeBuildModules) count(tool string) int {
f.mu.Lock()
defer f.mu.Unlock()
return f.calls[tool]
}
// digestOf is what every node builds an artifact to.
func digestOf(artifact string) string {
sum := sha256.Sum256([]byte(artifact))
return "sha256:" + hex.EncodeToString(sum[:])
}
func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) {
steps, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1})
if err != nil {
t.Fatal(err)
}
sub, err := nc.Subscribe("mesh.mod."+module+".call.>", func(m *nats.Msg) {
tokens := strings.Split(m.Subject, ".") // mesh mod <module> call <tool> <node> <caller>
tool, node := tokens[4], tokens[5]
f.mu.Lock()
f.calls[module+"."+tool]++
f.mu.Unlock()
var args map[string]any
_ = json.Unmarshal(m.Data, &args)
id := fmt.Sprintf("s%d", f.n.Add(1))
key, _ := delivery.StepKey(node, id)
put := func(s delivery.Step) {
data, err := delivery.EncodeStep(s)
if err == nil {
_, err = steps.Put(f.ctx, key, data)
}
if err != nil {
t.Errorf("writing step %s: %v", key, err)
}
}
step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: time.Now().UTC()}
put(step)
answer, _ := json.Marshal(map[string]any{"result": map[string]any{"step": id, "key": key, "state": "running"}, "error": "", "node": node})
_ = m.Respond(answer)
go func() {
time.Sleep(30 * time.Millisecond)
step.Ended = time.Now().UTC()
switch {
case strings.HasSuffix(tool, "_fetch"):
step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"workspace":"w`+id+`","commit":"`+commit+`"}`)
case f.failBuild.Load() && node == "n1":
step.State, step.Reason = delivery.StepFailed, fmt.Sprintf("go build of %v: boom", args["artifact"])
default:
step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"digest":"`+digestOf(fmt.Sprint(args["artifact"]))+`"}`)
}
put(step)
}()
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
type bed struct {
ctx context.Context
js jetstream.JetStream
modules jetstream.KeyValue
builds jetstream.KeyValue
tools map[string]jetstream.KeyValue
presence jetstream.KeyValue
have jetstream.KeyValue
fake *fakeBuildModules
srvURL string
dir string
}
// startBed starts the bus, the buckets the controller makes from the builder's and the build modules' manifests, the
// build modules' streams and the builder's durable consumers on them, two nodes' toolchains, the indexer's manifests, and
// the fake build modules. It runs no builder.
func startBed(t *testing.T) *bed {
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "b")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
conf := filepath.Join(dir, "bus.conf")
if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
t.Fatal(err)
}
opts, err := server.ProcessConfigFile(conf)
if err != nil {
t.Fatal(err)
}
opts.NoLog, opts.NoSigs = true, true
srv, err := server.NewServer(opts)
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("the bus did not come up")
}
t.Cleanup(srv.Shutdown)
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ctl.Close)
js, err := jetstream.New(ctl)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
t.Cleanup(cancel)
b := &bed{ctx: ctx, js: js, tools: map[string]jetstream.KeyValue{}, srvURL: srv.ClientURL(), dir: dir}
for module, stream := range toolchainStreams {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil {
t.Fatal(err)
}
for _, node := range []string{"n1", "n2"} {
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
t.Fatal(err)
}
}
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket(module), History: 1})
if err != nil {
t.Fatal(err)
}
b.tools[module] = kv
}
mk := func(name string, history int) jetstream.KeyValue {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
if err != nil {
t.Fatal(err)
}
return kv
}
b.modules = mk(delivery.ModulesBucket, delivery.ModulesHistory)
mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
presence := mk(delivery.PresenceBucket, delivery.PresenceHistory)
b.presence, b.have = presence, mk(delivery.HaveBucket, delivery.HaveHistory)
for _, node := range []string{"n1", "n2"} {
b.setOnline(t, node, true)
for _, module := range []string{"mesh-build-go", "mesh-build-node"} {
b.setHealth(t, node, module, delivery.HealthOK)
}
}
b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}}
b.fake.serve(t, ctl, "mesh-build-go")
b.fake.serve(t, ctl, "mesh-build-node")
b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "1")
b.report(t, "mesh-build-go", "n2", "go", "go1.27.1", "1")
b.report(t, "mesh-build-node", "n1", "typescript", "tsc5", "1")
b.report(t, "mesh-build-node", "n2", "typescript", "tsc5", "1")
return b
}
// setOnline writes the watchman's presence of a node.
func (b *bed) setOnline(t *testing.T, node string, online bool) {
t.Helper()
data, _ := delivery.EncodePresence(delivery.Presence{Online: online})
if _, err := b.presence.Put(b.ctx, node, data); err != nil {
t.Fatal(err)
}
}
// setHealth writes the node-engine's have of a build module on a node, running with this health.
func (b *bed) setHealth(t *testing.T, node, module string, health delivery.Health) {
t.Helper()
data, err := delivery.EncodeHave(delivery.Have{Phase: delivery.PhaseRunning, Health: health})
if err != nil {
t.Fatal(err)
}
key, _ := delivery.HaveKey(node, module)
if _, err := b.have.Put(b.ctx, key, data); err != nil {
t.Fatal(err)
}
}
// report writes a node's toolchains key of a build module and says so on the module's stream, as the module does.
func (b *bed) report(t *testing.T, module, node, language, version, recipe string) {
t.Helper()
data, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: recipe, At: time.Now().UTC()})
if err != nil {
t.Fatal(err)
}
key, _ := delivery.ToolchainKey(node, language)
if _, err := b.tools[module].Put(b.ctx, key, data); err != nil {
t.Fatal(err)
}
if _, err := b.js.Publish(b.ctx, "mesh.stream."+module+".event.toolchain.reported", []byte(`{}`)); err != nil {
t.Fatal(err)
}
}
// module writes modules.<name> as the indexer does for a module with a new-form fingerprint, and its manifest by digest.
func (b *bed) module(t *testing.T, name, fixture, path string) string {
return b.moduleWith(t, name, fixture, path, map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.1", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}})
}
// moduleWith is module with the toolchains its fingerprint was computed with.
func (b *bed) moduleWith(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed) string {
return b.moduleCommit(t, name, fixture, path, toolchains, commit)
}
func (b *bed) moduleCommit(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed, at string) string {
t.Helper()
raw, err := os.ReadFile(filepath.Join("testdata", fixture))
if err != nil {
t.Fatal(err)
}
digest := delivery.ManifestDigest(raw)
manifests, err := b.js.KeyValue(b.ctx, delivery.ManifestsBucket)
if err != nil {
t.Fatal(err)
}
mv, err := delivery.EncodeManifest(delivery.Manifest{Module: name, JSON: string(raw)})
if err != nil {
t.Fatal(err)
}
if _, err := manifests.Put(b.ctx, digest, mv); err != nil {
t.Fatal(err)
}
data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: at, FP: "src1:" + strings.Repeat("f", 64), FP2: fp,
Manifest: digest, Path: path, Toolchains: toolchains})
if err != nil {
t.Fatal(err)
}
if _, err := b.modules.Put(b.ctx, name, data); err != nil {
t.Fatal(err)
}
return digest
}
// archive writes modules.<name> as the real indexer does for a module of an archive only: its fingerprint names no
// toolchain (mesh-delivery-indexer internal/index inputs.go languagesOf gives an archive no language), so Toolchains is empty.
func (b *bed) archive(t *testing.T, name, path, at string) string {
t.Helper()
return b.moduleCommit(t, name, "catalogue-a67f0196-ca-trust.module.json", path, nil, at)
}
// start runs the program as the builder of a node.
func (b *bed) start(t *testing.T, node string, env map[string]string) {
t.Helper()
broker := filepath.Join(t.TempDir(), "broker")
cred, _ := json.Marshal(credential{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
e := map[string]string{"MESH_BROKER_FILE": broker}
for k, v := range env {
e[k] = v
}
runCtx, stop := context.WithCancel(b.ctx)
done := make(chan error, 1)
go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }()
t.Cleanup(func() { stop(); <-done })
}
// await reads the build key until pred holds of it, within 20 s.
func (b *bed) await(t *testing.T, module string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild {
t.Helper()
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
t.Fatal(err)
}
var last delivery.BuilderBuild
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
e, err := b.builds.Get(b.ctx, key)
if err != nil {
continue
}
if last, err = delivery.DecodeBuilderBuild(e.Value()); err == nil && pred(last) {
return last
}
}
t.Fatalf("builds.%s = %+v within 20 s", key, last)
return last
}
func isState(state string) func(delivery.BuilderBuild) bool {
return func(v delivery.BuilderBuild) bool { return v.State == state }
}
// TestBuilderMakesADoneKeyWithTheDigests: modules.netcheck (a Go and a TypeScript bundle) and its manifest are in the
// indexer's buckets, and two nodes report each build module's toolchain; the program writes builds.netcheck.<fp> done,
// the digests of both bundles, both build modules' toolchains, after a fetch and a build on each of the two nodes.
func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
b := startBed(t)
digest := b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
b.start(t, "n1", nil)
v := b.await(t, "netcheck", isState(delivery.BuildDone))
if v.By != "n1.mesh-delivery-builder" || v.Manifest != digest || len(v.Artifacts) != 2 {
t.Fatalf("build = %+v", v)
}
for _, a := range v.Artifacts {
if a.Digest != digestOf(a.Name) || a.Kind != "bundle" || a.Reference != "" {
t.Errorf("artifact = %+v, want the digest every node built and no reference", a)
}
}
if v.Toolchains["go"].Version != "go1.27.1" || v.Toolchains["typescript"].Version != "tsc5" {
t.Errorf("toolchains = %+v", v.Toolchains)
}
for tool, want := range map[string]int{"mesh-build-go.build_go_fetch": 2, "mesh-build-go.build_go_build": 2, "mesh-build-node.build_node_fetch": 2, "mesh-build-node.build_node_build": 2} {
if got := b.fake.count(tool); got != want {
t.Errorf("%s called %d times, want %d (once on each of two nodes)", tool, got, want)
}
}
}
// TestTwoBuilderInstancesStartingTogetherEndWithOneDoneKey: two builders on two nodes start together on one module key of an
// archive only (no toolchain named, as the indexer writes it); the second takes over the key the first holds building
// (ADR 0478 clause 6), and the key ends done, written once.
func TestTwoBuilderInstancesStartingTogetherEndWithOneDoneKey(t *testing.T) { // 1607/AC7
b := startBed(t)
b.archive(t, "ca-trust", "modules/ca-trust", commit)
b.start(t, "n1", nil)
b.start(t, "n2", nil)
v := b.await(t, "ca-trust", isState(delivery.BuildDone))
if len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "archive" || v.Artifacts[0].Digest != digestOf("files") {
t.Fatalf("build = %+v, want the archive's digest", v)
}
if n := b.fake.count("mesh-build-go.build_go_build"); n < 2 || n%2 != 0 { // each instance that built, on two nodes
t.Errorf("build_go_build was called %d times, want two for each instance that built", n)
}
}
// TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt: the indexer gives an archive no language, so modules.ca-trust has no
// toolchains; the builder builds it on the build module's nodes all the same (review round 4).
func TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt(t *testing.T) {
b := startBed(t)
b.archive(t, "ca-trust", "modules/ca-trust", commit)
b.start(t, "n1", nil)
v := b.await(t, "ca-trust", isState(delivery.BuildDone))
if len(v.Toolchains) != 0 || len(v.Artifacts) != 1 || strings.HasPrefix(v.Step, "one node (") {
t.Errorf("build = %+v, want done on two nodes naming no toolchain", v)
}
}
// TestAKeyBuiltOnOneNodeIsBuiltAgainWhenASecondNodeComesOnline: n2 is offline, so the key is done on n1 alone and marked;
// n2 coming online makes the builder build it again on both nodes, as a new revision (ADR 0478 clause 4).
func TestAKeyBuiltOnOneNodeIsBuiltAgainWhenASecondNodeComesOnline(t *testing.T) {
b := startBed(t)
b.setOnline(t, "n2", false)
b.archive(t, "ca-trust", "modules/ca-trust", commit)
b.start(t, "n1", nil)
one := b.await(t, "ca-trust", isState(delivery.BuildDone))
if !strings.HasPrefix(one.Step, "one node (") {
t.Fatalf("build = %+v, want it marked as built on one node", one)
}
key := mustKey(t, "ca-trust")
e, err := b.builds.Get(b.ctx, key)
if err != nil {
t.Fatal(err)
}
b.setOnline(t, "n2", true)
two := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
return v.State == delivery.BuildDone && !strings.HasPrefix(v.Step, "one node (")
})
if e2, _ := b.builds.Get(b.ctx, key); e2.Revision() <= e.Revision() || !strings.Contains(two.Step, "n1 and n2") {
t.Errorf("build = %+v at revision %d (was %d), want a new revision built on n1 and n2", two, e2.Revision(), e.Revision())
}
}
// TestANodeWhoseBuildModuleIsUnhealthyIsNotUsed: the node-engine's have of mesh-build-go on n2 says failed, so only n1 builds.
func TestANodeWhoseBuildModuleIsUnhealthyIsNotUsed(t *testing.T) {
b := startBed(t)
b.setHealth(t, "n2", "mesh-build-go", delivery.HealthFailed)
b.archive(t, "ca-trust", "modules/ca-trust", commit)
b.start(t, "n1", nil)
v := b.await(t, "ca-trust", isState(delivery.BuildDone))
if !strings.Contains(v.Step, "on one node, n1") {
t.Errorf("build = %+v, want it built on n1 alone", v)
}
}
// TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint: a build step fails on one node, so the
// key is failed naming it; the indexer writing modules.<m> again with the same fingerprint wakes the builder, which builds
// the key again as a new revision of the same key, now done.
func TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint(t *testing.T) { // 1607/AC6, AC7
b := startBed(t)
b.fake.failBuild.Store(true)
b.archive(t, "ca-trust", "modules/ca-trust", commit)
b.start(t, "n1", nil)
v := b.await(t, "ca-trust", isState(delivery.BuildFailed))
if !strings.Contains(v.Error, "n1") || !strings.Contains(v.Error, "boom") {
t.Errorf("build = %+v, want failed naming the node and the step's reason", v)
}
key, _ := delivery.BuilderBuildKey("ca-trust", fp)
failed, err := b.builds.Get(b.ctx, key)
if err != nil {
t.Fatal(err)
}
b.fake.failBuild.Store(false)
// The indexer writes the module again (here at a later commit), with the same fingerprint.
b.archive(t, "ca-trust", "modules/ca-trust", "89abcdef0123456789abcdef0123456789abcdef")
b.await(t, "ca-trust", isState(delivery.BuildDone))
done, err := b.builds.Get(b.ctx, key)
if err != nil {
t.Fatal(err)
}
if done.Revision() <= failed.Revision() {
t.Errorf("revision %d, was %d: the key was not written as a new revision", done.Revision(), failed.Revision())
}
}
// TestAToolchainChangeDoesNotBuildTheOldFingerprint: the fingerprint names go1.27.1, the nodes then agree on go1.27.2 (a
// report of both nodes), and the builder writes no key for the old fingerprint although it was woken by the report; a
// module value naming go1.27.2 (the new fingerprint, computed by the indexer) is built.
func TestAToolchainChangeDoesNotBuildTheOldFingerprint(t *testing.T) { // 1615/AC4
b := startBed(t)
b.report(t, "mesh-build-go", "n1", "go", "go1.27.2", "1")
b.report(t, "mesh-build-go", "n2", "go", "go1.27.2", "1")
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") // names go1.27.1
b.start(t, "n1", nil)
// Another module, whose fingerprint names the new toolchain, is built; by then the old one has been looked at.
b.moduleCommit(t, "netcheck2", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck",
map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.2", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}}, commit)
b.await(t, "netcheck2", isState(delivery.BuildDone))
if _, err := b.builds.Get(b.ctx, mustKey(t, "netcheck")); err == nil {
t.Error("the old fingerprint, naming go1.27.1, was built with go1.27.2")
}
if n := b.fake.count("mesh-build-go.build_go_build"); n != 2 {
t.Errorf("build_go_build was called %d times, want 2 (the new fingerprint on two nodes only)", n)
}
}
// TestARootModuleIsBuilt: a module whose path is "." is given to the build tools as it is.
func TestARootModuleIsBuilt(t *testing.T) { // 1615/AC4
b := startBed(t)
b.archive(t, "ca-trust", delivery.RootPath, commit)
b.start(t, "n1", nil)
b.await(t, "ca-trust", isState(delivery.BuildDone))
}
func mustKey(t *testing.T, module string) string {
k, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
t.Fatal(err)
}
return k
}
// moduleAt is module with the go toolchain version the fingerprint names and the commit it was computed at.
func (b *bed) moduleAt(t *testing.T, name, fixture, path, goVersion, at string) {
t.Helper()
b.moduleCommit(t, name, fixture, path, map[string]delivery.ToolchainUsed{"go": {Version: goVersion, Recipe: "1"}}, at)
}
// TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart: the builds bucket holds a key left building by another node for
// a module the indexer no longer has; the program, at start, fails it naming why (issue 1660, ADR 0478 clause 6).
func TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart(t *testing.T) {
b := startBed(t)
data, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "n2.mesh-delivery-builder", Manifest: strings.Repeat("a", 64)})
if err != nil {
t.Fatal(err)
}
if _, err := b.builds.Put(b.ctx, mustKey(t, "ghost"), data); err != nil {
t.Fatal(err)
}
b.start(t, "n1", nil)
v := b.await(t, "ghost", isState(delivery.BuildFailed))
if !strings.Contains(v.Error, "no longer has this fingerprint") || !strings.Contains(v.Error, "n2.mesh-delivery-builder") {
t.Errorf("build = %+v, want it closed naming who left it", v)
}
}