Merge pull request 'Reconcile a module from the machine's facts at start and on a down edge (issue 1957, ADR 0568)' (#166) from fix/1957-boot-reconcile into main
This commit was merged in pull request #166.
This commit is contained in:
@@ -69,7 +69,7 @@ func TestAModuleWithNothingLongRunningIsOkWithinALookOfItsApply(t *testing.T) {
|
||||
applying := fact
|
||||
applying.Phase = string(delivery.PhaseApplying)
|
||||
queue.NoteModule(ctx, applying)
|
||||
go judgeWhatRuns(ctx, j, queue, func(string) {}, nil)
|
||||
go judgeWhatRuns(ctx, j, queue, func(string) {}, nil, nil)
|
||||
// The facts look inside the apply, while the module is still applying.
|
||||
queue.SayFacts(ctx, link.Report{Health: &link.Health{Contract: link.LivenessContract, At: time.Now().UTC()}})
|
||||
if h, _ := bus.have("n1.xorg"); h.Health != delivery.HealthUnknown {
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/novox/mesh-host/internal/apply"
|
||||
"github.com/novox/mesh-host/internal/liveness"
|
||||
)
|
||||
|
||||
// heldBy is the containers an open maintenance window holds still, by runtime name (ADR 0189), at now.
|
||||
func heldBy(windows *apply.Windows) func(time.Time) map[string]bool {
|
||||
return func(now time.Time) map[string]bool {
|
||||
held := map[string]bool{}
|
||||
for _, w := range windows.Open(now) {
|
||||
for _, name := range w.Holds {
|
||||
held[name] = true
|
||||
}
|
||||
}
|
||||
return held
|
||||
}
|
||||
}
|
||||
|
||||
// inactiveIn is what the agent asks at its start of a module it believes runs (ADR 0568 §1): whether the machine shows a
|
||||
// long-running resource the composition declares running not active, read from the service manager and the container
|
||||
// runtime and from no record. A resource held by an open maintenance window, or waiting for a library, is not counted;
|
||||
// one that cannot be read is not counted either (unknown is not inactive), and neither is one the manager is restarting.
|
||||
func inactiveIn(rt liveness.Runtime, held func(time.Time) map[string]bool) func(context.Context, delivery.Composed) bool {
|
||||
return func(ctx context.Context, c delivery.Composed) bool {
|
||||
d, err := declarationOf(c)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
var holding map[string]bool
|
||||
if held != nil {
|
||||
holding = held(time.Now())
|
||||
}
|
||||
return anyInactive(ctx, rt, liveness.LongRunning(d, withWaiting(nil, apply.WaitingOnLibrary(d))), holding)
|
||||
}
|
||||
}
|
||||
|
||||
// anyInactive reads each group of resources once (the containers, each service manager's units) and says whether any is
|
||||
// not active.
|
||||
func anyInactive(ctx context.Context, rt liveness.Runtime, resources []liveness.Resource, held map[string]bool) bool {
|
||||
type group struct {
|
||||
kind, scope, user string
|
||||
targets []string
|
||||
}
|
||||
var groups []*group
|
||||
at := map[string]*group{}
|
||||
for _, r := range resources {
|
||||
if r.Kind == liveness.KindContainer && held[r.Target] {
|
||||
continue
|
||||
}
|
||||
g := &group{kind: r.Kind, scope: r.Scope, user: r.User}
|
||||
if r.Kind == liveness.KindContainer {
|
||||
g.scope, g.user = "", ""
|
||||
}
|
||||
key := g.kind + "/" + g.scope + "/" + g.user
|
||||
if r.Kind != liveness.KindContainer {
|
||||
key = "units/" + g.scope + "/" + g.user
|
||||
}
|
||||
if at[key] == nil {
|
||||
at[key] = g
|
||||
groups = append(groups, g)
|
||||
}
|
||||
at[key].targets = append(at[key].targets, r.Target)
|
||||
}
|
||||
for _, g := range groups {
|
||||
var got map[string]liveness.Observed
|
||||
var err error
|
||||
if g.kind == liveness.KindContainer {
|
||||
got, err = rt.Containers(ctx, g.targets)
|
||||
} else {
|
||||
got, err = rt.Units(ctx, g.scope, g.user, g.targets)
|
||||
}
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, t := range g.targets {
|
||||
if o := got[t]; !o.Found || !(o.Running || o.Restarting) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/novox/mesh-host/internal/liveness"
|
||||
)
|
||||
|
||||
// 1957/AC2 and AC4 (unit; ADR 0568 §2, §4): the judge's look acts on a resource that goes down, once per edge. A repeated
|
||||
// statement acts nothing, the edge clears when the resource is healthy again, and a resource held, restarting, starting or
|
||||
// missing (the other path's) is no edge.
|
||||
func testOnlyAResourceNewlyDownIsAnEdge(t *testing.T) {
|
||||
state := func(id, s, reason string, missing bool) liveness.State {
|
||||
return liveness.State{Resource: liveness.Resource{Module: "dnsmasq", ID: id}, State: s, Reason: reason, Missing: missing}
|
||||
}
|
||||
down := state("dnsmasq.unit", liveness.Unhealthy, liveness.ReasonDown, false)
|
||||
steps := []struct {
|
||||
name string
|
||||
st []liveness.State
|
||||
want []string
|
||||
}{
|
||||
{"down: an edge", []liveness.State{down}, []string{"dnsmasq"}},
|
||||
{"said again while down: none", []liveness.State{down}, nil},
|
||||
{"healthy again: clears", []liveness.State{state("dnsmasq.unit", liveness.Healthy, "", false)}, nil},
|
||||
{"down again: an edge", []liveness.State{down}, []string{"dnsmasq"}},
|
||||
{"held by a window: none", []liveness.State{state("dnsmasq.unit", liveness.Held, "", false)}, nil},
|
||||
{"restarting: none", []liveness.State{state("dnsmasq.unit", liveness.Unhealthy, liveness.ReasonRestarting, false)}, nil},
|
||||
{"starting: none", []liveness.State{state("dnsmasq.unit", liveness.Starting, "", false)}, nil},
|
||||
{"missing: the other path's", []liveness.State{state("dnsmasq.unit", liveness.Unhealthy, liveness.ReasonDown, true)}, nil},
|
||||
}
|
||||
var last map[string]bool
|
||||
for _, s := range steps {
|
||||
var got []string
|
||||
got, last = newlyInactive(last, liveness.Statement{Resources: s.st})
|
||||
if len(got) != len(s.want) || (len(got) == 1 && got[0] != s.want[0]) {
|
||||
t.Fatalf("%s: acted on %v, want %v", s.name, got, s.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type fakeRuntime struct {
|
||||
units map[string]liveness.Observed
|
||||
containers map[string]liveness.Observed
|
||||
err error
|
||||
}
|
||||
|
||||
func (f fakeRuntime) Containers(_ context.Context, _ []string) (map[string]liveness.Observed, error) {
|
||||
return f.containers, f.err
|
||||
}
|
||||
func (f fakeRuntime) Units(_ context.Context, _, _ string, _ []string) (map[string]liveness.Observed, error) {
|
||||
return f.units, f.err
|
||||
}
|
||||
|
||||
// 1957/AC1 (unit; ADR 0568 §1): the machine's facts say a declared-running resource is not active when its unit is stopped
|
||||
// or unknown to the manager; everything up, a restart in progress, a read that failed and a container held by a
|
||||
// maintenance window are not.
|
||||
func testTheMachineShowsAResourceNotActive(t *testing.T) {
|
||||
svc := []byte(`{"id":"dnsmasq.unit","type":"service","unit":"dnsmasq.service","state":"running","module":"dnsmasq"}`)
|
||||
ctr := []byte(`{"id":"dnsmasq.box","type":"container","name":"box","image":"x@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","module":"dnsmasq"}`)
|
||||
up := liveness.Observed{Found: true, Running: true}
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
rt fakeRuntime
|
||||
held map[string]bool
|
||||
want bool
|
||||
}{
|
||||
{"all up", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": up}, containers: map[string]liveness.Observed{"box": up}}, nil, false},
|
||||
{"unit inactive", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": {Found: true}}, containers: map[string]liveness.Observed{"box": up}}, nil, true},
|
||||
{"unit unknown to the manager", fakeRuntime{containers: map[string]liveness.Observed{"box": up}}, nil, true},
|
||||
{"unit restarting", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": {Found: true, Restarting: true}}, containers: map[string]liveness.Observed{"box": up}}, nil, false},
|
||||
{"container not running", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": up}, containers: map[string]liveness.Observed{"box": {Found: true}}}, nil, true},
|
||||
{"container held by a window", fakeRuntime{units: map[string]liveness.Observed{"dnsmasq.service": up}, containers: map[string]liveness.Observed{"box": {Found: true}}}, map[string]bool{"box": true}, false},
|
||||
{"unread", fakeRuntime{err: errors.New("the service manager could not be read")}, nil, false},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
d, err := declarationOf(delivery.Composed{Node: "n1", Module: "dnsmasq", Resources: []json.RawMessage{svc, ctr}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := anyInactive(context.Background(), c.rt, liveness.LongRunning(d, nil), c.held); got != c.want {
|
||||
t.Fatalf("inactive = %v, want %v", got, c.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+41
-11
@@ -1046,15 +1046,7 @@ func runLink(ctx context.Context, opts options) error {
|
||||
say(err.Error())
|
||||
}
|
||||
windows := apply.WindowsIn(filepath.Dir(opts.state))
|
||||
j.HeldNow = func(now time.Time) map[string]bool {
|
||||
held := map[string]bool{}
|
||||
for _, w := range windows.Open(now) {
|
||||
for _, name := range w.Holds {
|
||||
held[name] = true
|
||||
}
|
||||
}
|
||||
return held
|
||||
}
|
||||
j.HeldNow = heldBy(windows)
|
||||
judging = j
|
||||
// And which units its service managers say failed, and whose each is (novox/hq issue 315).
|
||||
unitJudge = units.New(units.Exec{Run: apply.ExecRunner, System: builtFor})
|
||||
@@ -1173,6 +1165,10 @@ func runLink(ctx context.Context, opts options) error {
|
||||
if ag := current.get(); ag != nil {
|
||||
ag.Retry(modules...)
|
||||
}
|
||||
}, func(modules ...string) {
|
||||
if ag := current.get(); ag != nil {
|
||||
ag.Reconcile(modules...)
|
||||
}
|
||||
})
|
||||
}
|
||||
// The node follows its wants: one module at a time from the composition its want names (novox/hq ADR 0539), and the
|
||||
@@ -1336,8 +1332,8 @@ const (
|
||||
// the bus when a state changed, again every minute while one is not healthy, and every five minutes
|
||||
// anyway. A statement the bus did not take is said at the next look. It reads; it never acts (ADR 0240
|
||||
// rule 6) — but for going on with a take act stopped at its bound (ADR 0328 §7), which runs beside it.
|
||||
func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session, say link.Announce, reapply func(modules ...string)) {
|
||||
var missingBefore map[string]bool
|
||||
func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session, say link.Announce, reapply, reconcile func(modules ...string)) {
|
||||
var missingBefore, inactiveBefore map[string]bool
|
||||
ticker := time.NewTicker(liveness.LookEvery)
|
||||
defer ticker.Stop()
|
||||
var lastSaid time.Time
|
||||
@@ -1439,6 +1435,14 @@ func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session,
|
||||
say(fmt.Sprintf("applying %s again: a resource it declares is not on this machine", strings.Join(m, ", ")))
|
||||
reapply(m...)
|
||||
}
|
||||
// A resource its module declares running that is down but still known to the manager is started by the same
|
||||
// apply, once each time it goes down, never on a clock (ADR 0568 §2): the judge reads, the agent acts.
|
||||
var inactive []string
|
||||
inactive, inactiveBefore = newlyInactive(inactiveBefore, st)
|
||||
if len(inactive) > 0 && reconcile != nil {
|
||||
say(fmt.Sprintf("applying %s again: a resource it declares running is not active", strings.Join(inactive, ", ")))
|
||||
reconcile(inactive...)
|
||||
}
|
||||
}
|
||||
if queue.SayHealth(ctx, *withAccesses(withFoundDirectories(withAccounts(withUnits(healthAsReported(st, netSt),
|
||||
unitSt), accountSt), found), accessSt)) {
|
||||
@@ -2134,6 +2138,32 @@ func newlyMissing(last map[string]bool, st liveness.Statement) ([]string, map[st
|
||||
return missingModules(fresh), now
|
||||
}
|
||||
|
||||
// newlyInactive are the modules of the resources a statement says down (not running after their grace, but still known to
|
||||
// their manager, and not restarting) that the last statement acted on did not, each once, sorted, and the resources down
|
||||
// now: a resource still down is not acted on again, and one that came back and went down again is (ADR 0568 §2).
|
||||
func newlyInactive(last map[string]bool, st liveness.Statement) ([]string, map[string]bool) {
|
||||
now := map[string]bool{}
|
||||
var fresh liveness.Statement
|
||||
for _, r := range st.Resources {
|
||||
if r.State != liveness.Unhealthy || r.Reason != liveness.ReasonDown || r.Missing {
|
||||
continue
|
||||
}
|
||||
now[r.ID] = true
|
||||
if !last[r.ID] {
|
||||
fresh.Resources = append(fresh.Resources, r)
|
||||
}
|
||||
}
|
||||
return missingModules(withMissing(fresh)), now
|
||||
}
|
||||
|
||||
// withMissing is the statement with each of its resources marked missing, so missingModules lists their modules.
|
||||
func withMissing(st liveness.Statement) liveness.Statement {
|
||||
for i := range st.Resources {
|
||||
st.Resources[i].Missing = true
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// recordKeys reads the public halves of the node's keys and the certificate the mesh issued for its internal name from
|
||||
// its state directory (novox/hq ADR 0555 §3): what the node's record says of them. A key that cannot be read is left
|
||||
// empty, never invented; the certificate is there only after the first declaration that carried it.
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
"github.com/novox/mesh-host/internal/apply"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
"github.com/novox/mesh-host/internal/liveness"
|
||||
"github.com/novox/mesh-host/internal/selfcompose"
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
)
|
||||
@@ -468,6 +469,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
// The same kept declaration's filled paths, so a module still running from it keeps its files filled.
|
||||
KeptFilled: func() map[string][]string { _, filled := kept(); return filled },
|
||||
Recorded: recordedIn(opts.state, say),
|
||||
Inactive: inactiveIn(&liveness.Exec{Run: apply.ExecRunner, System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))),
|
||||
Revisions: func(ctx context.Context, key string) ([]agent.Update, error) {
|
||||
got, err := session.KVHistory(ctx, composedBucket, key)
|
||||
if err != nil {
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
"github.com/novox/mesh-host/internal/selfcompose"
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
)
|
||||
|
||||
// The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition
|
||||
@@ -239,6 +240,7 @@ func within(t *testing.T, d time.Duration, what string, ok func() bool) {
|
||||
// applied within one second and its have names the fingerprint and the digest; a composition whose digest no want names
|
||||
// is not applied.
|
||||
func TestNatsAWantAndItsCompositionAreAppliedWithinASecond(t *testing.T) {
|
||||
t.Run("1957: an engine start starts a declared-running unit", testNatsAnEngineStartStartsADeclaredRunningUnitTheHaveSaysRuns)
|
||||
b := newItBed(t)
|
||||
stop := b.start(false)
|
||||
defer stop()
|
||||
@@ -708,3 +710,71 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str
|
||||
t.Errorf("a candidate moved what runs: %+v, %q", h, b.file("web"))
|
||||
}
|
||||
}
|
||||
|
||||
// 1957/AC1 (integration; ADR 0568 §1): a node agent starting on a bus that holds a `running` have for a module at its want,
|
||||
// with a service manager (a systemctl on the PATH) that shows the module's unit inactive, starts the unit once and writes
|
||||
// the have running; the same start with the unit active applies nothing.
|
||||
func testNatsAnEngineStartStartsADeclaredRunningUnitTheHaveSaysRuns(t *testing.T) {
|
||||
for _, active := range []bool{false, true} {
|
||||
b := newItBed(t)
|
||||
bin := t.TempDir()
|
||||
stateFile, calls := filepath.Join(bin, "state"), filepath.Join(bin, "calls")
|
||||
if active {
|
||||
os.WriteFile(stateFile, []byte("active"), 0o644)
|
||||
} else {
|
||||
os.WriteFile(stateFile, []byte("inactive"), 0o644)
|
||||
}
|
||||
script := "#!/bin/sh\necho \"$@\" >> " + calls + "\ncase \"$1\" in\n" +
|
||||
"show) printf 'LoadState=loaded\\nActiveState=%s\\nType=simple\\n' \"$(cat " + stateFile + ")\" ;;\n" +
|
||||
"start) printf active > " + stateFile + " ;;\nis-enabled) echo enabled ;;\nesac\nexit 0\n"
|
||||
if err := os.WriteFile(filepath.Join(bin, "systemctl"), []byte(script), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))
|
||||
|
||||
// The engine's own record of the unit, as a previous run left it, and the have it wrote then.
|
||||
res, _ := json.Marshal(map[string]any{"id": "web.unit", "type": "service", "unit": "web-it.service", "state": "running", "module": "web"})
|
||||
var known store.State
|
||||
known.Record(store.Applied{ID: "web.unit", Type: "service", Target: "web-it.service"})
|
||||
if err := store.Save(b.opts.state, known); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fp := fpIt("a")
|
||||
c := delivery.Composed{Node: itNode, Module: "web", FP: fp, Resources: []json.RawMessage{res}}
|
||||
signed, _ := delivery.SignComposed(c, b.controll)
|
||||
raw, _ := delivery.EncodeComposed(signed)
|
||||
dig, _ := delivery.ComposedDigestOf(signed)
|
||||
key, _ := delivery.ComposedKey(itNode, "web", fp)
|
||||
if _, err := b.composed.Put(key, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.want("web", fp, dig, "")
|
||||
hkey, _ := delivery.HaveKey(itNode, "web")
|
||||
have, err := delivery.EncodeHave(delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.haves.Put(hkey, have); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stop := b.start(false)
|
||||
within(t, 5*time.Second, "the unit was not started and the have not running", func() bool {
|
||||
got, _ := os.ReadFile(stateFile)
|
||||
h := b.have("web")
|
||||
return string(got) == "active" && h.Phase == delivery.PhaseRunning && h.Composed == dig
|
||||
})
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
stop()
|
||||
starts := 0
|
||||
log, _ := os.ReadFile(calls)
|
||||
for _, l := range strings.Split(string(log), "\n") {
|
||||
if strings.HasPrefix(l, "start") {
|
||||
starts++
|
||||
}
|
||||
}
|
||||
if want := map[bool]int{false: 1, true: 0}[active]; starts != want {
|
||||
t.Fatalf("unit active=%v: started %d times, want %d", active, starts, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,6 +36,7 @@ func TestACompositionIsRecordedOnlyWhenEveryResourceIs(t *testing.T) {
|
||||
|
||||
// The modules applied again are those with a missing resource, each once; one down but there is not among them.
|
||||
func TestTheModulesAppliedAgainAreThoseWithAMissingResource(t *testing.T) {
|
||||
t.Run("1957: the machine shows a resource not active", testTheMachineShowsAResourceNotActive)
|
||||
st := liveness.Statement{Resources: []liveness.State{
|
||||
{Resource: liveness.Resource{Module: "matrix", ID: "matrix.homeserver"}, State: liveness.Unhealthy, Missing: true},
|
||||
{Resource: liveness.Resource{Module: "matrix", ID: "matrix.db"}, State: liveness.Unhealthy, Missing: true},
|
||||
@@ -51,6 +52,7 @@ func TestTheModulesAppliedAgainAreThoseWithAMissingResource(t *testing.T) {
|
||||
// not acted on again, so a resource that cannot be made is not applied (and backed up) at every unrelated change. One
|
||||
// that comes back and goes again is newly missing again (the review of issue 1945).
|
||||
func TestOnlyANewlyMissingResourceAppliesItsModuleAgain(t *testing.T) {
|
||||
t.Run("1957: a resource newly down is an edge", testOnlyAResourceNewlyDownIsAnEdge)
|
||||
missing := func(ids ...string) liveness.Statement {
|
||||
var st liveness.Statement
|
||||
for _, id := range ids {
|
||||
|
||||
+61
-1
@@ -118,6 +118,11 @@ type Env struct {
|
||||
// Recorded says the engine holds a record of every resource of the composition, so it was applied and not removed
|
||||
// since; nil says yes. A composition the kept declaration or a have says runs is applied again when it says no.
|
||||
Recorded func(ctx context.Context, c delivery.Composed) bool
|
||||
// Inactive says a resource the composition declares running is not active on the machine now (a unit stopped, a
|
||||
// container not running), read from the machine and not from any record (ADR 0568 §1). A have or a kept declaration
|
||||
// that says the module runs is believed at the agent's start only when this says no; nil says no, and an unreadable
|
||||
// machine says no (unknown is not inactive).
|
||||
Inactive func(ctx context.Context, c delivery.Composed) bool
|
||||
// Revisions are a composition key's revisions the bucket keeps, oldest first; nil reads none. The composition a want
|
||||
// names is taken from them when a newer one replaced it under the same key (ADR 0539 clause 1).
|
||||
Revisions func(ctx context.Context, key string) ([]Update, error)
|
||||
@@ -167,6 +172,8 @@ type Agent struct {
|
||||
// composition is held out of the node's own apply until a composition of its own is handed (heldOut).
|
||||
heldAlone map[string]bool
|
||||
retry chan []string
|
||||
// reconcile carries the judge's edges: a resource found not active (ADR 0568 §2).
|
||||
reconcile chan []string
|
||||
// tries counts the transient failures in a row of each module; again carries a retry whose wait ran out, and
|
||||
// transient asks for every transiently failed module at once. stopped is closed when Run returns.
|
||||
tries map[string]int
|
||||
@@ -213,7 +220,7 @@ func New(env Env) *Agent {
|
||||
}
|
||||
return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{},
|
||||
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]notComposedMark{},
|
||||
retry: make(chan []string, 8), tries: map[string]int{}, again: make(chan againAsk, 64),
|
||||
retry: make(chan []string, 8), reconcile: make(chan []string, 64), tries: map[string]int{}, again: make(chan againAsk, 64),
|
||||
transient: make(chan struct{}, 1), stopped: make(chan struct{})}
|
||||
}
|
||||
|
||||
@@ -245,6 +252,17 @@ func (a *Agent) Run(ctx context.Context) error {
|
||||
dirty[m] = true
|
||||
}
|
||||
a.mu.Unlock()
|
||||
case modules := <-a.reconcile:
|
||||
a.mu.Lock()
|
||||
for _, m := range modules {
|
||||
if w := a.wants[m]; !w.Present || w.Want.Removed || a.attempt[m].Held {
|
||||
continue
|
||||
}
|
||||
delete(a.running, m)
|
||||
delete(a.attempt, m)
|
||||
dirty[m] = true
|
||||
}
|
||||
a.mu.Unlock()
|
||||
case ask := <-a.again:
|
||||
// The wait of a transiently failed apply ran out: tried again from the keys it has, if nothing tried it since.
|
||||
a.mu.Lock()
|
||||
@@ -303,6 +321,16 @@ func (a *Agent) Retry(modules ...string) {
|
||||
}
|
||||
}
|
||||
|
||||
// Reconcile asks for the modules to be applied again because a resource each declares running was found not active, once
|
||||
// per edge (the judge's look, ADR 0568 §2). It is Retry, but a module whose apply is held (a failed backup) is left
|
||||
// alone, as is one the node wants no more (ADR 0568 §4). It never blocks.
|
||||
func (a *Agent) Reconcile(modules ...string) {
|
||||
select {
|
||||
case a.reconcile <- modules:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// RetryTransient asks for every module whose last apply failed transiently to be applied again now: the engine says it
|
||||
// when it knows the bus answers again (after it wrote the bus's user list and the bus reloaded its accounts). It never
|
||||
// blocks.
|
||||
@@ -1010,6 +1038,18 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
a.mu.Lock()
|
||||
a.prior[module] = true
|
||||
a.mu.Unlock()
|
||||
if h.Phase == string(delivery.PhaseHeld) {
|
||||
// Held for a failed backup, and the operator's answer is still to come: the hold is kept, nothing is
|
||||
// backed up again or applied, and the next revision of the want is the answer (ADR 0568 §4).
|
||||
a.mu.Lock()
|
||||
a.attempt[module] = attempt{FP: w.Want.FP, Composed: w.Want.Composed, WantRev: w.Rev, CompRev: comp.Rev, Held: true,
|
||||
AnsweredRev: w.Rev}
|
||||
if h.Composed != "" {
|
||||
a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
|
||||
}
|
||||
a.mu.Unlock()
|
||||
return true
|
||||
}
|
||||
if h.Composed != "" {
|
||||
// This node-engine wrote it: what it says is what runs, and the kept declaration says nothing more. A
|
||||
// failed apply or a different composition is applied again.
|
||||
@@ -1022,6 +1062,9 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
}
|
||||
if h.FP == w.Want.FP && h.Composed == w.Want.Composed &&
|
||||
(h.Phase == "" || h.Phase == string(delivery.PhaseRunning) || marked) {
|
||||
if a.inactiveAtStart(ctx, module, comp) {
|
||||
return false
|
||||
}
|
||||
a.mu.Lock()
|
||||
a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
|
||||
a.mu.Unlock()
|
||||
@@ -1047,6 +1090,9 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
if !ok || !SameResources(kept, comp.C.Resources) {
|
||||
return false
|
||||
}
|
||||
if a.inactiveAtStart(ctx, module, comp) {
|
||||
return false
|
||||
}
|
||||
a.mu.Lock()
|
||||
a.running[module] = delivery.Have{FP: w.Want.FP, Composed: w.Want.Composed, Phase: delivery.PhaseRunning}
|
||||
a.mu.Unlock()
|
||||
@@ -1054,6 +1100,20 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
return true
|
||||
}
|
||||
|
||||
// inactiveAtStart says the machine shows a resource the module declares running not active, so what a have or the kept
|
||||
// declaration says runs is not believed and the module is applied once, as for a changed want (ADR 0568 §1).
|
||||
func (a *Agent) inactiveAtStart(ctx context.Context, module string, comp compNow) bool {
|
||||
if a.env.Inactive == nil || !a.env.Inactive(ctx, comp.C) {
|
||||
return false
|
||||
}
|
||||
a.mu.Lock()
|
||||
a.prior[module] = true
|
||||
a.sayOnce("inactive:"+module+":"+comp.Dig, fmt.Sprintf("%s is applied again: a resource it declares running is not "+
|
||||
"active on this machine", module))
|
||||
a.mu.Unlock()
|
||||
return true
|
||||
}
|
||||
|
||||
func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNow, others []delivery.Composed) {
|
||||
fp, dig := w.Want.FP, w.Want.Composed
|
||||
a.mu.Lock()
|
||||
|
||||
@@ -342,6 +342,7 @@ func TestStartingOnWhatTheKeptDeclarationHoldsAppliesNothing(t *testing.T) {
|
||||
|
||||
// What the node-engine itself last applied is what runs after a restart, not the kept declaration.
|
||||
func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) {
|
||||
t.Run("1957: an engine start reads the machine's facts", testAnEngineStartReconcilesAModuleFromTheMachinesFacts)
|
||||
t.Run("says the have again", testStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines)
|
||||
// An apply that failed is not what runs, though its have names the want's build and digest: the agent starting on
|
||||
// it applies again (review of the switch, round 1).
|
||||
@@ -431,6 +432,7 @@ func testStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines(t *testing
|
||||
// A module whose kept resources differ from the composition is applied, and a failed apply is retried only by a change
|
||||
// of either key.
|
||||
func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
||||
t.Run("1957: the judge's edge reconciles, and a failed re-apply is failed", testReconcileAppliesARunningModuleAgainAndLeavesTheHeldAndTheUnwantedAlone)
|
||||
// A real failure is not retried by the clock, however short the backoff: only by a change of either key.
|
||||
t.Run("a real failure", func(t *testing.T) {
|
||||
b := newBed(t)
|
||||
@@ -1233,3 +1235,122 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
|
||||
r.Phase == string(delivery.PhaseRunning) && r.FP == fpB && r.Composed == rdig
|
||||
})
|
||||
}
|
||||
|
||||
// 1957/AC1 and AC4 (unit; ADR 0568 §1, §4): an engine start takes a module from what the machine shows, not only from the
|
||||
// have or the kept declaration it kept. Every resource up applies nothing; one declared running and not active applies the
|
||||
// module once, from a have or from the kept declaration; a module with no want is not looked at.
|
||||
func testAnEngineStartReconcilesAModuleFromTheMachinesFacts(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
have bool // the engine's own have says running at the want
|
||||
held bool // else its have says held, from a failed backup
|
||||
kept bool // else the kept declaration says it runs
|
||||
inactive bool
|
||||
unwanted bool
|
||||
applied int
|
||||
asked bool // the machine was looked at
|
||||
}{
|
||||
{"all up: nothing is applied", true, false, false, false, false, 0, true},
|
||||
{"a unit not active: applied once", true, false, false, true, false, 1, true},
|
||||
{"kept declaration, a unit not active: applied once", false, false, true, true, false, 1, true},
|
||||
{"kept declaration, all up: nothing is applied", false, false, true, false, false, 0, true},
|
||||
{"a held module applies nothing at start", false, true, false, true, false, 0, false},
|
||||
{"no want: nothing is looked at or applied", true, false, false, true, true, 0, false},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
b := newBed(t)
|
||||
fp := fpOf("a")
|
||||
rs := []json.RawMessage{file("web", "x", "1")}
|
||||
ck, craw, dig := composition(t, "web", fp, false, rs...)
|
||||
wk, wraw := wantFor(t, "web", fp, dig, "")
|
||||
b.comps.before(ck, craw)
|
||||
if !c.unwanted {
|
||||
b.wants.before(wk, wraw)
|
||||
}
|
||||
if c.have {
|
||||
b.haves.stored = map[string]Have{"web": {Module: "web", FP: fp, Composed: dig, Phase: "running"}}
|
||||
}
|
||||
if c.held {
|
||||
b.haves.stored = map[string]Have{"web": {Module: "web", FP: fpOf("0"), Composed: "sha256:" + strings.Repeat("0", 64),
|
||||
Phase: string(delivery.PhaseHeld), Detail: "restic: the repository is locked"}}
|
||||
}
|
||||
if c.kept {
|
||||
b.kept = map[string][]json.RawMessage{"web": rs}
|
||||
}
|
||||
var mu sync.Mutex
|
||||
asked := false
|
||||
b.env.Inactive = func(context.Context, delivery.Composed) bool {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
asked = true
|
||||
return c.inactive
|
||||
}
|
||||
stop := b.run()
|
||||
defer stop()
|
||||
if !c.unwanted && !c.held {
|
||||
eventually(t, "the module was not settled", func() bool { return b.haves.last("web").Composed == dig })
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if got := b.appliedList(); len(got) != c.applied {
|
||||
t.Fatalf("applied %v, want %d", got, c.applied)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if asked != c.asked {
|
||||
t.Fatalf("the machine was asked: %v, want %v", asked, c.asked)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 1957/AC2, AC3 and AC4 (unit; ADR 0568 §2, §3, §4): the judge's edge, Reconcile, applies a running module again; one held
|
||||
// for a failed backup and one with no want are left alone; and an apply that fails writes the have `failed` with the
|
||||
// apply's own words.
|
||||
func testReconcileAppliesARunningModuleAgainAndLeavesTheHeldAndTheUnwantedAlone(t *testing.T) {
|
||||
b := newBed(t)
|
||||
fp := fpOf("a")
|
||||
ckW, crawW, digW := composition(t, "web", fp, false, file("web", "x", "1"))
|
||||
wkW, wrawW := wantFor(t, "web", fp, digW, "")
|
||||
ckD, crawD, digD := composition(t, "db", fp, true, file("db", "x", "1"))
|
||||
wkD, wrawD := wantFor(t, "db", fp, digD, "")
|
||||
b.comps.before(ckW, crawW)
|
||||
b.wants.before(wkW, wrawW)
|
||||
b.comps.before(ckD, crawD)
|
||||
b.wants.before(wkD, wrawD)
|
||||
b.haves.stored = map[string]Have{"web": {Module: "web", FP: fp, Composed: digW, Phase: "running"},
|
||||
"db": {Module: "db", FP: fpOf("0"), Composed: "sha256:" + strings.Repeat("0", 64), Phase: "running"}}
|
||||
b.backup = func(string) Backup { return Backup{Words: "restic: the repository is locked"} }
|
||||
stop := b.run()
|
||||
defer stop()
|
||||
eventually(t, "web was not settled", func() bool { return b.haves.last("web").Composed == digW })
|
||||
eventually(t, "db was not held", func() bool { return b.haves.last("db").Phase == string(delivery.PhaseHeld) })
|
||||
if got := b.appliedList(); len(got) != 0 {
|
||||
t.Fatalf("applied at the start: %v", got)
|
||||
}
|
||||
|
||||
b.ag.Reconcile("db", "ghost")
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if got := b.appliedList(); len(got) != 0 {
|
||||
t.Fatalf("a held module and an unwanted one were applied: %v", got)
|
||||
}
|
||||
if h := b.haves.last("db"); h.Phase != string(delivery.PhaseHeld) {
|
||||
t.Fatalf("the held module's have changed: %+v", h)
|
||||
}
|
||||
|
||||
b.ag.Reconcile("web")
|
||||
eventually(t, "web was not applied again", func() bool { return len(b.appliedList()) == 1 })
|
||||
eventually(t, "web is not running again", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) })
|
||||
|
||||
b.mu.Lock()
|
||||
b.failWith = "systemctl start web.service: exit status 1"
|
||||
b.mu.Unlock()
|
||||
b.ag.Reconcile("web")
|
||||
eventually(t, "the failed re-apply did not write the have failed with its words", func() bool {
|
||||
h := b.haves.last("web")
|
||||
return h.Phase == string(delivery.PhaseFailed) && h.Detail == "systemctl start web.service: exit status 1"
|
||||
})
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if got := b.appliedList(); len(got) != 2 {
|
||||
t.Fatalf("a failed apply was tried again with no edge: %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user