novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

disk-load: compose the node's data items through the sdk's runtime reader, not a copy of it (issue 1955)

Browse Source
This commit is contained in:
jschoubben
2026-10-29 23:46:57 +01:00
parent ce8ea7bbe2
commit 9d0ebe556e
8 changed files with 94 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/disk-load/cmd/disk-load/busread.go
-233
View File
@@ -1,233 +0,0 @@
package main
// The bus as disk-load composes a node's data items from it (frombus.Reader): placement, each node's record in its own
// have key, the indexer's modules and manifests, the builder's builds and mesh-config's resolved values, all through this
// module's runtime with its own `reads`. Items name no value, so no own secret and no credential is read. This is the
// reader mesh-provisioner has too; moving it into mesh-sdk's frombus is the way to keep one (code shared between
// modules belongs in the sdk).
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"net/http"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The states this module reads to compose from.
const (
placementRead = delivery.PlacementRead
modulesRead = delivery.IndexerModule + ".modules"
manifestsRead = delivery.IndexerModule + ".manifests"
buildsRead = delivery.BuilderModule + ".builds"
haveRead = "node-engine.have"
resolvedRead = "mesh-config.resolved"
)
// entryReader is a read state as the runtime gives it.
type entryReader interface {
Get(state, key string) (*stdio.StateEntry, error)
Keys(state string) ([]string, error)
}
type busState struct{}
func (busState) Get(state, key string) (*stdio.StateEntry, error) { return stdio.State(state).Get(key) }
func (busState) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
// busReader reads the bus for frombus through a Store.
type busReader struct {
s entryReader
fetch func(ctx context.Context, url string) ([]byte, error)
}
func (r busReader) each(read string, visit func(key string, value []byte)) error {
keys, err := r.s.Keys(read)
if err != nil {
return fmt.Errorf("%s: %w", read, err)
}
for _, k := range keys {
e, err := r.s.Get(read, k)
if err != nil {
return fmt.Errorf("%s %s: %w", read, k, err)
}
if e != nil {
visit(k, e.Value)
}
}
return nil
}
func (r busReader) Placements(context.Context) (map[string]delivery.Placement, error) {
out := map[string]delivery.Placement{}
err := r.each(placementRead, func(k string, v []byte) {
node, err := delivery.ParsePlacementKey(k)
if err != nil {
return
}
if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node {
out[node] = p
}
})
return out, err
}
func (r busReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
out := map[string]delivery.NodeRecord{}
err := r.each(haveRead, func(k string, v []byte) {
node, _, isNode, err := delivery.ParseHaveKey(k)
if err != nil || !isNode {
return
}
if h, err := delivery.DecodeHaveNode(v); err == nil && h.Record != nil {
out[node] = *h.Record
}
})
return out, err
}
func (r busReader) Modules(context.Context) (map[string]delivery.Module, error) {
out := map[string]delivery.Module{}
err := r.each(modulesRead, func(k string, v []byte) {
module, err := delivery.ParseModulesKey(k)
if err != nil {
return
}
if m, err := delivery.DecodeModule(v); err == nil {
out[module] = m
}
})
return out, err
}
func (r busReader) Manifest(_ context.Context, digest string) ([]byte, error) {
key, err := delivery.ManifestsKey(digest)
if err != nil {
return nil, err
}
e, err := r.s.Get(manifestsRead, key)
if err != nil {
return nil, err
}
if e == nil {
return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
}
m, err := delivery.DecodeManifest(e.Value)
if err != nil {
return nil, err
}
return []byte(m.JSON), nil
}
func (r busReader) Values(_ context.Context, node, module string) (map[string]any, error) {
var out map[string]any
prefix := node + "." + module + "."
err := r.each(resolvedRead, func(k string, v []byte) {
name, ok := strings.CutPrefix(k, prefix)
if !ok || strings.Contains(name, ".") {
return
}
if value, laid := catalogue.LaidValue(v); laid {
if out == nil {
out = map[string]any{}
}
out[name] = value
}
})
return out, err
}
func (busReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { return nil, nil }
func (busReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
return nil, nil, nil
}
func (r busReader) Build(_ context.Context, module, fp string) (*delivery.BuilderBuild, error) {
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
return nil, err
}
e, err := r.s.Get(buildsRead, key)
if err != nil || e == nil {
return nil, err
}
b, err := delivery.DecodeBuilderBuild(e.Value)
if err != nil {
return nil, err
}
return &b, nil
}
// archiveWithin bounds one read of a templates archive.
const archiveWithin = 30 * time.Second
func (r busReader) Archive(ctx context.Context, url, digest string) ([]byte, error) {
if r.fetch == nil {
return nil, fmt.Errorf("this holder reads no archive (%s)", url)
}
fetching, cancel := context.WithTimeout(ctx, archiveWithin)
defer cancel()
body, err := r.fetch(fetching, url)
if err != nil {
return nil, err
}
sum := sha256.Sum256(body)
if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
}
return body, nil
}
func (r busReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
all, err := r.Values(ctx, node, module)
var out map[string]any
for _, k := range keys {
if v, ok := all[k]; ok {
if out == nil {
out = map[string]any{}
}
out[k] = v
}
}
return out, err
}
// httpFetch reads an archive from the artifact store over HTTP.
func httpFetch(ctx context.Context, url string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, err
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("%s answered %s", url, resp.Status)
}
return io.ReadAll(io.LimitReader(resp.Body, 256<<20))
}
// PlacedItems is every data item the modules placed on node declare, composed from the bus as the node composes itself.
func PlacedItems(ctx context.Context, s entryReader, node string) (map[string][]catalogue.PlacedDataItem, error) {
r := busReader{s: s, fetch: httpFetch}
items, unplaced, err := frombus.DataItems(ctx, r, frombus.Mesh{}, frombus.Scope{Node: node})
if err != nil {
return nil, err
}
if len(unplaced) > 0 {
logf("data items left out as not placeable: %s", strings.Join(unplaced, "; "))
}
return items, nil
}
modules/disk-load/cmd/disk-load/main.go
+16 -1
View File
@@ -11,11 +11,14 @@ import (
"context"
"fmt"
"os"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
)
func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, "[disk-load] "+format+"\n", args...) }
@@ -28,7 +31,7 @@ func main() {
node := os.Getenv("MESH_NODE")
k := &Keeper{Node: node, Store: busStore{}, Say: logf, Measurer: Measurer{Run: RootRunner, Now: time.Now},
Items: func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) {
return PlacedItems(ctx, busState{}, node)
return placedItems(ctx, node)
}}
if node != "" {
every := time.Hour
@@ -44,3 +47,15 @@ func main() {
os.Exit(1)
}
}
// placedItems is every data item the modules placed on node declare, composed from the bus as the node composes itself.
func placedItems(ctx context.Context, node string) (map[string][]catalogue.PlacedDataItem, error) {
items, unplaced, err := frombus.DataItems(ctx, fromruntime.Reader(), frombus.Mesh{}, frombus.Scope{Node: node})
if err != nil {
return nil, err
}
if len(unplaced) > 0 {
logf("data items left out as not placeable: %s", strings.Join(unplaced, "; "))
}
return items, nil
}
modules/disk-load/cmd/disk-load/readings_test.go
+56
View File
@@ -69,3 +69,59 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
t.Fatal("a cache was measured")
}
}
// The redundant storage under an item is read: a ZFS pool healthy, then degraded; an md array with a member missing; and
// plain storage, which is no redundancy at all. A dataset is measured from the filesystem's counters and a shallow item
// from its top-level entries only; a walk that meets more than its bound stops and says so.
func TestTheStorageUnderAnItemAndTheLargeMethodsAreRead(t *testing.T) {
poolHealth, statusX, fs := "ONLINE\n", "pool 'storage' is healthy\n", "storage/media zfs\n"
run := func(_ context.Context, name string, args ...string) (string, error) {
line := name + " " + strings.Join(args, " ")
switch {
case name == "test":
return "", nil
case name == "findmnt":
return fs, nil
case name == "zpool" && args[0] == "list":
return poolHealth, nil
case name == "zpool":
return statusX, nil
case name == "cat":
return "Personalities : [raid1]\nmd127 : active raid1 sdb1[1] sda1[0]\n 976630464 blocks super 1.2 [2/1] [U_]\n\nunused devices: <none>\n", nil
case name == "zfs" && args[len(args)-1] == "/storage/media/movies":
return "storage/media\t97067690722560\n", nil
case name == "zfs":
return "", os.ErrNotExist
case name == "bash" && strings.Contains(line, "-maxdepth 1"):
return "1759744800 12673\n", nil
case name == "bash":
return "123 1759700000 2000000\n141\n", nil
}
return "", nil
}
m := Measurer{Run: run, Now: time.Now}
ctx := context.Background()
got := m.Measure(ctx, catalogue.PlacedDataItem{Module: "media", ID: "movies", Path: "/storage/media/movies", Measure: catalogue.MeasureDataset})
if got.SizeBytes == nil || *got.SizeBytes != 97067690722560 || !strings.HasPrefix(got.Precision, "dataset storage/media") ||
got.LastWrite == nil || got.Redundancy == nil || got.Redundancy.Kind != "zfs" || got.Redundancy.Healthy == nil || !*got.Redundancy.Healthy {
t.Fatalf("the library from its dataset: %+v", got)
}
poolHealth, statusX = "DEGRADED\n", " pool: storage\n state: DEGRADED\nstatus: One or more devices has been removed\n"
if r := m.redundancyOf(ctx, "/storage/media/movies"); r.Healthy == nil || *r.Healthy || !strings.Contains(r.Said, "DEGRADED") {
t.Fatalf("a degraded pool: %+v", r)
}
fs = "/dev/md127 ext4\n"
if r := m.redundancyOf(ctx, "/srv/x"); r == nil || r.Kind != "md" || r.Healthy == nil || *r.Healthy {
t.Fatalf("an array missing a member: %+v", r)
}
fs = "/dev/nvme0n1p3 ext4\n"
if r := m.redundancyOf(ctx, "/var/lib/x"); r != nil {
t.Fatalf("plain storage read as redundant: %+v", r)
}
if s := m.Measure(ctx, catalogue.PlacedDataItem{Path: "/mnt/plex/config", Measure: catalogue.MeasureShallow}); s.SizeBytes != nil || !strings.HasPrefix(s.Precision, "no size") {
t.Fatalf("a shallow item: %+v", s)
}
if w := m.Measure(ctx, catalogue.PlacedDataItem{Path: "/srv/big", Measure: catalogue.MeasureWalk}); !strings.HasPrefix(w.Precision, "partial") {
t.Fatalf("a walk stopped at its bound: %+v", w)
}
}
modules/disk-load/go.mod
+1 -1
View File
@@ -2,4 +2,4 @@ module diskload
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223134-08fc70783a7c
modules/disk-load/go.sum
+14 -2
View File
@@ -1,2 +1,14 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277 h1:ajjsH4MWcx45MGYASjUSRQKhLzEringuk3TA0oxZRCU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223134-08fc70783a7c h1:inVTVNIZI4/v7/aVhL+EkKzHCDlN73hJrWKz3jI59mY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223134-08fc70783a7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
modules/disk-load/module.json
+4 -4
View File
@@ -30,11 +30,11 @@
"readings"
],
"reads": [
"mesh-config.resolved",
"mesh-delivery-assigner.placement",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.manifests",
"mesh-delivery-builder.builds",
"node-engine.have",
"mesh-config.resolved"
"mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules",
"node-engine.have"
]
}