mesh-catalog/modules/disk-load/cmd/disk-load/keeper.go

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
}