A-37 the keeper: the module's one process bus account, a socket only root opens, and the root processes' bus through it

Builds: design 42 zfs-owns-zfs §18, §20, §26, §27, A-6, A-20, A-37
This commit is contained in:
2026-11-13 01:04:41 +01:00
parent c6844fc951
commit ec27f12524
5 changed files with 480 additions and 13 deletions
+38 -12
View File
@@ -1,7 +1,9 @@
package main
import (
"context"
"encoding/json"
"fmt"
"os"
"strings"
"sync"
@@ -33,25 +35,49 @@ func (b busStates) Get(state, key string) (json.RawMessage, bool, error) {
return e.Value, true, nil
}
// JoinBus gives the machine this process's events and states, as its own bus account says. Without the
// account or the bus the machine keeps none, and what it changes is said as a warning (§18: with the bus down
// the apply converges from the config file it last got). It answers how to leave the bus.
// JoinBus gives a root process the module's events and states through the keeper's socket (A-37): the process holds no
// bus account of its own. Without the socket the machine keeps none, and what it changes is said as a warning (§18:
// with the bus down the apply converges from the config file it last got). It answers how to leave the bus.
func (m *Machine) JoinBus() func() {
sock := strings.TrimSpace(os.Getenv(KeeperSocketEnv))
if sock == "" {
m.Warnings = append(m.Warnings, KeeperSocketEnv+" is not set, so the keeper is not reached and the bus is taken as down")
return func() {}
}
c := keeperClient{sock}
// The keeper answering is what the bus being up means: one call says it before the process relies on it.
if _, err := c.Keys(StateSchedule); err != nil {
m.Warnings = append(m.Warnings, err.Error()+": the bus is taken as down")
return func() {}
}
m.Emit = c.emit
m.States = c
m.Findings = meshconditionsclient.FindingWriter{State: findingState{c}}
m.Call = c.call
return func() {}
}
// RunKeeper is the process `zfs-keeper`: the module's one process bus account, serving the socket to root.
func RunKeeper(ctx context.Context) error {
sock := strings.TrimSpace(os.Getenv(KeeperSocketEnv))
if sock == "" {
return fmt.Errorf("%s is not set", KeeperSocketEnv)
}
acct, err := process.ReadBrokerAccount(os.Getenv)
if err != nil {
m.Warnings = append(m.Warnings, err.Error())
return func() {}
return err
}
c, err := process.Connect(acct)
if err != nil {
m.Warnings = append(m.Warnings, err.Error())
return func() {}
return err
}
m.Emit = func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }
m.States = busStates{c}
m.Findings = meshconditionsclient.FindingWriter{State: c.State(StateFindings)}
m.Call = c.Call
return c.Close
defer c.Close()
l, err := Listen(sock)
if err != nil {
return err
}
k := &Keeper{Emit: func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }, States: busStates{c}, Call: c.Call}
return k.Serve(ctx, l)
}
// toolStates is the module's states as the tools bundle reaches them, through the tool runner.
+312
View File
@@ -0,0 +1,312 @@
package main
import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"net"
"os"
"path/filepath"
"strings"
"time"
"golang.org/x/sys/unix"
)
// The keeper (design 42 zfs-owns-zfs A-37, §18, §20, §26, §27; ADR 0436 §1, §2).
//
// A process's bus account is a user the module makes, never root, so the three root processes (the apply, the run and
// the event follower) declare no `bus`. The process `zfs-keeper`, running as the module's user `zfs-keeper`, holds the
// module's one process bus account and serves a unix socket in the module's state directory that only root may open
// (the peer's uid is checked on every connection). Its calls are a closed list: emit an event of the module, put or
// delete a key of its states `applied`, `schedule` and `findings`, and invoke one of three (`zfs.zfs_snapshots_record`,
// `zfs.zfs_snapshots_schedule` and `seat:node-service-manager.start`, of the two units of the module only). The root
// processes say everything they say on the bus through it; with the keeper not answering they act as with the bus down
// (A-8), and warn.
//
// Reading a key or the keys of those states (`get`, `keys`) is also served: the run reads the hand acts waiting in the
// state `schedule` and the apply reads the state `applied` (A-17), and a root process has no other way to the bus.
// KeeperSocketEnv names the variable holding the socket's path, in every process of the module that uses it.
const KeeperSocketEnv = "ZFS_KEEPER_SOCKET"
// KeeperTimeout bounds one call through the keeper.
const KeeperTimeout = 30 * time.Second
// The calls of the keeper.
const (
OpEmit = "emit"
OpPut = "put"
OpDelete = "delete"
OpGet = "get"
OpKeys = "keys"
OpInvoke = "invoke"
)
// KeeperRequest is one call: one JSON line.
type KeeperRequest struct {
Op string `json:"op"`
Kind string `json:"kind,omitempty"`
State string `json:"state,omitempty"`
Key string `json:"key,omitempty"`
Body json.RawMessage `json:"body,omitempty"`
}
// KeeperAnswer is its answer: one JSON line.
type KeeperAnswer struct {
Error string `json:"error,omitempty"`
Found bool `json:"found,omitempty"`
Value json.RawMessage `json:"value,omitempty"`
Keys []string `json:"keys,omitempty"`
}
var keeperStates = map[string]bool{StateApplied: true, StateSchedule: true, StateFindings: true}
// Keeper serves the calls over the bus account of its own process.
type Keeper struct {
Emit func(kind string, body any) error
States StateStore
Call func(key string, body any) (json.RawMessage, error)
// PeerOf is the uid at the other end of a connection; nil reads it from the socket (SO_PEERCRED).
PeerOf func(net.Conn) (int, error)
}
// invokable says whether a call key is one of the three, and for the service manager whether it starts one of the
// module's two units.
func invokable(key string, body json.RawMessage) error {
switch {
case key == ToolRecord || key == ToolSchedule:
return nil
case strings.HasPrefix(key, "seat:node-service-manager.start@"):
var a map[string]any
if json.Unmarshal(body, &a) != nil || a["scope"] != "system" || (a["unit"] != RunService && a["unit"] != ApplyUnit) || len(a) != 2 {
return fmt.Errorf("the keeper starts only %s or %s, in the system scope", RunService, ApplyUnit)
}
return nil
}
return fmt.Errorf("%q is not one of the three calls the keeper invokes", key)
}
// Do answers one request.
func (k *Keeper) Do(r KeeperRequest) KeeperAnswer {
fail := func(err error) KeeperAnswer { return KeeperAnswer{Error: err.Error()} }
switch r.Op {
case OpEmit:
known := false
for _, e := range EventKinds {
known = known || (e == r.Kind && e != EventScrubRequested)
}
if !known {
return fail(fmt.Errorf("%q is not an event the module's processes say", r.Kind))
}
var body any
if len(r.Body) > 0 {
if err := json.Unmarshal(r.Body, &body); err != nil {
return fail(err)
}
}
if err := k.Emit(r.Kind, body); err != nil {
return fail(err)
}
return KeeperAnswer{}
case OpPut, OpDelete, OpGet, OpKeys:
if !keeperStates[r.State] {
return fail(fmt.Errorf("%q is not a state of the module's processes (applied, schedule, findings)", r.State))
}
switch r.Op {
case OpPut:
var v any
if err := json.Unmarshal(r.Body, &v); err != nil {
return fail(err)
}
if err := k.States.Put(r.State, r.Key, v); err != nil {
return fail(err)
}
case OpDelete:
if err := k.States.Delete(r.State, r.Key); err != nil {
return fail(err)
}
case OpGet:
v, ok, err := k.States.Get(r.State, r.Key)
if err != nil {
return fail(err)
}
return KeeperAnswer{Found: ok, Value: v}
case OpKeys:
keys, err := k.States.Keys(r.State)
if err != nil {
return fail(err)
}
return KeeperAnswer{Keys: keys}
}
return KeeperAnswer{}
case OpInvoke:
if err := invokable(r.Key, r.Body); err != nil {
return fail(err)
}
var body any
_ = json.Unmarshal(r.Body, &body)
out, err := k.Call(r.Key, body)
if err != nil {
return fail(err)
}
return KeeperAnswer{Value: out}
}
return fail(fmt.Errorf("%q is not a call of the keeper (emit, put, delete, get, keys, invoke)", r.Op))
}
// peerUID is the account at the other end of a unix socket.
func peerUID(c net.Conn) (int, error) {
uc, ok := c.(*net.UnixConn)
if !ok {
return -1, errors.New("not a unix socket")
}
raw, err := uc.SyscallConn()
if err != nil {
return -1, err
}
uid := -1
var gerr error
if err := raw.Control(func(fd uintptr) {
cred, err := unix.GetsockoptUcred(int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED)
if err != nil {
gerr = err
return
}
uid = int(cred.Uid)
}); err != nil {
return -1, err
}
return uid, gerr
}
// Serve answers each connection on l, one request each, until ctx ends: a peer that is not root is told so and
// nothing it sent is read.
func (k *Keeper) Serve(ctx context.Context, l net.Listener) error {
go func() { <-ctx.Done(); l.Close() }()
peerOf := k.PeerOf
if peerOf == nil {
peerOf = peerUID
}
for {
c, err := l.Accept()
if err != nil {
if ctx.Err() != nil {
return nil
}
return err
}
go func() {
defer c.Close()
_ = c.SetDeadline(time.Now().Add(KeeperTimeout))
enc := json.NewEncoder(c)
if uid, err := peerOf(c); err != nil || uid != 0 {
_ = enc.Encode(KeeperAnswer{Error: "only root opens the keeper's socket"})
return
}
line, err := bufio.NewReader(c).ReadBytes('\n')
if err != nil {
return
}
var r KeeperRequest
if err := json.Unmarshal(line, &r); err != nil {
_ = enc.Encode(KeeperAnswer{Error: "the request is not JSON: " + err.Error()})
return
}
_ = enc.Encode(k.Do(r))
}()
}
}
// Listen opens the socket: in a folder of the keeper's own, as the keeper's account, mode 0600 so that only the
// account itself and root open it; any older socket is replaced.
func Listen(sock string) (net.Listener, error) {
if err := os.MkdirAll(filepath.Dir(sock), 0o711); err != nil {
return nil, err
}
_ = os.Remove(sock)
l, err := net.Listen("unix", sock)
if err != nil {
return nil, err
}
if err := os.Chmod(sock, 0o600); err != nil {
l.Close()
return nil, err
}
return l, nil
}
// keeperClient is a root process's way to the bus: calls on the keeper's socket.
type keeperClient struct{ sock string }
func (c keeperClient) do(r KeeperRequest) (KeeperAnswer, error) {
conn, err := net.DialTimeout("unix", c.sock, 5*time.Second)
if err != nil {
return KeeperAnswer{}, fmt.Errorf("the keeper does not answer on %s: %v", c.sock, err)
}
defer conn.Close()
_ = conn.SetDeadline(time.Now().Add(KeeperTimeout))
if err := json.NewEncoder(conn).Encode(r); err != nil {
return KeeperAnswer{}, err
}
var a KeeperAnswer
if err := json.NewDecoder(conn).Decode(&a); err != nil {
return KeeperAnswer{}, fmt.Errorf("the keeper's answer: %v", err)
}
if a.Error != "" {
return a, errors.New(a.Error)
}
return a, nil
}
func (c keeperClient) emit(kind string, body any) error {
raw, err := json.Marshal(body)
if err != nil {
return err
}
_, err = c.do(KeeperRequest{Op: OpEmit, Kind: kind, Body: raw})
return err
}
func (c keeperClient) Put(state, key string, v any) error {
raw, err := json.Marshal(v)
if err != nil {
return err
}
_, err = c.do(KeeperRequest{Op: OpPut, State: state, Key: key, Body: raw})
return err
}
func (c keeperClient) Delete(state, key string) error {
_, err := c.do(KeeperRequest{Op: OpDelete, State: state, Key: key})
return err
}
func (c keeperClient) Keys(state string) ([]string, error) {
a, err := c.do(KeeperRequest{Op: OpKeys, State: state})
return a.Keys, err
}
func (c keeperClient) Get(state, key string) (json.RawMessage, bool, error) {
a, err := c.do(KeeperRequest{Op: OpGet, State: state, Key: key})
return a.Value, a.Found, err
}
func (c keeperClient) call(key string, body any) (json.RawMessage, error) {
raw, err := json.Marshal(body)
if err != nil {
return nil, err
}
a, err := c.do(KeeperRequest{Op: OpInvoke, Key: key, Body: raw})
return a.Value, err
}
// findingState is the state `findings` as the finding writer reaches it, through the keeper.
type findingState struct{ c keeperClient }
func (f findingState) Put(key string, value any) (uint64, error) {
return 0, f.c.Put(StateFindings, key, value)
}
func (f findingState) Delete(key string) error { return f.c.Delete(StateFindings, key) }
+124
View File
@@ -0,0 +1,124 @@
package main
import (
"context"
"encoding/json"
"errors"
"net"
"os"
"path/filepath"
"strings"
"testing"
)
// A-37: the keeper answers a closed list of calls: an event of the module (never scrub.requested), a key of its three
// states, and one of three invokes (the service manager only for the module's two units); anything else is refused.
func TestTheKeeperAnswersOnlyItsClosedListOfCalls(t *testing.T) {
var emitted []string
var called []string
st := memStates{}
k := &Keeper{
Emit: func(kind string, body any) error { emitted = append(emitted, kind); return nil },
States: st,
Call: func(key string, body any) (json.RawMessage, error) {
called = append(called, key)
return json.RawMessage(`{"ok":true}`), nil
},
}
req := func(op, kind, state, key, body string) KeeperRequest {
return KeeperRequest{Op: op, Kind: kind, State: state, Key: key, Body: json.RawMessage(body)}
}
cases := []struct {
name string
r KeeperRequest
refused string
}{
{"emit a kind", req(OpEmit, EventApplied, "", "", `{"key":"x"}`), ""},
{"emit the scrub's kind", req(OpEmit, EventScrubRequested, "", "", `{}`), "not an event"},
{"emit another module's kind", req(OpEmit, "package.changed", "", "", `{}`), "not an event"},
{"put applied", req(OpPut, "", StateApplied, "hostid", `{"outcome":"applied"}`), ""},
{"put schedule", req(OpPut, "", StateSchedule, "paused", `{}`), ""},
{"put another state", req(OpPut, "", "holds", "x", `{}`), "not a state"},
{"get findings", req(OpGet, "", StateFindings, "k", ``), ""},
{"keys of another state", req(OpKeys, "", "mirrors", "", ``), "not a state"},
{"delete schedule", req(OpDelete, "", StateSchedule, "paused", ``), ""},
{"invoke record", req(OpInvoke, "", "", ToolRecord, `{}`), ""},
{"invoke schedule", req(OpInvoke, "", "", ToolSchedule, `{}`), ""},
{"start the run", req(OpInvoke, "", "", "seat:node-service-manager.start@ace", `{"scope":"system","unit":"zfs-mesh-run.service"}`), ""},
{"start the apply", req(OpInvoke, "", "", "seat:node-service-manager.start@ace", `{"scope":"system","unit":"zfs-config-apply.service"}`), ""},
{"start another unit", req(OpInvoke, "", "", "seat:node-service-manager.start@ace", `{"scope":"system","unit":"sshd.service"}`), "starts only"},
{"stop a unit", req(OpInvoke, "", "", "seat:node-service-manager.stop@ace", `{"scope":"system","unit":"zfs-mesh-run.service"}`), "not one of the three"},
{"invoke a tool of the module", req(OpInvoke, "", "", "zfs.zfs_snapshot", `{}`), "not one of the three"},
{"another op", req("shell", "", "", "", ``), "not a call"},
}
for _, c := range cases {
a := k.Do(c.r)
if (c.refused == "") != (a.Error == "") || (c.refused != "" && !strings.Contains(a.Error, c.refused)) {
t.Errorf("%s: %+v", c.name, a)
}
}
if len(emitted) != 1 || len(called) != 4 {
t.Errorf("emitted %v, called %v", emitted, called)
}
if _, left := st["applied/hostid"]; !left {
t.Errorf("the put was not kept")
}
}
// A-37: only root opens the socket; a root process says and reads through it as through a bus of its own, and with no
// keeper answering it acts as with the bus down, with a warning (A-8).
func TestOnlyRootOpensTheKeepersSocketAndANeedlessKeeperIsTheBusDown(t *testing.T) {
st := memStates{}
uid := 0
k := &Keeper{Emit: func(string, any) error { return nil }, States: st,
Call: func(string, any) (json.RawMessage, error) { return nil, errors.New("down") },
PeerOf: func(net.Conn) (int, error) { return uid, nil }}
dir, err := os.MkdirTemp("/var/tmp", "zk")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(dir)
sock := filepath.Join(dir, "k", "k.sock")
l, err := Listen(sock)
if err != nil {
t.Fatal(err)
}
ctx, stop := context.WithCancel(context.Background())
defer stop()
go k.Serve(ctx, l)
c := keeperClient{sock}
if err := c.Put(StateSchedule, "snapshots", map[string]string{"by": "x"}); err != nil {
t.Fatal(err)
}
if keys, err := c.Keys(StateSchedule); err != nil || len(keys) != 1 || keys[0] != "snapshots" {
t.Errorf("%v %v", keys, err)
}
if v, ok, err := c.Get(StateSchedule, "snapshots"); err != nil || !ok || !strings.Contains(string(v), `"by":"x"`) {
t.Errorf("%s %v %v", v, ok, err)
}
if _, err := c.call(ToolSchedule, map[string]any{}); err == nil || !strings.Contains(err.Error(), "down") {
t.Errorf("an invoke that failed: %v", err)
}
uid = 1000
if err := c.Put(StateSchedule, "x", 1); err == nil || !strings.Contains(err.Error(), "only root") {
t.Errorf("a peer that is not root: %v", err)
}
if _, left := st["schedule/x"]; left {
t.Errorf("a refused peer's call was read")
}
// No keeper: the machine keeps no bus and says so.
t.Setenv(KeeperSocketEnv, filepath.Join(dir, "none.sock"))
m := newFake(t).machine()
m.Emit = nil
m.JoinBus()
if m.Emit != nil || m.States != nil || len(m.Warnings) != 1 || !strings.Contains(m.Warnings[0], "bus is taken as down") {
t.Errorf("no keeper: %+v", m.Warnings)
}
t.Setenv(KeeperSocketEnv, sock)
uid = 0
m2 := newFake(t).machine()
m2.JoinBus()
if m2.Emit == nil || m2.States == nil || m2.Call == nil || m2.Findings == nil {
t.Errorf("a keeper answering: the bus is up")
}
}
+5
View File
@@ -53,6 +53,11 @@ func main() {
os.Exit(1)
}
fmt.Println(said)
case "keeper":
if err := RunKeeper(ctx); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
case "events":
if err := Follower(ctx, os.Stdout); err != nil {
fmt.Fprintln(os.Stderr, err)
+1 -1
View File
@@ -15,5 +15,5 @@ require (
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/sys v0.48.0 // indirect
golang.org/x/sys v0.48.0
)