diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 255e9014..0e0558df 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index ea781e21..6c03f9b8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -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}) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 633b02a4..0140c9e5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -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 (``) with the node's online or not; the keys of +// the parts (`part..`) 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) {