diff --git a/checks/health/go.mod b/checks/health/go.mod index 9a268a24..fd5c93e8 100644 --- a/checks/health/go.mod +++ b/checks/health/go.mod @@ -2,4 +2,4 @@ module checks/health 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 diff --git a/checks/health/go.sum b/checks/health/go.sum index fc894705..fbfd7431 100644 --- a/checks/health/go.sum +++ b/checks/health/go.sum @@ -1,2 +1,2 @@ -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= diff --git a/modules/disk-load/cmd/disk-load/busread.go b/modules/disk-load/cmd/disk-load/busread.go deleted file mode 100644 index 5542fca0..00000000 --- a/modules/disk-load/cmd/disk-load/busread.go +++ /dev/null @@ -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 -} diff --git a/modules/disk-load/cmd/disk-load/main.go b/modules/disk-load/cmd/disk-load/main.go index 6946af88..2eeff0b8 100644 --- a/modules/disk-load/cmd/disk-load/main.go +++ b/modules/disk-load/cmd/disk-load/main.go @@ -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 +} diff --git a/modules/disk-load/cmd/disk-load/readings_test.go b/modules/disk-load/cmd/disk-load/readings_test.go index e4970ca1..e6b23514 100644 --- a/modules/disk-load/cmd/disk-load/readings_test.go +++ b/modules/disk-load/cmd/disk-load/readings_test.go @@ -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: \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) + } +} diff --git a/modules/disk-load/go.mod b/modules/disk-load/go.mod index 936c7c3e..530147e3 100644 --- a/modules/disk-load/go.mod +++ b/modules/disk-load/go.mod @@ -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 diff --git a/modules/disk-load/go.sum b/modules/disk-load/go.sum index fc894705..2d4cc3b2 100644 --- a/modules/disk-load/go.sum +++ b/modules/disk-load/go.sum @@ -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= diff --git a/modules/disk-load/module.json b/modules/disk-load/module.json index edc915d0..4b4313ca 100644 --- a/modules/disk-load/module.json +++ b/modules/disk-load/module.json @@ -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" ] }