package main import ( "bytes" "context" "encoding/json" "errors" "fmt" "git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings" "git.novox.be/novox/mesh-sdk/go/catalogue" "os" "os/exec" "sort" "strings" "sync" "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" ) // Store is the state `readings` as the keeper uses it. type Store interface { Get(key string) (json.RawMessage, error) Put(key string, value any) error Keys() ([]string, error) } // busStore is one of disk-load's states through the runtime: readings, or rounds. type busStore struct{ state string } func (b busStore) Get(key string) (json.RawMessage, error) { e, err := stdio.State(b.state).Get(key) if err != nil || e == nil { return nil, err } return e.Value, nil } func (b busStore) Put(key string, v any) error { _, err := stdio.State(b.state).Put(key, v) return err } func (b busStore) Keys() ([]string, error) { return stdio.State(b.state).Keys() } // Keeper measures the items of its node and keeps the readings. type Keeper struct { Node string // Items is every data item the modules placed on this node declare, as the node-engine composed them and its state // data holds them (Watched.All, novox/hq ADR 0577 clause 2). Items func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) Measurer Measurer Store Store // Rounds is the state rounds, where each round says how far it got (ADR 0577 clause 3); nil keeps none. Rounds Store Say func(string, ...any) mu sync.Mutex // one round at a time // pending are the modules whose items changed since they were last measured (Due), measured by Work. pmu sync.Mutex pending map[string]bool wake chan struct{} // unwatched says a round was refused because the watch had not caught up (ErrNotWatchedYet): one runs as soon as // it has (Watched.Follow), not an hour later. unwatched atomic.Bool } // RoundIfRefused runs one round now when the last was refused for a watch not caught up, and says whether it did. func (k *Keeper) RoundIfRefused(ctx context.Context) bool { if !k.unwatched.CompareAndSwap(true, false) { return false } _, _, _ = k.Round(ctx, false) return true } // SayEvery is how many items a round measures between two lines saying how far it got. const SayEvery = 50 // RoundVersion is the version of a round's layout. const RoundVersion = 1 // Round is the value of a node's key of the state rounds: when its last round started and ended, how many items it // had to measure, how many it measured, wrote and kept, and each item that failed and why. Ended is nil while it runs; // Error says why a round measured nothing at all. type Round struct { V int `json:"v"` Node string `json:"node"` Walks bool `json:"walks,omitempty"` Started time.Time `json:"started"` Ended *time.Time `json:"ended,omitempty"` Items int `json:"items"` Measured int `json:"measured"` Written int `json:"written"` Kept int `json:"kept"` Failed []FailedItem `json:"failed,omitempty"` Error string `json:"error,omitempty"` } // FailedItem is one item a round could not measure or keep, `.`, and why. type FailedItem struct { Item string `json:"item"` Why string `json:"why"` } // putRound writes the round to this node's key; a round that cannot be written is said, and measures on. func (k *Keeper) putRound(r Round) { if k.Rounds == nil { return } if err := k.Rounds.Put(k.Node, r); err != nil { k.Say("the round's outcome cannot be kept in the state %s: %v", StateRounds, err) } } // Round measures every item that is not a cache, one at a time, and writes the readings that changed. An item // measured by a walk is walked only when walks is true or its last walk is older than WalkEvery: the hourly round // reads counters, and a walk runs at most once a day (ADR 0233 §22). // // A round says how far it got (novox/hq ADR 0577 clause 3): it writes its outcome to this node's key of the state rounds // when it starts, every SayEvery items and when it ends, and says the same in the journal. An item that fails is a // failed item of the round, never its end. func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) { k.mu.Lock() defer k.mu.Unlock() r := Round{V: RoundVersion, Node: k.Node, Walks: walks, Started: k.Measurer.Now().UTC()} end := func() { at := k.Measurer.Now().UTC() r.Ended = &at k.putRound(r) } k.putRound(r) k.Say("a round of measuring the data declared here starts (walks: %v)", walks) items, err := k.Items(ctx) if err != nil { if errors.Is(err, ErrNotWatchedYet) { k.unwatched.Store(true) } r.Error = err.Error() end() k.Say("the round measured nothing: %v", err) return 0, 0, err } modules := make([]string, 0, len(items)) for m := range items { modules = append(modules, m) for _, it := range items[m] { if it.Class != ClassCache { r.Items++ } } } sort.Strings(modules) var failed []string for _, module := range modules { for _, it := range items[module] { if it.Class == ClassCache { continue } w, said, err := k.measureOne(ctx, it, walks) r.Measured++ switch { case err != nil: failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err)) r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: err.Error()}) case said != "": // Measured, and the measurement failed: its reading says so, and the round counts it failed. r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: said}) } if err == nil && w { written++ } else if err == nil { kept++ } r.Written, r.Kept = written, kept if r.Measured%SayEvery == 0 { k.Say("the round has measured %d of %d items: %d written, %d unchanged, %d failed", r.Measured, r.Items, written, kept, len(r.Failed)) k.putRound(r) } } } // A round retires nothing: a reading is retired only when its module's key says so (MeasureModule), never because // an item is missing from what the watch holds now, which may be empty or partial. end() k.Say("the round ended: %d of %d items measured, %d readings written, %d unchanged, %d failed", r.Measured, r.Items, written, kept, len(r.Failed)) if len(failed) > 0 { return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; ")) } return written, kept, nil } // Due asks Work to measure a module's items: its key of the engine's state data changed (novox/hq ADR 0577 clause 2). func (k *Keeper) Due(module string) { k.pmu.Lock() if k.pending == nil { k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1) } k.pending[module] = true wake := k.wake k.pmu.Unlock() select { case wake <- struct{}{}: default: } } // Work measures each module Due names, one at a time, until ctx ends. func (k *Keeper) Work(ctx context.Context) { k.pmu.Lock() if k.pending == nil { k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1) } wake := k.wake k.pmu.Unlock() for { select { case <-ctx.Done(): return case <-wake: } for { k.pmu.Lock() var module string for m := range k.pending { module = m break } delete(k.pending, module) k.pmu.Unlock() if module == "" { break } k.MeasureModule(ctx, module) } } } // MeasureModule measures one module's items now, as its key of the engine's state data changed, and retires the // readings of a module that no longer declares any here. A walk is not repeated within a day for an item that did not // change (measureOne). func (k *Keeper) MeasureModule(ctx context.Context, module string) { k.mu.Lock() defer k.mu.Unlock() items, err := k.Items(ctx) if err != nil { k.Say("%s's data items changed and are not measured: %v", module, err) return } n, err := k.retireGone(module, items[module]) if err != nil { k.Say("%s's readings of items it no longer declares here are not retired: %v", module, err) } else if n > 0 { k.Say("%s no longer declares %d item(s) here: their readings are retired, measured no more", module, n) } if len(items[module]) == 0 { return } written, kept := 0, 0 var failed []string for _, it := range items[module] { if it.Class == ClassCache { continue } w, said, err := k.measureOne(ctx, it, false) switch { case err != nil: failed = append(failed, it.ID+": "+err.Error()) case said != "": failed = append(failed, it.ID+": "+said) case w: written++ default: kept++ } } k.Say("%s's data items changed and were measured: %d readings written, %d unchanged, %d failed%s", module, written, kept, len(failed), map[bool]string{true: " (" + strings.Join(failed, "; ") + ")", false: ""}[len(failed) > 0]) } // retireGone marks retired every reading of this node of module whose item its key of the engine's state data no longer // names — the key was deleted (the module was unassigned there, or is no longer pulled in) or names fewer items — with // when and why, and keeps it: the record of what the module left behind, until a person deletes it (ADR 0233). It is // asked only on a change of that module's key (novox/hq ADR 0577 clause 2), never from a round. It is cleared when the // item is declared again (measureOne). func (k *Keeper) retireGone(module string, declared []catalogue.PlacedDataItem) (int, error) { return k.retire(func(r diskreadings.Reading) bool { return r.Module == module }, declared) } // RetireUnkeyed retires, once the watch handed over every current key, the readings of this node whose module has no // key in that set: a key deleted while the watch was not up (disk-load restarted, the machine rebooted or slept) never // reaches disk-load as a delete. An empty set is skipped and said: the engine has not written its keys yet, and an // empty set would retire every reading. func (k *Keeper) RetireUnkeyed(ctx context.Context) { k.mu.Lock() defer k.mu.Unlock() items, err := k.Items(ctx) if err != nil { k.Say("the readings of modules with no key are not looked for: %v", err) return } if len(items) == 0 { k.Say("the node-engine's state %s holds no key of this machine yet: no reading is retired for it", engineDataBucket) return } var declared []catalogue.PlacedDataItem for _, its := range items { declared = append(declared, its...) } n, err := k.retire(func(r diskreadings.Reading) bool { _, keyed := items[r.Module]; return !keyed }, declared) switch { case err != nil: k.Say("the readings of modules with no key are not all retired: %v", err) case n > 0: k.Say("%d reading(s) of modules with no key of this machine retired, measured no more", n) } } // retire marks retired every reading of this node that of picks and declared does not name. func (k *Keeper) retire(of func(diskreadings.Reading) bool, declared []catalogue.PlacedDataItem) (int, error) { keys, err := k.Store.Keys() if err != nil { return 0, err } declaredNow := map[string]bool{} for _, it := range declared { declaredNow[diskreadings.ReadingKey(k.Node, it.Module, it.ID)] = true } now := k.Measurer.Now().UTC() n := 0 for _, key := range keys { if !strings.HasPrefix(key, k.Node+".") || declaredNow[key] { continue } raw, err := k.Store.Get(key) if err != nil || len(raw) == 0 { continue } r, err := diskreadings.DecodeReading(raw) if err != nil || r.Node != k.Node || !of(r) || !r.RetiredAt.IsZero() { continue } r.RetiredAt = now r.RetiredWhy = fmt.Sprintf("no longer declared on %s: its module was unassigned there, or is no longer pulled in", k.Node) if err := k.Store.Put(key, r); err != nil { return n, err } n++ } return n, nil } // measureOne measures one item, unless it is a walk done within WalkEvery of an unchanged item, and writes its reading // when it changed. It answers whether it wrote the reading, what the measurement said failed ("" when it did not, or // when the item was not measured now), and an error when the reading could not be read or kept. func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, string, error) { key := diskreadings.ReadingKey(k.Node, it.Module, it.ID) var prev *diskreadings.Reading raw, err := k.Store.Get(key) if err != nil { return false, "", err } if len(raw) > 0 { if p, err := diskreadings.DecodeReading(raw); err == nil { prev = &p } } now := k.Measurer.Now() if prev != nil && prev.RetiredAt.IsZero() && prev.PlacedDataItem == it && it.Measure == catalogue.MeasureWalk && !walks && prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery { return false, "", nil } m := k.Measurer.Measure(ctx, it) next, changed := diskreadings.NextReading(prev, k.Node, it, m, now) if prev != nil && !prev.RetiredAt.IsZero() { // Declared again: it is no longer retired. next.RetiredAt, next.RetiredWhy, changed = time.Time{}, "", true } if !changed { return false, m.Error, nil } return true, m.Error, k.Store.Put(key, next) } // MeasureNow measures one item now, walk or not, and writes its reading when it changed. func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (diskreadings.Reading, bool, error) { k.mu.Lock() defer k.mu.Unlock() items, err := k.Items(ctx) if err != nil { return diskreadings.Reading{}, false, err } for _, it := range items[module] { if it.ID != item { continue } if it.Class == ClassCache { return diskreadings.Reading{}, false, fmt.Errorf("%s.%s is a cache, which is listed and never measured", module, item) } w, _, err := k.measureOne(ctx, it, true) if err != nil { return diskreadings.Reading{}, false, err } raw, err := k.Store.Get(diskreadings.ReadingKey(k.Node, module, item)) if err != nil { return diskreadings.Reading{}, w, err } r, err := diskreadings.DecodeReading(raw) return r, w, err } return diskreadings.Reading{}, false, fmt.Errorf("%s declares no data item %q on this machine", module, item) } // Loop runs a round after a few minutes and then every `every`, with the walks of the night's round when it is the // hour after the night; ctx ends it. func (k *Keeper) Loop(ctx context.Context, every time.Duration, first time.Duration) { select { case <-ctx.Done(): return case <-time.After(first): } for { _, _, _ = k.Round(ctx, false) // it says its outcome itself, in the journal and its key of the state rounds select { case <-ctx.Done(): return case <-time.After(every): } } } // RootRunner runs a command as root, through sudo without a prompt when this process is not root. A walk is bounded // by WalkFor itself; the runner's own bound is a little longer. func RootRunner(ctx context.Context, name string, args ...string) (string, error) { ctx, cancel := context.WithTimeout(ctx, WalkFor+time.Minute) defer cancel() program, argv := name, args if os.Getuid() != 0 { program, argv = "sudo", append([]string{"-n", name}, args...) } var stdout, stderr bytes.Buffer cmd := exec.CommandContext(ctx, program, argv...) cmd.Stdout, cmd.Stderr = &stdout, &stderr if err := cmd.Run(); err != nil { said := strings.TrimSpace(stderr.String()) if said == "" { said = strings.TrimSpace(stdout.String()) } if program == "sudo" && strings.HasPrefix(said, "sudo:") { return "", fmt.Errorf("%s needs root and the runtime's account may not run it without a prompt: %s", name, said) } lines := strings.Split(said, "\n") if len(lines) > 3 { lines = lines[len(lines)-3:] } if said == "" { return "", fmt.Errorf("%s: %w", name, err) } return "", errors.New(strings.Join(lines, " / ")) } return stdout.String(), nil }