Files
mesh-controller/cmd/mesh-controller/gate_integration_test.go
T
mesh-tools 013dad4688 Take the catalogue, overlay and validator from mesh-sdk in cmd/mesh-builder and cmd/mesh-controller, a to g (issue 1381, 1 of 6 import swaps)
Swaps three import paths for mesh-sdk's go/catalogue, go/overlay and go/validate (the same packages, copied from this repository at 6c2a8d62) and lets gofmt order the groups. The six swap commits compile together once the pin commit (go.mod, modules.txt) is on the branch; each alone does not, because a file on the SDK's types cannot call one still on the old package.
2026-10-20 08:22:07 +00:00

483 lines
19 KiB
Go

//go:build integration
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/lease"
"github.com/novox/mesh-controller/internal/link"
)
func aGateMesh(t *testing.T) *gateMesh {
t.Helper()
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
g := &gateMesh{open: open, health: map[string]health{}}
// The gate's own tests judge a first machine, as walks a controller before novox/hq ADR 0360 started still are.
wasGated := gateFirstMachine
gateFirstMachine = true
t.Cleanup(func() { gateFirstMachine = wasGated })
g.keeper, _ = withConditionsInMemory(t)
g.told = &conditions.Told{}
was := doctorFrom
doctorFrom = &doctor{open: open, keeper: g.keeper, teller: g.told}
t.Cleanup(func() { doctorFrom = was })
for _, b := range []inventory.Build{
{ID: "build-1", Module: "app", Commit: "c1", Repository: "novox/mesh-catalog", Path: "modules/app",
Asked: time.Now().Add(-2 * time.Hour), At: time.Now().Add(-2 * time.Hour)},
{ID: "build-2", Module: "app", Commit: "c2", Repository: "novox/mesh-catalog", Path: "modules/app",
Asked: time.Now().Add(-time.Minute), At: time.Now().Add(-time.Minute)},
} {
manifest, _ := json.Marshal(catalogue.Manifest{Module: "app", Version: b.Commit})
b.Manifest = manifest
if err := inv.RecordBuild(ctx, b); err != nil {
t.Fatal(err)
}
}
register := func(commit string, asked time.Time) {
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "app", Version: commit},
inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/app", BuiltFrom: commit,
Head: commit, Asked: asked}); err != nil {
t.Fatal(err)
}
}
register("c1", time.Now().Add(-2*time.Hour))
for _, n := range []string{"anchor", "laptop"} {
if _, err := inv.Assign(ctx, n, "app"); err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{"app": "c1"}); err != nil {
t.Fatal(err)
}
}
register("c2", time.Now().Add(-time.Minute))
// Every send: recorded with the build the module is at, applied and reported by the machine.
n := 0
wasSend := sendRollout
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
g.sent = append(g.sent, append([]string(nil), names...))
current, err := open.inventory.CurrentBuilds(ctx)
if err != nil {
return nil, err
}
for _, node := range names {
n++
digest := fmt.Sprintf("d-%s-%d", node, n)
if err := open.inventory.RecordSent(ctx, nodeID(t, open, node), digest, map[string]string{"app": current["app"].Commit}); err != nil {
return nil, err
}
if _, err := open.inventory.RecordDoing(ctx, nodeID(t, open, node), inventory.Doing{Node: node,
Outcome: inventory.OutcomeApplied, Declared: digest, Applied: 1, At: time.Now()}); err != nil {
return nil, err
}
}
return names, nil
}
t.Cleanup(func() { sendRollout = wasSend })
// What a judging reads: the store's reports, and a health the test says per machine.
wasGather := gatherGateFacts
gatherGateFacts = func(ctx context.Context, open *stores, component string) (gateFacts, error) {
f := gateFacts{now: time.Now(), reports: map[string]inventory.Reported{}, engines: map[string]string{},
rolledBack: map[string][]lease.Rollback{}, served: map[string]served{}}
reports, err := open.inventory.LastReports(ctx)
if err != nil {
return f, err
}
for _, r := range reports {
if g.health[r.Node] == healthBroken {
r.Outcome = inventory.OutcomeFailed
}
f.reports[r.Node] = r
}
for node, h := range g.health {
if h == healthNotYet {
// Not reported on the send: neither the send made last, nor the gate's own (issue 352).
f.rolledBack[node] = nil
r := f.reports[node]
r.Current, r.Declared, r.ReportedSequence = false, "", 0
f.reports[node] = r
}
}
return f, nil
}
t.Cleanup(func() { gatherGateFacts = wasGather })
wasWithin, wasEvery, wasBound := gateWithin, gateEvery, gateBound
// The witness window is the node-engine's, from a start no test makes (novox/hq ADR 0351).
wasWindow := gateWitnessWindow
gateWitnessWindow = 0
t.Cleanup(func() { gateWitnessWindow = wasWindow })
// One judging per advance until a test says otherwise: a pass waits gateEvery for the next.
gateWithin, gateEvery = time.Hour, time.Hour
t.Cleanup(func() { gateWithin, gateEvery, gateBound = wasWithin, wasEvery, wasBound })
built := time.Now().UTC()
plan := inventory.Plan{ID: "plan-gate", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2",
Created: built, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
Modules: map[string]*inventory.PlanModule{"app": {State: "built", BuiltAt: &built, Commit: "c2",
Build: "build-2"}}}
if err := inv.SavePlan(ctx, &plan); err != nil {
t.Fatal(err)
}
return g
}
func TestABuildThatFailsItsGateIsRolledBackOnItsFirstMachineAndGoesNoFurther(t *testing.T) {
judgeLeavesAsTheCore(t)
g := aGateMesh(t)
ctx := t.Context()
inv := g.open.inventory
advancePlans(ctx, g.open) // the first machine is sent the new build
if !reflect.DeepEqual(g.sent, [][]string{{"anchor"}}) {
t.Fatalf("sent %v, not the first machine alone", g.sent)
}
if p := g.plan(t); p.Modules["app"].Previous != "c1" {
t.Fatalf("the build the first machine ran before was not kept: %+v", p.Modules["app"])
}
g.health["anchor"] = healthBroken // the new build breaks its first machine, after one good judging
gateEvery = 0
advancePlans(ctx, g.open)
p := g.plan(t)
gate := p.Modules["app"].Gate
if p.State != inventory.PlanFailed || gate == nil || gate.Verdict != inventory.GateFailed ||
gate.Rollback != inventory.RolledBack {
t.Fatalf("the plan is %s (%s), its gate %+v", p.State, p.Note, gate)
}
if !reflect.DeepEqual(g.sent, [][]string{{"anchor"}, {"anchor"}}) {
t.Fatalf("sent %v: the rollback goes to the first machine, and nothing reaches laptop", g.sent)
}
if current, _ := inv.CurrentBuilds(ctx); current["app"].Commit != "c1" {
t.Fatalf("the module is registered at %s, not put back to c1", current["app"].Commit)
}
if sent, _, _ := inv.SentBuilds(ctx, "anchor"); sent["app"] != "c1" {
t.Fatalf("anchor was last sent %v, not the previous build", sent)
}
if sent, _, _ := inv.SentBuilds(ctx, "laptop"); sent["app"] != "c1" {
t.Fatalf("laptop was sent the failed build: %v", sent)
}
if failed, err := inv.GateFailed(ctx, "build-2"); err != nil || !failed {
t.Fatalf("the build is not marked failed at its gate: %v %v", failed, err)
}
// Said: a condition for the operator, and the event.
open, err := g.keeper.Open(ctx)
if err != nil {
t.Fatal(err)
}
var key string
for _, c := range open {
if c.Kind == kindRolledBack {
key = c.Key
}
}
if key != "build.app.anchor.rolled-back" {
t.Fatalf("no rolled-back condition for app on anchor: %+v", open)
}
saidIt := false
for _, e := range g.told.Said() {
saidIt = saidIt || e.Event == link.KeyRolledBack
}
if !saidIt {
t.Fatalf("the rollback was not said as an event: %+v", g.told.Said())
}
// Never again: more passes send nothing, a second judging rolls nothing back, and the failed build
// heard again is not registered.
advancePlans(ctx, g.open)
state := p.Modules["app"]
gateFailed(ctx, g.open, &p, "app", state, []string{"anchor"}, "again")
if len(g.sent) != 2 {
t.Fatalf("sent again after the rollback: %v", g.sent)
}
_, _, err = takeIn(ctx, inv, link.BuildResult{ID: "build-2", Repository: "novox/mesh-catalog", Path: "modules/app",
Commit: "c2", Manifest: mustJSON(t, catalogue.Manifest{Module: "app", Version: "c2"}),
Source: &link.SourceOnSeat{Seat: "git", Repository: "novox/mesh-catalog"}})
if err == nil || !strings.Contains(err.Error(), "failed its gate") {
t.Fatalf("the failed build was registered again: %v", err)
}
if current, _ := inv.CurrentBuilds(ctx); current["app"].Commit != "c1" {
t.Fatalf("the module moved to %s", current["app"].Commit)
}
if _, err := retryPlan(ctx, g.open, "plan-gate"); err == nil || !strings.Contains(err.Error(), "failed its gate") {
t.Fatalf("a plan stopped at a failed gate was retried: %v", err)
}
// The probe keeps the condition while the newest verdict is the failure.
obs, err := probeGates(ctx, doctorFrom)
if err != nil || len(obs) != 1 || obs[0].Key() != key {
t.Fatalf("the gate's probe found %+v %v", obs, err)
}
}
func TestABuildThatPassesItsGateRollsEverywhereUnattended(t *testing.T) {
g := aGateMesh(t)
ctx := t.Context()
gateEvery = 0
for i := 0; i < 6; i++ {
advancePlans(ctx, g.open)
}
p := g.plan(t)
if !reflect.DeepEqual(g.sent, [][]string{{"anchor"}, {"laptop"}}) {
t.Fatalf("sent %v", g.sent)
}
gate := p.Modules["app"].Gate
if p.State != inventory.PlanDone || gate == nil || gate.Verdict != inventory.GatePassed || gate.Passes < gateWanted(gate) {
t.Fatalf("the plan is %s (%s), its gate %+v", p.State, p.Note, gate)
}
if v, found, err := g.open.inventory.GateOf(ctx, "build-2"); err != nil || !found || v.Verdict != inventory.GatePassed {
t.Fatalf("the verdict was not kept: %+v %v %v", v, found, err)
}
if obs, err := probeGates(ctx, doctorFrom); err != nil || len(obs) != 0 {
t.Fatalf("a passing build left a condition: %+v %v", obs, err)
}
}
func TestABuildRollsToEveryMachineAtOnceWithNoGate(t *testing.T) {
g := aGateMesh(t)
gateFirstMachine = false
ctx := t.Context()
judged := false
wasGather := gatherGateFacts
gatherGateFacts = func(ctx context.Context, open *stores, component string) (gateFacts, error) {
judged = true
return wasGather(ctx, open, component)
}
t.Cleanup(func() { gatherGateFacts = wasGather })
for i := 0; i < 6; i++ {
advancePlans(ctx, g.open)
}
p := g.plan(t)
if !reflect.DeepEqual(g.sent, [][]string{{"anchor", "laptop"}}) {
t.Fatalf("sent %v, not both machines in one send", g.sent)
}
if s := p.Modules["app"]; p.State != inventory.PlanDone || s.Gate != nil || s.FirstAt != nil || judged {
t.Fatalf("the plan is %s (%s), app %+v, judged %v", p.State, p.Note, s, judged)
}
if obs, err := probeGates(ctx, doctorFrom); err != nil || len(obs) != 0 {
t.Fatalf("a rolled build left a condition: %+v %v", obs, err)
}
}
func TestALeafModulePassesOnOneHealthyReading(t *testing.T) {
g := aGateMesh(t)
ctx := t.Context()
gateWithin = time.Hour // a core module could not pass on fewer readings for an hour
advancePlans(ctx, g.open) // the first machine is sent the new build
advancePlans(ctx, g.open) // one healthy judging
p := g.plan(t)
gate := p.Modules["app"].Gate
if gate == nil || gate.Verdict != inventory.GatePassed || gate.Passes != 1 || gate.Wants != 1 {
t.Fatalf("a leaf module's gate after one healthy judging: %+v (plan %s: %s)", gate, p.State, p.Note)
}
if !strings.Contains(gate.Why, "healthy 1 times") {
t.Errorf("the verdict says %q", gate.Why)
}
}
func TestABuildNeverHealthyFailsAtTheBound(t *testing.T) {
judgeLeavesAsTheCore(t)
g := aGateMesh(t)
ctx := t.Context()
advancePlans(ctx, g.open)
g.health["anchor"] = healthNotYet
gateEvery = 0
advancePlans(ctx, g.open)
if p := g.plan(t); !p.Open() || len(g.sent) != 1 {
t.Fatalf("judged before the bound: %s %v", p.State, g.sent)
}
gateBound = -time.Second
advancePlans(ctx, g.open)
p := g.plan(t)
if gate := p.Modules["app"].Gate; p.State != inventory.PlanFailed || gate.Verdict != inventory.GateFailed ||
!strings.Contains(gate.Why, "not healthy within") || gate.Rollback != inventory.RolledBack {
t.Fatalf("the plan is %s, its gate %+v", p.State, gate)
}
}
func TestTheBusIsNeverRolledOutAutomatically(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
bus := catalogue.Manifest{Module: "nats", Version: "2", Provides: []catalogue.Offer{{Name: "mesh-bus"}},
Upgrade: &catalogue.UpgradePolicy{Policy: catalogue.PolicyRoll}}
if err := inv.RegisterModule(ctx, bus, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git",
Path: "modules/nats", BuiltFrom: "n2", Head: "n2"}); err != nil {
t.Fatal(err)
}
if _, err := inv.Assign(ctx, "anchor", "nats"); err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, nodeID(t, open, "anchor"), "d-anchor", map[string]string{"nats": "n1"}); err != nil {
t.Fatal(err)
}
u, err := inv.UpgradeOf(ctx, "nats")
if err != nil || u.RollOut || u.From != catalogue.FromBus {
t.Fatalf("the bus's policy is %+v %v", u, err)
}
if err := inv.SetUpgradeOf(ctx, "nats", inventory.Upgrade{RollOut: true}); !errors.Is(err, inventory.ErrBusIsPlanned) {
t.Fatalf("a person rolled the bus out: %v", err)
}
var sent [][]string
was := sendRollout
sendRollout = func(_ context.Context, _ *stores, names []string) ([]string, error) {
sent = append(sent, names)
return names, nil
}
t.Cleanup(func() { sendRollout = was })
now := time.Now().UTC()
plan := inventory.Plan{ID: "plan-bus", Repository: "novox/mesh-catalog", Commit: "n2", Created: now,
State: inventory.PlanBuilding, Tiers: [][]string{{"nats"}},
Modules: map[string]*inventory.PlanModule{"nats": {State: "built", BuiltAt: &now, Commit: "n2", Build: "b"}}}
if err := inv.SavePlan(ctx, &plan); err != nil {
t.Fatal(err)
}
advancePlans(ctx, open)
if p, _ := inv.PlanByID(ctx, "plan-bus"); p.State != inventory.PlanDone || len(sent) != 0 {
t.Fatalf("a plan sent the bus: %s %v", p.State, sent)
}
held, err := busHeld(ctx, inv, []string{"anchor", "laptop"})
if err != nil || !strings.Contains(held["anchor"], "planned step") || held["laptop"] != "" {
t.Fatalf("a push may send the bus's machine: %v %v", held, err)
}
// Nor may any other send — a plan's for another module on that machine carried the bus with it.
if _, err := sendToEach(ctx, open, []string{"laptop", "anchor"}); !errors.Is(err, errBusWaits) {
t.Fatalf("a send to the bus's machine was not refused: %v", err)
}
// A rebuild that made the same artifacts is no move.
for _, b := range []inventory.Build{{ID: "nb1", Module: "nats", Commit: "n1"}, {ID: "nb2", Module: "nats", Commit: "n2"}} {
b.Made = []inventory.Artifact{{Name: "server", Kind: "image", Reference: "registry/nats@sha256:same"}}
b.Asked, b.At = time.Now(), time.Now()
if err := inv.RecordBuild(ctx, b); err != nil {
t.Fatal(err)
}
}
if held, err := busHeld(ctx, inv, []string{"anchor"}); err != nil || len(held) != 0 {
t.Fatalf("a rebuild that changes nothing held the bus's machine: %v %v", held, err)
}
if err := inv.RecordBuild(ctx, inventory.Build{ID: "nb3", Module: "nats", Commit: "n2", Asked: time.Now().Add(time.Second),
At: time.Now().Add(time.Second), Made: []inventory.Artifact{{Name: "server", Kind: "image",
Reference: "registry/nats@sha256:new"}}}); err != nil {
t.Fatal(err)
}
// The planned step refuses to start without its word on reversibility, and without a snapshot taken
// first by the bus machine's backup holder.
if err := busCommand(ctx, []string{"upgrade", "--why", "2.11"}); err == nil || !strings.Contains(err.Error(), "reversible") {
t.Fatalf("a bus upgrade started without saying whether it can be reverted: %v", err)
}
wasSnapshot := takeBusSnapshot
t.Cleanup(func() { takeBusSnapshot = wasSnapshot })
takeBusSnapshot = func(context.Context, string, string) (string, error) { return "", errors.New("no holder answers") }
if err := busCommand(ctx, []string{"upgrade", "--why", "2.11", "--reversible"}); err == nil ||
!strings.Contains(err.Error(), "snapshotted") || len(sent) != 0 {
t.Fatalf("a bus upgrade started without its snapshot: %v, sent %v", err, sent)
}
takeBusSnapshot = func(_ context.Context, module, node string) (string, error) {
return node + "'s restore point of " + module, nil
}
if err := busCommand(ctx, []string{"upgrade", "--why", "2.11", "--reversible"}); err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(sent, [][]string{{"anchor"}}) {
t.Fatalf("the step sent %v, not the bus's machine", sent)
}
s, found, err := inv.LatestBusStep(ctx)
if err != nil || !found || s.Snapshot != "anchor's restore point of nats" || s.Ended != nil ||
!reflect.DeepEqual(s.Machines, []string{"anchor"}) {
t.Fatalf("the step is %+v %v %v", s, found, err)
}
}
func TestAMergeThatDeletesAModulePlansNothingToBuildForIt(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
asked := asksRecorded(t)
for _, name := range []string{"gone", "kept"} {
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: name, Version: "1"}, inventory.Source{
Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/" + name, BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
}
m := link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: "c1deadbeef",
Paths: []string{"modules/gone/module.json", "modules/gone/index.ts"},
Removed: []string{"modules/gone/module.json", "modules/gone/index.ts"}}
if err := (following{open: open}).SourceMoved(ctx, m); err != nil {
t.Fatal(err)
}
if len(*asked) != 0 {
t.Fatalf("a deleted module was asked to build: %v", *asked)
}
if plans, _ := inv.OpenPlans(ctx); len(plans) != 0 {
t.Fatalf("a merge that only deleted a module made a plan: %+v", plans)
}
shelf, err := inv.Catalogue(ctx)
if err != nil {
t.Fatal(err)
}
if _, still := shelf["gone"]; still {
t.Fatal("a deleted module nothing holds was not forgotten")
}
// Without the announcer saying what went: the build finds no manifest, and the plan goes on.
asked0 := time.Now().UTC().Add(-time.Minute)
plan := inventory.Plan{ID: "plan-deleted", Repository: "novox/mesh-catalog", Commit: "c2", Created: asked0,
State: inventory.PlanBuilding, Tiers: [][]string{{"kept"}},
Modules: map[string]*inventory.PlanModule{"kept": {State: "asked", AskedAt: &asked0, Build: "build-k"}}}
if err := inv.SavePlan(ctx, &plan); err != nil {
t.Fatal(err)
}
planBuilt(ctx, open, "kept", "", "http://forge/novox/mesh-catalog.git has no module.json at modules/kept, so "+
"there is nothing saying what it is: open …/module.json: no such file or directory", asked0, "build-k")
p, _ := inv.PlanByID(ctx, "plan-deleted")
if p.State == inventory.PlanFailed || p.Modules["kept"].State != planDeleted {
t.Fatalf("a module deleted at its source failed the plan: %s %+v", p.State, p.Modules["kept"])
}
}
func TestAWitnessesVerdictIsAConditionWhileItsReportsSayIt(t *testing.T) {
open := aMesh(t)
d := &doctor{open: open}
at := time.Now().UTC()
witnessed.heard("control", []lease.Rollback{
{Component: lease.ComponentController, From: "sha256:new", To: "sha256:old", Outcome: lease.OutcomeRolledBack,
Why: "the new controller did not take the lease within 60s", At: at},
{Component: lease.ComponentNodeTools, From: "sha256:t2", Outcome: lease.OutcomeUnwitnessed, Why: "no grant", At: at},
}, at)
t.Cleanup(func() { witnessed.heard("control", nil, time.Now()) })
obs, err := probeGates(t.Context(), d)
if err != nil {
t.Fatal(err)
}
got := map[string]conditions.Severity{}
for _, o := range obs {
got[o.Key()] = o.Severity
}
want := map[string]conditions.Severity{"core.controller.control.rolled-back": conditions.Urgent,
"core.node-tools.control.unwitnessed": conditions.Warning}
if !reflect.DeepEqual(got, want) {
t.Fatalf("the witness's verdicts are %v", got)
}
witnessed.heard("control", nil, time.Now())
if obs, err := probeGates(t.Context(), d); err != nil || len(obs) != 0 {
t.Fatalf("a verdict the reports no longer carry is still said: %+v %v", obs, err)
}
}