Merge pull request 'Write each module's have and the node's own facts beside the report (issue 1132); the check reads build-agent from the controller (issue 1142)' (#118) from feat/delivery-have into main
This commit was merged in pull request #118.
This commit is contained in:
+30
-1
@@ -1663,7 +1663,7 @@ func declaredResolvConf(d *declaration.Declaration) (string, string, bool) {
|
||||
if !ok || f.Path != network.ResolvConf || f.Seeded() || f.Into != "" {
|
||||
continue
|
||||
}
|
||||
module, ok := liveness.ModuleOf(f.ID)
|
||||
module, ok := d.ModuleOfID(f.ID)
|
||||
if !ok {
|
||||
return "", "", false
|
||||
}
|
||||
@@ -2323,6 +2323,9 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto
|
||||
report.Failed = map[string]string{"apply": applyErr.Error()}
|
||||
}
|
||||
report.Failed = withUnanswered(report.Failed, probe)
|
||||
// And how each module of the declaration came out, for the have keys written beside this report (novox/hq ADR 0429
|
||||
// §4, issue 1132): not on the wire.
|
||||
report.Modules = moduleFacts(declared, outcome, applyErr)
|
||||
// Which resources did not apply, for whether a reconcile's account is news (issue 233): not said on the wire.
|
||||
report.Unapplied = unapplied(declared, outcome, applyErr)
|
||||
return report
|
||||
@@ -2575,3 +2578,29 @@ func unhealthyWords(r liveness.State) string {
|
||||
return fmt.Sprintf("%s (%s %s) is unhealthy: %s, %d automatic restart(s) since this engine began watching it",
|
||||
r.ID, r.Kind, r.Target, r.Reason, r.Restarts)
|
||||
}
|
||||
|
||||
// moduleFacts is how each module of a declaration came out of an apply, with the build the controller sent for it
|
||||
// (novox/hq ADR 0429 §4, issue 1132): sorted by module so the facts do not move between applies. A module an adopted
|
||||
// node has not taken is held as found and not applied, so it is no fact.
|
||||
func moduleFacts(d *declaration.Declaration, outcome apply.Report, applyErr error) []link.ModuleFact {
|
||||
if d == nil {
|
||||
return nil
|
||||
}
|
||||
results := apply.ModuleResults(d, outcome, applyErr)
|
||||
facts := make([]link.ModuleFact, 0, len(results))
|
||||
for module, result := range results {
|
||||
if _, untaken := untakenModules(d)[module]; untaken {
|
||||
continue
|
||||
}
|
||||
facts = append(facts, link.ModuleFact{Module: module, Build: d.Builds[module], Failed: result.Failed, Detail: result.Detail})
|
||||
}
|
||||
sort.Slice(facts, func(i, j int) bool { return facts[i].Module < facts[j].Module })
|
||||
return facts
|
||||
}
|
||||
|
||||
func untakenModules(d *declaration.Declaration) map[string][]string {
|
||||
if d.Adoption == nil {
|
||||
return nil
|
||||
}
|
||||
return d.Adoption.Untaken
|
||||
}
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-host/internal/apply"
|
||||
"github.com/novox/mesh-host/internal/declaration"
|
||||
)
|
||||
|
||||
// 1132/AC1, AC3: the report carries one fact per module the declaration names, with the build the controller sent for
|
||||
// it; a module not yet taken on an adopted node is not one.
|
||||
func TestTheReportCarriesOneFactPerModuleWithItsBuild(t *testing.T) {
|
||||
d, err := declaration.Parse([]byte(`{"declaration":1,"builds":{"a":"b-1","b":"b-2"},"resources":[
|
||||
{"id":"a.dir","type":"directory","path":"/srv/a","module":"a"},
|
||||
{"id":"b.dir","type":"directory","path":"/srv/b","module":"b"}]}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
failure := &apply.Error{Resource: "b.dir", Err: errors.New("no space"), Failures: []apply.Failure{{Resource: "b.dir", Err: errors.New("no space")}}}
|
||||
facts := moduleFacts(d, apply.Report{}, failure)
|
||||
if len(facts) != 2 || facts[0].Module != "a" || facts[0].Build != "b-1" || facts[0].Failed ||
|
||||
facts[1].Module != "b" || facts[1].Build != "b-2" || !facts[1].Failed || facts[1].Detail == "" {
|
||||
t.Fatalf("facts %+v", facts)
|
||||
}
|
||||
adopted, err := declaration.Parse([]byte(`{"declaration":1,"adoption":{"taken":["a"],"untaken":{"b":["b.dir"]}},"resources":[
|
||||
{"id":"a.dir","type":"directory","path":"/srv/a","module":"a"},
|
||||
{"id":"b.dir","type":"directory","path":"/srv/b","module":"b"}]}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if facts := moduleFacts(adopted, apply.Report{}, nil); len(facts) != 1 || facts[0].Module != "a" {
|
||||
t.Fatalf("an untaken module is a fact: %+v", facts)
|
||||
}
|
||||
if moduleFacts(nil, apply.Report{}, nil) != nil {
|
||||
t.Fatal("facts without a declaration")
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
golang.org/x/crypto v0.57.0
|
||||
golang.org/x/sys v0.48.0
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 h1:mAMt3KFmrHzDiBv4MLL94opWcG9y73ISMrBUyisapQE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 h1:BrHyE3ifFMUQLIza6I/iNlEgdcysB9/Y+1FO6Q9xws8=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
|
||||
+27
-13
@@ -179,6 +179,23 @@ type Error struct {
|
||||
// (novox/hq ADR 0136) or a directory the engine refused (issue 495). Named, because "everything
|
||||
// was attempted" is then untrue, and a report that says so after skipping a module lies.
|
||||
Modules []string
|
||||
|
||||
// Failures is every resource that failed, the first among them, each with its error (issue 1132, novox/hq ADR
|
||||
// 0429 §4): Others counts the rest and names none, and a module's state on the bus names what failed in it.
|
||||
Failures []Failure
|
||||
}
|
||||
|
||||
// Failure is one resource that failed in an apply, and why.
|
||||
type Failure struct {
|
||||
Resource string
|
||||
Err error
|
||||
}
|
||||
|
||||
// ModuleResult is how one module's part of an apply ended (issue 1132).
|
||||
type ModuleResult struct {
|
||||
Failed bool
|
||||
// Detail names each failing resource with its error, or why the module was not attempted.
|
||||
Detail string
|
||||
}
|
||||
|
||||
func (e *Error) Error() string {
|
||||
@@ -747,6 +764,7 @@ func ApplyMindingWindows(
|
||||
first := failures[0]
|
||||
first.Done = report
|
||||
first.Others = len(failures) - 1
|
||||
first.Failures = failuresOf(failures)
|
||||
first.Modules = gatedModules()
|
||||
log(" kept " + guardPrefix + "*: the guard is not up, so what it replaces was left in force")
|
||||
return report, known, first
|
||||
@@ -758,7 +776,7 @@ func ApplyMindingWindows(
|
||||
// **A module whose step did not complete is skipped from there on** (novox/hq ADR 0136).
|
||||
// Reported rather than passed over in silence: "not attempted" and "nothing to do" are
|
||||
// different answers, and only one of them is somebody's to fix.
|
||||
if module, ours := moduleOf(resource.Identity()); ours && gated[module] {
|
||||
if module, ours := d.ModuleOfID(resource.Identity()); ours && gated[module] {
|
||||
report.Outcomes = append(report.Outcomes, Outcome{
|
||||
ID: resource.Identity(), Type: string(resource.Kind()), Target: resource.Target(),
|
||||
Action: "skipped", Detail: gatedWhy[module],
|
||||
@@ -920,7 +938,7 @@ func ApplyMindingWindows(
|
||||
in.refused[filepath.Clean(dir.Path)] = err
|
||||
}
|
||||
if dir, ok := resource.(*declaration.Directory); ok && in.refused[filepath.Clean(dir.Path)] != nil {
|
||||
if module, ours := moduleOf(resource.Identity()); ours {
|
||||
if module, ours := d.ModuleOfID(resource.Identity()); ours {
|
||||
gated[module] = true
|
||||
gatedWhy[module] = fmt.Sprintf("its directory %s was refused (novox/hq issue 495), so "+
|
||||
"nothing more of this module was attempted", filepath.Clean(dir.Path))
|
||||
@@ -969,7 +987,7 @@ func ApplyMindingWindows(
|
||||
//
|
||||
// An action still gates the machine: the bootstrap is a row of them, each making the
|
||||
// next possible, and they belong to no module.
|
||||
if module, ours := moduleOf(resource.Identity()); ours {
|
||||
if module, ours := d.ModuleOfID(resource.Identity()); ours {
|
||||
gated[module] = true
|
||||
gatedWhy[module] = "a step this module declares did not complete"
|
||||
log(fmt.Sprintf(" gated %s.*: a step it declares did not complete, so the rest "+
|
||||
@@ -979,6 +997,7 @@ func ApplyMindingWindows(
|
||||
failed.Stopped = true
|
||||
failed.Done = report
|
||||
failed.Others = len(failures) - 1
|
||||
failed.Failures = failuresOf(failures)
|
||||
failed.Modules = gatedModules()
|
||||
return report, known, failed
|
||||
}
|
||||
@@ -1162,6 +1181,7 @@ func ApplyMindingWindows(
|
||||
first := failures[0]
|
||||
first.Done = report
|
||||
first.Others = len(failures) - 1
|
||||
first.Failures = failuresOf(failures)
|
||||
first.Modules = gatedModules()
|
||||
return report, known, first
|
||||
}
|
||||
@@ -3696,16 +3716,10 @@ func unitKey(scope, user, unit string) string {
|
||||
// False for what the mesh declares in its own right: the foundation's resources carry no dot at all,
|
||||
// and the adoption's are named for the mesh rather than for a module. Both belong to no module, and
|
||||
// their gate is therefore the machine's.
|
||||
func moduleOf(identity string) (string, bool) {
|
||||
if strings.HasPrefix(identity, declaration.AdoptionPrefix) {
|
||||
return "", false
|
||||
}
|
||||
at := strings.LastIndex(identity, ".")
|
||||
if at <= 0 {
|
||||
return "", false
|
||||
}
|
||||
return identity[:at], true
|
||||
}
|
||||
//
|
||||
// This is the guess from an id alone, for a caller with no declaration in hand; a caller with one asks
|
||||
// Declaration.ModuleOfID, which reads the module the controller said (issue 1132).
|
||||
func moduleOf(identity string) (string, bool) { return declaration.ModuleGuess(identity) }
|
||||
|
||||
// handOver removes what a process replaces, once the process is running and still is a moment later
|
||||
// (novox/hq issue 213, ADR 0184). A replacement that is not up keeps what it replaces in place
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package apply
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-host/internal/declaration"
|
||||
)
|
||||
|
||||
// failuresOf is every failure of an apply as the error carries it: the first among them, then the rest, so a module
|
||||
// whose resource failed second is still named (issue 1132, and issue 500, which counted the rest and named none).
|
||||
func failuresOf(failures []*Error) []Failure {
|
||||
out := make([]Failure, 0, len(failures))
|
||||
for _, f := range failures {
|
||||
out = append(out, Failure{Resource: f.Resource, Err: f.Err})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ModuleResults groups an apply's outcome per module the declaration names (novox/hq ADR 0429 §4, issue 1132): the
|
||||
// module's resources all applied, or it failed and Detail says which resource failed with its error, which resources
|
||||
// were skipped behind a step that did not complete, or that its resources were not attempted because a step of no
|
||||
// module stopped the apply. A module is named by what the declaration says of each resource (ModuleOfID), so one
|
||||
// whose id would guess another is still its own.
|
||||
//
|
||||
// A failure that is no resource's (an error that is not an Error, or a resource of no module) belongs to no module and
|
||||
// changes none: the node says it in its report. The report is the outcome of an apply that did not fail; an apply that
|
||||
// failed carries its own account in the Error, and that is read in its place.
|
||||
func ModuleResults(d *declaration.Declaration, report Report, err error) map[string]ModuleResult {
|
||||
if d == nil {
|
||||
return nil
|
||||
}
|
||||
out := map[string]ModuleResult{}
|
||||
attempted := map[string]bool{}
|
||||
var failed *Error
|
||||
errors.As(err, &failed)
|
||||
outcomes := report.Outcomes
|
||||
if failed != nil {
|
||||
outcomes = append(append([]Outcome(nil), outcomes...), failed.Done.Outcomes...)
|
||||
}
|
||||
skipped := map[string]string{}
|
||||
for _, o := range outcomes {
|
||||
module, ok := d.ModuleOfID(o.ID)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if o.Action == "skipped" {
|
||||
skipped[module] = o.Detail
|
||||
continue
|
||||
}
|
||||
attempted[o.ID] = true
|
||||
}
|
||||
var reasons = map[string][]string{}
|
||||
for _, r := range d.Resources {
|
||||
if module, ok := d.ModuleOfID(r.Identity()); ok {
|
||||
out[module] = ModuleResult{}
|
||||
if failed != nil && failed.Stopped && !attempted[r.Identity()] {
|
||||
reasons[module] = append(reasons[module], fmt.Sprintf("%s was not attempted: %s stopped the apply", r.Identity(), failed.Resource))
|
||||
}
|
||||
}
|
||||
}
|
||||
if failed != nil {
|
||||
for _, f := range failed.Failures {
|
||||
if module, ok := d.ModuleOfID(f.Resource); ok {
|
||||
if _, named := out[module]; named {
|
||||
reasons[module] = append(reasons[module], fmt.Sprintf("%s failed: %v", f.Resource, f.Err))
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(failed.Failures) == 0 {
|
||||
if module, ok := d.ModuleOfID(failed.Resource); ok {
|
||||
if _, named := out[module]; named {
|
||||
reasons[module] = append(reasons[module], fmt.Sprintf("%s failed: %v", failed.Resource, failed.Err))
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, module := range failed.Modules {
|
||||
if _, named := out[module]; named {
|
||||
why := skipped[module]
|
||||
if why == "" {
|
||||
why = "a step it declares did not complete"
|
||||
}
|
||||
reasons[module] = append(reasons[module], "gated: "+why)
|
||||
}
|
||||
}
|
||||
}
|
||||
for module, why := range skipped {
|
||||
if _, named := out[module]; named && len(reasons[module]) == 0 {
|
||||
reasons[module] = append(reasons[module], "gated: "+why)
|
||||
}
|
||||
}
|
||||
for module, why := range reasons {
|
||||
sort.Strings(why)
|
||||
out[module] = ModuleResult{Failed: true, Detail: strings.Join(why, "; ")}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package apply
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-host/internal/declaration"
|
||||
)
|
||||
|
||||
func twoModules(t *testing.T) *declaration.Declaration {
|
||||
t.Helper()
|
||||
d, err := declaration.Parse([]byte(`{"declaration":1,"builds":{"a":"b-1","b":"b-2"},"resources":[
|
||||
{"id":"a.dir","type":"directory","path":"/srv/a","module":"a"},
|
||||
{"id":"b.dir","type":"directory","path":"/srv/b","module":"b"},
|
||||
{"id":"b.conf","type":"file","path":"/srv/b/c","content":"x","module":"b"},
|
||||
{"id":"c.dir","type":"directory","path":"/srv/c","module":"c"}]}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// 1132/AC3: every resource of module A applied and one of module B failed: A is a success, B failed with the failing
|
||||
// resource and its error named; a module gated by another's failure is failed too.
|
||||
func TestAnApplyIsGroupedPerModule(t *testing.T) {
|
||||
d := twoModules(t)
|
||||
done := Report{Outcomes: []Outcome{
|
||||
{ID: "a.dir", Action: "created"},
|
||||
{ID: "c.dir", Action: "skipped", Detail: "a step this module declares did not complete"},
|
||||
}}
|
||||
err := &Error{Resource: "b.dir", Err: errors.New("no space left"), Done: done, Others: 1,
|
||||
Failures: []Failure{{Resource: "b.dir", Err: errors.New("no space left")}, {Resource: "b.conf", Err: errors.New("denied")}},
|
||||
Modules: []string{"c"}}
|
||||
got := ModuleResults(d, Report{}, err)
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("modules %v, want a, b and c", got)
|
||||
}
|
||||
if got["a"].Failed || got["a"].Detail != "" {
|
||||
t.Errorf("a: %+v, want a clean success", got["a"])
|
||||
}
|
||||
b := got["b"]
|
||||
if !b.Failed || !strings.Contains(b.Detail, "b.dir") || !strings.Contains(b.Detail, "no space left") ||
|
||||
!strings.Contains(b.Detail, "b.conf") || !strings.Contains(b.Detail, "denied") {
|
||||
t.Errorf("b: %+v, want both failing resources and their errors", b)
|
||||
}
|
||||
c := got["c"]
|
||||
if !c.Failed || !strings.Contains(c.Detail, "did not complete") {
|
||||
t.Errorf("c: %+v, want it failed as gated, with why", c)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC3: an apply that did not fail is a success for every module the declaration names.
|
||||
func TestACleanApplyIsASuccessForEveryModule(t *testing.T) {
|
||||
got := ModuleResults(twoModules(t), Report{}, nil)
|
||||
for _, m := range []string{"a", "b", "c"} {
|
||||
if r, ok := got[m]; !ok || r.Failed {
|
||||
t.Errorf("%s: %+v (%v), want a success", m, r, ok)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC3: a step that belongs to no module stops the apply there; a module with a resource not attempted is not a
|
||||
// success.
|
||||
func TestAModuleNotAttemptedIsNotASuccess(t *testing.T) {
|
||||
d := twoModules(t)
|
||||
err := &Error{Resource: "boot", Err: errors.New("boom"), Stopped: true, Failures: []Failure{{Resource: "boot", Err: errors.New("boom")}},
|
||||
Done: Report{Outcomes: []Outcome{{ID: "a.dir", Action: "created"}}}}
|
||||
got := ModuleResults(d, Report{}, err)
|
||||
if got["a"].Failed {
|
||||
t.Errorf("a was done: %+v", got["a"])
|
||||
}
|
||||
if !got["b"].Failed || !strings.Contains(got["b"].Detail, "boot") {
|
||||
t.Errorf("b was not attempted: %+v", got["b"])
|
||||
}
|
||||
}
|
||||
@@ -350,7 +350,7 @@ func moveOutPass(d *declaration.Declaration, known store.State, applied map[stri
|
||||
if unknown != declaration.UnknownMoveOut || !applied[r.Identity()] {
|
||||
continue
|
||||
}
|
||||
module, _ := moduleOf(r.Identity())
|
||||
module, _ := d.ModuleOfID(r.Identity())
|
||||
if module == "" {
|
||||
module = r.Identity()
|
||||
}
|
||||
|
||||
@@ -148,21 +148,26 @@ func TestAForgeManifestTheCatalogueDoesNotHaveIsNamed(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// theCataloguesModule is the catalogue's own manifest of a module, never a copy: the clone beside a merge
|
||||
// check (MESH_CHECK_BESIDE), where it must be; else a checkout beside this one; else the test skips aloud.
|
||||
// theCataloguesModule is a module's own manifest, never a copy: the clone beside a merge check (MESH_CHECK_BESIDE), where it
|
||||
// must be; else a checkout beside this one; else the test skips aloud. The catalogue holds it, but for the build agent,
|
||||
// which moved whole into the controller's repository (ADR 0420, issue 1019, issue 1151) and is read from there.
|
||||
func theCataloguesModule(t *testing.T, module string) []byte {
|
||||
t.Helper()
|
||||
inside := filepath.Join("mesh-catalog", "modules", module, "module.json")
|
||||
repository := "mesh-catalog"
|
||||
if module == BuilderModule {
|
||||
repository = "mesh-controller"
|
||||
}
|
||||
inside := filepath.Join(repository, "modules", module, "module.json")
|
||||
if root := os.Getenv("MESH_CHECK_BESIDE"); root != "" {
|
||||
raw, err := os.ReadFile(filepath.Join(root, inside))
|
||||
if err != nil {
|
||||
t.Fatalf("the catalogue beside this check (MESH_CHECK_BESIDE=%s) has no %s: %v", root, inside, err)
|
||||
t.Fatalf("%s beside this check (MESH_CHECK_BESIDE=%s) has no %s: %v", repository, root, inside, err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
raw, err := os.ReadFile(filepath.Join("..", "..", "..", inside))
|
||||
if err != nil {
|
||||
t.Skipf("NOT JUDGED: no catalogue beside this checkout and no MESH_CHECK_BESIDE; the merge check judges it")
|
||||
t.Skipf("NOT JUDGED: no %s beside this checkout and no MESH_CHECK_BESIDE; the merge check judges it", repository)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
@@ -217,3 +222,29 @@ func TestABuildAgentManifestWithNowhereForTheCarriedBuilderIsRefused(t *testing.
|
||||
t.Fatalf("a manifest with no image artifact was pinned: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1151: the build agent's manifest lives in the controller's repository now (ADR 0420, issue 1019), and a merge check of
|
||||
// this repository has the controller cloned beside it, so a tree shaped like that check's, whose catalogue no longer holds
|
||||
// the build agent, is read from the controller.
|
||||
func TestTheBuildAgentsManifestIsReadFromTheControllerBesideTheCheck(t *testing.T) {
|
||||
beside := t.TempDir()
|
||||
for path, body := range map[string]string{
|
||||
"mesh-catalog/modules/nats/module.json": `{"module":"nats"}`,
|
||||
"mesh-controller/modules/build-agent/module.json": `{"module":"build-agent"}`,
|
||||
} {
|
||||
full := filepath.Join(beside, path)
|
||||
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(full, []byte(body), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
t.Setenv("MESH_CHECK_BESIDE", beside)
|
||||
if got := string(theCataloguesModule(t, BuilderModule)); got != `{"module":"build-agent"}` {
|
||||
t.Errorf("read %s", got)
|
||||
}
|
||||
if got := string(theCataloguesModule(t, "nats")); got != `{"module":"nats"}` {
|
||||
t.Errorf("a module the catalogue holds was read as %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -103,6 +103,8 @@ type Resource interface {
|
||||
// and not a hash of the content: it is what lets the store say *this is the same resource
|
||||
// I applied last time*, which is what makes removal possible at all.
|
||||
Identity() string
|
||||
// DeclaredModule is the module the controller says the resource belongs to, or "" (issue 1132).
|
||||
DeclaredModule() string
|
||||
// Kind is the resource's type, for the store and for reporting.
|
||||
Kind() Type
|
||||
// Target is what the resource acts on, for a person reading a report.
|
||||
@@ -115,12 +117,30 @@ type Resource interface {
|
||||
// strict decoder would otherwise refuse. `Kind()` returns the constant and is what anything
|
||||
// else should read.
|
||||
|
||||
// DeclaredModule is on every kind: the module the controller says the resource belongs to, or "" when it does not say
|
||||
// (an older controller). Read through Declaration.ModuleOfID, which falls back to the guess.
|
||||
func (d *Directory) DeclaredModule() string { return d.Module }
|
||||
func (f *File) DeclaredModule() string { return f.Module }
|
||||
func (u *User) DeclaredModule() string { return u.Module }
|
||||
func (n *Network) DeclaredModule() string { return n.Module }
|
||||
func (a *Access) DeclaredModule() string { return a.Module }
|
||||
func (a *Archive) DeclaredModule() string { return a.Module }
|
||||
func (d *Process) DeclaredModule() string { return d.Module }
|
||||
func (s *Service) DeclaredModule() string { return s.Module }
|
||||
func (o *Opening) DeclaredModule() string { return o.Module }
|
||||
func (p *Package) DeclaredModule() string { return p.Module }
|
||||
func (c *Container) DeclaredModule() string { return c.Module }
|
||||
func (a *Action) DeclaredModule() string { return a.Module }
|
||||
|
||||
// Directory is a directory that should exist, with a mode.
|
||||
type Directory struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Path string `json:"path"`
|
||||
Mode string `json:"mode,omitempty"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Path string `json:"path"`
|
||||
Mode string `json:"mode,omitempty"`
|
||||
// Owner is the user this belongs to, by name. Absent means root, which is what everything
|
||||
// managed was until users existed.
|
||||
Owner string `json:"owner,omitempty"`
|
||||
@@ -201,8 +221,11 @@ func (d *Directory) validate(where string, _ bool) []string {
|
||||
|
||||
// File is a file with literal content. The host renders nothing.
|
||||
type File struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Path string `json:"path"`
|
||||
Content string `json:"content"`
|
||||
Mode string `json:"mode,omitempty"`
|
||||
@@ -520,7 +543,10 @@ func (f *File) checkOwnership(where string) []string {
|
||||
type User struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Name string `json:"name"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Name string `json:"name"`
|
||||
|
||||
// Shell this user logs in with. Absent means the host asserts nothing and leaves whatever is
|
||||
// there — the same rule Service.Boot follows, for the same reason: a field that always
|
||||
@@ -571,7 +597,10 @@ const RootNever = "never"
|
||||
type Network struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Name string `json:"name"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
func (n *Network) Identity() string { return n.ID }
|
||||
@@ -616,7 +645,10 @@ const (
|
||||
type Access struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Path string `json:"path"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Path string `json:"path"`
|
||||
// Mode is how this module reaches the path: read or read-write. Absent narrows to read.
|
||||
Mode string `json:"mode,omitempty"`
|
||||
|
||||
@@ -743,6 +775,9 @@ func (u *User) validate(where string, _ bool) []string {
|
||||
type Archive struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
// Source is where to fetch it from.
|
||||
Source string `json:"source"`
|
||||
// Digest is sha256 of the archive, as "sha256:<hex>".
|
||||
@@ -796,6 +831,9 @@ func (a *Archive) validate(where string, _ bool) []string {
|
||||
type Process struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
// Name is what the unit is called, and what an operator will see in the process table.
|
||||
Name string `json:"name"`
|
||||
// Source is where to fetch the bundle from, and Digest is what it must hash to. The same
|
||||
@@ -985,7 +1023,10 @@ func (d *Process) validate(where string, _ bool) []string {
|
||||
type Service struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Unit string `json:"unit"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Unit string `json:"unit"`
|
||||
// State is "running" or "stopped" — or absent, and then **the unit's lifecycle is the
|
||||
// machine's; the mesh only reflects its triggers** (novox/hq ADR 0117). The uplink modules
|
||||
// declare the machine's own network manager this way: the mesh writes into its configuration
|
||||
@@ -1157,8 +1198,11 @@ func BlockMarkers(id string) (begin, end string) {
|
||||
// firewall sees a published port after the runtime has translated it, so a forwarded opening names
|
||||
// the container's own port in To as well as the machine's in Port.
|
||||
type Opening struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Port int `json:"port"`
|
||||
Protocol string `json:"protocol"`
|
||||
From string `json:"from"`
|
||||
@@ -1223,8 +1267,11 @@ func (o *Opening) validate(where string, _ bool) []string {
|
||||
// Present is the whole of what it asserts, never a version: version is the package manager's
|
||||
// business and the mesh does not hold a second opinion about it.
|
||||
type Package struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Package string `json:"package"`
|
||||
// Absent declares that the package is NOT installed (novox/hq ADR 0180): the host removes it
|
||||
// when it is, and leaves a machine that never had it alone. For the one case a module replaces
|
||||
@@ -1249,7 +1296,10 @@ func (p *Package) validate(where string, _ bool) []string {
|
||||
type Container struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
Name string `json:"name"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Name string `json:"name"`
|
||||
// Image is pinned by digest (novox/hq ADR 0006) — a tag moves and a digest does not.
|
||||
Image string `json:"image"`
|
||||
Env map[string]string `json:"env,omitempty"`
|
||||
@@ -1555,8 +1605,11 @@ func (c *Container) validate(where string, _ bool) []string {
|
||||
|
||||
// Action runs something the bundle declared, and the host never learns what it means.
|
||||
type Action struct {
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Type Type `json:"type"`
|
||||
// Module is the module this resource belongs to, as the controller says it (novox/hq ADR 0429 §4, issue 1132). Absent
|
||||
// is an older controller, and the module is then guessed from the id (ModuleGuess).
|
||||
Module string `json:"module,omitempty"`
|
||||
Command []string `json:"command"`
|
||||
// Verify is not optional and is not a courtesy. It is the read-back AND the idempotency
|
||||
// check: the host does not know what a database is, so "is it already there" is a question
|
||||
@@ -1727,6 +1780,58 @@ type Declaration struct {
|
||||
// the only account whose write of a filled file the node-engine's local write socket accepts. Absent is a
|
||||
// controller that names none, and every such write is refused.
|
||||
Provisioner string
|
||||
|
||||
// Builds is the build id of every module this declaration names, by module (novox/hq ADR 0429 §4, issue 1132).
|
||||
Builds map[string]string
|
||||
|
||||
// declared is what each resource says of its module, by id; nil on a declaration built by hand.
|
||||
declared map[string]string
|
||||
}
|
||||
|
||||
// ModuleGuess is the module a resource id names when the declaration does not say: everything before its last dot.
|
||||
//
|
||||
// The mesh composes a module's resource ids as `<module>.<its own id>`, and **a module's name may contain a dot**
|
||||
// (`novox.be` is one on this mesh) while a resource's own id never does, so the owner is everything before the *last*
|
||||
// dot. False for what the mesh declares in its own right: the foundation's resources carry no dot, and the
|
||||
// adoption's are named for the mesh. This is the guess an older controller's declaration is read by; a declaration
|
||||
// that says `module` is read by ModuleOfID, which never guesses for a resource that says (issue 1132).
|
||||
func ModuleGuess(id string) (string, bool) {
|
||||
if strings.HasPrefix(id, AdoptionPrefix) {
|
||||
return "", false
|
||||
}
|
||||
at := strings.LastIndex(id, ".")
|
||||
if at <= 0 {
|
||||
return "", false
|
||||
}
|
||||
return id[:at], true
|
||||
}
|
||||
|
||||
// ModuleOfID is the module a resource of this declaration belongs to: the module it says (novox/hq ADR 0429 §4,
|
||||
// issue 1132), and otherwise the guess from its id (ModuleGuess), so a declaration from an older controller reads as
|
||||
// it always did. False for what belongs to no module.
|
||||
func (d *Declaration) ModuleOfID(id string) (string, bool) {
|
||||
if d != nil {
|
||||
if d.declared == nil {
|
||||
for _, r := range d.Resources {
|
||||
if r.Identity() == id && r.DeclaredModule() != "" {
|
||||
return r.DeclaredModule(), true
|
||||
}
|
||||
}
|
||||
} else if m, ok := d.declared[id]; ok {
|
||||
return m, true
|
||||
}
|
||||
}
|
||||
return ModuleGuess(id)
|
||||
}
|
||||
|
||||
// indexModules keeps what each resource says of its module, so ModuleOfID is a lookup.
|
||||
func (d *Declaration) indexModules() {
|
||||
d.declared = map[string]string{}
|
||||
for _, r := range d.Resources {
|
||||
if m := r.DeclaredModule(); m != "" {
|
||||
d.declared[r.Identity()] = m
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkProvisioner refuses a provisioner client's account that is no account name, or is root: the account whose
|
||||
@@ -2126,6 +2231,10 @@ type envelope struct {
|
||||
// Provisioner is optional too (novox/hq ADR 0349 §4). **An older host refuses this key**; a controller sends it
|
||||
// only to a host whose health says link.ProvisionerContract.
|
||||
Provisioner string `json:"provisioner_account,omitempty"`
|
||||
// Builds is optional too (novox/hq ADR 0429 §4, issue 1132): the build id of each module the declaration names, and
|
||||
// every resource's `module`. **An older host refuses these keys**; a controller sends them only to a host whose
|
||||
// reports say `reads_modules`.
|
||||
Builds map[string]string `json:"builds,omitempty"`
|
||||
}
|
||||
|
||||
func parse(raw []byte, allowActions bool) (*Declaration, error) {
|
||||
@@ -2145,7 +2254,7 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) {
|
||||
d := &Declaration{Version: env.Version, For: env.For, Adoption: env.Adoption, Sequence: env.Sequence,
|
||||
Epoch: env.Epoch, LeftOut: env.LeftOut, Generation: env.Generation, PutBack: env.PutBack,
|
||||
Identities: env.Identities, Groups: env.Groups, Libraries: env.Libraries, Filled: env.Filled,
|
||||
Provisioner: env.Provisioner}
|
||||
Provisioner: env.Provisioner, Builds: env.Builds}
|
||||
problems := checkIdentities(env.Identities, env.Groups)
|
||||
problems = append(problems, checkProvisioner(env.Provisioner)...)
|
||||
problems = append(problems, checkLibraries(env.Groups, env.Libraries)...)
|
||||
@@ -2264,6 +2373,7 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) {
|
||||
if len(problems) > 0 {
|
||||
return nil, &RefusalError{Problems: problems}
|
||||
}
|
||||
d.indexModules()
|
||||
return d, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
package declaration
|
||||
|
||||
import "testing"
|
||||
|
||||
// 1132/AC1: a resource says its module, and the declaration says each module's build; a resource that does
|
||||
// not say gets today's guess, so an older controller still works.
|
||||
func TestAResourceSaysItsModuleAndTheDeclarationItsBuilds(t *testing.T) {
|
||||
d, err := Parse([]byte(`{"declaration":1,
|
||||
"builds":{"novox.be":"b-9","web":"b-4"},
|
||||
"resources":[
|
||||
{"id":"novox.be.data","type":"directory","path":"/srv/a","module":"novox.be"},
|
||||
{"id":"web.conf","type":"file","path":"/etc/w","content":"x","module":"web"},
|
||||
{"id":"old.dir","type":"directory","path":"/srv/old"},
|
||||
{"id":"base","type":"directory","path":"/srv/base"}]}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for id, want := range map[string]string{"novox.be.data": "novox.be", "web.conf": "web", "old.dir": "old"} {
|
||||
got, ok := d.ModuleOfID(id)
|
||||
if !ok || got != want {
|
||||
t.Errorf("%s belongs to %q (%v), want %q", id, got, ok, want)
|
||||
}
|
||||
}
|
||||
if m, ok := d.ModuleOfID("base"); ok {
|
||||
t.Errorf("a resource with no module and no dot belongs to %q, want none", m)
|
||||
}
|
||||
if d.Builds["web"] != "b-4" || d.Builds["novox.be"] != "b-9" || len(d.Builds) != 2 {
|
||||
t.Errorf("builds read as %v", d.Builds)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC1: the module the declaration gives wins over the guess, even when the id would guess another.
|
||||
func TestTheDeclaredModuleWinsOverTheGuess(t *testing.T) {
|
||||
d, err := Parse([]byte(`{"declaration":1,"resources":[
|
||||
{"id":"a.b.c","type":"directory","path":"/srv/a","module":"a.b"},
|
||||
{"id":"x.y","type":"directory","path":"/srv/x","module":"other"}]}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m, _ := d.ModuleOfID("x.y"); m != "other" {
|
||||
t.Errorf("x.y belongs to %q, want the declared %q", m, "other")
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC1: every kind accepts the field.
|
||||
func TestEveryKindAcceptsAModule(t *testing.T) {
|
||||
for _, typ := range []Type{TypeDirectory, TypeFile, TypeUser, TypeNetwork, TypeAccess, TypeArchive, TypeProcess,
|
||||
TypeService, TypeOpening, TypePackage, TypeContainer, TypeAction} {
|
||||
r := newOf(typ)
|
||||
if r == nil {
|
||||
t.Fatalf("no kind %s", typ)
|
||||
}
|
||||
if _, ok := r.(interface{ DeclaredModule() string }); !ok {
|
||||
t.Errorf("%s does not say its module", typ)
|
||||
}
|
||||
if unknown := unknownFields([]byte(`{"id":"m.x","type":"`+string(typ)+`","module":"m"}`), r); len(unknown) != 0 {
|
||||
t.Errorf("%s refuses %v", typ, unknown)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2,9 +2,11 @@ package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
@@ -115,3 +117,56 @@ func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error {
|
||||
defer cancel()
|
||||
return b.Conn.FlushWithContext(flush)
|
||||
}
|
||||
|
||||
// HaveBind binds the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132), once for a pass of writes.
|
||||
// **Only binds**: a node-engine's account reaches no part of the JetStream API beyond what it is granted, and the bucket
|
||||
// is the controller's to create and to configure, so a bucket that is not there is an error said as what it is. Every
|
||||
// call on the bucket waits no longer than ctx has left: a refused publish is answered by nothing, and the client's own
|
||||
// wait is five seconds.
|
||||
func (b OverNATS) HaveBind(ctx context.Context) (HaveBucket, error) {
|
||||
js := b.JS
|
||||
wait := 5 * time.Second
|
||||
if at, ok := ctx.Deadline(); ok {
|
||||
wait = time.Until(at)
|
||||
if wait <= 0 {
|
||||
return nil, context.DeadlineExceeded
|
||||
}
|
||||
}
|
||||
if b.Conn != nil {
|
||||
bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
js = bounded
|
||||
}
|
||||
kv, err := js.KeyValue(delivery.HaveBucket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err)
|
||||
}
|
||||
return natsHaveBucket{kv}, nil
|
||||
}
|
||||
|
||||
type natsHaveBucket struct{ kv nats.KeyValue }
|
||||
|
||||
func (n natsHaveBucket) Get(_ context.Context, key string) ([]byte, uint64, error) {
|
||||
entry, err := n.kv.Get(key)
|
||||
if errors.Is(err, nats.ErrKeyNotFound) {
|
||||
return nil, 0, ErrHaveAbsent
|
||||
}
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return entry.Value(), entry.Revision(), nil
|
||||
}
|
||||
|
||||
func (n natsHaveBucket) Set(_ context.Context, key string, value []byte, revision uint64) error {
|
||||
var err error
|
||||
if revision == 0 {
|
||||
_, err = n.kv.Create(key, value)
|
||||
return err
|
||||
}
|
||||
_, err = n.kv.Update(key, value, revision)
|
||||
return err
|
||||
}
|
||||
|
||||
var _ HaveBus = OverNATS{}
|
||||
|
||||
@@ -0,0 +1,242 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold.
|
||||
var ErrHaveAbsent = errors.New("link: no such key")
|
||||
|
||||
// HaveBus is a link that can bind the have bucket (novox/hq ADR 0429 §4). Separate from Bus, as HealthBus is: a link that
|
||||
// cannot is still a link, and the write waits for the next apply or look.
|
||||
type HaveBus interface {
|
||||
// HaveBind binds the bucket the controller made, once for a pass of writes, within ctx. A node whose grant is not there
|
||||
// yet gets an error here (or the context's end), and writes nothing.
|
||||
HaveBind(ctx context.Context) (HaveBucket, error)
|
||||
}
|
||||
|
||||
// HaveBucket is the bound bucket.
|
||||
type HaveBucket interface {
|
||||
// Get is a key's value and last revision, or ErrHaveAbsent.
|
||||
Get(ctx context.Context, key string) (value []byte, revision uint64, err error)
|
||||
// Set writes a key over the revision read: zero creates it, and a key that moved since is refused.
|
||||
Set(ctx context.Context, key string, value []byte, revision uint64) error
|
||||
}
|
||||
|
||||
// haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health
|
||||
// as the last look found it — kept so a look that changes only a health can write only that module's have (issue 1132).
|
||||
type haveState struct {
|
||||
mu sync.Mutex
|
||||
modules map[string]ModuleFact
|
||||
health map[string]delivery.Health
|
||||
node *delivery.HaveNode
|
||||
}
|
||||
|
||||
// moduleHealths is the health of each module a look names, from the states of its resources: failed when any is
|
||||
// unhealthy, waits when any waits (waiting, starting, held), ok when all are healthy, unknown otherwise.
|
||||
func moduleHealths(h *Health) map[string]delivery.Health {
|
||||
out := map[string]delivery.Health{}
|
||||
if h == nil {
|
||||
return out
|
||||
}
|
||||
rank := map[string]int{}
|
||||
for _, r := range h.Resources {
|
||||
if r.Module == "" {
|
||||
continue
|
||||
}
|
||||
var this int
|
||||
switch r.State {
|
||||
case StateUnhealthy:
|
||||
this = 3
|
||||
case StateWaiting, StateStarting, StateHeld:
|
||||
this = 2
|
||||
case StateHealthy:
|
||||
this = 0
|
||||
default:
|
||||
this = 1
|
||||
}
|
||||
if was, seen := rank[r.Module]; !seen || this > was {
|
||||
rank[r.Module] = this
|
||||
}
|
||||
}
|
||||
for module, r := range rank {
|
||||
out[module] = [...]delivery.Health{delivery.HealthOK, delivery.HealthUnknown, delivery.HealthWaits, delivery.HealthFailed}[r]
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// nodeFacts is the node's own facts of a report, each as the JSON the report carries it in.
|
||||
func nodeFacts(r Report) delivery.HaveNode {
|
||||
n := delivery.HaveNode{V: delivery.Version}
|
||||
carry := func(into *json.RawMessage, empty bool, v any) {
|
||||
if empty {
|
||||
return
|
||||
}
|
||||
if raw, err := json.Marshal(v); err == nil {
|
||||
*into = raw
|
||||
}
|
||||
}
|
||||
carry(&n.Windows, len(r.Windows) == 0, r.Windows)
|
||||
carry(&n.Busy, len(r.Busy) == 0, r.Busy)
|
||||
carry(&n.Strays, len(r.Strays) == 0, r.Strays)
|
||||
carry(&n.Filters, len(r.Filters) == 0, r.Filters)
|
||||
carry(&n.Profile, len(r.Profile) == 0, r.Profile)
|
||||
return n
|
||||
}
|
||||
|
||||
// noteApply takes what an apply's report says of the modules and the node.
|
||||
func (q *Queue) noteApply(r Report) bool {
|
||||
if r.Modules == nil {
|
||||
return false
|
||||
}
|
||||
s := &q.haves
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.modules = map[string]ModuleFact{}
|
||||
for _, m := range r.Modules {
|
||||
s.modules[m.Module] = m
|
||||
}
|
||||
if r.Health != nil {
|
||||
s.health = moduleHealths(r.Health)
|
||||
}
|
||||
n := nodeFacts(r)
|
||||
s.node = &n
|
||||
return true
|
||||
}
|
||||
|
||||
// noteHealth takes a look's health, and says whether any module's changed.
|
||||
func (q *Queue) noteHealth(h Health) bool {
|
||||
s := &q.haves
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
next := moduleHealths(&h)
|
||||
changed := false
|
||||
for module := range s.modules {
|
||||
want, ok := next[module]
|
||||
if !ok {
|
||||
want = delivery.HealthUnknown
|
||||
}
|
||||
have, known := s.health[module]
|
||||
if !known {
|
||||
have = delivery.HealthUnknown
|
||||
}
|
||||
if have != want {
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
s.health = next
|
||||
return changed
|
||||
}
|
||||
|
||||
// sayHave is a line in the node-engine's log, if it has one.
|
||||
func (q *Queue) sayHave(s string) {
|
||||
if q.Say != nil {
|
||||
q.Say(s)
|
||||
}
|
||||
}
|
||||
|
||||
// writeHaves writes this node's keys in the have bucket where they differ from what the bucket holds (novox/hq ADR 0429
|
||||
// §4, issue 1132): each module's have, and the node's own facts. It reads the key first and writes nothing when the value
|
||||
// is the same, otherwise writes over the revision it read, and creates the key when there is none. A write the bucket
|
||||
// refuses is said and left: the next apply or look writes again, and the report was sent before this ran.
|
||||
func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
|
||||
hb, ok := bus.(HaveBus)
|
||||
if !ok || bus == nil {
|
||||
return
|
||||
}
|
||||
s := &q.haves
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.modules == nil && s.node == nil {
|
||||
return
|
||||
}
|
||||
node := q.Membership.Node
|
||||
// The whole pass is bounded by the queue's timeout, and it stops at the first error: a node whose grant is not there
|
||||
// yet would otherwise make the worker wait the bound again for every key (issue 1132).
|
||||
write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout())
|
||||
defer cancel()
|
||||
bucket, err := hb.HaveBind(write)
|
||||
if err != nil {
|
||||
q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err))
|
||||
return
|
||||
}
|
||||
modules := make([]string, 0, len(s.modules))
|
||||
for module := range s.modules {
|
||||
modules = append(modules, module)
|
||||
}
|
||||
sort.Strings(modules)
|
||||
for _, module := range modules {
|
||||
f := s.modules[module]
|
||||
h := delivery.Have{V: delivery.Version, Build: f.Build, Phase: delivery.PhaseRunning,
|
||||
Health: delivery.HealthUnknown, Detail: f.Detail}
|
||||
if f.Failed {
|
||||
h.Phase = delivery.PhaseFailed
|
||||
}
|
||||
if got, known := s.health[module]; known {
|
||||
h.Health = got
|
||||
}
|
||||
key, err := delivery.HaveKey(node, module)
|
||||
if err != nil {
|
||||
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
|
||||
continue
|
||||
}
|
||||
raw, err := delivery.EncodeHave(h)
|
||||
if err != nil {
|
||||
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
|
||||
continue
|
||||
}
|
||||
if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
|
||||
c, err := delivery.DecodeHave(current)
|
||||
return err == nil && delivery.SameHave(c, h)
|
||||
}) {
|
||||
return
|
||||
}
|
||||
}
|
||||
if s.node != nil {
|
||||
key, err := delivery.HaveNodeKey(node)
|
||||
if err == nil {
|
||||
var raw []byte
|
||||
if raw, err = delivery.EncodeHaveNode(*s.node); err == nil {
|
||||
want := *s.node
|
||||
q.putHave(write, bucket, key, raw, func(current []byte) bool {
|
||||
c, err := delivery.DecodeHaveNode(current)
|
||||
return err == nil && delivery.SameHaveNode(c, want)
|
||||
})
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the
|
||||
// bucket failed, the error is said once, and the rest wait for the next pass.
|
||||
func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool {
|
||||
if ctx.Err() != nil {
|
||||
q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
|
||||
return false
|
||||
}
|
||||
current, revision, err := b.Get(ctx, key)
|
||||
switch {
|
||||
case errors.Is(err, ErrHaveAbsent):
|
||||
revision = 0
|
||||
case err != nil:
|
||||
q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
|
||||
return false
|
||||
case same(current):
|
||||
return true
|
||||
}
|
||||
if err := b.Set(ctx, key, value, revision); err != nil {
|
||||
q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
//go:build integration
|
||||
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// 1132/AC8: against a real bus with the bucket node-engine_have, a node-engine that applies a declaration of two
|
||||
// modules writes <node>.<a>, <node>.<b> and <node>.node, and a second apply with nothing changed adds no revision.
|
||||
//
|
||||
// MESH_TEST_NATS=nats://127.0.0.1:4222 go test -tags integration ./internal/link/ -run TestNatsHavesAreWrittenOnceAndOnlyOnChange
|
||||
//
|
||||
// The bucket is made here by the test: in the mesh the controller makes it, and the node-engine only binds to it.
|
||||
func TestNatsHavesAreWrittenOnceAndOnlyOnChange(t *testing.T) {
|
||||
conn, js := aBus(t)
|
||||
const node = "haveok"
|
||||
kv, err := js.KeyValue(delivery.HaveBucket)
|
||||
if err != nil {
|
||||
if kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: delivery.HaveBucket, History: delivery.HaveHistory}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, name := range []string{"a", "b", "node"} {
|
||||
_ = kv.Purge(node + "." + name)
|
||||
}
|
||||
|
||||
m, key := aMember(t)
|
||||
m.Node = node
|
||||
var failB atomic.Bool
|
||||
var applied atomic.Int32
|
||||
failB.Store(true)
|
||||
apply := func(_ context.Context, raw, _ []byte) Report {
|
||||
applied.Add(1)
|
||||
var d struct{ Order }
|
||||
_ = json.Unmarshal(raw, &d)
|
||||
return Report{Declared: digest(raw), Order: d.Order, Applied: []string{"a.x", "b.x"},
|
||||
Modules: []ModuleFact{{Module: "a", Build: "b-1"},
|
||||
{Module: "b", Build: "b-2", Failed: failB.Load(), Detail: "b.x failed: no space"}},
|
||||
Health: healthOf(map[string]string{"a": StateHealthy, "b": StateHealthy}),
|
||||
Windows: []Window{{Step: "backup", Holds: []string{"db"}}}}
|
||||
}
|
||||
q := aQueue(m, apply, &keptInMemory{})
|
||||
q.attach(context.Background(), OverNATS{Conn: conn, JS: js})
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
worker := runWorker(ctx, q)
|
||||
defer func() { stop(); <-worker }()
|
||||
|
||||
revisions := func() map[string]uint64 {
|
||||
out := map[string]uint64{}
|
||||
for _, name := range []string{"a", "b", "node"} {
|
||||
if e, err := kv.Get(node + "." + name); err == nil {
|
||||
out[name] = e.Revision()
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
arrives := func(what string, id string, seq int64) {
|
||||
want := applied.Load() + 1
|
||||
q.Deliver(ordered(t, key, id, 7, seq))
|
||||
eventually(t, what, func() bool { return applied.Load() >= want })
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
arrives("the first apply", "one", 1)
|
||||
eventually(t, "the three keys were never written", func() bool { return len(revisions()) == 3 })
|
||||
first := revisions()
|
||||
|
||||
a, _ := kv.Get(node + ".a")
|
||||
haveA, err := delivery.DecodeHave(a.Value())
|
||||
if err != nil || haveA.Phase != delivery.PhaseRunning || haveA.Build != "b-1" || haveA.Health != delivery.HealthOK {
|
||||
t.Fatalf("a holds %+v (%v)", haveA, err)
|
||||
}
|
||||
b, _ := kv.Get(node + ".b")
|
||||
haveB, err := delivery.DecodeHave(b.Value())
|
||||
if err != nil || haveB.Phase != delivery.PhaseFailed || haveB.Detail == "" {
|
||||
t.Fatalf("b holds %+v (%v)", haveB, err)
|
||||
}
|
||||
n, _ := kv.Get(node + ".node")
|
||||
if hn, err := delivery.DecodeHaveNode(n.Value()); err != nil || len(hn.Windows) == 0 {
|
||||
t.Fatalf("the node key holds %+v (%v)", hn, err)
|
||||
}
|
||||
|
||||
arrives("the second apply", "two", 2)
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
for name, rev := range revisions() {
|
||||
if rev != first[name] {
|
||||
t.Errorf("%s moved from revision %d to %d with nothing changed", name, first[name], rev)
|
||||
}
|
||||
}
|
||||
|
||||
failB.Store(false)
|
||||
arrives("the third apply", "three", 3)
|
||||
eventually(t, "b was never updated", func() bool { return revisions()["b"] > first["b"] })
|
||||
if now := revisions(); now["a"] != first["a"] || now["node"] != first["node"] {
|
||||
t.Errorf("a key that did not change moved: %v from %v", now, first)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,341 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// What the node-engine writes beside its report: a module's have, and the node's own facts (novox/hq ADR 0429 §4,
|
||||
// issue 1132).
|
||||
|
||||
// haveLink is a link with a bucket behind it that keeps each key's revision as the bus does.
|
||||
type haveLink struct {
|
||||
*quietLink
|
||||
hmu sync.Mutex
|
||||
kv map[string]haveEntry
|
||||
sets []string // keys written, in order
|
||||
events []string // "report" and "have <key>", in the order they happened
|
||||
failing bool
|
||||
binds int
|
||||
setTries int
|
||||
bindFails, bindStuck bool
|
||||
}
|
||||
|
||||
type haveEntry struct {
|
||||
value []byte
|
||||
rev uint64
|
||||
}
|
||||
|
||||
func newHaveLink() *haveLink { return &haveLink{quietLink: newQuietLink(), kv: map[string]haveEntry{}} }
|
||||
|
||||
func (l *haveLink) Report(ctx context.Context, node string, body []byte) error {
|
||||
err := l.quietLink.Report(ctx, node, body)
|
||||
if err == nil {
|
||||
l.hmu.Lock()
|
||||
l.events = append(l.events, "report")
|
||||
l.hmu.Unlock()
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (l *haveLink) Health(context.Context, string, []byte) error { return nil }
|
||||
|
||||
func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) {
|
||||
l.hmu.Lock()
|
||||
l.binds++
|
||||
stuck, refuse := l.bindStuck, l.bindFails
|
||||
l.hmu.Unlock()
|
||||
if stuck {
|
||||
<-ctx.Done()
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
if refuse {
|
||||
return nil, errors.New("no grant for the have bucket")
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
func (l *haveLink) Get(_ context.Context, key string) ([]byte, uint64, error) {
|
||||
l.hmu.Lock()
|
||||
defer l.hmu.Unlock()
|
||||
e, ok := l.kv[key]
|
||||
if !ok {
|
||||
return nil, 0, ErrHaveAbsent
|
||||
}
|
||||
return e.value, e.rev, nil
|
||||
}
|
||||
|
||||
func (l *haveLink) Set(_ context.Context, key string, value []byte, revision uint64) error {
|
||||
l.hmu.Lock()
|
||||
defer l.hmu.Unlock()
|
||||
l.setTries++
|
||||
if l.failing {
|
||||
return errors.New("the bucket refused")
|
||||
}
|
||||
e, ok := l.kv[key]
|
||||
if (revision == 0) == ok || (ok && e.rev != revision) {
|
||||
return errors.New("wrong last sequence")
|
||||
}
|
||||
l.kv[key] = haveEntry{value: value, rev: e.rev + 1}
|
||||
l.sets = append(l.sets, key)
|
||||
l.events = append(l.events, "have "+key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *haveLink) have(t *testing.T, key string) delivery.Have {
|
||||
t.Helper()
|
||||
l.hmu.Lock()
|
||||
defer l.hmu.Unlock()
|
||||
h, err := delivery.DecodeHave(l.kv[key].value)
|
||||
if err != nil {
|
||||
t.Fatalf("%s holds no have: %v", key, err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func (l *haveLink) setCount() int {
|
||||
l.hmu.Lock()
|
||||
defer l.hmu.Unlock()
|
||||
return len(l.sets)
|
||||
}
|
||||
|
||||
func (l *haveLink) put(t *testing.T, key string, h delivery.Have) {
|
||||
t.Helper()
|
||||
raw, err := delivery.EncodeHave(h)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
l.hmu.Lock()
|
||||
defer l.hmu.Unlock()
|
||||
l.kv[key] = haveEntry{value: raw, rev: l.kv[key].rev + 1}
|
||||
}
|
||||
|
||||
func healthOf(states map[string]string) *Health {
|
||||
h := &Health{Contract: LivenessContract, At: time.Now().UTC()}
|
||||
for module, state := range states {
|
||||
h.Resources = append(h.Resources, ResourceHealth{Module: module, Resource: module + ".server", State: state})
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func appliedModules() Report {
|
||||
return Report{Declared: "d1", Applied: []string{"a.x", "b.x"},
|
||||
Modules: []ModuleFact{{Module: "a", Build: "b-1"}, {Module: "b", Build: "b-2", Failed: true, Detail: "b.x failed: no space"}},
|
||||
Health: healthOf(map[string]string{"a": StateHealthy, "b": StateUnhealthy})}
|
||||
}
|
||||
|
||||
func aHaveQueue(l *haveLink) (*Queue, *[]string) {
|
||||
var said []string
|
||||
var mu sync.Mutex
|
||||
q := &Queue{Membership: Membership{Node: "n1"}, Timeout: time.Second, Say: func(s string) {
|
||||
mu.Lock()
|
||||
said = append(said, s)
|
||||
mu.Unlock()
|
||||
}}
|
||||
q.attach(context.Background(), l)
|
||||
return q, &said
|
||||
}
|
||||
|
||||
// 1132/AC3, AC5: a module that applied is running with its build and a module that failed is failed with the detail;
|
||||
// the key does not exist yet, so it is created.
|
||||
func TestAnAppliedModuleIsWrittenRunningAndAFailedOneFailed(t *testing.T) {
|
||||
l := newHaveLink()
|
||||
q, _ := aHaveQueue(l)
|
||||
q.tell(context.Background(), appliedModules(), applied)
|
||||
a, b := l.have(t, "n1.a"), l.have(t, "n1.b")
|
||||
if a.Phase != delivery.PhaseRunning || a.Build != "b-1" || a.Health != delivery.HealthOK {
|
||||
t.Errorf("a: %+v", a)
|
||||
}
|
||||
if b.Phase != delivery.PhaseFailed || b.Build != "b-2" || !strings.Contains(b.Detail, "no space") ||
|
||||
b.Health != delivery.HealthFailed {
|
||||
t.Errorf("b: %+v", b)
|
||||
}
|
||||
if l.kv["n1.a"].rev != 1 {
|
||||
t.Errorf("a was created at revision %d", l.kv["n1.a"].rev)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC5: a have equal to the bucket's value is not written; a different one is written over the last revision.
|
||||
func TestAHaveIsWrittenOnlyWhenItDiffers(t *testing.T) {
|
||||
l := newHaveLink()
|
||||
l.put(t, "n1.a", delivery.Have{V: delivery.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
l.put(t, "n1.b", delivery.Have{V: delivery.Version, Build: "b-0", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
q, _ := aHaveQueue(l)
|
||||
q.tell(context.Background(), appliedModules(), applied)
|
||||
if l.kv["n1.a"].rev != 1 {
|
||||
t.Errorf("a equal to the bucket was written again, to revision %d", l.kv["n1.a"].rev)
|
||||
}
|
||||
if l.kv["n1.b"].rev != 2 || l.have(t, "n1.b").Phase != delivery.PhaseFailed {
|
||||
t.Errorf("b was not updated over its last revision: %+v at %d", l.have(t, "n1.b"), l.kv["n1.b"].rev)
|
||||
}
|
||||
before := l.setCount()
|
||||
q.tell(context.Background(), appliedModules(), applied)
|
||||
if l.setCount() != before {
|
||||
t.Errorf("an apply that changed nothing wrote %d keys", l.setCount()-before)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC4: a look's health reaches the module's have: ok, failed, waits or unknown, and only a change is written.
|
||||
func TestAModulesHealthFollowsTheLook(t *testing.T) {
|
||||
l := newHaveLink()
|
||||
q, _ := aHaveQueue(l)
|
||||
rep := Report{Declared: "d1", Modules: []ModuleFact{{Module: "a", Build: "b"}, {Module: "b", Build: "b"},
|
||||
{Module: "c", Build: "b"}, {Module: "d", Build: "b"}}}
|
||||
q.tell(context.Background(), rep, applied)
|
||||
for _, m := range []string{"a", "b", "c", "d"} {
|
||||
if h := l.have(t, "n1."+m).Health; h != delivery.HealthUnknown {
|
||||
t.Errorf("%s with no look is %s, want unknown", m, h)
|
||||
}
|
||||
}
|
||||
h := healthOf(map[string]string{"a": StateHealthy, "b": StateUnhealthy, "c": StateStarting})
|
||||
h.Resources = append(h.Resources, ResourceHealth{Module: "a", Resource: "a.two", State: StateWaiting},
|
||||
ResourceHealth{Module: "b", Resource: "b.two", State: StateHealthy}, ResourceHealth{Module: "d", Resource: "d.x", State: StateUnknown})
|
||||
if !q.SayHealth(context.Background(), *h) {
|
||||
t.Fatal("not said")
|
||||
}
|
||||
for m, want := range map[string]delivery.Health{"a": delivery.HealthWaits, "b": delivery.HealthFailed,
|
||||
"c": delivery.HealthWaits, "d": delivery.HealthUnknown} {
|
||||
if got := l.have(t, "n1."+m).Health; got != want {
|
||||
t.Errorf("%s is %s, want %s", m, got, want)
|
||||
}
|
||||
}
|
||||
before := l.setCount()
|
||||
h2 := *h
|
||||
q.SayHealth(context.Background(), h2)
|
||||
if l.setCount() != before {
|
||||
t.Errorf("the same look wrote %d keys", l.setCount()-before)
|
||||
}
|
||||
h2.Resources = []ResourceHealth{{Module: "a", Resource: "a.server", State: StateHealthy}}
|
||||
q.SayHealth(context.Background(), h2)
|
||||
if got := l.have(t, "n1.a").Health; got != delivery.HealthOK {
|
||||
t.Errorf("a after a healthy look is %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC6: the report is sent first and carries no new key; the node's key holds its windows, busy, strays, filters
|
||||
// and profile.
|
||||
func TestTheReportIsSentFirstAndTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
l := newHaveLink()
|
||||
q, _ := aHaveQueue(l)
|
||||
rep := appliedModules()
|
||||
rep.Windows = []Window{{Step: "backup", Holds: []string{"db"}}}
|
||||
rep.Busy = []Busy{{ID: "x", Container: "x", State: "waiting"}}
|
||||
rep.Strays = []Stray{{Kind: "container", Name: "old"}}
|
||||
rep.Filters = []Filter{{Where: "inet mesh", Owner: "mesh"}}
|
||||
rep.Profile = map[string]any{"cores": 8}
|
||||
q.tell(context.Background(), rep, applied)
|
||||
if len(l.events) < 2 || l.events[0] != "report" {
|
||||
t.Fatalf("the report was not first: %v", l.events)
|
||||
}
|
||||
raw, _ := json.Marshal(l.said()[0])
|
||||
if strings.Contains(string(raw), "Modules") || strings.Contains(string(raw), "\"modules\"") {
|
||||
t.Errorf("the report carries the module facts: %s", raw)
|
||||
}
|
||||
got, err := delivery.DecodeHaveNode(l.kv["n1.node"].value)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
enc := func(v any) json.RawMessage { b, _ := json.Marshal(v); return b }
|
||||
want := delivery.HaveNode{V: delivery.Version, Windows: enc(rep.Windows), Busy: enc(rep.Busy), Strays: enc(rep.Strays),
|
||||
Filters: enc(rep.Filters), Profile: enc(rep.Profile)}
|
||||
if !delivery.SameHaveNode(got, want) {
|
||||
t.Errorf("the node key holds %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC7: a write the bucket refuses is said in the log; the report went out, and the next apply writes again.
|
||||
func TestAFailedHaveWriteIsSaidAndNeverBlocksTheReport(t *testing.T) {
|
||||
l := newHaveLink()
|
||||
l.failing = true
|
||||
q, said := aHaveQueue(l)
|
||||
if !q.tell(context.Background(), appliedModules(), applied) {
|
||||
t.Fatal("the report did not go out")
|
||||
}
|
||||
if len(l.said()) != 1 {
|
||||
t.Fatalf("reports sent: %d", len(l.said()))
|
||||
}
|
||||
if !strings.Contains(strings.Join(*said, "\n"), "n1.a") || !strings.Contains(strings.Join(*said, "\n"), "refused") {
|
||||
t.Errorf("the failure is not in the log: %v", *said)
|
||||
}
|
||||
l.hmu.Lock()
|
||||
l.failing = false
|
||||
l.hmu.Unlock()
|
||||
q.tell(context.Background(), appliedModules(), applied)
|
||||
if l.have(t, "n1.a").Phase != delivery.PhaseRunning {
|
||||
t.Error("the next apply did not write again")
|
||||
}
|
||||
}
|
||||
|
||||
// A link without a bucket, and a report with no module facts (a refusal), write nothing and say nothing wrong.
|
||||
func TestNoBucketAndNoFactsWriteNothing(t *testing.T) {
|
||||
plain := newQuietLink()
|
||||
q := &Queue{Membership: Membership{Node: "n1"}, Timeout: time.Second}
|
||||
q.attach(context.Background(), plain)
|
||||
if !q.tell(context.Background(), appliedModules(), applied) {
|
||||
t.Fatal("the report did not go out")
|
||||
}
|
||||
l := newHaveLink()
|
||||
q2, _ := aHaveQueue(l)
|
||||
q2.tell(context.Background(), Report{Refused: "no"}, applied)
|
||||
if l.setCount() != 0 {
|
||||
t.Errorf("a refusal wrote %d keys", l.setCount())
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC5: a reconcile whose report is not news still writes the have keys that differ, so a node-engine that
|
||||
// restarted writes them again; and writes nothing when they are the same.
|
||||
func TestASilentReconcileStillWritesTheHaves(t *testing.T) {
|
||||
l := newHaveLink()
|
||||
q, _ := aHaveQueue(l)
|
||||
q.Reconcile = func(context.Context) (Report, bool) { return appliedModules(), true }
|
||||
q.reconcile(context.Background(), false)
|
||||
if len(l.said()) != 0 {
|
||||
t.Fatalf("a reconcile that is not news said %d reports", len(l.said()))
|
||||
}
|
||||
if l.have(t, "n1.a").Phase != delivery.PhaseRunning {
|
||||
t.Error("the reconcile wrote no have")
|
||||
}
|
||||
before := l.setCount()
|
||||
q.reconcile(context.Background(), false)
|
||||
if l.setCount() != before {
|
||||
t.Errorf("an unchanged reconcile wrote %d keys", l.setCount()-before)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC7: a bind that fails, or never answers, writes nothing and returns within the queue's timeout; a write that
|
||||
// fails ends the pass, and is said once.
|
||||
func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) {
|
||||
for name, set := range map[string]func(*haveLink){
|
||||
"a bind refused": func(l *haveLink) { l.bindFails = true },
|
||||
"a bind that never answers": func(l *haveLink) { l.bindStuck = true },
|
||||
} {
|
||||
l := newHaveLink()
|
||||
set(l)
|
||||
q, said := aHaveQueue(l)
|
||||
q.Timeout = 100 * time.Millisecond
|
||||
start := time.Now()
|
||||
if !q.tell(context.Background(), appliedModules(), applied) {
|
||||
t.Fatalf("%s: the report did not go out", name)
|
||||
}
|
||||
if took := time.Since(start); took > time.Second {
|
||||
t.Errorf("%s: the pass took %s", name, took)
|
||||
}
|
||||
if l.binds != 1 || l.setTries != 0 || len(*said) != 1 {
|
||||
t.Errorf("%s: binds %d, writes tried %d, said %v; want one bind, no write, one line", name, l.binds, l.setTries, *said)
|
||||
}
|
||||
}
|
||||
l := newHaveLink()
|
||||
l.failing = true
|
||||
q, said := aHaveQueue(l)
|
||||
q.tell(context.Background(), appliedModules(), applied)
|
||||
if l.binds != 1 || l.setTries != 1 || len(*said) != 1 {
|
||||
t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said)
|
||||
}
|
||||
}
|
||||
@@ -342,6 +342,10 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error {
|
||||
return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body)
|
||||
}
|
||||
|
||||
func (l *natsLink) HaveBind(ctx context.Context) (HaveBucket, error) {
|
||||
return OverNATS{Conn: l.conn, JS: l.js}.HaveBind(ctx)
|
||||
}
|
||||
|
||||
// Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the
|
||||
// node-engine holds is this one, not OverNATS: without it here the queue's ToolBus assertion failed on
|
||||
// every live link and each tool check read "no link to the bus is open" while the link was up.
|
||||
@@ -370,6 +374,7 @@ func (l *natsLink) Ask(ctx context.Context, subject string, body []byte) ([]byte
|
||||
var (
|
||||
_ Bus = (*natsLink)(nil)
|
||||
_ HealthBus = (*natsLink)(nil)
|
||||
_ HaveBus = (*natsLink)(nil)
|
||||
_ ToolBus = (*natsLink)(nil)
|
||||
_ Asker = (*natsLink)(nil)
|
||||
_ Asked = (*natsLink)(nil)
|
||||
|
||||
@@ -134,6 +134,11 @@ type Report struct {
|
||||
// know, whole. A node-engine rolled back says so no longer.
|
||||
ReadsGeneration bool `json:"reads_generation,omitempty"`
|
||||
|
||||
// ReadsModules says this node-engine reads a resource's `module` and a declaration's `builds` (novox/hq ADR 0429
|
||||
// §4, issue 1132), on every report it makes, beside ReadsGeneration: the controller sends them only to a machine
|
||||
// that said so, for the same reason.
|
||||
ReadsModules bool `json:"reads_modules,omitempty"`
|
||||
|
||||
// OlderGeneration is set on a report refusing a declaration composed from an older assignment
|
||||
// generation than this machine applied (novox/hq issue 234). The refused declaration is the report's
|
||||
// own Declared and Order, so the controller can name who sent it from its record of sends.
|
||||
@@ -196,6 +201,11 @@ type Report struct {
|
||||
// a network manager switched, reaches the mesh at the next push rather than never.
|
||||
Profile map[string]any `json:"profile,omitempty"`
|
||||
|
||||
// Modules is how each module of the declaration this report is about came out (novox/hq ADR 0429 §4, issue 1132). **Not
|
||||
// on the wire**: the report is sent as it always was, and the node-engine writes these beside it as the module's
|
||||
// have. Nil on a report that is not an apply's.
|
||||
Modules []ModuleFact `json:"-"`
|
||||
|
||||
// Host is the version of the host that produced this report (novox/hq ADR 0141).
|
||||
//
|
||||
// Without it nothing can say a machine is behind, so "every machine current with its source"
|
||||
@@ -851,3 +861,12 @@ type GenerationRefusal struct {
|
||||
Generation int64 `json:"generation"`
|
||||
Applied int64 `json:"applied"`
|
||||
}
|
||||
|
||||
// ModuleFact is how one module's part of an apply came out: the build the node was sent for it, and whether every
|
||||
// resource of it applied. Detail names the failing resource and its error (issue 1132).
|
||||
type ModuleFact struct {
|
||||
Module string
|
||||
Build string
|
||||
Failed bool
|
||||
Detail string
|
||||
}
|
||||
|
||||
@@ -93,6 +93,9 @@ type Queue struct {
|
||||
saying sync.Mutex
|
||||
counted bool
|
||||
numbers numbered
|
||||
|
||||
// haves is what this node-engine writes beside its reports (novox/hq ADR 0429 §4, issue 1132).
|
||||
haves haveState
|
||||
}
|
||||
|
||||
// Reconcile is the reconcile's act, run by the queue's worker. See Queue.Reconcile.
|
||||
@@ -265,6 +268,14 @@ func (q *Queue) reconcile(ctx context.Context, asked bool) {
|
||||
return
|
||||
}
|
||||
if !asked && (q.News == nil || !q.News(report)) {
|
||||
// Not worth a report, but a reconcile found how each module is now: the have keys say it (issue 1132), so a
|
||||
// node-engine that restarted writes them again without waiting for a change.
|
||||
if q.noteApply(report) {
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
q.writeHaves(ctx, bus)
|
||||
}
|
||||
return
|
||||
}
|
||||
if asked {
|
||||
@@ -324,6 +335,10 @@ func (q *Queue) tell(ctx context.Context, r Report, kind kindOf, settle ...Decla
|
||||
q.unasked = &r
|
||||
}
|
||||
if published {
|
||||
// After the report, which is sent first and unchanged; a write the bucket refuses never undoes it (issue 1132).
|
||||
if q.noteApply(r) {
|
||||
q.writeHaves(ctx, bus)
|
||||
}
|
||||
if keep && q.Unsaid != nil {
|
||||
if err := q.Unsaid.Said(r.Declared); err != nil {
|
||||
q.Say("said this apply's report, and cannot forget it: " + err.Error())
|
||||
@@ -354,6 +369,8 @@ func (q *Queue) stamp(r Report) Report {
|
||||
// Every report of a node-engine that orders its reports says it reads a generation (novox/hq issue
|
||||
// 234): this build does, and the controller sends one only to a machine that said so.
|
||||
r.ReadsGeneration = true
|
||||
// And the module of each resource and the build of each module (novox/hq ADR 0429 §4, issue 1132), the same way.
|
||||
r.ReadsModules = true
|
||||
return r
|
||||
}
|
||||
|
||||
@@ -538,6 +555,10 @@ func (q *Queue) SayHealth(ctx context.Context, h Health) bool {
|
||||
q.Say("could not say this machine's health: " + err.Error())
|
||||
return false
|
||||
}
|
||||
// A look that changed a module's health changes its have (issue 1132).
|
||||
if q.noteHealth(h) {
|
||||
q.writeHaves(ctx, bus)
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
|
||||
@@ -563,3 +563,26 @@ func TestEveryReportSaysItReadsAGeneration(t *testing.T) {
|
||||
t.Fatalf("the report does not say it reads a generation: %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC2: every report says this node-engine reads a resource's module and a declaration's builds, beside the
|
||||
// generation: the controller sends them only to a machine that said so, because an older node-engine refuses the keys.
|
||||
func TestEveryReportSaysItReadsModules(t *testing.T) {
|
||||
m, key := aMember(t)
|
||||
l := newQuietLink()
|
||||
a := &applies{}
|
||||
q := aQueue(m, a.apply, &keptInMemory{})
|
||||
q.attach(context.Background(), l)
|
||||
q.Deliver(ordered(t, key, "new", 7, 12))
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
worker := runWorker(ctx, q)
|
||||
eventually(t, "the delivery was never reported", func() bool { return len(l.said()) == 1 })
|
||||
stop()
|
||||
<-worker
|
||||
raw, err := json.Marshal(l.said()[0])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(string(raw), `"reads_modules":true`) || !strings.Contains(string(raw), `"reads_generation":true`) {
|
||||
t.Fatalf("the report does not say it reads modules beside the generation: %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -100,7 +100,7 @@ func LongRunning(d *declaration.Declaration, held map[string]bool) []Resource {
|
||||
return nil
|
||||
}
|
||||
for _, r := range d.Resources {
|
||||
module, ok := ModuleOf(r.Identity())
|
||||
module, ok := d.ModuleOfID(r.Identity())
|
||||
if !ok || held[r.Identity()] {
|
||||
continue
|
||||
}
|
||||
@@ -133,16 +133,7 @@ func LongRunning(d *declaration.Declaration, held map[string]bool) []Resource {
|
||||
// ModuleOf is the module a resource's id belongs to: everything before its last dot — a module's name may
|
||||
// carry a dot, its resources' own ids never do (the apply's moduleOf, for the same reason). False for
|
||||
// what the mesh declares in its own right.
|
||||
func ModuleOf(id string) (string, bool) {
|
||||
if strings.HasPrefix(id, declaration.AdoptionPrefix) {
|
||||
return "", false
|
||||
}
|
||||
at := strings.LastIndex(id, ".")
|
||||
if at <= 0 {
|
||||
return "", false
|
||||
}
|
||||
return id[:at], true
|
||||
}
|
||||
func ModuleOf(id string) (string, bool) { return declaration.ModuleGuess(id) }
|
||||
|
||||
// Observed is one resource as one read found it.
|
||||
type Observed struct {
|
||||
|
||||
Reference in New Issue
Block a user