jschoubben
1c117d1189
A keep wait on a filesystem that cannot carry ACLs is not excused, and nothing is asked; one whose filesystem must first enable them says in the ask that the take sets it. The tool runner's account is sent with no uid, for the engine to resolve by name.
377 lines
16 KiB
Go
377 lines
16 KiB
Go
package main
|
|
|
|
// The mesh takes a shared folder that already holds data once, on the operator's word (novox/hq ADR 0325 §7, §8).
|
|
//
|
|
// **What waits.** A node-engine that keeps folders states an access waiting, naming it (inventory.Wait.Access), while
|
|
// its folder is there and has not been taken; the controller excuses that wait only for a declared, placed folder not
|
|
// yet taken (checkAccessWait), and the module is said as its needs-operator condition, in words with no path.
|
|
//
|
|
// **One ask per folder and machine.** The serving controller's asker asks the operator once for each such folder on
|
|
// each machine — however many modules wait on it — at the level approve (ADR 0259), in the clause's words: "Let the
|
|
// mesh keep <path> on <machine>, now for <modules>: …". It is kept in the controller's asks beside a proposal
|
|
// (asked.Take), never a condition's: the condition is the module's, and two modules waiting on one folder are one
|
|
// question. While it is open no second is asked; one that expired unanswered is asked again while the folder waits;
|
|
// one about a folder that no longer waits is taken back. None is asked for a folder already taken, so a module that
|
|
// joins a taken folder is asked nothing (its access wait is not excused, and its entries are added at apply).
|
|
//
|
|
// **On the warrant, and only on it**, the controller asks the machine's engine to take the folder (link.AskTake),
|
|
// naming the access by its resource id and the warrant; the engine adds the entries of the users it was declared for
|
|
// that folder. On its word that it started, the folder is recorded taken (kept_folder), and from the next send every
|
|
// access to it says `taken`. Where the engine's wait says the filesystem must first allow ACLs (acl
|
|
// "needs-enabling"), the ask says the take sets that; where it cannot carry them (acl "unsupported") the wait is not
|
|
// excused and nothing is asked.
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
"github.com/novox/mesh-controller/internal/outward"
|
|
)
|
|
|
|
// folderTake is one folder the operator is asked to let the mesh keep, as the controller's ask keeps it
|
|
// (asked.Take). Every field the ask is composed from is here, so the act on the warrant is composed from the record.
|
|
type folderTake struct {
|
|
Node string `json:"node"`
|
|
Path string `json:"path"`
|
|
// Access is the resource id of one access to it in the machine's declaration: what the take act names.
|
|
Access string `json:"access"`
|
|
// Accesses are the access ids (each module's own) that place it, Modules the modules waiting on it, both sorted.
|
|
Accesses []string `json:"accesses"`
|
|
Modules []string `json:"modules"`
|
|
// EnablesACL says the folder's filesystem must first allow access entries (a ZFS dataset's acltype), which the
|
|
// take act sets: the ask says so.
|
|
EnablesACL bool `json:"enables-acl,omitempty"`
|
|
// Placed is the filesystem the folder is on, as the engine names it without a path: the warrant is for that one.
|
|
Placed string `json:"placed,omitempty"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// takeVerb is the verb a take's answer binds: the controller's own, performed by itself on the warrant.
|
|
const takeVerb = askerName + ".take"
|
|
|
|
// key is the folder on its machine: what one ask is kept for.
|
|
// The filesystem is part of it: a folder that moved to another disk is asked about again.
|
|
func (t folderTake) key() string { return t.Node + "\x00" + filepath.Clean(t.Path) + "\x00" + t.Placed }
|
|
|
|
// digest binds everything the take does and was said for: the machine, the folder, the access it names and the
|
|
// modules the operator was told.
|
|
func (t folderTake) digest() string {
|
|
raw, _ := json.Marshal(struct {
|
|
Node, Path, Access, Placed string
|
|
Modules []string
|
|
EnablesACL bool
|
|
}{t.Node, filepath.Clean(t.Path), t.Access, t.Placed, t.Modules, t.EnablesACL})
|
|
sum := sha256.Sum256(raw)
|
|
return "sha256:" + hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
// actions is the take's one answer, at the level approve, binding the exact act through boundAct.
|
|
func (t folderTake) actions(id string) []conditions.Action {
|
|
return []conditions.Action{{Label: "Let the mesh keep it", Verb: takeVerb, Machine: t.Node,
|
|
Level: conditions.LevelApprove, Arguments: map[string]string{"take": id, "node": t.Node,
|
|
"path": filepath.Clean(t.Path), "access": t.Access, "digest": t.digest()}}}
|
|
}
|
|
|
|
// takeWords is the clause's ask (ADR 0325 §8), with the folder as shown: "Let the mesh keep <path> on <machine>, now
|
|
// for <modules>: …".
|
|
func takeWords(shown, machine string, modules []string, enablesACL bool) string {
|
|
words := fmt.Sprintf("Let the mesh keep %s on %s, now for %s: it adds their access entries, and those of any "+
|
|
"module given access later, to every folder and file there; the group bits `ls -l` shows become the "+
|
|
"entries' mask, and owners, groups and every other right stay as they are", shown, machine,
|
|
strings.Join(modules, ", "))
|
|
if enablesACL {
|
|
words += ". Its filesystem does not allow access entries yet (a ZFS dataset's acltype): approving sets that first"
|
|
}
|
|
return words
|
|
}
|
|
|
|
// ask is the take's ask, composed from it alone. The folder's path is said whole where the router's content rule
|
|
// lets it leave the mesh, and otherwise named by the shared folder it is ("the shared folder movies"), which is the
|
|
// access the module's wait names.
|
|
func (t folderTake) ask(id string, machines []string) (asks.Ask, map[string]int) {
|
|
shown := filepath.Clean(t.Path)
|
|
if _, ok := outward.Check(shown, machines...); !ok {
|
|
shown = "the shared folder " + strings.Join(t.Accesses, ", ")
|
|
}
|
|
headline := fmt.Sprintf("Let the mesh keep a shared folder on %s?", t.Node)
|
|
if len([]rune(headline)) > asks.HeadlineLength {
|
|
headline = "Let the mesh keep a shared folder?"
|
|
}
|
|
explanation := takeWords(shown, t.Node, t.Modules, t.EnablesACL) + "."
|
|
if _, ok := outward.Check(explanation, machines...); !ok {
|
|
explanation = sayable(explanation, machines)
|
|
}
|
|
q := asks.Ask{ID: id, Headline: headline, Explanation: explanation,
|
|
Who: asks.Operator, Expires: t.At.Add(askApproveFor),
|
|
OnExpiry: "nothing is changed, and you are asked again while the folder waits", About: "take." + t.Node}
|
|
options := map[string]int{}
|
|
for i, act := range t.actions(id) {
|
|
binds, _ := asks.ActDigest(boundAct(act))
|
|
oid := optionID(act.Label)
|
|
options[oid] = i
|
|
q.Options = append(q.Options, asks.Option{ID: oid, Label: act.Label, Level: asks.Level(act.Level), Binds: binds,
|
|
Does: "the mesh adds the access entries there, and keeps them from then on without asking"})
|
|
}
|
|
return q, options
|
|
}
|
|
|
|
// takeOnWarrant performs a take the operator approved: the engine is asked, and the folder recorded taken on its
|
|
// word. It answers what the engine said.
|
|
type takeOnWarrant func(ctx context.Context, t folderTake, askID, by string) (string, error)
|
|
|
|
// askFolders asks the operator about every folder waiting to be kept that is not asked about yet, and takes back an
|
|
// open ask about one that no longer waits. Called at the end of each reconcile, after the conditions' asks, within
|
|
// the asks the router holds open at once.
|
|
func (a *asker) askFolders(ctx context.Context) error {
|
|
if a.folders == nil {
|
|
return nil
|
|
}
|
|
wanted, err := a.folders(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
now := a.now()
|
|
all, err := a.store.All(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
open, openNow := map[string]asked{}, 0
|
|
for _, r := range all {
|
|
if r.State != askOpen || !now.Before(r.Ask.Expires) {
|
|
continue
|
|
}
|
|
openNow++
|
|
if r.Take != nil {
|
|
open[r.Take.key()] = r
|
|
}
|
|
}
|
|
want := map[string]bool{}
|
|
for _, t := range wanted {
|
|
want[t.key()] = true
|
|
}
|
|
for key, r := range open {
|
|
if !want[key] {
|
|
if err := a.cancel(ctx, r, "the folder no longer waits to be kept"); err != nil {
|
|
return err
|
|
}
|
|
openNow--
|
|
}
|
|
}
|
|
var machines []string
|
|
if a.machines != nil {
|
|
if machines, err = a.machines(ctx); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, t := range wanted {
|
|
if _, asking := open[t.key()]; asking {
|
|
continue // one ask per folder and machine (§8)
|
|
}
|
|
if openNow >= askMostOpen {
|
|
return nil // asked when one of the open ones ends
|
|
}
|
|
t.At = now
|
|
id := newAskID()
|
|
q, options := t.ask(id, machines)
|
|
if err := q.Check(now); err != nil {
|
|
a.logf("the operator could not be asked to let the mesh keep a folder on %s: %v", t.Node, err)
|
|
continue
|
|
}
|
|
body, err := json.Marshal(q)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
take := t
|
|
if err := a.store.Create(ctx, asked{ID: id, Condition: q.About, Ask: q, Actions: t.actions(id),
|
|
Options: options, State: askOpen, Opened: now, Take: &take}); err != nil {
|
|
return fmt.Errorf("the take could not be kept in the controller's asks, so it was not asked: %w", err)
|
|
}
|
|
if err := a.publish(ctx, asks.AskSubject(askerName), body, "ask."+id); err != nil {
|
|
_, _ = a.store.Change(ctx, id, func(x *asked) bool {
|
|
if x.State != askOpen || x.Acted != "" {
|
|
return false
|
|
}
|
|
x.State, x.Ended, x.Acted = askUnsent, a.now(), "nothing: it could not be published: "+err.Error()
|
|
return true
|
|
})
|
|
a.logf("the operator could not be asked to let the mesh keep a folder on %s: %v", t.Node, err)
|
|
continue
|
|
}
|
|
openNow++
|
|
a.logf("asked the operator to let the mesh keep a folder on %s for %s (%s)", t.Node,
|
|
strings.Join(t.Modules, ", "), id)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// decideTake is what the asker does with a warrant for a take (asker.Decided): the act only when it is the one the
|
|
// option bound — composed again from the record, the users' digest among it — and then the engine is asked.
|
|
func (a *asker) decideTake(ctx context.Context, r asked, act conditions.Action, w asks.Warrant) (string, error) {
|
|
t := *r.Take
|
|
recomposed := t.actions(r.ID)
|
|
option, offered := r.Ask.Option(w.Option)
|
|
if !offered || len(recomposed) != 1 {
|
|
return "", fmt.Errorf("the take %s offers no answer %q: nothing is taken", r.ID, w.Option)
|
|
}
|
|
if err := option.Performs(boundAct(recomposed[0])); err != nil {
|
|
return "", fmt.Errorf("the take kept for %s is not the one the operator was shown: %v", r.ID, err)
|
|
}
|
|
if err := option.Performs(boundAct(act)); err != nil {
|
|
return "", fmt.Errorf("the act chosen for %s is not the one the operator was shown: %v", r.ID, err)
|
|
}
|
|
if a.take == nil {
|
|
return "", errors.New("this controller cannot take a folder on a warrant")
|
|
}
|
|
by := fmt.Sprintf("approved by %s via %s at %s (ask %s)", byWords(w), viaWords(w),
|
|
w.At.Local().Format("2006-01-02 15:04"), r.ID)
|
|
return a.take(ctx, t, r.ID, by)
|
|
}
|
|
|
|
// takeFolderIn is takeOnWarrant on this controller's stores and bus: the engine is asked with the mesh's key, naming the
|
|
// ask the warrant answered; on its word that it started, the folder is recorded taken. A refusal records nothing.
|
|
func takeFolderIn(open *stores, ask func(ctx context.Context, act link.AccessActAsk) (link.HandOverAnswer, error)) takeOnWarrant {
|
|
return func(ctx context.Context, t folderTake, askID, by string) (string, error) {
|
|
// Who asked, as one short line the engine takes (at most link.MaxHandOverBy): the ask names the rest.
|
|
short := by
|
|
if len(short) > 180 {
|
|
short = "the operator, on the warrant for ask " + askID
|
|
}
|
|
answer, err := ask(ctx, link.AccessActAsk{Node: t.Node, Path: filepath.Clean(t.Path), By: short,
|
|
Access: t.Access, Warrant: askID})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if answer.Refused != "" {
|
|
return "", fmt.Errorf("%s refused to take the folder: %s", t.Node, answer.Refused)
|
|
}
|
|
if err := open.inventory.RecordTaken(ctx, t.Node, inventory.TakenFolder{Path: t.Path, By: by, Ask: askID,
|
|
Warrant: askID, Said: answer.Said, Placed: t.Placed}); err != nil {
|
|
return "", fmt.Errorf("%s started the take (%s), and it could NOT be recorded: %w", t.Node, answer.Said, err)
|
|
}
|
|
return fmt.Sprintf("%s takes the folder for %s: %s", t.Node, strings.Join(t.Modules, ", "), answer.Said), nil
|
|
}
|
|
}
|
|
|
|
// waitingFolder is one access wait excused for a folder not yet taken, as foldersWaiting gathers them.
|
|
type waitingFolder struct {
|
|
module, access string
|
|
// resource is the access's resource id on the machine, as the engine stated it; placed the filesystem its folder
|
|
// is on, without its path.
|
|
resource, placed string
|
|
enablesACL bool
|
|
}
|
|
|
|
// foldersWaiting is every folder waiting to be kept across the mesh (ADR 0325 §7, §8): from each machine's newest
|
|
// statement, each access wait of a module that is excused (checkWaiting) and waits for the operator's word rather
|
|
// than for a disk, grouped by machine and folder.
|
|
func foldersWaiting(ctx context.Context, inv *inventory.Inventory, healths map[string]inventory.NodeHealth) (
|
|
map[string]map[string][]waitingFolder, operatorWaitFacts) {
|
|
var f operatorWaitFacts
|
|
out := map[string]map[string][]waitingFolder{}
|
|
machines := make([]string, 0, len(healths))
|
|
for m := range healths {
|
|
machines = append(machines, m)
|
|
}
|
|
sort.Strings(machines)
|
|
for _, machine := range machines {
|
|
rs := healths[machine].Resources
|
|
var modules []string
|
|
for _, r := range rs {
|
|
for _, w := range r.Waits {
|
|
if r.State == link.StateWaiting && w.Access != "" && !w.Disk {
|
|
modules = append(modules, r.Module)
|
|
}
|
|
}
|
|
}
|
|
if len(modules) == 0 {
|
|
continue
|
|
}
|
|
readWaitFacts(ctx, inv, machine, modules, nil, &f)
|
|
for _, r := range checkWaiting(machine, rs, f) {
|
|
if r.State != link.StateWaiting {
|
|
continue
|
|
}
|
|
for _, w := range r.Waits {
|
|
if w.Access == "" || w.Disk {
|
|
continue
|
|
}
|
|
path, placed := f.placed[r.Module+"@"+machine][w.Access]
|
|
if !placed {
|
|
continue
|
|
}
|
|
if out[machine] == nil {
|
|
out[machine] = map[string][]waitingFolder{}
|
|
}
|
|
out[machine][path] = append(out[machine][path], waitingFolder{module: r.Module, access: w.Access,
|
|
resource: r.Resource, placed: w.Placed, enablesACL: w.ACL == link.ACLNeedsEnabling})
|
|
}
|
|
}
|
|
}
|
|
return out, f
|
|
}
|
|
|
|
// foldersIn is the asker's folders on this controller's stores: every folder waiting to be kept, each with the
|
|
// modules waiting on it and the access the take names (the first waiting module's, by its resource id). Who the
|
|
// take adds entries for is what the engine was declared for that folder: the ask names the modules.
|
|
func foldersIn(open *stores) func(ctx context.Context) ([]folderTake, error) {
|
|
return func(ctx context.Context) ([]folderTake, error) {
|
|
healths, err := open.inventory.Healths(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
waiting, _ := foldersWaiting(ctx, open.inventory, healths)
|
|
return takesOf(waiting), nil
|
|
}
|
|
}
|
|
|
|
// takesOf is one take per machine and folder from the waits gathered, sorted.
|
|
func takesOf(waiting map[string]map[string][]waitingFolder) []folderTake {
|
|
var out []folderTake
|
|
for machine, folders := range waiting {
|
|
for path, all := range folders {
|
|
// One take per filesystem the waits name: they agree but for a moment while a disk is swapped.
|
|
byPlaced := map[string][]waitingFolder{}
|
|
for _, w := range all {
|
|
byPlaced[w.placed] = append(byPlaced[w.placed], w)
|
|
}
|
|
for placed, ws := range byPlaced {
|
|
t := folderTake{Node: machine, Path: path, Placed: placed}
|
|
seenM, seenA := map[string]bool{}, map[string]bool{}
|
|
var resources []string
|
|
for _, w := range ws {
|
|
if !seenM[w.module] {
|
|
seenM[w.module] = true
|
|
t.Modules = append(t.Modules, w.module)
|
|
}
|
|
if !seenA[w.access] {
|
|
seenA[w.access] = true
|
|
t.Accesses = append(t.Accesses, w.access)
|
|
}
|
|
resources = append(resources, w.resource)
|
|
t.EnablesACL = t.EnablesACL || w.enablesACL
|
|
}
|
|
sort.Strings(t.Modules)
|
|
sort.Strings(t.Accesses)
|
|
sort.Strings(resources)
|
|
t.Access = resources[0]
|
|
out = append(out, t)
|
|
}
|
|
}
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].key() < out[j].key() })
|
|
return out
|
|
}
|