118 lines
3.7 KiB
Go
118 lines
3.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
)
|
|
|
|
// What a machine has, against what it is wanted to have (novox/hq ADR 0539 §1, §9). Nothing is sent, so nothing records that a
|
|
// machine was sent anything: whether a machine has "acted on the current words" is read from its haves, the node-engine's
|
|
// own bucket, beside the assigner's wants. A node is current when every module wanted on it is had at the wanted fingerprint
|
|
// and composition, applied (not failed, not held, not still applying), and every module wanted removed is gone. A node that
|
|
// nothing is wanted on has nothing to apply and is current.
|
|
|
|
// currentNodes is, for each node a want names, whether it has everything wanted of it. Pure.
|
|
func currentNodes(wants map[string]sdkdelivery.Want, haves map[string]sdkdelivery.Have) map[string]bool {
|
|
out := map[string]bool{}
|
|
for key, w := range wants {
|
|
node, _, err := sdkdelivery.ParseWantKey(key)
|
|
if err != nil {
|
|
continue // the node's own want: its composition is the node-engine's own
|
|
}
|
|
if _, seen := out[node]; !seen {
|
|
out[node] = true
|
|
}
|
|
h, had := haves[key]
|
|
switch {
|
|
case w.Removed:
|
|
if had && h.Phase != sdkdelivery.PhaseRemoved {
|
|
out[node] = false
|
|
}
|
|
case !had, h.FP != w.FP, h.Composed != w.Composed, h.Phase != sdkdelivery.PhaseRunning:
|
|
out[node] = false
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// readWantsAndHaves reads the assigner's wants and the node-engines' haves from the bus; a bucket not there yet is empty.
|
|
func readWantsAndHaves(ctx context.Context, js jetstream.JetStream) (map[string]sdkdelivery.Want, map[string]sdkdelivery.Have, error) {
|
|
wants := map[string]sdkdelivery.Want{}
|
|
haves := map[string]sdkdelivery.Have{}
|
|
each := func(bucket string, take func(key string, value []byte)) error {
|
|
kv, err := js.KeyValue(ctx, bucket)
|
|
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keys, err := kv.Keys(ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, key := range keys {
|
|
e, err := kv.Get(ctx, key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
take(key, e.Value())
|
|
}
|
|
return nil
|
|
}
|
|
if err := each(sdkdelivery.AssignerWantsBucket, func(key string, value []byte) {
|
|
if w, err := sdkdelivery.DecodeWant(value); err == nil {
|
|
wants[key] = w
|
|
}
|
|
}); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if err := each(sdkdelivery.HaveBucket, func(key string, value []byte) {
|
|
if h, err := sdkdelivery.DecodeHave(value); err == nil {
|
|
haves[key] = h
|
|
}
|
|
}); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return wants, haves, nil
|
|
}
|
|
|
|
// lastReports is inventory.LastReports with Current read from the haves: every machine's last report, and whether the machine
|
|
// has everything the assigner wants of it. A process with no bus cannot tell, and leaves Current false.
|
|
func lastReports(ctx context.Context, inv *inventory.Inventory) ([]inventory.Reported, error) {
|
|
reports, err := inv.LastReports(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
bus, err := aBus()
|
|
if err != nil {
|
|
return reports, nil // empty-on-error: no bus to read the haves from; Current stays false, which is "not known to be"
|
|
}
|
|
defer bus.Close()
|
|
js, err := jetstream.New(bus.Conn())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
wants, haves, err := readWantsAndHaves(ctx, js)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
current := currentNodes(wants, haves)
|
|
for i := range reports {
|
|
c, wanted := current[reports[i].Node]
|
|
reports[i].Current = c || !wanted && reports[i].At != nil
|
|
}
|
|
return reports, nil
|
|
}
|