Assigner: a node's presence is read through the watchdogs seat's keys and values (ADR 0634 §1; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -405,7 +405,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
runs := map[string]string{"ace": fOld, "g14": "build-9", "shanks": fOld}
|
||||
for _, node := range []string{"ace", "g14", "shanks"} {
|
||||
b.compose(t, node, 'c')
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
b.putJSON(t, buckets.Presence, node, on)
|
||||
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: runs[node], Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
b.putJSON(t, buckets.Have, node+".web", have)
|
||||
@@ -687,7 +687,7 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) {
|
||||
digest := b.manifest(t, "web", `{"module":"web","version":"1"}`)
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
b.putJSON(t, buckets.Modules, "web", mod)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
b.putJSON(t, buckets.Presence, "ace", on)
|
||||
failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fNew, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed, Detail: "connection refused"})
|
||||
b.putJSON(t, buckets.Have, "ace.web", failed)
|
||||
@@ -745,7 +745,7 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) {
|
||||
digest := b.manifest(t, "mail", `{"module":"mail","version":"1","resources":[{"id":"data","type":"directory","path":"/var/lib/mail"}],"data":{"own":[{"id":"mail","path":"${dir:data}","class":"irreplaceable","backup":"copy"}]}}`)
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fNew, Manifest: digest, Repository: "novox/mail", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
b.putJSON(t, buckets.Modules, "mail", mod)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
b.putJSON(t, buckets.Presence, "ace", on)
|
||||
failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fNew, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed, Detail: "connection refused"})
|
||||
b.putJSON(t, buckets.Have, "ace.mail", failed)
|
||||
@@ -785,7 +785,7 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) {
|
||||
func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
const modules = 40
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
b.putJSON(t, buckets.Presence, "ace", on)
|
||||
names := make([]string, modules)
|
||||
for i := range names {
|
||||
@@ -847,7 +847,7 @@ func (b *bed) fixture(t *testing.T, name, fp, manifest string) {
|
||||
gkey, _ := grantsstate.PartGrantKey(name, fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
b.putJSON(t, buckets.Grants, gkey, grant)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
b.putJSON(t, buckets.Presence, "ace", on)
|
||||
b.place(t, "ace", name)
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
@@ -191,7 +192,7 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
|
||||
return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: time.Now().Format(time.RFC3339Nano)})
|
||||
}
|
||||
p := &passer{w: w, reads: &enginereads.Keeper{Call: call, Fallback: w.recordReads, Sync: true}}
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: false})
|
||||
on, _ := presenceValue(false)
|
||||
w.put(buckets.Presence, "ace", on)
|
||||
n = w.nodes(nil, nil)
|
||||
p.engineReads(n)
|
||||
@@ -202,7 +203,7 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
|
||||
t.Errorf("the statements are %+v", s)
|
||||
}
|
||||
p.engineReads(w.nodes(nil, nil))
|
||||
on, _ = delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ = presenceValue(true)
|
||||
w.put(buckets.Presence, "ace", on)
|
||||
p.engineReads(w.nodes(nil, nil))
|
||||
if asked["ace"] != 2 {
|
||||
@@ -662,7 +663,7 @@ func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) {
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
p.w.put(buckets.Modules, "web", mod)
|
||||
placeOn(t, p.w, "g14", "web")
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
p.w.put(buckets.Presence, "g14", on)
|
||||
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpTest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
p.w.put(buckets.Have, "g14.web", have)
|
||||
@@ -700,7 +701,7 @@ func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) {
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/resolved", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
p.w.put(buckets.Modules, "systemd-resolved", mod)
|
||||
placeOn(t, p.w, "shanks", "systemd-resolved")
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
p.w.put(buckets.Presence, "shanks", on)
|
||||
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpTest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed})
|
||||
p.w.put(buckets.Have, "shanks.systemd-resolved", have)
|
||||
@@ -996,7 +997,7 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
|
||||
placeOn(t, p.w, "g14", "web")
|
||||
pl, _ := mmac.EncodePlacement(mma.Placement{Node: "g14", Modules: []mma.PlacedModule{{Module: "web"}}})
|
||||
put(plc.PlacementBucket, "placement.g14", pl)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
on, _ := presenceValue(true)
|
||||
put(buckets.Presence, "g14", on)
|
||||
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpOld, Composed: was, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
put(buckets.Have, "g14.web", have)
|
||||
@@ -1181,3 +1182,8 @@ func mustAskID(a delivery.AssignerAsk) string {
|
||||
}
|
||||
return ask.ID
|
||||
}
|
||||
|
||||
// presenceValue is a node's presence as the watchman writes it.
|
||||
func presenceValue(online bool) ([]byte, error) {
|
||||
return json.Marshal(watchdogs.PresenceByNodeValue{V: 1, Online: online})
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
@@ -13,6 +14,7 @@ import (
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -325,16 +327,29 @@ func (w *world) records() map[string]nodeengine.NodeRecord {
|
||||
return out
|
||||
}
|
||||
|
||||
// presenceOf reads one key of the watchman's presence: a node's key (`<node>`) with the node's online or not; the keys of
|
||||
// the parts (`part.<part>.<instance>`) are not a node's and are none here.
|
||||
func presenceOf(key string, value []byte) (node string, online, ok bool) {
|
||||
if strings.Contains(key, ".") {
|
||||
return "", false, false
|
||||
}
|
||||
node, ok = watchdogs.ParsePresenceByNodeKey(key)
|
||||
if !ok {
|
||||
return "", false, false
|
||||
}
|
||||
p, err := watchdogs.DecodePresenceByNodeValue(value)
|
||||
if err != nil {
|
||||
return "", false, false
|
||||
}
|
||||
return node, p.Online, true
|
||||
}
|
||||
|
||||
// online is, by node, whether the watchman holds the node online.
|
||||
func (w *world) online() map[string]bool {
|
||||
out := map[string]bool{}
|
||||
for key, value := range w.read(buckets.Presence) {
|
||||
node, _, _, err := delivery.ParsePresenceKey(key)
|
||||
if err != nil || node == "" {
|
||||
continue
|
||||
}
|
||||
if p, err := delivery.DecodePresence(value); err == nil {
|
||||
out[node] = p.Online
|
||||
if node, online, ok := presenceOf(key, value); ok {
|
||||
out[node] = online
|
||||
}
|
||||
}
|
||||
return out
|
||||
@@ -365,12 +380,8 @@ func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[stri
|
||||
n := assign.Nodes{Online: map[string]bool{}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{}, Providers: map[string]map[string][]string{},
|
||||
Reads: map[string][]string{}}
|
||||
for key, value := range w.read(buckets.Presence) {
|
||||
node, _, _, err := delivery.ParsePresenceKey(key)
|
||||
if err != nil || node == "" {
|
||||
continue
|
||||
}
|
||||
if p, err := delivery.DecodePresence(value); err == nil {
|
||||
n.Online[node] = p.Online
|
||||
if node, online, ok := presenceOf(key, value); ok {
|
||||
n.Online[node] = online
|
||||
}
|
||||
}
|
||||
for key, value := range w.read(buckets.Have) {
|
||||
|
||||
Reference in New Issue
Block a user