diff --git a/modules/zfs/cmd/zfs-tools/bus.go b/modules/zfs/cmd/zfs-tools/bus.go index 7d0654061..af6ffb6c5 100644 --- a/modules/zfs/cmd/zfs-tools/bus.go +++ b/modules/zfs/cmd/zfs-tools/bus.go @@ -13,6 +13,8 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" ) // busStates is the module's states as this process reaches them (its bus account's `state`). @@ -73,10 +75,41 @@ func RunKeeper(ctx context.Context) error { 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} + _, js, err := acct.Dial() + if err != nil { + return err + } + 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) } +// eventPublisher is the part of JetStream the keeper publishes with. +type eventPublisher interface { + PublishMsg(ctx context.Context, m *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error) +} + +// 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. +func publishEvent(js eventPublisher, acct process.Account, kind string, body any) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + id := fmt.Sprintf("%d-%d", time.Now().UnixNano(), os.Getpid()) + msg := nats.NewMsg(acct.BusSubject(EventsStream, kind)) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", acct.Module) + msg.Header.Set("x-node", acct.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) + return err +} + // toolStates is the module's states as the tools bundle reaches them, through the tool runner. type toolStates struct{} diff --git a/modules/zfs/cmd/zfs-tools/publish_event_test.go b/modules/zfs/cmd/zfs-tools/publish_event_test.go new file mode 100644 index 000000000..df3177b67 --- /dev/null +++ b/modules/zfs/cmd/zfs-tools/publish_event_test.go @@ -0,0 +1,30 @@ +package main + +import ( + "context" + "testing" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" +) + +type recordingPublisher struct{ msgs []*nats.Msg } + +func (r *recordingPublisher) PublishMsg(_ context.Context, m *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) { + r.msgs = append(r.msgs, m) + return &jetstream.PubAck{}, nil +} + +// A-41, review finding 1: the keeper publishes on the subject the module's events stream captures. +func TestTheKeeperPublishesOnTheEventsStreamsSubject(t *testing.T) { + p := &recordingPublisher{} + acct := process.Account{Node: "n1", Module: "zfs"} + if err := publishEvent(p, acct, EventApplied, map[string]string{"key": "x"}); err != nil { + t.Fatal(err) + } + if len(p.msgs) != 1 || p.msgs[0].Subject != "mesh.bus.zfs.events."+EventApplied || p.msgs[0].Header.Get("x-event-id") == "" { + t.Errorf("%+v", p.msgs) + } +} diff --git a/modules/zfs/go.mod b/modules/zfs/go.mod index 8b1859664..9078972fa 100644 --- a/modules/zfs/go.mod +++ b/modules/zfs/go.mod @@ -11,7 +11,7 @@ require ( require ( github.com/klauspost/compress v1.20.1 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/nats-io/nats.go v1.54.0 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