diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes.go b/modules/mesh-self-check/cmd/mesh-self-check/probes.go index 84760902..e4dbce45 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes.go @@ -34,7 +34,7 @@ const ( // `reads`, which its test holds to these. var ( Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb} - Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead, HealthRead} + Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead, HealthRead, ReadingsRead, GoodRead, ModulesRead, ManifestsRead} ) // Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe @@ -65,7 +65,9 @@ func Registry() []Probe { Deferred: "the bindings of provisions to their consumers: " + storeOnly}, {ID: "D13", Asserts: "every item of data a machine declares is measured, is there, holds what it held, and is " + "backed up within its bound", From: "issue 273, ADR 0233", Kind: "data-shrank", - Deferred: "the data items' measurements and their history: " + storeOnly}, + Raises: []string{kindDataMissing, kindDataQuiet, kindBackupStale, kindArrayDegraded, kindProtectionMissing, + kindDataUnmeasured}, + Run: probeData}, {ID: "D15", Asserts: "no user of the bus had its connection dropped more than twelve times in the last hour: the " + "bus module's nats_closed_connections", From: "issue 327", Kind: "bus-reconnects", Run: probeReconnects}, {ID: "DA", Asserts: "every machine that names an agent account has it judged, on its node-engine's newest " + @@ -76,9 +78,7 @@ func Registry() []Probe { Deferred: "the node-engines' statements of root and the tools that run as root: " + storeOnly}, {ID: "D-fill", Asserts: "every module's bucket and log holds less than three quarters of its cap; one said " + "filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", - Kind: "bucket-or-log-filling", Raises: []string{"bucket-or-log-unread"}, - Deferred: "each bucket's and log's cap: the bus module's nats_buckets answers bytes and no cap, and the " + - "declared caps are " + storeOnly}, + Kind: "bucket-or-log-filling", Run: probeFill}, {ID: "D-heard", Asserts: "no grant of a restricted stream names a module that composes nothing for it", From: "ADR 0376 §2", Kind: "grant-composes-nothing", Deferred: "the grants of restricted streams with what each machine composed for them: " + storeOnly}, diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go new file mode 100644 index 00000000..23f74c52 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go @@ -0,0 +1,342 @@ +package main + +// D13, the data the modules declare (novox/hq ADR 0233, ADR 0571 clause 3, ADR 0564 §6): every item of data a machine +// declares is measured, is there, holds what it held, and is backed up within its bound. The controller composed what +// every machine declares and asked each machine's backup holder what it measured; the facts now live on the bus: the +// readings disk-load keeps (its state `readings`, with the item's seven-day peak), the last good night per item restic +// keeps (its state `good`), and the manifests the indexer holds (for the bounds an item declares). Each kind, key and +// word is the controller's (mesh-controller cmd/mesh-controller/data.go), so the condition the keeper raises from it is +// the one the controller raised. + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The states D13 reads (declared under `reads`). +const ( + ReadingsRead = health.ReadingsRead + GoodRead = health.GoodRead + ModulesRead = delivery.IndexerModule + ".modules" + ManifestsRead = delivery.IndexerModule + ".manifests" +) + +// The kinds D13 raises. +const ( + kindDataShrank = "data-shrank" + kindEmptyReplacement = "empty-replacement" + kindDataHeldTwice = "data-held-twice" + kindDataQuiet = "data-quiet" + kindBackupStale = "backup-stale" + kindDataUnmeasured = "data-unmeasured" + kindDataMissing = "data-missing" + kindArrayDegraded = "array-degraded" + kindProtectionMissing = "protection-missing" + kindCleanupWaiting = "cleanup-waiting" + shrinkWindowDays = 7 + shrinkFloor int64 = 16 << 20 +) + +// dataFacts is everything D13 judges: the readings, the last good nights, each module's manifest, and the machines whose +// backup holder was heard from. +type dataFacts struct { + readings []health.Reading + good map[string]time.Time // by health.GoodKey + manifests map[string]catalogue.Manifest + // unmeasured are the machines that hold node-backup, are heard from, and have no reading of anything. + unmeasured []string + now time.Time +} + +// dataFacts reads them from the bus. +func (c *Checker) dataFacts() (dataFacts, error) { + f := dataFacts{good: map[string]time.Time{}, manifests: map[string]catalogue.Manifest{}, now: c.Now()} + keys, err := c.Store.Keys(ReadingsRead) + if err != nil { + return f, fmt.Errorf("disk-load's readings cannot be read: %w", err) + } + byNode := map[string]int{} + for _, k := range keys { + raw, ok, err := c.Store.Get(ReadingsRead, k) + if err != nil { + return f, fmt.Errorf("the reading %s cannot be read: %w", k, err) + } + if !ok { + continue + } + r, err := health.DecodeReading(raw) + if err != nil { + continue + } + f.readings = append(f.readings, r) + byNode[r.Node]++ + } + gkeys, err := c.Store.Keys(GoodRead) + if err != nil { + return f, fmt.Errorf("restic's last good nights cannot be read: %w", err) + } + for _, k := range gkeys { + raw, ok, err := c.Store.Get(GoodRead, k) + if err != nil { + return f, fmt.Errorf("the last good night %s cannot be read: %w", k, err) + } + if g, derr := health.DecodeGoodRow(raw); ok && derr == nil && !g.LastGood.IsZero() { + f.good[health.GoodKey(g.Node, g.Module, g.Item)] = g.LastGood + } + } + mods, err := c.Store.Keys(ModulesRead) + if err != nil { + return f, fmt.Errorf("the indexer's modules cannot be read: %w", err) + } + for _, name := range mods { + raw, ok, err := c.Store.Get(ModulesRead, name) + if err != nil || !ok { + continue + } + // Only the manifest's digest of the indexer's module, and the manifest's text: what the bounds are read from. + var m struct { + Manifest string `json:"manifest"` + } + if json.Unmarshal(raw, &m) != nil || m.Manifest == "" { + continue + } + mraw, ok, err := c.Store.Get(ManifestsRead, m.Manifest) + if err != nil || !ok { + continue + } + var doc struct { + JSON string `json:"manifest"` + } + if json.Unmarshal(mraw, &doc) != nil { + continue + } + var man catalogue.Manifest + if json.Unmarshal([]byte(doc.JSON), &man) == nil { + f.manifests[name] = man + } + } + heard, err := c.heard() + if err != nil { + return f, err + } + holders, err := c.holdersOf("node-backup") + if err != nil { + return f, err + } + for _, n := range holders { + if heard[n] && byNode[n] == 0 { + f.unmeasured = append(f.unmeasured, n) + } + } + return f, nil +} + +// probeData is D13. +func probeData(ctx context.Context, c *Checker) ([]Observation, error) { + f, err := c.dataFacts() + if err != nil { + return nil, err + } + return dataFindings(f), nil +} + +func severityOf(class string) health.Severity { + if class == catalogue.ClassIrreplaceable { + return Urgent + } + return Warning +} + +func keyOfItem(machine, module, item string) string { return machine + "." + module + "." + item } + +func comparable(precision string) bool { + return precision == "" || precision == "exact" || strings.HasPrefix(precision, "dataset ") +} + +func datasetOf(precision string) string { + rest, ok := strings.CutPrefix(precision, "dataset ") + if !ok { + return "" + } + name, _, _ := strings.Cut(rest, ":") + return name +} + +func orUnknownPath(p string) string { + if p == "" { + return "a path its backup holder never named" + } + return p +} + +func sizeWords(size *int64) string { + if size == nil || *size < 0 { + return "size unknown" + } + b := float64(*size) + for _, unit := range []string{"B", "KB", "MB", "GB"} { + if b < 1024 || unit == "GB" { + if unit == "B" { + return fmt.Sprintf("%d B", *size) + } + return fmt.Sprintf("%.1f %s", b, unit) + } + b /= 1024 + } + return "" +} + +// firstSeen is when an item was first read: the oldest day its seven-day peak is made of, else its measurement. A lower +// bound on how long it has been declared (the controller kept the day it first saw it). +func firstSeen(r health.Reading, now time.Time) time.Time { + if len(r.Days) > 0 { + if t, err := time.Parse("2006-01-02", r.Days[0].Day); err == nil { + return t + } + } + if r.MeasuredAt != nil { + return *r.MeasuredAt + } + return now +} + +func obs(kind string, id, machine string, sev health.Severity, summary, said string) Observation { + return Observation{Kind: kind, Token: kind, Severity: sev, Summary: summary, Said: said, + Subject: Subject{Scope: ScopeMachine, ID: id, Machine: machine}} +} + +// dataFindings is every condition the data on record raises now: a function of what is known, as the controller's was. +func dataFindings(f dataFacts) []Observation { + var out []Observation + arrays := map[string][]health.Reading{} + type shrunk struct { + machine, dataset, class string + size, peak int64 + items []string + } + shrunkDatasets := map[string]shrunk{} + for _, m := range f.unmeasured { + o := obs(kindDataUnmeasured, m, m, Warning, fmt.Sprintf("%s's backup holder did not say what it measured of the data "+ + "declared there, so nothing about that data is known", m), "disk-load keeps no reading of this machine's data") + o.Confirm = true + out = append(out, o) + } + rs := append([]health.Reading(nil), f.readings...) + sort.Slice(rs, func(i, j int) bool { + return keyOfItem(rs[i].Node, rs[i].Module, rs[i].ID) < keyOfItem(rs[j].Node, rs[j].Module, rs[j].ID) + }) + for _, r := range rs { + class := r.Class + item, declared := f.manifests[r.Module].DataItem(r.ID) + id := keyOfItem(r.Node, r.Module, r.ID) + if !catalogue.Watched(class) { + continue + } + severity := severityOf(class) + if r.Redundancy != nil { + where := r.Node + "/" + r.Redundancy.Kind + ":" + r.Redundancy.Where + arrays[where] = append(arrays[where], r) + } else if declared && item.Redundancy != "" && r.MeasuredAt != nil && r.Error == "" { + out = append(out, obs(kindProtectionMissing, id, r.Node, severity, + fmt.Sprintf("%s of %s on %s (%s) is said to be protected by the redundancy of the storage it is on, "+ + "and it is on nothing the backup holder can read as redundant: it has no protection the mesh can see", + r.ID, r.Module, r.Node, class), + orUnknownPath(r.Path)+" is on no redundant storage the backup holder can read")) + } + peak := r.Peak + shrank := peak != nil && r.SizeBytes != nil && comparable(r.Precision) && *r.SizeBytes*2 < *peak && *peak-*r.SizeBytes >= shrinkFloor + switch { + case r.Error != "" && strings.Contains(r.Error, "does not exist"): + out = append(out, obs(kindDataMissing, id, r.Node, severity, + fmt.Sprintf("%s of %s on %s (%s) is gone: where it was declared, nothing exists any more", r.ID, r.Module, r.Node, class), + orUnknownPath(r.Path)+" does not exist: "+firstLine(r.Error))) + case shrank && datasetOf(r.Precision) != "": + // Several items on one dataset share its size: one condition for the dataset, as loud as the most + // precious item on it. + k := r.Node + "/" + datasetOf(r.Precision) + ds := shrunkDatasets[k] + ds.machine, ds.dataset, ds.size, ds.peak = r.Node, datasetOf(r.Precision), *r.SizeBytes, *peak + ds.class = catalogue.StricterClass(ds.class, class) + ds.items = append(ds.items, r.Module+"/"+r.ID) + shrunkDatasets[k] = ds + case shrank: + out = append(out, obs(kindDataShrank, id, r.Node, severity, + fmt.Sprintf("%s of %s on %s (%s) shrank to %s from %s within %d days — more than half of what it "+ + "held is gone. If that was meant, silence this with why; if not, `node-backup.restore` puts the last "+ + "good copy beside it", r.ID, r.Module, r.Node, class, sizeWords(r.SizeBytes), sizeWords(peak), shrinkWindowDays), "")) + } + if !declared { + continue + } + if within := item.ActiveWithin(); within > 0 && r.LastWrite != nil && f.now.Sub(*r.LastWrite) > within { + out = append(out, obs(kindDataQuiet, id, r.Node, severity, + fmt.Sprintf("%s of %s on %s is written all the time, and has not been since %s (its bound is %s): "+ + "whatever writes it has stopped", r.ID, r.Module, r.Node, r.LastWrite.UTC().Format(time.RFC3339), within), "")) + } + // A backup is required of what is irreplaceable and copied; of what is valuable it is the standard + // plan, said only where the machine was measured — where a holder is there to take it. + if item.BackedUp() && (class == catalogue.ClassIrreplaceable || r.MeasuredAt != nil) { + within := item.BackupWithin() + last, has := f.good[health.GoodKey(r.Node, r.Module, r.ID)] + switch { + case !has && f.now.Sub(firstSeen(r, f.now)) > within: + out = append(out, obs(kindBackupStale, id, r.Node, severity, + fmt.Sprintf("%s of %s on %s is %s and has no good backup on record, %s after it was first "+ + "declared — is node-backup held there, and do its nights succeed? (`node-backup.backed-up`)", + r.ID, r.Module, r.Node, class, f.now.Sub(firstSeen(r, f.now)).Round(time.Hour)), "")) + case has && f.now.Sub(last) > within: + out = append(out, obs(kindBackupStale, id, r.Node, severity, + fmt.Sprintf("%s of %s on %s is %s and its newest good backup is from %s, older than its bound "+ + "of %s", r.ID, r.Module, r.Node, class, last.UTC().Format(time.RFC3339), within), "")) + } + } + } + keys := make([]string, 0, len(shrunkDatasets)) + for k := range shrunkDatasets { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + ds := shrunkDatasets[k] + out = append(out, obs(kindDataShrank, ds.machine+".dataset."+strings.ReplaceAll(ds.dataset, "/", "-"), ds.machine, + severityOf(ds.class), fmt.Sprintf("the dataset on %s that holds %s shrank to %s from %s within %d days — more than "+ + "half of what it held is gone", ds.machine, strings.Join(ds.items, ", "), sizeWords(&ds.size), sizeWords(&ds.peak), + shrinkWindowDays), "the dataset "+ds.dataset)) + } + // The redundant storage watched data is on: one condition per array, as loud as the most precious item on it — + // the array, not each item, is what degrades. + wheres := make([]string, 0, len(arrays)) + for w := range arrays { + wheres = append(wheres, w) + } + sort.Strings(wheres) + for _, where := range wheres { + items := arrays[where] + red := items[0].Redundancy + if red.Healthy != nil && *red.Healthy { + continue + } + class, machine := catalogue.ClassValuable, items[0].Node + var names []string + for _, r := range items { + class = catalogue.StricterClass(class, r.Class) + names = append(names, r.Module+"/"+r.ID) + } + state := "could not be read" + if red.Healthy != nil { + state = "is NOT healthy" + } + out = append(out, obs(kindArrayDegraded, machine+".array."+strings.NewReplacer("/", "-", ":", "-").Replace(red.Kind+"-"+red.Where), + machine, severityOf(class), fmt.Sprintf("the %s storage on %s that protects %s %s", red.Kind, machine, + strings.Join(names, ", "), state), fmt.Sprintf("the %s storage %s %s: %s", red.Kind, red.Where, state, firstLine(red.Said)))) + } + return out +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go index 8e5e6401..6b76380b 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go @@ -7,9 +7,12 @@ import ( "slices" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) // bus is the bus module's answers as a probe asks them: every mesh stream and consumer as defined, unless changed. @@ -21,6 +24,8 @@ type bus struct { asksFail error // banned is each node's answer to node-intrusion-prevention.banned; a node absent does not answer. banned map[string]string + // fills are rows nats_streams answers beside the mesh's streams: a module's bucket or log with its size and cap. + fills []fillRow } func (b bus) rows() map[string]*streamRow { @@ -73,6 +78,16 @@ func (b bus) ask(key string, body any) (json.RawMessage, error) { detail = append(detail, *r) } } + if stream == "" && len(b.fills) > 0 { + var rows []any + for _, r := range detail { + rows = append(rows, r) + } + for _, f := range b.fills { + rows = append(rows, f) + } + return json.Marshal(map[string]any{"detail": rows}) + } if stream != "" && len(detail) == 0 { return nil, errors.New("no stream called " + stream) } @@ -152,6 +167,40 @@ func TestTheBusProbes(t *testing.T) { "ace": `{"banned":[]}`}}, nil, "", machines()}, {"D8 a heard holder that does not answer is no pass", probeBans, bus{banned: map[string]string{}}, nil, "a ban list could not be read", machines()}, + // D13 (issue 273, ADR 0233): the readings disk-load keeps and the last good nights restic keeps, against the bounds + // the manifests declare; the kinds, keys and words are the controller's. + {"D13 all held", probeData, bus{}, nil, "", dataMesh(func(*dataRig) {})}, + {"D13 shrank, gone, quiet, unprotected and stale", probeData, bus{}, []string{ + "machine.ace.pg.data.data-quiet", "machine.ace.pg.data.backup-stale", "machine.ace.pg.dumps.data-missing", + "machine.ace.plex.movies.protection-missing", "machine.ace.plex.thumbs.data-shrank", + "machine.ace.array.md-md127.array-degraded"}, + "", dataMesh(func(r *dataRig) { + r.size("ace", "pg", "data", 100<<20, 100<<20).write("ace", "pg", "data", -80*time.Hour) + r.gone("ace", "pg", "dumps") + r.size("ace", "plex", "thumbs", 10<<20, 600<<20) + r.size("ace", "plex", "movies", 1<<30, 1<<30).noArray("ace", "plex", "movies") + r.size("ace", "plex", "library", 1<<30, 1<<30).array("ace", "plex", "library", "md", "md127", false) + r.lastGood("ace", "pg", "data", -100*time.Hour) + })}, + {"D13 a dataset shrank once for its items", probeData, bus{}, []string{"machine.ace.dataset.storage-media.data-shrank"}, + "", dataMesh(func(r *dataRig) { + r.size("ace", "plex", "thumbs", 10<<20, 600<<20).dataset("ace", "plex", "thumbs", "storage/media") + r.size("ace", "plex", "meta", 10<<20, 600<<20).dataset("ace", "plex", "meta", "storage/media") + })}, + {"D13 a machine with a backup holder and no reading", probeData, bus{}, []string{"machine.g14.data-unmeasured"}, "", + dataMesh(func(r *dataRig) { r.holder("g14") })}, + // D-fill (ADR 0297 §6): a module's bucket or log at three quarters of its cap, kept until under seven tenths. + {"D-fill all with room", probeFill, bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 74, MaxBytes: 100}, + {Name: "LOG_mesh-delivery-builder_build-log", Bytes: 10, MaxBytes: 100}, {Name: "KV_nocap_x", Bytes: 999}}}, nil, "", nil}, + {"D-fill a bucket and a log filling", probeFill, bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 75, MaxBytes: 100}, + {Name: "LOG_mesh-delivery-builder_build-log", Bytes: 99, MaxBytes: 100}}}, + []string{"bus.KV_mesh-issues_issues.filling", "bus.LOG_mesh-delivery-builder_build-log.filling"}, "", nil}, + {"D-fill between the thresholds raises nothing new", probeFill, + bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 72, MaxBytes: 100}}}, nil, "", nil}, + {"D-fill between the thresholds keeps what is said", probeFill, + bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 72, MaxBytes: 100}}}, + []string{"bus.KV_mesh-issues_issues.filling"}, "", filling("KV_mesh-issues_issues")}, + {"D-fill the bus module silent", probeFill, bus{asksFail: errors.New("no responders")}, nil, "did not answer", nil}, // DA (ADR 0266): the agent account each machine names, judged on its engine's report while it is heard from. {"DA confined", probeAgentAccounts, bus{}, nil, "", agents(map[string]string{"novox": stateHealthy})}, {"DA a way to root found", probeAgentAccounts, bus{}, []string{"machine.novox.agent-root"}, "", @@ -166,7 +215,7 @@ func TestTheBusProbes(t *testing.T) { agents(map[string]string{"novox": "old"})}, } for _, tc := range cases { - c := &Checker{Ask: tc.bus.ask, Store: tc.mesh} + c := &Checker{Ask: tc.bus.ask, Store: tc.mesh, Now: func() time.Time { return dataNow }} if tc.mesh == nil { c.Store = newStore() } @@ -205,6 +254,16 @@ func machines() *memStore { return s } +// filling is a store holding the self-check's written fill finding of a stream: the condition open. +func filling(stream string) *memStore { + s := newStore() + o := Observation{Kind: "bucket-or-log-filling", Token: "filling", Subject: Subject{Scope: ScopeBus, ID: stream}, + Summary: "filling", Source: "D-fill"} + raw, _ := health.EncodeFinding(o.Finding("run", time.Now())) + _ = s.Put(FindingsState, o.Key(), json.RawMessage(raw)) + return s +} + // agents is machines() where each node named names the agent account "agent" and its report holds the verdict on it: // a state, "" for none, "pending" for one waiting for its search, "old" for an engine older than the judging. func agents(verdicts map[string]string) *memStore { @@ -255,3 +314,115 @@ func TestTheRegistryIsADR0564sTable(t *testing.T) { t.Errorf("the registry holds %v, ADR 0564 moves %v", ids, moved) } } + +// ---- D13's rig ------------------------------------------------------------------------------------------------ + +var dataNow = time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC) + +// dataRig is the mesh D13 reads: a manifest per module with the items it declares, a reading per item, a last good night. +type dataRig struct { + s *memStore + readings map[string]*health.Reading + good map[string]time.Time +} + +func (r *dataRig) key(node, module, item string) string { return health.ReadingKey(node, module, item) } + +func (r *dataRig) reading(node, module, item string) *health.Reading { + k := r.key(node, module, item) + if r.readings[k] == nil { + at := dataNow + r.readings[k] = &health.Reading{V: health.ReadingVersion, Node: node, PlacedDataItem: catalogue.PlacedDataItem{Module: module, + ID: item, Class: catalogue.ClassValuable, Path: "/srv/" + module + "/" + item}, Measured: health.Measured{MeasuredAt: &at, Precision: "exact"}} + } + return r.readings[k] +} + +func (r *dataRig) size(node, module, item string, size, peak int64) *dataRig { + x := r.reading(node, module, item) + x.SizeBytes, x.Peak = &size, &peak + return r +} + +func (r *dataRig) write(node, module, item string, ago time.Duration) *dataRig { + t := dataNow.Add(ago) + r.reading(node, module, item).LastWrite = &t + return r +} + +func (r *dataRig) gone(node, module, item string) *dataRig { + r.reading(node, module, item).Error = "/srv/" + module + "/" + item + " does not exist" + return r +} + +func (r *dataRig) dataset(node, module, item, name string) *dataRig { + r.reading(node, module, item).Precision = "dataset " + name + ": its whole size" + return r +} + +func (r *dataRig) array(node, module, item, kind, where string, healthy bool) *dataRig { + r.reading(node, module, item).Redundancy = &health.Redundancy{Kind: kind, Where: where, Healthy: &healthy, Said: "a member is missing"} + return r +} + +func (r *dataRig) noArray(node, module, item string) *dataRig { + r.reading(node, module, item).Redundancy = nil + return r +} + +func (r *dataRig) lastGood(node, module, item string, ago time.Duration) *dataRig { + r.good[r.key(node, module, item)] = dataNow.Add(ago) + return r +} + +func (r *dataRig) holder(node string) { + key, _ := delivery.SeatHoldersKey("node-backup", node) + _ = r.s.Put(SeatHoldersRead, key, delivery.SeatHolders{V: 1, Seat: "node-backup", Scope: "node", Node: node, + Holders: []delivery.SeatHolder{{Module: "restic", Node: node}}}) + _ = r.s.Put(PresenceRead, node, delivery.Presence{V: 1, Online: true}) +} + +// dataMesh builds the mesh: pg declares data (backed up within 48 h, written all the time within 24 h) and dumps; plex +// declares movies (protected by redundancy), thumbs, meta and library (all valuable); every item is read as held, and what +// is given changes it. +func dataMesh(change func(*dataRig)) *memStore { + r := &dataRig{s: newStore(), readings: map[string]*health.Reading{}, good: map[string]time.Time{}} + r.holder("ace") + declare := func(module string, items map[string]catalogue.DataItem) { + var man catalogue.Manifest + man.Module = module + man.Data = &catalogue.Data{} + for id, it := range items { + it.ID = id + man.Data.Own = append(man.Data.Own, it) + } + raw, _ := json.Marshal(man) + digest := delivery.ManifestDigest(raw) + _ = r.s.Put(ManifestsRead, digest, delivery.Manifest{V: 1, Module: module, JSON: string(raw)}) + _ = r.s.Put(ModulesRead, module, delivery.Module{V: 1, Manifest: digest}) + } + declare("pg", map[string]catalogue.DataItem{ + "data": {Class: catalogue.ClassValuable, Within: "48h", Active: "24h"}, + "dumps": {Class: catalogue.ClassValuable}}) + declare("plex", map[string]catalogue.DataItem{ + "movies": {Class: catalogue.ClassValuable, Redundancy: "the array"}, + "thumbs": {Class: catalogue.ClassValuable}, + "meta": {Class: catalogue.ClassValuable}, + "library": {Class: catalogue.ClassValuable, Redundancy: "the array"}}) + for _, i := range [][2]string{{"pg", "data"}, {"pg", "dumps"}, {"plex", "movies"}, {"plex", "thumbs"}, {"plex", "meta"}, {"plex", "library"}} { + r.size("ace", i[0], i[1], 1<<30, 1<<30).lastGood("ace", i[0], i[1], -time.Hour).write("ace", i[0], i[1], -time.Minute) + if i[0] == "plex" && (i[1] == "movies" || i[1] == "library") { + r.array("ace", i[0], i[1], "md", "md127", true) + } + } + change(r) + for k, x := range r.readings { + _ = r.s.Put(ReadingsRead, k, x) + } + for k, t := range r.good { + node, rest, _ := strings.Cut(k, ".") + module, item, _ := strings.Cut(rest, ".") + _ = r.s.Put(GoodRead, k, health.GoodRow{V: health.GoodVersion, Node: node, Module: module, Item: item, CoveredBy: "/x", LastGood: t}) + } + return r.s +} diff --git a/modules/mesh-self-check/module.json b/modules/mesh-self-check/module.json index cc47e9c5..a1a892ea 100644 --- a/modules/mesh-self-check/module.json +++ b/modules/mesh-self-check/module.json @@ -22,7 +22,11 @@ "mesh-delivery-assigner.placement", "mesh-delivery-assigner.seat-holders", "mesh-delivery-watchman.presence", - "mesh-host.health" + "mesh-host.health", + "disk-load.readings", + "restic.good", + "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.manifests" ], "state": [ "findings",