mounts: learn where the mesh keeps data from disk-load's readings, not the controller's data (issue 1955, ADR 0571 clause 3)
This commit is contained in:
@@ -14,6 +14,7 @@ package main
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
@@ -122,22 +123,22 @@ func (m *Machine) leave(p string, v any) error {
|
||||
// Asker calls a tool on the mesh as this module.
|
||||
type Asker func(key string, body any) (json.RawMessage, error)
|
||||
|
||||
// DataVerb is the controller's verb that says where the mesh keeps data on a machine.
|
||||
const DataVerb = "seat:mesh-controller.data"
|
||||
// Readings reads disk-load's readings of one machine's data items (its state `readings`, ADR 0571 clause 3).
|
||||
type Readings func(node string) ([]health.Reading, error)
|
||||
|
||||
// LeaveData asks the controller where the mesh keeps data on this machine and leaves the paths for the
|
||||
// LeaveData reads disk-load for where the mesh keeps data on this machine and leaves the paths for the
|
||||
// process (finding 4d). A failed ask keeps the last paths, with why; with none, the process sets up
|
||||
// nothing new.
|
||||
func (m *Machine) LeaveData(node string, ask Asker, now time.Time) []string {
|
||||
if node == "" || ask == nil {
|
||||
func (m *Machine) LeaveData(node string, read Readings, now time.Time) []string {
|
||||
if node == "" || read == nil {
|
||||
return nil
|
||||
}
|
||||
previous, _ := readJSON[DataPaths](m, DataFile)
|
||||
out := DataPaths{Machine: node, Asked: now.Format(time.RFC3339)}
|
||||
raw, err := ask(DataVerb, map[string]any{"machine": node})
|
||||
readings, err := read(node)
|
||||
var paths []string
|
||||
if err == nil {
|
||||
paths, err = DataFrom(raw)
|
||||
paths, err = PathsFrom(readings)
|
||||
}
|
||||
var said []string
|
||||
if err != nil {
|
||||
|
||||
@@ -6,9 +6,9 @@ package main
|
||||
// mesh keeps data in on this machine, from the controller's `data` verb over the bus.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -63,44 +63,16 @@ func sortWants(w []Want) {
|
||||
}
|
||||
}
|
||||
|
||||
// DataFrom reads the paths out of mesh-controller.data's answer: `answer.data`, `data`, or the same as
|
||||
// text in `output`. An answer with no data list is an error, never an empty list: an empty list would
|
||||
// let every mount point through.
|
||||
func DataFrom(raw json.RawMessage) ([]string, error) {
|
||||
type item struct {
|
||||
Path string `json:"path"`
|
||||
}
|
||||
var whole struct {
|
||||
Answer *struct {
|
||||
Data *[]item `json:"data"`
|
||||
} `json:"answer"`
|
||||
Data *[]item `json:"data"`
|
||||
Output string `json:"output"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &whole); err != nil {
|
||||
return nil, fmt.Errorf("the controller's data answer is not JSON: %v", err)
|
||||
}
|
||||
var items *[]item
|
||||
switch {
|
||||
case whole.Answer != nil && whole.Answer.Data != nil:
|
||||
items = whole.Answer.Data
|
||||
case whole.Data != nil:
|
||||
items = whole.Data
|
||||
case whole.Output != "":
|
||||
var inner struct {
|
||||
Data *[]item `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(whole.Output), &inner); err == nil && inner.Data != nil {
|
||||
items = inner.Data
|
||||
}
|
||||
}
|
||||
if items == nil {
|
||||
return nil, errors.New("the controller's data answer names no data list")
|
||||
// PathsFrom is the paths disk-load's readings name: where the mesh keeps data here. No reading is an error, never an
|
||||
// empty list: an empty list would let every mount point through.
|
||||
func PathsFrom(readings []health.Reading) ([]string, error) {
|
||||
if len(readings) == 0 {
|
||||
return nil, errors.New("disk-load keeps no reading of this machine's data yet")
|
||||
}
|
||||
paths := []string{}
|
||||
for _, it := range *items {
|
||||
if strings.HasPrefix(it.Path, "/") {
|
||||
paths = append(paths, filepath.Clean(it.Path))
|
||||
for _, r := range readings {
|
||||
if strings.HasPrefix(r.Path, "/") {
|
||||
paths = append(paths, filepath.Clean(r.Path))
|
||||
}
|
||||
}
|
||||
return paths, nil
|
||||
|
||||
@@ -13,8 +13,10 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -62,13 +64,36 @@ func shareOnTheBus(m *Machine) {
|
||||
_, err := stdio.State(SharesState).Put(key, value)
|
||||
return err
|
||||
}
|
||||
ask := func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }
|
||||
read := func(node string) ([]health.Reading, error) {
|
||||
st := stdio.State(health.ReadingsRead)
|
||||
keys, err := st.Keys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []health.Reading
|
||||
for _, k := range keys {
|
||||
if !strings.HasPrefix(k, node+".") {
|
||||
continue
|
||||
}
|
||||
e, err := st.Get(k)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if e == nil {
|
||||
continue
|
||||
}
|
||||
if r, err := health.DecodeReading(e.Value); err == nil {
|
||||
out = append(out, r)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
last, lastSaid := "", ""
|
||||
for n := 0; ; n++ {
|
||||
said := m.Share(node, get, put, &last, time.Now())
|
||||
// Where the mesh keeps data here changes only with an assignment: asked at start and every five minutes.
|
||||
if n%10 == 0 {
|
||||
said = append(said, m.LeaveData(node, ask, time.Now())...)
|
||||
said = append(said, m.LeaveData(node, read, time.Now())...)
|
||||
}
|
||||
if s := fmt.Sprint(said); len(said) > 0 && s != lastSaid {
|
||||
fmt.Fprintln(os.Stderr, "mounts:", s)
|
||||
|
||||
@@ -77,12 +77,10 @@ func TestTheManifestNamesWhatTheBundleServesAndWhatItNeeds(t *testing.T) {
|
||||
if strings.Join(m.Requires, ",") != "nfs-share" {
|
||||
t.Errorf("requires %v", m.Requires)
|
||||
}
|
||||
if strings.Join(m.Reads, ",") != ExportsState {
|
||||
t.Errorf("the module reads nfs-server's exports from the bus: reads %v", m.Reads)
|
||||
}
|
||||
// It calls one verb: the controller's, for where the mesh keeps data on this machine (review finding 4d).
|
||||
if strings.Join(m.Invokes, ",") != DataVerb {
|
||||
t.Errorf("invokes %v, want %s", m.Invokes, DataVerb)
|
||||
// It reads nfs-server's exports, and disk-load's readings for where the mesh keeps data on this machine (review
|
||||
// finding 4d, ADR 0571 clause 3); it calls no verb.
|
||||
if strings.Join(m.Reads, ",") != "disk-load.readings,"+ExportsState || len(m.Invokes) != 0 {
|
||||
t.Errorf("reads %v, invokes %v", m.Reads, m.Invokes)
|
||||
}
|
||||
if !strings.Contains(raw, `node=${machine:name}`) {
|
||||
t.Error("the config gives the process this machine's name, so it never mounts a share onto the machine that serves it")
|
||||
|
||||
@@ -5,6 +5,8 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -198,18 +200,13 @@ func TestWithoutTheMeshsDataNothingNewIsSetUp(t *testing.T) {
|
||||
func TestTheBundleLeavesTheMeshsDataForTheProcess(t *testing.T) {
|
||||
f := newSystem(t)
|
||||
f.write(ConfigFile, libraryOnly)
|
||||
answer := json.RawMessage(`{"answer":{"data":[{"machine":"workstation","module":"plex","item":"data","path":"/mnt/plex/data"}]},"ok":true}`)
|
||||
paths, err := DataFrom(answer)
|
||||
rs := []health.Reading{{V: 1, Node: "workstation", PlacedDataItem: catalogue.PlacedDataItem{Module: "plex", ID: "data", Path: "/mnt/plex/data"}}}
|
||||
paths, err := PathsFrom(rs)
|
||||
if err != nil || len(paths) != 1 || paths[0] != "/mnt/plex/data" {
|
||||
t.Fatalf("paths %q %v", paths, err)
|
||||
}
|
||||
// The controller's verbs also answer as text in output.
|
||||
paths, err = DataFrom(json.RawMessage(`{"ok":true,"output":"{\"data\":[{\"path\":\"/srv/x\"}]}"}`))
|
||||
if err != nil || len(paths) != 1 || paths[0] != "/srv/x" {
|
||||
t.Fatalf("paths from output %q %v", paths, err)
|
||||
}
|
||||
if _, err := DataFrom(json.RawMessage(`{"ok":false}`)); err == nil {
|
||||
t.Error("an answer with no data list is an error, not an empty list")
|
||||
if _, err := PathsFrom(nil); err == nil {
|
||||
t.Error("no reading is an error, not an empty list")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -287,23 +284,17 @@ func callVerb(t *testing.T, m *Machine, name string, args map[string]any) (any,
|
||||
func TestTheBundleKeepsTheLastDataWhenAnAskFails(t *testing.T) {
|
||||
f := newSystem(t)
|
||||
m := f.machine()
|
||||
var asked []string
|
||||
ok := func(key string, body any) (json.RawMessage, error) {
|
||||
raw, _ := json.Marshal(body)
|
||||
asked = append(asked, key+" "+string(raw))
|
||||
return json.RawMessage(`{"answer":{"data":[{"path":"/mnt/plex/data"}]}}`), nil
|
||||
ok := func(node string) ([]health.Reading, error) {
|
||||
return []health.Reading{{V: 1, Node: node, PlacedDataItem: catalogue.PlacedDataItem{Module: "plex", ID: "data", Path: "/mnt/plex/data"}}}, nil
|
||||
}
|
||||
if said := m.LeaveData("workstation", ok, time.Now()); len(said) != 0 {
|
||||
t.Fatalf("said %q", said)
|
||||
}
|
||||
if len(asked) != 1 || asked[0] != `seat:mesh-controller.data {"machine":"workstation"}` {
|
||||
t.Errorf("asked %q", asked)
|
||||
}
|
||||
failing := func(string, any) (json.RawMessage, error) { return nil, os.ErrDeadlineExceeded }
|
||||
failing := func(string) ([]health.Reading, error) { return nil, os.ErrDeadlineExceeded }
|
||||
said := m.LeaveData("workstation", failing, time.Now())
|
||||
d, err := readJSON[DataPaths](m, DataFile)
|
||||
if err != nil || len(d.Paths) != 1 || d.Paths[0] != "/mnt/plex/data" || d.Error == "" || len(said) == 0 {
|
||||
t.Errorf("a failed ask keeps the last paths and says why: %+v %v %q", d, err, said)
|
||||
t.Errorf("a failed read keeps the last paths and says why: %+v %v %q", d, err, said)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user