Take the rehearsal tool out of the nats module: a new nats build is held until the bus upgrade itself, so it could only run after the step it rehearses (ADR 0236 section 36)
This commit is contained in:
+6
-12
@@ -7,7 +7,7 @@ artifact (`nats:<release>-alpine@sha256:<digest>`: the digest names the image, t
|
||||
and `cmd/nats-tools`' release test keeps the tag equal to the server its tests run). The module's own two
|
||||
programs are mounted into it read-only (novox/hq ADR 0480): the snapshot program (a Go bundle of
|
||||
`cmd/mesh-nats-snapshot`) at `/opt/mesh`, and the entrypoint (`program/entrypoint.sh`, an archive) at
|
||||
`/opt/mesh-entrypoint`, run through `sh`. Both archive resources are in the container's `restart-on`. Its tools (`cmd/nats-tools`) are served by the node's runtime and only read, but for `nats_rehearse`, which starts containers of its own on a copy of the last snapshot and nothing else.
|
||||
`/opt/mesh-entrypoint`, run through `sh`. Both archive resources are in the container's `restart-on`. Its tools (`cmd/nats-tools`) are served by the node's runtime and only read.
|
||||
|
||||
## Its data, and how it is backed up
|
||||
|
||||
@@ -87,29 +87,23 @@ The bus is replaced only as a planned step a person starts (novox/hq ADR 0236, A
|
||||
merge of a change to this module builds the bus and sends nothing. The steps, in this order:
|
||||
|
||||
1. **Nothing else runs.** No delivery plan is open (`mesh-controller.plans`).
|
||||
2. **Rehearse on a copy.** `<node>/nats.nats_rehearse` on the bus's machine, given the new build's image and
|
||||
the directories holding its unpacked program and entrypoint (and the old image, which defaults to the
|
||||
running bus's): it verifies the latest `bus.tar`, restores it into a throwaway directory, starts the new
|
||||
declaration on it on a loopback port and checks every stream is served, stops it, starts the old
|
||||
declaration on the same copy and checks again, then removes everything and answers each step. It touches
|
||||
nothing of the running bus, and refuses while a bus upgrade is open. Go on only when it answers `ok`.
|
||||
3. **Snapshot with the dump the bus runs now**, through the machine's backup holder
|
||||
2. **Snapshot with the dump the bus runs now**, through the machine's backup holder
|
||||
(`<node>/node-backup.now` for `nats` on the bus's machine), and verify the `bus.tar` it wrote
|
||||
(`docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot verify`, or the old path
|
||||
`mesh-nats-snapshot verify` on the declaration before ADR 0480). Note where it is: that is `snapshot-taken`.
|
||||
4. **Ask the operator for the moment** on the proven channel, naming what it raises: bus maintenance is
|
||||
3. **Ask the operator for the moment** on the proven channel, naming what it raises: bus maintenance is
|
||||
open while the step runs; every bus connection is dropped for seconds and reconnects; conditions of
|
||||
other modules may flash and clear. Go on only on their yes.
|
||||
5. **Start it:** `mesh-controller.bus` with `upgrade` true, `why`, `reversible` true and `snapshot-taken`.
|
||||
4. **Start it:** `mesh-controller.bus` with `upgrade` true, `why`, `reversible` true and `snapshot-taken`.
|
||||
`reversible` is true because the server binary (the same digest), the data directory and the
|
||||
configuration do not change; `TestTheStoreSurvivesTheNewDeclarationTheOldAndTheNewAgain` shows it:
|
||||
the new declaration, then the old one, then the new again run on one store, each reading what the
|
||||
one before wrote.
|
||||
6. **After:** `nats_server`, `nats_streams` (every stream and bucket as before) and the controller's
|
||||
5. **After:** `nats_server`, `nats_streams` (every stream and bucket as before) and the controller's
|
||||
`doctor`; the step ends done when the controller has checked every stream, every consumer and a
|
||||
round trip (ADR 0236 section 40). Then wait for the next nightly dump and verify it: the upgrade is
|
||||
finished only once that dump verified.
|
||||
7. **Back:** a bus upgrade to the previous build puts the old declaration back on the same store; if the
|
||||
6. **Back:** a bus upgrade to the previous build puts the old declaration back on the same store; if the
|
||||
store itself is hurt, restore the snapshot as described above.
|
||||
|
||||
## Tests
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
// nats-tools: the mesh bus's own tools (novox/hq design 25), a Go bundle the node's runtime launches on the
|
||||
// machine that runs the bus. Everything here reads — the server's monitoring API and the user list the
|
||||
// controller composed — and nothing changes the bus; the one tool that starts anything, nats_rehearse, starts
|
||||
// only containers of its own on a copy of a snapshot. stdout is the MCP channel; this says nothing else.
|
||||
// controller composed — and nothing changes the bus. stdout is the MCP channel; this says nothing else.
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -23,13 +22,6 @@ func num(description string) map[string]any {
|
||||
|
||||
func arg(a map[string]any, k string) string { s, _ := a[k].(string); return s }
|
||||
|
||||
func first(a, b string) string {
|
||||
if a != "" {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func limitOf(a map[string]any, fallback int) int {
|
||||
if v, ok := a["limit"].(float64); ok && v > 0 {
|
||||
return int(v)
|
||||
@@ -148,27 +140,6 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
|
||||
varz, verr := get("/varz")
|
||||
return InForce(answer, at, werr, varz, verr), nil
|
||||
}},
|
||||
{Name: "nats_rehearse",
|
||||
Description: "Rehearse the bus upgrade on a copy (novox/hq ADR 0480 clause 6): verify the latest snapshot (the dump's bus.tar), restore it into a throwaway directory, start the NEW declaration (the upstream image with the program and entrypoint mounted) on it on a loopback port and check it serves every stream, stop it, start the OLD declaration on the same copy and check again, and remove everything. Answers each step. It touches nothing of the running bus (container, directory, ports) and refuses while a bus upgrade is open or when that cannot be told.",
|
||||
Input: map[string]any{
|
||||
"snapshot": str("the snapshot's path on this machine (default: the nats module's bus.tar)"),
|
||||
"new_image": str("the new declaration's image, name:tag@sha256:digest"),
|
||||
"new_program": str("the directory on this machine holding mesh-nats-snapshot, mounted at /opt/mesh (default: the module's unpacked program)"),
|
||||
"new_entrypoint": str("the directory on this machine holding entrypoint.sh, mounted at /opt/mesh-entrypoint (default: the module's unpacked entrypoint)"),
|
||||
"old_image": str("the old declaration's image (default: the image of the running bus's container, read only)"),
|
||||
},
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
args := RehearseArgs{
|
||||
Snapshot: first(arg(a, "snapshot"), env("MESH_NATS_SNAPSHOT", "")),
|
||||
NewImage: arg(a, "new_image"),
|
||||
NewProgram: first(arg(a, "new_program"), env("MESH_NATS_PROGRAM_DIR", "")),
|
||||
NewEntrypoint: first(arg(a, "new_entrypoint"), env("MESH_NATS_ENTRYPOINT_DIR", "")),
|
||||
OldImage: arg(a, "old_image"),
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
|
||||
defer cancel()
|
||||
return LiveRehearsal().Run(ctx, args)
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,359 +0,0 @@
|
||||
package main
|
||||
|
||||
// nats_rehearse: the bus upgrade rehearsed on a copy, before the real step (novox/hq ADR 0480 clause 6, ADR 0236).
|
||||
//
|
||||
// It takes the latest snapshot the dump wrote (bus.tar), verifies it, restores it into a throwaway directory, starts
|
||||
// the NEW declaration's container on that copy on a loopback port and checks that it serves every stream the snapshot
|
||||
// holds, stops it, starts the OLD declaration on the same copy and checks again, and removes everything it made. Each
|
||||
// step is answered. That is what makes `reversible` true for the bus upgrade: the new declaration and then the old
|
||||
// one run on one store, both well.
|
||||
//
|
||||
// **It touches only its own copy.** Every container it starts is named mesh-rehearsal-<id> and publishes only a random
|
||||
// port on the loopback; its directory is a new temporary one; the running bus's container, directory and ports are
|
||||
// never named, started, stopped, mounted or published. The one thing it reads of the running bus is the image
|
||||
// reference of its container (`docker inspect`), as the default for the old declaration. It refuses to run while a bus
|
||||
// upgrade is open, and when it cannot tell, so a rehearsal never loads the machine in the middle of the step.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// StreamWant is one stream the snapshot holds and the copy must serve with at least its messages.
|
||||
type StreamWant struct {
|
||||
Name string `json:"name"`
|
||||
Messages uint64 `json:"messages"`
|
||||
}
|
||||
|
||||
// Step is one answered step of the rehearsal.
|
||||
type Step struct {
|
||||
Step string `json:"step"`
|
||||
OK bool `json:"ok"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
}
|
||||
|
||||
// Declaration is a way to run the bus: an image, what is mounted into it and the arguments it is given.
|
||||
type Declaration struct {
|
||||
Name string
|
||||
Image string
|
||||
Mounts []string // host:container[:ro], besides the copy's own
|
||||
Args []string
|
||||
}
|
||||
|
||||
// Rehearsal is the rehearsal and what it reaches outside itself; the tests replace the reaches.
|
||||
type Rehearsal struct {
|
||||
// Docker runs the docker command line and answers its standard output.
|
||||
Docker func(ctx context.Context, args ...string) (string, error)
|
||||
// Serves connects to the copy and checks it serves the streams.
|
||||
Serves func(ctx context.Context, port, user, password string, want []StreamWant) (string, error)
|
||||
// UpgradeOpen says whether a bus upgrade is open, and why not if it cannot say.
|
||||
UpgradeOpen func(ctx context.Context) (bool, string, error)
|
||||
// TempDir makes the throwaway directory; RemoveAll removes it.
|
||||
TempDir func() (string, error)
|
||||
RemoveAll func(string) error
|
||||
// ID is a fresh word for the names; UID the user the containers run as, so what they write can be removed.
|
||||
ID func() string
|
||||
UID string
|
||||
// BusContainer is the running bus's container, which is never touched.
|
||||
BusContainer string
|
||||
}
|
||||
|
||||
// RehearseArgs are the call's arguments.
|
||||
type RehearseArgs struct {
|
||||
Snapshot string // host path of bus.tar
|
||||
NewImage string // the new declaration's image
|
||||
NewProgram string // host directory holding mesh-nats-snapshot, mounted at /opt/mesh
|
||||
NewEntrypoint string // host directory holding entrypoint.sh, mounted at /opt/mesh-entrypoint
|
||||
OldImage string // the old declaration's image; the running bus's when empty
|
||||
}
|
||||
|
||||
const (
|
||||
rehearsalUser = "rehearsal"
|
||||
rehearsalConf = "listen: \"0.0.0.0:4222\"\nhttp: 0.0.0.0:8222\njetstream { store_dir: \"/data\" }\ninclude \"accounts.conf\"\n"
|
||||
rehearsalAccounts = "accounts { MESH { jetstream: enabled, users: [ { user: \"%s\", password: \"%s\" } ] } }\n"
|
||||
)
|
||||
|
||||
// Run is the rehearsal. It answers every step; a failed step stops it, everything is removed, and the error says which.
|
||||
func (r Rehearsal) Run(ctx context.Context, a RehearseArgs) (map[string]any, error) {
|
||||
var steps []Step
|
||||
done := func(name string, err error, detail string) error {
|
||||
s := Step{Step: name, OK: err == nil, Detail: detail}
|
||||
if err != nil {
|
||||
s.Detail = err.Error()
|
||||
}
|
||||
steps = append(steps, s)
|
||||
return err
|
||||
}
|
||||
fail := func(err error) (map[string]any, error) {
|
||||
return map[string]any{"ok": false, "steps": steps}, fmt.Errorf("the rehearsal failed at %q: %v (steps: %s)",
|
||||
steps[len(steps)-1].Step, err, spoken(steps))
|
||||
}
|
||||
|
||||
if a.Snapshot == "" || a.NewImage == "" || a.NewProgram == "" || a.NewEntrypoint == "" {
|
||||
return nil, errors.New("snapshot, new_image, new_program and new_entrypoint are needed")
|
||||
}
|
||||
for _, p := range []string{a.Snapshot, a.NewProgram, a.NewEntrypoint} {
|
||||
if !filepath.IsAbs(p) || strings.ContainsAny(p, ":,\n") {
|
||||
return nil, fmt.Errorf("%q is not an absolute path without a colon, comma or newline", p)
|
||||
}
|
||||
}
|
||||
open, why, err := r.UpgradeOpen(ctx)
|
||||
if err != nil {
|
||||
_ = done("no bus upgrade is open", err, "")
|
||||
return fail(fmt.Errorf("it cannot tell whether a bus upgrade is open, and refuses until it can: %w", err))
|
||||
}
|
||||
if open {
|
||||
_ = done("no bus upgrade is open", errors.New("a bus upgrade is open: "+why), "")
|
||||
return fail(errors.New("a bus upgrade is open: " + why))
|
||||
}
|
||||
_ = done("no bus upgrade is open", nil, "")
|
||||
|
||||
oldImage := a.OldImage
|
||||
if oldImage == "" {
|
||||
out, err := r.Docker(ctx, "inspect", "-f", "{{.Config.Image}}", r.BusContainer)
|
||||
oldImage = strings.TrimSpace(out)
|
||||
if err != nil || oldImage == "" {
|
||||
_ = done("the old declaration's image", fmt.Errorf("the running bus's image could not be read and old_image is not given: %v", err), "")
|
||||
return fail(errors.New("no old image"))
|
||||
}
|
||||
}
|
||||
_ = done("the old declaration's image", nil, oldImage)
|
||||
|
||||
dir, err := r.TempDir()
|
||||
if err != nil {
|
||||
_ = done("a throwaway directory", err, "")
|
||||
return fail(err)
|
||||
}
|
||||
id := r.ID()
|
||||
var started []string
|
||||
defer func() {
|
||||
for _, n := range started {
|
||||
_, _ = r.Docker(context.Background(), "rm", "-f", n)
|
||||
}
|
||||
_ = r.RemoveAll(dir)
|
||||
}()
|
||||
conf, store := filepath.Join(dir, "conf"), filepath.Join(dir, "store")
|
||||
for _, d := range []string{conf, store} {
|
||||
if err := os.MkdirAll(d, 0o700); err != nil {
|
||||
_ = done("a throwaway directory", err, "")
|
||||
return fail(err)
|
||||
}
|
||||
}
|
||||
password := id + id
|
||||
// nats.conf is what the mesh's entrypoint reads; nats-server.conf is what the upstream image runs with no arguments,
|
||||
// which is how an old declaration without an entrypoint of its own starts.
|
||||
for _, f := range []string{"nats.conf", "nats-server.conf"} {
|
||||
if err := os.WriteFile(filepath.Join(conf, f), []byte(rehearsalConf), 0o600); err != nil {
|
||||
_ = done("a throwaway directory", err, "")
|
||||
return fail(err)
|
||||
}
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(fmt.Sprintf(rehearsalAccounts, rehearsalUser, password)), 0o600); err != nil {
|
||||
_ = done("a throwaway directory", err, "")
|
||||
return fail(err)
|
||||
}
|
||||
_ = done("a throwaway directory", nil, dir)
|
||||
|
||||
snapDir, snapFile := filepath.Dir(a.Snapshot), filepath.Base(a.Snapshot)
|
||||
// The snapshot's directory is the module's, readable by the machine's root only; what reads it runs as root with
|
||||
// the directory mounted read-only, and writes nothing there. What writes (the copy) is handed to this user.
|
||||
program := []string{"-v", a.NewProgram + ":/opt/mesh:ro", "-v", snapDir + ":/in:ro"}
|
||||
tool := func(extra ...string) []string {
|
||||
return append(append([]string{"run", "--rm"}, program...), extra...)
|
||||
}
|
||||
|
||||
// 1. The snapshot is whole.
|
||||
out, err := r.Docker(ctx, tool("--entrypoint", "/opt/mesh/mesh-nats-snapshot", a.NewImage, "verify", "--from", "/in/"+snapFile)...)
|
||||
if err := done("the snapshot verifies", err, strings.TrimSpace(out)); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
// 2. What it holds.
|
||||
out, err = r.Docker(ctx, "run", "--rm", "-v", snapDir+":/in:ro", "--entrypoint", "tar", a.NewImage, "-xOf", "/in/"+snapFile, "manifest.json")
|
||||
var want []StreamWant
|
||||
if err == nil {
|
||||
var m struct {
|
||||
Streams []StreamWant `json:"streams"`
|
||||
}
|
||||
if err = json.Unmarshal([]byte(out), &m); err == nil {
|
||||
want = m.Streams
|
||||
if len(want) == 0 {
|
||||
err = errors.New("the snapshot holds no stream")
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := done("the snapshot's streams are read", err, fmt.Sprintf("%d stream(s)", len(want))); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
// 3. A new store from it, beside nothing: the throwaway directory.
|
||||
out, err = r.Docker(ctx, "run", "--rm", "-v", a.NewProgram+":/opt/mesh:ro", "-v", snapDir+":/in:ro", "-v", store+":/out",
|
||||
"--entrypoint", "/opt/mesh/mesh-nats-snapshot", a.NewImage, "restore", "--into", "/out/jetstream-copy", "--from", "/in/"+snapFile)
|
||||
if err := done("the snapshot is restored into the copy", err, lastLine(out)); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
// The restore ran as root (it reads the snapshot's directory); the copy is handed to this user, and its
|
||||
// jetstream directory moved to where the server looks.
|
||||
_, err = r.Docker(ctx, "run", "--rm", "-v", store+":/out", "--entrypoint", "sh", a.NewImage, "-c",
|
||||
"mv /out/jetstream-copy/jetstream /out/jetstream && rmdir /out/jetstream-copy && chown -R "+r.UID+" /out")
|
||||
if err := done("the copy is handed to the rehearsal's user", err, ""); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
|
||||
// 4. The new declaration, then the old one, on the same copy.
|
||||
decls := []Declaration{
|
||||
{Name: "new", Image: a.NewImage,
|
||||
Mounts: []string{a.NewProgram + ":/opt/mesh:ro", a.NewEntrypoint + ":/opt/mesh-entrypoint:ro"},
|
||||
Args: []string{"sh", "/opt/mesh-entrypoint/entrypoint.sh"}},
|
||||
{Name: "old", Image: oldImage},
|
||||
}
|
||||
for _, d := range decls {
|
||||
name := "mesh-rehearsal-" + d.Name + "-" + id
|
||||
if name == r.BusContainer {
|
||||
return nil, errors.New("a rehearsal never names the running bus's container")
|
||||
}
|
||||
args := []string{"run", "-d", "--name", name, "--user", r.UID, "-p", "127.0.0.1::4222", "-e", "MESH_NATS_CONF_POLL_SECONDS=1",
|
||||
"-v", store + ":/data", "-v", conf + ":/etc/nats:ro"}
|
||||
for _, m := range d.Mounts {
|
||||
args = append(args, "-v", m)
|
||||
}
|
||||
args = append(append(args, d.Image), d.Args...)
|
||||
started = append(started, name)
|
||||
if _, err := r.Docker(ctx, args...); err != nil {
|
||||
_ = done("the "+d.Name+" declaration starts on the copy", err, "")
|
||||
return fail(err)
|
||||
}
|
||||
mapped, err := r.Docker(ctx, "port", name, "4222/tcp")
|
||||
_, port, ok := strings.Cut(strings.TrimSpace(strings.SplitN(mapped, "\n", 2)[0]), ":")
|
||||
if err != nil || !ok {
|
||||
_ = done("the "+d.Name+" declaration starts on the copy", fmt.Errorf("its port is not known: %v %q", err, mapped), "")
|
||||
return fail(errors.New("no port"))
|
||||
}
|
||||
detail, err := r.Serves(ctx, port, rehearsalUser, password, want)
|
||||
if err := done("the "+d.Name+" declaration serves every stream of the copy", err, detail); err != nil {
|
||||
logs, _ := r.Docker(context.Background(), "logs", "--tail", "20", name)
|
||||
steps[len(steps)-1].Detail += " | its last log lines: " + strings.TrimSpace(logs)
|
||||
return fail(err)
|
||||
}
|
||||
if _, err := r.Docker(ctx, "stop", "-t", "10", name); err != nil {
|
||||
_ = done("the "+d.Name+" declaration stops", err, "")
|
||||
return fail(err)
|
||||
}
|
||||
_ = done("the "+d.Name+" declaration stops", nil, "")
|
||||
}
|
||||
_ = done("everything made is removed", nil, "containers "+strings.Join(started, ", ")+" and "+dir)
|
||||
return map[string]any{"ok": true, "reversible": true, "steps": steps,
|
||||
"says": "the new declaration and then the old one ran healthy on one copy of the snapshot's store"}, nil
|
||||
}
|
||||
|
||||
func lastLine(s string) string {
|
||||
lines := strings.Split(strings.TrimSpace(s), "\n")
|
||||
return lines[len(lines)-1]
|
||||
}
|
||||
|
||||
func spoken(steps []Step) string {
|
||||
var parts []string
|
||||
for _, s := range steps {
|
||||
mark := "ok"
|
||||
if !s.OK {
|
||||
mark = "FAILED"
|
||||
}
|
||||
parts = append(parts, s.Step+": "+mark)
|
||||
}
|
||||
return strings.Join(parts, "; ")
|
||||
}
|
||||
|
||||
// LiveRehearsal is the rehearsal as this machine reaches what it needs.
|
||||
func LiveRehearsal() Rehearsal {
|
||||
return Rehearsal{
|
||||
Docker: func(ctx context.Context, args ...string) (string, error) {
|
||||
cmd := exec.CommandContext(ctx, "docker", args...)
|
||||
var stderr strings.Builder
|
||||
cmd.Stderr = &stderr
|
||||
out, err := cmd.Output()
|
||||
if err != nil {
|
||||
return string(out), fmt.Errorf("docker %s: %v: %s", args[0], err, strings.TrimSpace(stderr.String()))
|
||||
}
|
||||
return string(out), nil
|
||||
},
|
||||
Serves: servesStreams,
|
||||
UpgradeOpen: func(context.Context) (bool, string, error) {
|
||||
// The controller's own word, through the runtime: `bus` says the last step and whether it still runs.
|
||||
raw, err := stdio.Ask("mesh-controller.bus", map[string]any{})
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
return upgradeRunning(string(raw))
|
||||
},
|
||||
TempDir: func() (string, error) {
|
||||
return os.MkdirTemp(env("MESH_NATS_REHEARSAL_DIR", os.TempDir()), "mesh-nats-rehearsal-*")
|
||||
},
|
||||
RemoveAll: os.RemoveAll,
|
||||
ID: func() string { b := make([]byte, 4); _, _ = rand.Read(b); return hex.EncodeToString(b) },
|
||||
UID: fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid()),
|
||||
BusContainer: env("MESH_NATS_CONTAINER", "mesh-broker-nats"),
|
||||
}
|
||||
}
|
||||
|
||||
// upgradeRunning reads the controller's answer for the last bus step: running, or ended. An answer it cannot read is
|
||||
// an error, so the rehearsal refuses rather than guesses.
|
||||
func upgradeRunning(answer string) (bool, string, error) {
|
||||
if strings.Contains(answer, "running since") {
|
||||
i := strings.Index(answer, "last step")
|
||||
if i < 0 {
|
||||
i = 0
|
||||
}
|
||||
return true, strings.TrimSpace(answer[i:min(len(answer), i+200)]), nil
|
||||
}
|
||||
if strings.Contains(answer, "the bus is ") || strings.Contains(answer, "last step") || strings.Contains(answer, "bus upgrade") {
|
||||
return false, "", nil
|
||||
}
|
||||
return false, "", fmt.Errorf("the controller's answer for the bus is not one this tool reads: %.120q", answer)
|
||||
}
|
||||
|
||||
// servesStreams connects to the copy and checks each stream is there with at least its messages.
|
||||
func servesStreams(ctx context.Context, port, user, password string, want []StreamWant) (string, error) {
|
||||
var nc *nats.Conn
|
||||
var err error
|
||||
deadline := time.Now().Add(60 * time.Second)
|
||||
for {
|
||||
nc, err = nats.Connect("nats://127.0.0.1:"+port, nats.UserInfo(user, password), nats.Timeout(2*time.Second))
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) || ctx.Err() != nil {
|
||||
return "", fmt.Errorf("the copy did not come up: %w", err)
|
||||
}
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
defer nc.Close()
|
||||
js, err := nc.JetStream()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var missing []string
|
||||
for _, w := range want {
|
||||
info, err := js.StreamInfo(w.Name)
|
||||
if err != nil {
|
||||
missing = append(missing, w.Name+" ("+err.Error()+")")
|
||||
continue
|
||||
}
|
||||
if info.State.Msgs < w.Messages {
|
||||
missing = append(missing, fmt.Sprintf("%s holds %d of %d messages", w.Name, info.State.Msgs, w.Messages))
|
||||
}
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return "", errors.New("the copy lacks: " + strings.Join(missing, "; "))
|
||||
}
|
||||
return fmt.Sprintf("%d stream(s), each with at least the messages the snapshot holds", len(want)), nil
|
||||
}
|
||||
@@ -1,136 +0,0 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
// The rehearsal against real containers: a source server filled and snapshotted by the real program, then the
|
||||
// rehearsal run on that bus.tar with the manifest's pinned image, the real program and the real entrypoint. Run through
|
||||
// mesh-controller.try (the merge check runs it with -tags integration); never by hand on a workstation.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func sh(t *testing.T, name string, args ...string) string {
|
||||
t.Helper()
|
||||
out, err := exec.Command(name, args...).CombinedOutput()
|
||||
if err != nil {
|
||||
t.Fatalf("%s %s: %v\n%s", name, strings.Join(args, " "), err, out)
|
||||
}
|
||||
return strings.TrimSpace(string(out))
|
||||
}
|
||||
|
||||
func TestARehearsalRunsTheNewAndTheOldDeclarationOnACopyOfARealSnapshot(t *testing.T) {
|
||||
if _, err := exec.LookPath("docker"); err != nil {
|
||||
t.Fatalf("this test needs docker on the machine that runs it: %v", err)
|
||||
}
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m struct {
|
||||
Build struct {
|
||||
Artifacts []struct{ Name, From string } `json:"artifacts"`
|
||||
} `json:"build"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
image := ""
|
||||
for _, a := range m.Build.Artifacts {
|
||||
if a.Name == "server" {
|
||||
image = a.From
|
||||
}
|
||||
}
|
||||
work := t.TempDir()
|
||||
program, entry, snaps := filepath.Join(work, "program"), filepath.Join(work, "entry"), filepath.Join(work, "snaps")
|
||||
for _, d := range []string{program, entry, snaps, filepath.Join(work, "srcdata")} {
|
||||
if err := os.MkdirAll(d, 0o777); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = os.Chmod(d, 0o777)
|
||||
}
|
||||
build := exec.Command("go", "build", "-trimpath", "-o", filepath.Join(program, "mesh-nats-snapshot"), "../mesh-nats-snapshot")
|
||||
build.Env = append(os.Environ(), "CGO_ENABLED=0", "GOOS=linux")
|
||||
if out, err := build.CombinedOutput(); err != nil {
|
||||
t.Fatalf("building the snapshot program: %v\n%s", err, out)
|
||||
}
|
||||
script, err := os.ReadFile("../../program/entrypoint.sh")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(entry, "entrypoint.sh"), script, 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The source: a plain server with the one account, filled, and snapshotted by the real program.
|
||||
conf := filepath.Join(work, "srcconf")
|
||||
if err := os.MkdirAll(conf, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = os.WriteFile(filepath.Join(conf, "nats.conf"), []byte(rehearsalConf), 0o644)
|
||||
_ = os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(fmt.Sprintf(rehearsalAccounts, "admin", "admin")), 0o644)
|
||||
src := fmt.Sprintf("mesh-rehearsal-source-%d", os.Getpid())
|
||||
uid := fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid())
|
||||
sh(t, "docker", "run", "-d", "--rm", "--name", src, "--user", uid, "-p", "127.0.0.1::4222",
|
||||
"-v", filepath.Join(work, "srcdata")+":/data", "-v", conf+":/etc/nats:ro", image, "-c", "/etc/nats/nats.conf")
|
||||
t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", src).Run() })
|
||||
_, port, _ := strings.Cut(sh(t, "docker", "port", src, "4222/tcp"), ":")
|
||||
var nc *nats.Conn
|
||||
for deadline := time.Now().Add(30 * time.Second); ; time.Sleep(300 * time.Millisecond) {
|
||||
nc, err = nats.Connect("nats://127.0.0.1:"+port, nats.UserInfo("admin", "admin"))
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
defer nc.Close()
|
||||
js, _ := nc.JetStream()
|
||||
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-rehearsal"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := 0; i < 5; i++ {
|
||||
if _, err := kv.Put(fmt.Sprintf("k%d", i), []byte("v")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
cred, _ := json.Marshal(map[string]string{"url": "nats://127.0.0.1:" + port, "user": "admin", "password": "admin"})
|
||||
cmd := exec.Command(filepath.Join(program, "mesh-nats-snapshot"), "snapshot", "--server", "nats://127.0.0.1:"+port, "--credential", "-")
|
||||
cmd.Stdin = strings.NewReader(string(cred))
|
||||
tarball, err := cmd.Output()
|
||||
if err != nil {
|
||||
t.Fatalf("the snapshot: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(snaps, "bus.tar"), tarball, 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
r := LiveRehearsal()
|
||||
r.UpgradeOpen = func(context.Context) (bool, string, error) { return false, "", nil }
|
||||
r.BusContainer = "mesh-broker-nats-not-here"
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
|
||||
defer cancel()
|
||||
got, err := r.Run(ctx, RehearseArgs{Snapshot: filepath.Join(snaps, "bus.tar"), NewImage: image,
|
||||
NewProgram: program, NewEntrypoint: entry, OldImage: image})
|
||||
if err != nil {
|
||||
t.Fatalf("%v\n%+v", err, got)
|
||||
}
|
||||
if got["ok"] != true {
|
||||
t.Fatalf("answer %+v", got)
|
||||
}
|
||||
if left := sh(t, "docker", "ps", "-a", "-q", "--filter", "name=mesh-rehearsal-new", "--filter", "name=mesh-rehearsal-old"); left != "" {
|
||||
t.Errorf("containers are left: %s", left)
|
||||
}
|
||||
}
|
||||
@@ -1,262 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// fake is a scripted docker: every call is recorded; a call whose first argument and one later word match a key of
|
||||
// fail answers that error.
|
||||
type fake struct {
|
||||
calls []string
|
||||
outputs map[string]string // first word of the call -> output
|
||||
fail map[string]error
|
||||
removed []string
|
||||
}
|
||||
|
||||
func (f *fake) docker(_ context.Context, args ...string) (string, error) {
|
||||
line := strings.Join(args, " ")
|
||||
f.calls = append(f.calls, line)
|
||||
for key, err := range f.fail {
|
||||
if strings.Contains(line, key) {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
switch args[0] {
|
||||
case "inspect":
|
||||
return "nats:old@sha256:0\n", nil
|
||||
case "port":
|
||||
return "127.0.0.1:41000\n", nil
|
||||
}
|
||||
for _, a := range args {
|
||||
if a == "tar" {
|
||||
return `{"streams":[{"name":"KV_a","messages":3},{"name":"EVENTS","messages":10}]}`, nil
|
||||
}
|
||||
}
|
||||
return f.outputs[args[0]], nil
|
||||
}
|
||||
|
||||
func rehearsal(t *testing.T, f *fake) (Rehearsal, *[]string) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
var serves []string
|
||||
return Rehearsal{
|
||||
Docker: f.docker,
|
||||
Serves: func(_ context.Context, port, user, password string, want []StreamWant) (string, error) {
|
||||
serves = append(serves, port+" "+user)
|
||||
if len(want) != 2 || want[0].Name != "KV_a" || want[0].Messages != 3 {
|
||||
t.Errorf("the streams it was asked to find: %+v", want)
|
||||
}
|
||||
return "ok", nil
|
||||
},
|
||||
UpgradeOpen: func(context.Context) (bool, string, error) { return false, "", nil },
|
||||
TempDir: func() (string, error) { return dir, nil },
|
||||
RemoveAll: func(p string) error { f.removed = append(f.removed, p); return os.RemoveAll(p) },
|
||||
ID: func() string { return "abcd" },
|
||||
UID: "1000:1000",
|
||||
BusContainer: "mesh-broker-nats",
|
||||
}, &serves
|
||||
}
|
||||
|
||||
var args = RehearseArgs{Snapshot: "/var/lib/snap/bus.tar", NewImage: "nats:2.11.17-alpine@sha256:aa", NewProgram: "/var/lib/p", NewEntrypoint: "/var/lib/e"}
|
||||
|
||||
func TestARehearsalRunsTheNewDeclarationThenTheOldOnOneCopyAndRemovesEverything(t *testing.T) {
|
||||
f := &fake{}
|
||||
r, serves := rehearsal(t, f)
|
||||
got, err := r.Run(context.Background(), args)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got["ok"] != true || got["reversible"] != true {
|
||||
t.Fatalf("answer %v", got)
|
||||
}
|
||||
var names []string
|
||||
for _, s := range got["steps"].([]Step) {
|
||||
if !s.OK {
|
||||
t.Errorf("step %q failed: %s", s.Step, s.Detail)
|
||||
}
|
||||
names = append(names, s.Step)
|
||||
}
|
||||
for _, want := range []string{"no bus upgrade is open", "the snapshot verifies", "the snapshot is restored into the copy",
|
||||
"the new declaration serves every stream of the copy", "the old declaration serves every stream of the copy", "everything made is removed"} {
|
||||
found := false
|
||||
for _, n := range names {
|
||||
found = found || n == want
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("no step %q in %v", want, names)
|
||||
}
|
||||
}
|
||||
if len(*serves) != 2 {
|
||||
t.Errorf("the copy was checked %d times, not twice (new, then old)", len(*serves))
|
||||
}
|
||||
// New before old, on the same store; the old one is the running bus's image.
|
||||
var runs []string
|
||||
for _, c := range f.calls {
|
||||
if strings.HasPrefix(c, "run -d") {
|
||||
runs = append(runs, c)
|
||||
}
|
||||
}
|
||||
if len(runs) != 2 || !strings.Contains(runs[0], "mesh-rehearsal-new-abcd") || !strings.Contains(runs[1], "mesh-rehearsal-old-abcd") {
|
||||
t.Fatalf("containers started: %v", runs)
|
||||
}
|
||||
if !strings.Contains(runs[0], "/opt/mesh-entrypoint") || !strings.HasSuffix(runs[0], "sh /opt/mesh-entrypoint/entrypoint.sh") {
|
||||
t.Errorf("the new declaration is not the mounted entrypoint: %s", runs[0])
|
||||
}
|
||||
if !strings.HasSuffix(runs[1], "nats:old@sha256:0") {
|
||||
t.Errorf("the old declaration is not the running bus's image: %s", runs[1])
|
||||
}
|
||||
store := func(s string) string { return s[strings.Index(s, "-v ")+3 : strings.Index(s, ":/data")] }
|
||||
if store(runs[0]) != store(runs[1]) {
|
||||
t.Errorf("the two declarations ran on different stores: %s / %s", store(runs[0]), store(runs[1]))
|
||||
}
|
||||
// Everything is removed: both containers, and the directory.
|
||||
for _, n := range []string{"mesh-rehearsal-new-abcd", "mesh-rehearsal-old-abcd"} {
|
||||
if !contains(f.calls, "rm -f "+n) {
|
||||
t.Errorf("%s was not removed", n)
|
||||
}
|
||||
}
|
||||
if len(f.removed) != 1 {
|
||||
t.Errorf("the throwaway directory was removed %d times", len(f.removed))
|
||||
}
|
||||
}
|
||||
|
||||
// Nothing the rehearsal runs names the running bus's container, directory or ports.
|
||||
func TestARehearsalNeverTouchesTheRunningBus(t *testing.T) {
|
||||
f := &fake{}
|
||||
r, _ := rehearsal(t, f)
|
||||
if _, err := r.Run(context.Background(), args); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, c := range f.calls {
|
||||
if strings.Contains(c, "mesh-broker-nats") && !strings.HasPrefix(c, "inspect -f {{.Config.Image}} mesh-broker-nats") {
|
||||
t.Errorf("a call names the running bus: %s", c)
|
||||
}
|
||||
for _, bad := range []string{"/var/lib/mesh-broker-nats", "4222:4222", "8222", "exec "} {
|
||||
if strings.Contains(c, bad) {
|
||||
t.Errorf("a call reaches %q of the running bus: %s", bad, c)
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(c, "run -d") && !strings.Contains(c, "-p 127.0.0.1::4222") {
|
||||
t.Errorf("a container publishes more than a random loopback port: %s", c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestARehearsalRefusesWhileABusUpgradeIsOpen(t *testing.T) {
|
||||
f := &fake{}
|
||||
r, _ := rehearsal(t, f)
|
||||
r.UpgradeOpen = func(context.Context) (bool, string, error) { return true, "step 3 running since 12:00", nil }
|
||||
got, err := r.Run(context.Background(), args)
|
||||
if err == nil || !strings.Contains(err.Error(), "a bus upgrade is open") {
|
||||
t.Fatalf("err %v", err)
|
||||
}
|
||||
if got["ok"] != false {
|
||||
t.Errorf("answer %v", got)
|
||||
}
|
||||
if len(f.calls) != 0 {
|
||||
t.Errorf("it ran docker while an upgrade is open: %v", f.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestARehearsalRefusesWhenItCannotTellWhetherAnUpgradeIsOpen(t *testing.T) {
|
||||
f := &fake{}
|
||||
r, _ := rehearsal(t, f)
|
||||
r.UpgradeOpen = func(context.Context) (bool, string, error) { return false, "", errors.New("no answer") }
|
||||
if _, err := r.Run(context.Background(), args); err == nil || len(f.calls) != 0 {
|
||||
t.Fatalf("err %v, calls %v", err, f.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAFailedDeclarationStopsTheRehearsalSaysWhichAndStillRemovesEverything(t *testing.T) {
|
||||
f := &fake{}
|
||||
r, _ := rehearsal(t, f)
|
||||
r.Serves = func(_ context.Context, port, _, _ string, _ []StreamWant) (string, error) {
|
||||
return "", errors.New("KV_a holds 1 of 3 messages")
|
||||
}
|
||||
got, err := r.Run(context.Background(), args)
|
||||
if err == nil || !strings.Contains(err.Error(), "the new declaration serves every stream of the copy") || !strings.Contains(err.Error(), "KV_a holds 1 of 3") {
|
||||
t.Fatalf("err %v", err)
|
||||
}
|
||||
steps := got["steps"].([]Step)
|
||||
if steps[len(steps)-1].OK {
|
||||
t.Error("the last step is not the failed one")
|
||||
}
|
||||
for _, c := range f.calls {
|
||||
if strings.Contains(c, "mesh-rehearsal-old") && strings.HasPrefix(c, "run") {
|
||||
t.Error("the old declaration ran after the new one failed")
|
||||
}
|
||||
}
|
||||
if !contains(f.calls, "rm -f mesh-rehearsal-new-abcd") || len(f.removed) != 1 {
|
||||
t.Errorf("not cleaned up: %v %v", f.calls, f.removed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestABadSnapshotStopsTheRehearsalBeforeAnythingStarts(t *testing.T) {
|
||||
f := &fake{fail: map[string]error{"verify": errors.New("archive does not match its manifest")}}
|
||||
r, _ := rehearsal(t, f)
|
||||
if _, err := r.Run(context.Background(), args); err == nil || !strings.Contains(err.Error(), "the snapshot verifies") {
|
||||
t.Fatalf("err %v", err)
|
||||
}
|
||||
for _, c := range f.calls {
|
||||
if strings.HasPrefix(c, "run -d") {
|
||||
t.Error("a container started on an unverified snapshot")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheOldImageNeedsToBeKnown(t *testing.T) {
|
||||
f := &fake{fail: map[string]error{"inspect": errors.New("no such container")}}
|
||||
r, _ := rehearsal(t, f)
|
||||
if _, err := r.Run(context.Background(), args); err == nil || !strings.Contains(err.Error(), "no old image") {
|
||||
t.Fatalf("err %v", err)
|
||||
}
|
||||
a := args
|
||||
a.OldImage = "nats:given"
|
||||
f2 := &fake{fail: map[string]error{"inspect": errors.New("no such container")}}
|
||||
r2, _ := rehearsal(t, f2)
|
||||
if _, err := r2.Run(context.Background(), a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPathsAreCheckedBeforeAnythingRuns(t *testing.T) {
|
||||
f := &fake{}
|
||||
r, _ := rehearsal(t, f)
|
||||
for _, bad := range []RehearseArgs{{}, {Snapshot: "bus.tar", NewImage: "i", NewProgram: "/p", NewEntrypoint: "/e"}, {Snapshot: "/a:b/bus.tar", NewImage: "i", NewProgram: "/p", NewEntrypoint: "/e"}} {
|
||||
if _, err := r.Run(context.Background(), bad); err == nil {
|
||||
t.Errorf("%+v was accepted", bad)
|
||||
}
|
||||
}
|
||||
if len(f.calls) != 0 {
|
||||
t.Errorf("docker ran: %v", f.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheControllersWordForTheBusIsReadOrRefused(t *testing.T) {
|
||||
for answer, want := range map[string]bool{
|
||||
`the bus is nats, built abc, on novox; never rolled out`: false,
|
||||
"the bus is nats\nlast step 4: a → b on novox by x (y): running since 2026-10-22 12:00; snapshot s": true,
|
||||
"last step 4: a → b on novox by x (y): done at 2026-10-22 12:00; snapshot s": false,
|
||||
} {
|
||||
open, _, err := upgradeRunning(answer)
|
||||
if err != nil || open != want {
|
||||
t.Errorf("%q: open %v, err %v", answer, open, err)
|
||||
}
|
||||
}
|
||||
if _, _, err := upgradeRunning("{}"); err == nil {
|
||||
t.Error("an answer it cannot read was taken for a closed upgrade")
|
||||
}
|
||||
}
|
||||
|
||||
func contains(calls []string, want string) bool {
|
||||
for _, c := range calls {
|
||||
if c == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -44,8 +44,7 @@
|
||||
"nats_backlog",
|
||||
"nats_buckets",
|
||||
"nats_users",
|
||||
"nats_user_can",
|
||||
"nats_rehearse"
|
||||
"nats_user_can"
|
||||
],
|
||||
"guards": [
|
||||
8222
|
||||
@@ -186,10 +185,7 @@
|
||||
"env": {
|
||||
"MESH_NATS_MONITOR": "http://127.0.0.1:8222",
|
||||
"MESH_NATS_CONTAINER": "mesh-broker-nats",
|
||||
"MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf",
|
||||
"MESH_NATS_SNAPSHOT": "${dir:snapshots}/bus.tar",
|
||||
"MESH_NATS_PROGRAM_DIR": "/var/lib/nats-module/snapshot",
|
||||
"MESH_NATS_ENTRYPOINT_DIR": "/var/lib/nats-module/entrypoint"
|
||||
"MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf"
|
||||
}
|
||||
},
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user