diff --git a/modules/zfs/cmd/zfs-tools/bus.go b/modules/zfs/cmd/zfs-tools/bus.go index af6ffb6c5..b5da08576 100644 --- a/modules/zfs/cmd/zfs-tools/bus.go +++ b/modules/zfs/cmd/zfs-tools/bus.go @@ -48,7 +48,13 @@ func (m *Machine) JoinBus() func() { } // Every call reaches the socket anew, so a keeper that comes up after this process is used from then on; one that // is down at a call answers that call with an error, never the bus taken as down for the process's life. - c := keeperClient{sock} + // A call the keeper does not answer is said once, as a warning: the root process then acts as with the bus down (A-37). + var once sync.Once + c := keeperClient{sock: sock, down: func(err error) { + once.Do(func() { + m.Warnings = append(m.Warnings, "the bus is taken as down: "+err.Error()) + }) + }} m.Emit = c.emit m.States = c m.Findings = meshconditionsclient.FindingWriter{State: findingState{c}} @@ -75,10 +81,11 @@ func RunKeeper(ctx context.Context) error { if err != nil { return err } - _, js, err := acct.Dial() + nc, js, err := acct.Dial() if err != nil { return err } + defer nc.Close() k := &Keeper{Emit: func(kind string, body any) error { return publishEvent(js, acct, kind, body) }, States: busStates{c}, Call: c.Call} return k.Serve(ctx, l) } @@ -90,7 +97,7 @@ type eventPublisher interface { // publishEvent puts one event on the module's events stream as a JetStream publish the bus acknowledges, on the // subject the stream captures, mesh.bus..events. (A-41); never process.Conn.Emit, whose subject the -// keeper's account is not granted. The message id is a random one, as before. +// keeper's account is not granted. The message id (nanotime-pid) is the time in nanoseconds and the process id, so two events of one process never share one. func publishEvent(js eventPublisher, acct process.Account, kind string, body any) error { data, err := json.Marshal(body) if err != nil { diff --git a/modules/zfs/cmd/zfs-tools/keeper.go b/modules/zfs/cmd/zfs-tools/keeper.go index fc63abf7e..0ffa12d79 100644 --- a/modules/zfs/cmd/zfs-tools/keeper.go +++ b/modules/zfs/cmd/zfs-tools/keeper.go @@ -239,26 +239,38 @@ func Listen(sock string) (net.Listener, error) { } // keeperClient is a root process's way to the bus: calls on the keeper's socket. -type keeperClient struct{ sock string } +type keeperClient struct { + sock string + // down, when set, is told of a call the keeper did not answer (not of an answer that is an error). + down func(error) +} func (c keeperClient) do(r KeeperRequest) (KeeperAnswer, error) { + a, err, unanswered := c.exchange(r) + if unanswered && c.down != nil { + c.down(err) + } + return a, err +} + +// exchange is one call on the socket; unanswered says the keeper gave no answer (as opposed to an answer that is an error). +func (c keeperClient) exchange(r KeeperRequest) (a KeeperAnswer, err error, unanswered bool) { 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) + return KeeperAnswer{}, fmt.Errorf("the keeper does not answer on %s: %v", c.sock, err), true } defer conn.Close() _ = conn.SetDeadline(time.Now().Add(KeeperTimeout)) if err := json.NewEncoder(conn).Encode(r); err != nil { - return KeeperAnswer{}, err + return KeeperAnswer{}, err, true } - var a KeeperAnswer if err := json.NewDecoder(conn).Decode(&a); err != nil { - return KeeperAnswer{}, fmt.Errorf("the keeper's answer: %v", err) + return KeeperAnswer{}, fmt.Errorf("the keeper's answer: %v", err), true } if a.Error != "" { - return a, errors.New(a.Error) + return a, errors.New(a.Error), false } - return a, nil + return a, nil, false } func (c keeperClient) emit(kind string, body any) error { diff --git a/modules/zfs/cmd/zfs-tools/keeper_test.go b/modules/zfs/cmd/zfs-tools/keeper_test.go index fda12c202..c27ee3e72 100644 --- a/modules/zfs/cmd/zfs-tools/keeper_test.go +++ b/modules/zfs/cmd/zfs-tools/keeper_test.go @@ -86,7 +86,7 @@ func TestOnlyRootOpensTheKeepersSocketAndANeedlessKeeperIsTheBusDown(t *testing. ctx, stop := context.WithCancel(context.Background()) defer stop() go k.Serve(ctx, l) - c := keeperClient{sock} + c := keeperClient{sock: sock} if err := c.Put(StateSchedule, "snapshots", map[string]string{"by": "x"}); err != nil { t.Fatal(err) } @@ -144,3 +144,29 @@ func TestOnlyRootOpensTheKeepersSocketAndANeedlessKeeperIsTheBusDown(t *testing. t.Errorf("a keeper answering: the bus is up") } } + +// A-37: a root process whose keeper does not answer acts as with the bus down and says so, once, in a warning, for the +// hand acts, the forgetting of applied keys and the pause alike. +func TestAKeeperThatDoesNotAnswerIsWarnedOnce(t *testing.T) { + dir, err := os.MkdirTemp("/var/tmp", "keeper-down") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + t.Setenv(KeeperSocketEnv, filepath.Join(dir, "none.sock")) + m := newFake(t).machine() + m.JoinBus() + if got := m.handActs(); got != nil { + t.Errorf("hand acts with no keeper: %v", got) + } + if len(m.Warnings) != 1 || !strings.Contains(m.Warnings[0], "bus is taken as down") { + t.Fatalf("hand acts: %v", m.Warnings) + } + m.forget(&Config{}) + if m.Paused() { + t.Errorf("paused with no keeper") + } + if len(m.Warnings) != 1 { + t.Errorf("said more than once: %v", m.Warnings) + } +}