package main import ( "bytes" "context" "encoding/json" "errors" "fmt" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/health" "os" "os/exec" "sort" "strings" "sync" "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 the state through the runtime. type busStore struct{} func (busStore) Get(key string) (json.RawMessage, error) { e, err := stdio.State(StateReadings).Get(key) if err != nil || e == nil { return nil, err } return e.Value, nil } func (busStore) Put(key string, v any) error { _, err := stdio.State(StateReadings).Put(key, v) return err } func (busStore) Keys() ([]string, error) { return stdio.State(StateReadings).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, composed from the bus (frombus.DataItems). Items func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) Measurer Measurer Store Store Say func(string, ...any) mu sync.Mutex // one round at a time } // 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). func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) { k.mu.Lock() defer k.mu.Unlock() items, err := k.Items(ctx) if err != nil { return 0, 0, err } modules := make([]string, 0, len(items)) for m := range items { modules = append(modules, m) } sort.Strings(modules) var failed []string for _, module := range modules { for _, it := range items[module] { if it.Class == ClassCache { continue } w, err := k.measureOne(ctx, it, walks) if err != nil { failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err)) continue } if w { written++ } else { kept++ } } } retired, err := k.retireGone(items) written += retired if err != nil { failed = append(failed, err.Error()) } if len(failed) > 0 { return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; ")) } return written, kept, nil } // retireGone marks retired every reading of this node whose item the node no longer declares — its module was unassigned // there, or is no longer pulled in — with when and why, and keeps it: the record of what the module left behind, until a // person deletes it (ADR 0233). It is cleared when the item is declared again (measureOne). func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int, error) { keys, err := k.Store.Keys() if err != nil { return 0, err } declaredNow := map[string]bool{} for module, its := range declared { for _, it := range its { declaredNow[health.ReadingKey(k.Node, 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 := health.DecodeReading(raw) if err != nil || r.Node != k.Node || !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 } func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, error) { key := health.ReadingKey(k.Node, it.Module, it.ID) var prev *health.Reading raw, err := k.Store.Get(key) if err != nil { return false, err } if len(raw) > 0 { if p, err := health.DecodeReading(raw); err == nil { prev = &p } } now := k.Measurer.Now() if prev != nil && prev.RetiredAt.IsZero() && it.Measure == catalogue.MeasureWalk && !walks && prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery { return false, nil } m := k.Measurer.Measure(ctx, it) next, changed := health.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, nil } return true, 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) (health.Reading, bool, error) { k.mu.Lock() defer k.mu.Unlock() items, err := k.Items(ctx) if err != nil { return health.Reading{}, false, err } for _, it := range items[module] { if it.ID != item { continue } if it.Class == ClassCache { return health.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 health.Reading{}, false, err } raw, err := k.Store.Get(health.ReadingKey(k.Node, module, item)) if err != nil { return health.Reading{}, w, err } r, err := health.DecodeReading(raw) return r, w, err } return health.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 { w, kept, err := k.Round(ctx, false) if err != nil { k.Say("measuring the data declared here: %v", err) } else { k.Say("measured the data declared here: %d readings changed, %d unchanged", w, kept) } 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 }