Write each module's have and the node's own facts beside the report, grouped per module and keyed by the module the controller names (issue 1132)

The node-engine reads a resource's module and a declaration's builds, says it does in every report (reads_modules), groups the apply's outcome per module with every failing resource named, and writes have.<node>.<module> and have.<node>.node to the bucket only when a value differs, never ahead of or in place of the report (novox/hq ADR 0429 section 4).
This commit is contained in:
2026-10-18 19:10:05 +02:00
parent 6469e82384
commit 0b391f49a2
19 changed files with 1193 additions and 45 deletions
+30 -1
View File
@@ -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
}
+38
View File
@@ -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")
}
}
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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
View File
@@ -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
+99
View File
@@ -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
}
+76
View File
@@ -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"])
}
}
+1 -1
View File
@@ -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()
}
+126 -16
View File
@@ -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
}
+60
View File
@@ -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)
}
}
}
+45
View File
@@ -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,46 @@ func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error {
defer cancel()
return b.Conn.FlushWithContext(flush)
}
// haveBucket binds to the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132). **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.
func (b OverNATS) haveBucket() (nats.KeyValue, error) {
kv, err := b.JS.KeyValue(delivery.HaveBucket)
if err != nil {
return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err)
}
return kv, nil
}
// HaveGet reads a key of the have bucket with its revision.
func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) {
kv, err := b.haveBucket()
if err != nil {
return nil, 0, err
}
entry, err := 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
}
// HaveSet writes a key of the have bucket over the revision read, or creates it when the revision is zero.
func (b OverNATS) HaveSet(_ context.Context, key string, value []byte, revision uint64) error {
kv, err := b.haveBucket()
if err != nil {
return err
}
if revision == 0 {
_, err = kv.Create(key, value)
return err
}
_, err = kv.Update(key, value, revision)
return err
}
var _ HaveBus = OverNATS{}
+219
View File
@@ -0,0 +1,219 @@
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 read and write the node-engine's own keys in 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 {
// HaveGet is a key's value and last revision, or ErrHaveAbsent.
HaveGet(ctx context.Context, key string) (value []byte, revision uint64, err error)
// HaveSet writes a key over the revision read: zero creates it, and a key that moved since is refused.
HaveSet(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
write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout())
defer cancel()
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
}
q.putHave(write, hb, key, raw, func(current []byte) bool {
c, err := delivery.DecodeHave(current)
return err == nil && delivery.SameHave(c, h)
})
}
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, hb, 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.
func (q *Queue) putHave(ctx context.Context, hb HaveBus, key string, value []byte, same func([]byte) bool) {
current, revision, err := hb.HaveGet(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 there: %v", key, err))
return
case same(current):
return
}
if err := hb.HaveSet(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))
}
}
+104
View File
@@ -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)
}
}
+291
View File
@@ -0,0 +1,291 @@
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
}
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) HaveGet(_ 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) HaveSet(_ context.Context, key string, value []byte, revision uint64) error {
l.hmu.Lock()
defer l.hmu.Unlock()
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)
}
}
+9
View File
@@ -342,6 +342,14 @@ 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) HaveGet(ctx context.Context, key string) ([]byte, uint64, error) {
return OverNATS{Conn: l.conn, JS: l.js}.HaveGet(ctx, key)
}
func (l *natsLink) HaveSet(ctx context.Context, key string, value []byte, revision uint64) error {
return OverNATS{Conn: l.conn, JS: l.js}.HaveSet(ctx, key, value, revision)
}
// 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 +378,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)
+19
View File
@@ -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
}
+21
View File
@@ -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
}
+23
View File
@@ -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)
}
}
+2 -11
View File
@@ -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 {