337 lines
10 KiB
Go
337 lines
10 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
// The condition keeper whole against a real bus (novox/hq ADR 0564 clauses 2 and 3, issue 1952 C6): the built binary,
|
|
// played over its real stdin and stdout as the node's tool runtime plays it (tools/call; mesh/state.* answered from a
|
|
// real nats-server's key-value buckets, watches handed over as the runtime hands them, mesh/seat.publish onto its
|
|
// subjects). An observer's finding written to its bucket raises a condition the keeper keeps in its bucket and says as
|
|
// condition-raised with the controller's body; the seat's verb lists it; the finding deleted clears it, said as
|
|
// condition-cleared.
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
|
seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
)
|
|
|
|
type wire struct {
|
|
ID json.RawMessage `json:"id,omitempty"`
|
|
Method string `json:"method,omitempty"`
|
|
Params json.RawMessage `json:"params,omitempty"`
|
|
Result json.RawMessage `json:"result,omitempty"`
|
|
Error *struct {
|
|
Message string `json:"message"`
|
|
} `json:"error,omitempty"`
|
|
}
|
|
|
|
// player is the node's tool runtime, as far as the keeper needs it, over a real bus.
|
|
type player struct {
|
|
ctx context.Context
|
|
nc *nats.Conn
|
|
js jetstream.JetStream
|
|
mu sync.Mutex
|
|
in io.Writer
|
|
next int
|
|
pending map[string]chan wire
|
|
}
|
|
|
|
func (p *player) send(m map[string]any) {
|
|
raw, _ := json.Marshal(m)
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
_, _ = p.in.Write(append(raw, '\n'))
|
|
}
|
|
|
|
func (p *player) answer(id json.RawMessage, result any, err error) {
|
|
if err != nil {
|
|
p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
|
|
return
|
|
}
|
|
p.send(map[string]any{"id": id, "result": result})
|
|
}
|
|
|
|
func (p *player) serve(out io.Reader) {
|
|
scanner := bufio.NewScanner(out)
|
|
scanner.Buffer(make([]byte, 1<<20), 1<<24)
|
|
for scanner.Scan() {
|
|
var m wire
|
|
if json.Unmarshal(scanner.Bytes(), &m) != nil {
|
|
continue
|
|
}
|
|
if m.Method == "" {
|
|
var key string
|
|
_ = json.Unmarshal(m.ID, &key)
|
|
p.mu.Lock()
|
|
ch := p.pending[key]
|
|
delete(p.pending, key)
|
|
p.mu.Unlock()
|
|
if ch != nil {
|
|
ch <- m
|
|
}
|
|
continue
|
|
}
|
|
go p.handle(m)
|
|
}
|
|
}
|
|
|
|
// kv is a state's bucket: the keeper's own by its local name, another's by `<module>.<name>`.
|
|
func (p *player) kv(state string) (jetstream.KeyValue, error) {
|
|
if owner, name, read := strings.Cut(state, "."); read {
|
|
return p.js.KeyValue(p.ctx, owner+"_"+name)
|
|
}
|
|
return p.js.KeyValue(p.ctx, Module+"_"+state)
|
|
}
|
|
|
|
func (p *player) handle(m wire) {
|
|
var a struct {
|
|
State string `json:"state"`
|
|
Key string `json:"key"`
|
|
Value json.RawMessage `json:"value"`
|
|
Revision uint64 `json:"revision"`
|
|
Subject string `json:"subject"`
|
|
Body json.RawMessage `json:"body"`
|
|
}
|
|
_ = json.Unmarshal(m.Params, &a)
|
|
kv, kvErr := p.kv(a.State)
|
|
switch m.Method {
|
|
case "mesh/state.get":
|
|
if kvErr != nil {
|
|
p.answer(m.ID, nil, kvErr)
|
|
return
|
|
}
|
|
e, err := kv.Get(p.ctx, a.Key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
p.answer(m.ID, nil, nil)
|
|
return
|
|
}
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
|
|
case "mesh/state.create", "mesh/state.update", "mesh/state.put":
|
|
if kvErr != nil {
|
|
p.answer(m.ID, nil, kvErr)
|
|
return
|
|
}
|
|
var rev uint64
|
|
var err error
|
|
switch m.Method {
|
|
case "mesh/state.create":
|
|
rev, err = kv.Create(p.ctx, a.Key, a.Value)
|
|
case "mesh/state.update":
|
|
rev, err = kv.Update(p.ctx, a.Key, a.Value, a.Revision)
|
|
default:
|
|
rev, err = kv.Put(p.ctx, a.Key, a.Value)
|
|
}
|
|
if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
|
p.send(map[string]any{"id": m.ID, "error": map[string]any{"code": -32010, "message": "changed: " + err.Error()}})
|
|
return
|
|
}
|
|
p.answer(m.ID, map[string]any{"revision": rev}, err)
|
|
case "mesh/state.keys":
|
|
if kvErr != nil {
|
|
p.answer(m.ID, nil, kvErr)
|
|
return
|
|
}
|
|
keys, err := kv.Keys(p.ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
keys, err = []string{}, nil
|
|
}
|
|
p.answer(m.ID, keys, err)
|
|
case "mesh/state.delete":
|
|
err := kvErr
|
|
if err == nil {
|
|
err = kv.Delete(p.ctx, a.Key)
|
|
}
|
|
p.answer(m.ID, map[string]any{}, err)
|
|
case "mesh/state.watch":
|
|
if kvErr != nil {
|
|
p.answer(m.ID, nil, kvErr)
|
|
return
|
|
}
|
|
w, err := kv.WatchAll(p.ctx)
|
|
if err != nil {
|
|
p.answer(m.ID, nil, err)
|
|
return
|
|
}
|
|
var watchID string
|
|
_ = json.Unmarshal(m.ID, &watchID)
|
|
current := true
|
|
for e := range w.Updates() {
|
|
if e == nil {
|
|
current = false
|
|
p.answer(m.ID, map[string]any{}, nil)
|
|
continue
|
|
}
|
|
op := "put"
|
|
if e.Operation() != jetstream.KeyValuePut {
|
|
op = "delete"
|
|
}
|
|
p.mu.Lock()
|
|
p.next++
|
|
id := fmt.Sprintf("w-%d", p.next)
|
|
p.mu.Unlock()
|
|
p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watchID,
|
|
"change": map[string]any{"state": a.State, "key": e.Key(), "op": op, "value": json.RawMessage(nullIfEmpty(e.Value())),
|
|
"revision": e.Revision(), "current": current}}})
|
|
}
|
|
case "mesh/seat.publish":
|
|
p.answer(m.ID, map[string]any{"sequence": 1}, p.nc.Publish(a.Subject, a.Body))
|
|
case "mesh/subscribe":
|
|
p.answer(m.ID, map[string]any{}, nil)
|
|
default:
|
|
p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
|
|
}
|
|
}
|
|
|
|
func nullIfEmpty(b []byte) []byte {
|
|
if len(b) == 0 {
|
|
return []byte("null")
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (p *player) call(name string, args map[string]any) (json.RawMessage, error) {
|
|
p.mu.Lock()
|
|
p.next++
|
|
id := fmt.Sprintf("c-%d", p.next)
|
|
ch := make(chan wire, 1)
|
|
p.pending[id] = ch
|
|
p.mu.Unlock()
|
|
p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args,
|
|
"_meta": map[string]any{"mesh/caller": "person.jochen"}}})
|
|
select {
|
|
case m := <-ch:
|
|
if m.Error != nil {
|
|
return nil, errors.New(m.Error.Message)
|
|
}
|
|
var r struct {
|
|
Content []struct {
|
|
Text string `json:"text"`
|
|
} `json:"content"`
|
|
IsError bool `json:"isError"`
|
|
}
|
|
if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 {
|
|
return nil, fmt.Errorf("%s answered %s", name, m.Result)
|
|
}
|
|
if r.IsError {
|
|
return nil, errors.New(r.Content[0].Text)
|
|
}
|
|
return json.RawMessage(r.Content[0].Text), nil
|
|
case <-time.After(60 * time.Second):
|
|
return nil, fmt.Errorf("%s did not answer in 60s", name)
|
|
}
|
|
}
|
|
|
|
// 1952/C6: an observer's finding raises a condition on the bus, said with the controller's body, and its deletion clears
|
|
// it.
|
|
func TestTheKeeperWholeRaisesAndClearsOnTheBus(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
t.Cleanup(cancel)
|
|
bin := filepath.Join(t.TempDir(), Module)
|
|
build := exec.Command("go", "build", "-o", bin, ".")
|
|
build.Stderr = os.Stderr
|
|
if err := build.Run(); err != nil {
|
|
t.Fatalf("the keeper does not build: %v", err)
|
|
}
|
|
nc, err := nats.Connect(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
buckets := []string{"mesh-controller_conditions", "mesh-delivery-assigner_wants", "node-engine_have", "mesh-delivery-assigner_placement",
|
|
"mesh-healers_attempts"}
|
|
for _, o := range Observers {
|
|
buckets = append(buckets, o+"_findings")
|
|
}
|
|
for _, s := range []string{ConditionsState, HistoryState, HeldState, BoundsState} {
|
|
buckets = append(buckets, Module+"_"+s)
|
|
}
|
|
for _, b := range buckets {
|
|
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
events, err := nc.SubscribeSync("mesh.seat." + Seat + ".event.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cmd := exec.CommandContext(ctx, bin)
|
|
cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE="+Module)
|
|
stdin, _ := cmd.StdinPipe()
|
|
stdout, _ := cmd.StdoutPipe()
|
|
cmd.Stderr = os.Stderr
|
|
if err := cmd.Start(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
|
|
p := &player{ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}}
|
|
go p.serve(stdout)
|
|
|
|
// The first call serves; the watches start after it.
|
|
if _, err := p.call(InputsTool, map[string]any{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
time.Sleep(time.Second)
|
|
at := time.Now().UTC()
|
|
f := seat.FindingsByKeyValue{Key: "machine.g14.silent", Kind: "silent", Subject: seat.Subject{Scope: cc.ScopeMachine, ID: "g14",
|
|
Machine: "g14"}, Severity: cc.Warning, Summary: "g14 is silent", Headline: "g14 is offline",
|
|
Explanation: "g14 said nothing for three minutes.", Resolved: "g14 is back", Source: "S1",
|
|
Observer: watchmanModule, At: at, Evidence: []seat.Evidence{{At: at, Said: "no presence for 3m"}}}
|
|
raw, err := encodeFinding(f)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
findings, _ := js.KeyValue(ctx, watchmanModule+"_findings")
|
|
if _, err := findings.Put(ctx, f.Key, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
msg, err := events.NextMsg(10 * time.Second)
|
|
if err != nil {
|
|
t.Fatalf("no condition-raised was said: %v", err)
|
|
}
|
|
var e Event
|
|
if err := json.Unmarshal(msg.Data, &e); err != nil || msg.Subject != "mesh.seat.mesh-conditions.event.condition-raised" ||
|
|
e.Key != f.Key || e.Change != cc.ChangeRaised || e.Show != cc.ShowKeeper(f.Key) || e.Headline != f.Headline {
|
|
t.Fatalf("said %s on %s", msg.Data, msg.Subject)
|
|
}
|
|
listed, err := p.call(Seat+".conditions", map[string]any{})
|
|
if err != nil || !strings.Contains(string(listed), `"key":"machine.g14.silent"`) {
|
|
t.Fatalf("conditions answered %s, %v", listed, err)
|
|
}
|
|
if err := findings.Delete(ctx, f.Key); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
msg, err = events.NextMsg(10 * time.Second)
|
|
if err != nil || msg.Subject != "mesh.seat.mesh-conditions.event.condition-cleared" {
|
|
t.Fatalf("no condition-cleared was said: %v", err)
|
|
}
|
|
conditions, _ := js.KeyValue(ctx, Module+"_"+ConditionsState)
|
|
if _, err := conditions.Get(ctx, f.Key); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
t.Errorf("the cleared condition is still kept: %v", err)
|
|
}
|
|
}
|