Merge pull request 'The bus runs its upstream image with the mesh's programs mounted (issues 1640, 1641, 1642, ADR 0480 §1-§3, §5, §6; ADR 0497 §5; ADR 0506)' (#434) from feat/r4d into main
This commit was merged in pull request #434.
This commit is contained in:
@@ -196,6 +196,19 @@ for m in $touched; do
|
||||
# another process) run here and nowhere else: the unit suite stays under a second (novox/hq playbook 07).
|
||||
(cd "modules/$m" && CGO_ENABLED=$cgo GOPRIVATE=git.novox.be go vet -tags integration ${modfile:+"-modfile=$modfile"} ./... &&
|
||||
CGO_ENABLED=$cgo GOPRIVATE=git.novox.be go test -tags integration ${modfile:+"-modfile=$modfile"} $race -count=1 ./...)
|
||||
# The docker tests of the bus's snapshot program and entrypoint (novox/hq ADR 0480 clause 6) start throwaway nats
|
||||
# containers, and the Go toolchain container holds no docker, so above they skip with a "NOT RUN" line. Where the
|
||||
# build seat says docker exists (MESH_CHECK_DOCKER=1) they run here by name, and a missing docker fails the check
|
||||
# (MESH_REQUIRE_DOCKER=1 makes the tests fail instead of skip). Without it the step says it did not run.
|
||||
if [ "$m" = nats ]; then
|
||||
if [ "${MESH_CHECK_DOCKER:-}" = 1 ]; then
|
||||
(cd modules/nats && MESH_REQUIRE_DOCKER=1 CGO_ENABLED=$cgo GOPRIVATE=git.novox.be go test -tags integration ${modfile:+"-modfile=$modfile"} -count=1 \
|
||||
-run 'TestTheDeclaredDumpRunsInTheBussImage|TestTheEntrypointReloadsTheServerWithoutDroppingAConnection|TestTheStoreSurvivesTheNewDeclarationTheOldAndTheNewAgain|TestASnapshotOfALiveBusRestoresToTheSameContent|TestTheSnapshotUserCannotChangeTheBus' \
|
||||
./cmd/mesh-nats-snapshot) || { echo "the bus's docker tests failed"; exit 1; }
|
||||
else
|
||||
echo "NOT RUN: the bus's five docker tests (modules/nats/cmd/mesh-nats-snapshot) skipped: this check has no docker (MESH_CHECK_DOCKER is not 1)"
|
||||
fi
|
||||
fi
|
||||
done
|
||||
if [ -n "$untestable" ]; then
|
||||
echo "the check fails: the tests of$untestable could not run (novox/hq issue 385)"
|
||||
|
||||
@@ -107,8 +107,9 @@ func TestAutomxIsAGoBundleForAlpineMountedAtOptMeshAndRestartedWithIt(t *testing
|
||||
}
|
||||
}
|
||||
|
||||
// Clause 3: an upstream artifact pinned by digest, the one route-proxy's trust artifact runs; never an image
|
||||
// built from a Dockerfile (clause 11), and no Dockerfile or base to build one on.
|
||||
// Clause 3: an upstream artifact pinned by digest, the one route-proxy's trust artifact runs. automx's own design
|
||||
// builds no image from a Dockerfile and has no Dockerfile or base to build one on; the catalogue as such does not
|
||||
// refuse the image kind (novox/hq ADR 0487 clause 13).
|
||||
func TestAutomxRunsTheMirroredAlpineImageAndBuildsNoImage(t *testing.T) {
|
||||
m := readManifest(t)
|
||||
c := m.resource("automx")
|
||||
@@ -121,7 +122,7 @@ func TestAutomxRunsTheMirroredAlpineImageAndBuildsNoImage(t *testing.T) {
|
||||
}
|
||||
for _, a := range m.Build["artifacts"].([]any) {
|
||||
if a.(map[string]any)["kind"] == "image" {
|
||||
t.Errorf("artifact %v is an image built from a Dockerfile (ADR 0452)", a)
|
||||
t.Errorf("artifact %v is an image built from a Dockerfile, which automx's design (ADR 0480) does not use", a)
|
||||
}
|
||||
}
|
||||
if _, has := m.Build["on"]; has {
|
||||
|
||||
@@ -1,47 +0,0 @@
|
||||
# nats's server image: the upstream server, plus an entrypoint that reloads it in place when the
|
||||
# mesh rewrites its configuration. See entrypoint.sh for why that belongs here and not in the host.
|
||||
#
|
||||
# **Pinned to the multi-architecture index digest, not a platform's.** `docker manifest inspect`
|
||||
# reports a platform manifest per architecture and the index that lists them; pinning a platform's
|
||||
# digest builds on this workstation and fails on any node of another architecture, with an error
|
||||
# that names a manifest rather than the mistake. This is the index — `docker pull` reports the same
|
||||
# one, and `RepoDigests` confirms it.
|
||||
#
|
||||
# **The release the digest is, said here because a digest does not say it:**
|
||||
#
|
||||
# upstream: nats 2.11.17-alpine
|
||||
#
|
||||
# Kept equal to the server version cmd/nats-tools tests against (its go.mod), and a test there fails
|
||||
# when they differ: that test is what says the server delivers every message to a consumer with
|
||||
# several filters. 2.10.29 did not — it moved such a consumer past a message now and then without
|
||||
# handing it over, and the controller never heard of a merge (novox/hq issue 266).
|
||||
#
|
||||
# Unlike every other module's Dockerfile, this builds no TypeScript and uses no mesh base image:
|
||||
# the module's code is the server, which upstream already built. There is no BUILD_BASE here on
|
||||
# purpose — the server is not compiled; only the snapshot program below is. The upstream image is
|
||||
# declared in the manifest under build.on and arrives as NATS_BASE, like every other base the mesh
|
||||
# copies into its own store before a build (novox/hq ADR 0097); the digest above is the index one
|
||||
# for the reason given. So does GO_BASE, the toolchain the snapshot program is built with.
|
||||
#
|
||||
# **One thing is compiled: the bus's snapshot program** (novox/hq ADR 0235). The machine's backup
|
||||
# holder runs the module's declared dump — `docker exec` into this container — and the program asks
|
||||
# the server for each stream through the snapshot API, writing a tar on stdout. It is here, beside
|
||||
# the server, for two reasons: the dump runs where the server is without mounting anything into it,
|
||||
# and a restore needs nats-server itself — the very binary this image already carries — to fill a new
|
||||
# store. Its own Go module (snapshot/go.mod), so this build fetches only public modules.
|
||||
ARG NATS_BASE
|
||||
ARG GO_BASE
|
||||
FROM ${GO_BASE} AS snapshot
|
||||
WORKDIR /src
|
||||
COPY snapshot/go.mod snapshot/go.sum ./
|
||||
RUN go mod download
|
||||
COPY snapshot/*.go ./
|
||||
RUN CGO_ENABLED=0 go build -trimpath -ldflags '-s -w' -o /mesh-nats-snapshot .
|
||||
|
||||
FROM ${NATS_BASE}
|
||||
|
||||
COPY entrypoint.sh /usr/local/bin/mesh-nats-entrypoint
|
||||
RUN chmod 0755 /usr/local/bin/mesh-nats-entrypoint
|
||||
COPY --from=snapshot /mesh-nats-snapshot /usr/local/bin/mesh-nats-snapshot
|
||||
|
||||
ENTRYPOINT ["/usr/local/bin/mesh-nats-entrypoint"]
|
||||
+49
-12
@@ -2,7 +2,12 @@
|
||||
|
||||
The bus server (novox/hq design 25), the configuration the module owns (`nats.conf`: ports, TLS,
|
||||
JetStream's store) and the user list the controller composes (`accounts.conf`), reloaded in place by
|
||||
the entrypoint. Its tools (`cmd/nats-tools`) are served by the node's runtime and only read.
|
||||
the entrypoint. The server is the upstream nats image, pinned by digest in the manifest's `server`
|
||||
artifact (`nats:<release>-alpine@sha256:<digest>`: the digest names the image, the tag says the release,
|
||||
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.
|
||||
|
||||
## Its data, and how it is backed up
|
||||
|
||||
@@ -14,9 +19,9 @@ events, every module's state.
|
||||
while the server writes them may not restore. The `jetstream` data item is protected by a dump
|
||||
instead (novox/hq ADR 0235, to-be 43): each night the machine's backup holder runs
|
||||
|
||||
docker exec -i mesh-broker-nats mesh-nats-snapshot snapshot < <the module's broker credential> > <snapshots>/bus.tar
|
||||
docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot snapshot < <the module's broker credential> > <snapshots>/bus.tar
|
||||
|
||||
and backs up the `snapshots` directory. `mesh-nats-snapshot` (`snapshot/`, built into this image)
|
||||
and backs up the `snapshots` directory. `mesh-nats-snapshot` (`cmd/mesh-nats-snapshot`, mounted into the server's container)
|
||||
asks the server for each stream through JetStream's snapshot API, one at a time, acknowledging each
|
||||
chunk so the server's flow control keeps moving; the server goes on taking writes and no stream is
|
||||
paused or reconfigured. It writes one tar on stdout:
|
||||
@@ -47,14 +52,15 @@ live one**, which a person swaps in.
|
||||
1. **Take the night back, beside the live data.** `node-backup.restore` for `nats` on the bus's
|
||||
machine (optionally naming a restore point) restores the snapshots directory as
|
||||
`<snapshots>.restored-<stamp>/`, holding that night's `bus.tar`.
|
||||
2. **Check it.** `docker exec -i mesh-broker-nats mesh-nats-snapshot verify < <restored>/bus.tar` —
|
||||
2. **Check it.** `docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot verify < <restored>/bus.tar` —
|
||||
every file against the manifest's checksum, every archive read to its end.
|
||||
3. **Build the new store**, with the bus's own image (the server that fills it is the same release
|
||||
the bus runs):
|
||||
3. **Build the new store**, with the bus's own image and the mounted program (the server that fills
|
||||
it is the same release the bus runs):
|
||||
|
||||
image=$(docker inspect -f '{{.Config.Image}}' mesh-broker-nats)
|
||||
docker run --rm -v <restored>:/in:ro -v <new-store>:/out \
|
||||
--entrypoint mesh-nats-snapshot "$image" restore --into /out --from /in/bus.tar
|
||||
-v <the module's state directory>/snapshot:/opt/mesh:ro \
|
||||
--entrypoint /opt/mesh/mesh-nats-snapshot "$image" restore --into /out --from /in/bus.tar
|
||||
|
||||
It starts a server of its own on the container's loopback with the bus's one account (`MESH`),
|
||||
restores every stream, holds each to the manifest, stops it, and leaves
|
||||
@@ -75,16 +81,45 @@ A single stream can be restored into any server that does not hold it with
|
||||
`mesh-nats-snapshot restore --server <url> --credential <file>`; the archive is also what the nats CLI
|
||||
reads (`nats stream restore <dir>` on an unpacked `streams/<name>/`).
|
||||
|
||||
## Upgrading the bus, and going back
|
||||
|
||||
The bus is replaced only as a planned step a person starts (novox/hq ADR 0236, ADR 0480 clause 6): the
|
||||
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. **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`.
|
||||
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.
|
||||
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.
|
||||
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.
|
||||
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
|
||||
|
||||
- `snapshot/`: `MESH_TEST_DOCKER=1 go test -race ./...` — a throwaway `nats:2.11` filled like the
|
||||
- `cmd/mesh-nats-snapshot`: `go test -tags integration -race ./...`, run by `mesh-controller.try` (they start throwaway containers; nobody runs them by hand) — a throwaway `nats:2.11` filled like the
|
||||
mesh's bus (events with a gap deleted, a work queue part worked, buckets with history, deletes and a
|
||||
purge, an object, durable consumers, a memory stream) and written to throughout the snapshot; the
|
||||
snapshot restored into a fresh server and into a new store served by a third, each compared message
|
||||
by message; a damaged archive refused; and the snapshot user refused every write.
|
||||
`MESH_TEST_NATS_IMAGE=<this image>` also runs the manifest's declared dump, exactly, against the
|
||||
module's image and configuration.
|
||||
- `cmd/nats-tools`: `go test ./...`.
|
||||
by message; a damaged archive refused; and the snapshot user refused every write. The same run starts
|
||||
the manifest's pinned upstream image with the program and the entrypoint mounted as the manifest
|
||||
declares them: the declared dump, exactly, writes a `bus.tar` that verifies; a change of the user list
|
||||
reloads the server and a connection open meanwhile survives; and the new declaration, the old one and
|
||||
the new again run on one store.
|
||||
- `cmd/nats-tools`: `go test ./...` — the tools, the release test (the manifest's tag equals the server
|
||||
tested) and the manifest's shape (programs mounted read-only, named in `restart-on`, no image built from
|
||||
a Dockerfile).
|
||||
|
||||
## Files it does not know
|
||||
|
||||
@@ -94,3 +129,5 @@ Each directory or archive this module declares says how it treats a file it did
|
||||
- `/var/lib/nats-module/conf`: everything in it is the program's; the mesh moves nothing out (ADR 0422 §2).
|
||||
- `${dir:snapshots}`: everything in it is the program's; the mesh moves nothing out (ADR 0422 §2).
|
||||
- `${dir:mesh-state}`: everything in it is the program's; the mesh moves nothing out (ADR 0422 §2).
|
||||
- `${dir:state}/snapshot`: the unpacked snapshot program; everything in it is the program's; the mesh moves nothing out (ADR 0422 §2).
|
||||
- `${dir:state}/entrypoint`: the unpacked entrypoint; everything in it is the program's; the mesh moves nothing out (ADR 0422 §2).
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
// The declared dump and the mounted entrypoint, exactly as the manifest declares them, against the upstream image
|
||||
// the manifest pins: the mesh's snapshot program and entrypoint mounted at /opt (novox/hq ADR 0480 clauses 1, 2, 5).
|
||||
//
|
||||
// go test -tags integration -run 'TestTheDeclaredDump|TestTheEntrypoint|TestTheStoreSurvives' . (through mesh-controller.try)
|
||||
//
|
||||
// The image is pulled by the manifest's reference, so the server these run is the one the bus runs.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// manifestOf is the nats module's own definition, beside this program.
|
||||
func manifestOf(t *testing.T) map[string]any {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func declaredDump(t *testing.T, m map[string]any) string {
|
||||
t.Helper()
|
||||
for _, item := range m["data"].(map[string]any)["own"].([]any) {
|
||||
it := item.(map[string]any)
|
||||
if it["id"] != "jetstream" {
|
||||
continue
|
||||
}
|
||||
backup, _ := it["backup"].(map[string]any)
|
||||
if backup == nil || backup["into"] != "snapshots" {
|
||||
t.Fatalf("the bus's streams are not protected by a dump into its snapshots: %v", it["backup"])
|
||||
}
|
||||
return backup["dump"].(string)
|
||||
}
|
||||
t.Fatal("the nats module declares no jetstream data item")
|
||||
return ""
|
||||
}
|
||||
|
||||
// serverConf is the module's own server settings, from the template its resource names.
|
||||
func serverConf(t *testing.T, m map[string]any) string {
|
||||
t.Helper()
|
||||
for _, r := range m["resources"].([]any) {
|
||||
res := r.(map[string]any)
|
||||
if res["id"] == "server-conf" {
|
||||
return templateText(t, res)
|
||||
}
|
||||
}
|
||||
t.Fatal("the nats module declares no server configuration")
|
||||
return ""
|
||||
}
|
||||
|
||||
// busImage is the upstream image the manifest pins, as a reference docker pulls (name:tag@digest).
|
||||
func busImage(t *testing.T, m map[string]any) string {
|
||||
t.Helper()
|
||||
for _, a := range m["build"].(map[string]any)["artifacts"].([]any) {
|
||||
art := a.(map[string]any)
|
||||
if art["name"] == "server" {
|
||||
if art["kind"] != "upstream" {
|
||||
t.Fatalf("the server artifact is a %v, not an upstream image", art["kind"])
|
||||
}
|
||||
return art["from"].(string)
|
||||
}
|
||||
}
|
||||
t.Fatal("the nats module declares no server artifact")
|
||||
return ""
|
||||
}
|
||||
|
||||
// programs builds the snapshot program for alpine (static, as the mesh's build does) and lays out the two
|
||||
// directories the manifest mounts: the program at /opt/mesh and the entrypoint at /opt/mesh-entrypoint.
|
||||
func programs(t *testing.T) (program, entry string) {
|
||||
t.Helper()
|
||||
program, entry = t.TempDir(), t.TempDir()
|
||||
build := exec.Command("go", "build", "-trimpath", "-o", filepath.Join(program, "mesh-nats-snapshot"), ".")
|
||||
build.Env = append(os.Environ(), "CGO_ENABLED=0", "GOOS=linux")
|
||||
if out, err := build.CombinedOutput(); err != nil {
|
||||
t.Fatalf("building the program: %v\n%s", err, out)
|
||||
}
|
||||
script, err := os.ReadFile(filepath.Join(moduleDir, "program", "entrypoint.sh"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Mode 0644: the manifest runs the script through sh, so no execute bit is relied on.
|
||||
if err := os.WriteFile(filepath.Join(entry, "entrypoint.sh"), script, 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return program, entry
|
||||
}
|
||||
|
||||
// runBus starts the upstream image the way the manifest does: its volumes, its args, the module's configuration
|
||||
// and the given user list. The poll is a second so a test need not wait out the production five.
|
||||
func runBus(t *testing.T, m map[string]any, name, accounts string) (port, pin, conf string) {
|
||||
t.Helper()
|
||||
if out, _ := exec.Command("docker", "ps", "-a", "-q", "--filter", "name=^"+name+"$").Output(); len(strings.TrimSpace(string(out))) > 0 {
|
||||
t.Skipf("a container called %s already runs here; this test will not touch it", name)
|
||||
}
|
||||
conf, tlsDir, data := t.TempDir(), t.TempDir(), t.TempDir()
|
||||
pin = selfSigned(t, tlsDir)
|
||||
_ = os.WriteFile(filepath.Join(conf, "nats.conf"), []byte(serverConf(t, m)), 0o644)
|
||||
_ = os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(accounts), 0o644)
|
||||
program, entry := programs(t)
|
||||
docker(t, "run", "-d", "--rm", "--name", name, "--user", owner(), "-p", "127.0.0.1::4222",
|
||||
"-e", "MESH_NATS_CONF_POLL_SECONDS=1",
|
||||
"-v", data+":/data", "-v", conf+":/etc/nats:ro", "-v", tlsDir+":/tls:ro",
|
||||
"-v", program+":/opt/mesh:ro", "-v", entry+":/opt/mesh-entrypoint:ro",
|
||||
busImage(t, m), "sh", "/opt/mesh-entrypoint/entrypoint.sh")
|
||||
t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", name).Run() })
|
||||
_, port, _ = strings.Cut(docker(t, "port", name, "4222/tcp"), ":")
|
||||
return port, pin, conf
|
||||
}
|
||||
|
||||
// The dump is the one the manifest declares, run by `sh -c` as the holder runs it, with the module's
|
||||
// directories filled in; the server is the module's image with the module's own configuration and a
|
||||
// user list holding the bus's own module with the controller's grants for it.
|
||||
func TestTheDeclaredDumpRunsInTheBussImage(t *testing.T) {
|
||||
needDocker(t)
|
||||
const name = "mesh-broker-nats" // the container the dump names
|
||||
m := manifestOf(t)
|
||||
dump := declaredDump(t, m)
|
||||
|
||||
state, snapshots := t.TempDir(), t.TempDir()
|
||||
accounts := liveBusConf()
|
||||
accounts = accounts[strings.Index(accounts, "accounts {"):]
|
||||
port, pin, _ := runBus(t, m, name, accounts)
|
||||
|
||||
admin := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
|
||||
js, _ := admin.JetStream()
|
||||
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_hand-acts"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, k := range []string{"a", "b", "c"} {
|
||||
if _, err := kv.Put(k, []byte("act "+k)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
run := func(cred Credential) (string, error) {
|
||||
raw, _ := json.Marshal(cred)
|
||||
if err := os.WriteFile(filepath.Join(state, "broker"), raw, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
command := strings.NewReplacer("${dir:mesh-state}", state, "${dir:snapshots}", snapshots).Replace(dump)
|
||||
out, err := exec.Command("sh", "-c", command).CombinedOutput()
|
||||
return string(out), err
|
||||
}
|
||||
|
||||
began := time.Now()
|
||||
out, err := run(Credential{URL: "nats://bus.example:4222", Fingerprint: pin, User: snapshotUser, Password: "snap"})
|
||||
if err != nil {
|
||||
t.Fatalf("the declared dump failed: %v\n%s", err, out)
|
||||
}
|
||||
t.Logf("the declared dump took %s:\n%s", time.Since(began).Round(time.Millisecond), out)
|
||||
tarPath := filepath.Join(snapshots, "bus.tar")
|
||||
f, err := os.Open(tarPath)
|
||||
if err != nil {
|
||||
t.Fatalf("the dump left no bus.tar: %v", err)
|
||||
}
|
||||
a, err := ReadArchive(f, "")
|
||||
f.Close()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer a.Close()
|
||||
if err := a.Verify(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(a.Manifest.Streams) != 1 || a.Manifest.Streams[0].Name != "KV_mesh-controller_hand-acts" || a.Manifest.Streams[0].Messages != 3 {
|
||||
t.Fatalf("the snapshot holds %+v", a.Manifest.Streams)
|
||||
}
|
||||
before, _ := os.ReadFile(tarPath)
|
||||
|
||||
// A night that cannot snapshot fails, says why, and leaves last night's snapshot as it was.
|
||||
out, err = run(Credential{Fingerprint: pin, User: snapshotUser, Password: "wrong"})
|
||||
if err == nil {
|
||||
t.Fatalf("a dump the bus refused succeeded:\n%s", out)
|
||||
}
|
||||
if !strings.Contains(out, "Authorization Violation") && !strings.Contains(strings.ToLower(out), "authorization") {
|
||||
t.Errorf("the failure does not say the bus refused the credential: %s", out)
|
||||
}
|
||||
after, _ := os.ReadFile(tarPath)
|
||||
if string(after) != string(before) {
|
||||
t.Error("a failed night replaced the last good snapshot")
|
||||
}
|
||||
if _, err := os.Stat(tarPath + ".partial"); !os.IsNotExist(err) {
|
||||
t.Error("a failed night left its partial file behind")
|
||||
}
|
||||
}
|
||||
|
||||
// **A change of the user list reloads the server and keeps its connections** (ADR 0480 clause 5, ADR 0236 §37).
|
||||
// The mounted entrypoint watches accounts.conf and signals the server; a client connected meanwhile is not
|
||||
// dropped, and a user added by the change can then connect.
|
||||
func TestTheEntrypointReloadsTheServerWithoutDroppingAConnection(t *testing.T) {
|
||||
needDocker(t)
|
||||
m := manifestOf(t)
|
||||
accounts := plainConf()
|
||||
accounts = accounts[strings.Index(accounts, "accounts {"):]
|
||||
port, pin, conf := runBus(t, m, fmt.Sprintf("mesh-reload-test-%d", os.Getpid()), accounts)
|
||||
|
||||
admin := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
|
||||
if err := admin.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
changed := strings.Replace(accounts, `{ user: "admin", password: "admin" }`,
|
||||
`{ user: "admin", password: "admin" }, { user: "late", password: "late" }`, 1)
|
||||
if changed == accounts {
|
||||
t.Fatal("the test's user list did not change")
|
||||
}
|
||||
// Written as the mesh writes it: replaced, not edited in place, so the watcher sees whole content.
|
||||
tmp := filepath.Join(conf, "accounts.conf.new")
|
||||
if err := os.WriteFile(tmp, []byte(changed), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Rename(tmp, filepath.Join(conf, "accounts.conf")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// dial retries until the server takes the user: the reload happens within the poll interval.
|
||||
late := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "late", Password: "late"}, nil)
|
||||
_ = late
|
||||
if !admin.IsConnected() {
|
||||
t.Fatal("the connection open before the reload was dropped by it")
|
||||
}
|
||||
if err := admin.Flush(); err != nil {
|
||||
t.Fatalf("the connection open before the reload no longer works: %v", err)
|
||||
}
|
||||
if n := admin.Stats().Reconnects; n != 0 {
|
||||
t.Fatalf("the connection open before the reload reconnected %d times", n)
|
||||
}
|
||||
}
|
||||
|
||||
// **The bus can go back** (ADR 0480 clause 6): the new declaration and then the old one run on the same store, both
|
||||
// well, which is what makes `reversible` true for the bus upgrade. The server binary (the upstream image's digest),
|
||||
// the data directory and the configuration are the same in both; only where the entrypoint and the program come
|
||||
// from differs. The old declaration is run as the image it replaced ran it: the same script as the image's own
|
||||
// entrypoint, the program on the image's path. Then the new one again, so a way there, back and there again is
|
||||
// shown, each time with the stream's messages as the one before left them.
|
||||
func TestTheStoreSurvivesTheNewDeclarationTheOldAndTheNewAgain(t *testing.T) {
|
||||
needDocker(t)
|
||||
m := manifestOf(t)
|
||||
accounts := liveBusConf()
|
||||
accounts = accounts[strings.Index(accounts, "accounts {"):]
|
||||
conf, tlsDir, data := t.TempDir(), t.TempDir(), t.TempDir()
|
||||
pin := selfSigned(t, tlsDir)
|
||||
_ = os.WriteFile(filepath.Join(conf, "nats.conf"), []byte(serverConf(t, m)), 0o644)
|
||||
_ = os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(accounts), 0o644)
|
||||
program, entry := programs(t)
|
||||
script := filepath.Join(entry, "entrypoint.sh")
|
||||
if err := os.Chmod(script, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
common := []string{"-d", "--rm", "--user", owner(), "-p", "127.0.0.1::4222", "-e", "MESH_NATS_CONF_POLL_SECONDS=1",
|
||||
"-v", data + ":/data", "-v", conf + ":/etc/nats:ro", "-v", tlsDir + ":/tls:ro"}
|
||||
declarations := map[string][]string{
|
||||
"new": {"-v", program + ":/opt/mesh:ro", "-v", entry + ":/opt/mesh-entrypoint:ro", busImage(t, m), "sh", "/opt/mesh-entrypoint/entrypoint.sh"},
|
||||
"old": {"-v", script + ":/usr/local/bin/mesh-nats-entrypoint:ro",
|
||||
"-v", filepath.Join(program, "mesh-nats-snapshot") + ":/usr/local/bin/mesh-nats-snapshot:ro",
|
||||
"--entrypoint", "/usr/local/bin/mesh-nats-entrypoint", busImage(t, m)},
|
||||
}
|
||||
var kv nats.KeyValue
|
||||
puts := 0
|
||||
step := func(which string) {
|
||||
name := fmt.Sprintf("mesh-rehearsal-%s-%d", which, os.Getpid())
|
||||
docker(t, append(append([]string{"run", "--name", name}, common...), declarations[which]...)...)
|
||||
t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", name).Run() })
|
||||
_, port, _ := strings.Cut(docker(t, "port", name, "4222/tcp"), ":")
|
||||
c := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
|
||||
js, _ := c.JetStream()
|
||||
var err error
|
||||
if puts == 0 {
|
||||
kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-rehearsal"})
|
||||
} else {
|
||||
kv, err = js.KeyValue("mesh-rehearsal")
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("%s declaration: the store's bucket: %v", which, err)
|
||||
}
|
||||
for i := 1; i <= puts; i++ {
|
||||
e, err := kv.Get(fmt.Sprintf("k%d", i))
|
||||
if err != nil || string(e.Value()) != fmt.Sprintf("v%d", i) {
|
||||
t.Fatalf("%s declaration: key k%d reads %v, %v", which, i, e, err)
|
||||
}
|
||||
}
|
||||
puts++
|
||||
if _, err := kv.Put(fmt.Sprintf("k%d", puts), []byte(fmt.Sprintf("v%d", puts))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.Close()
|
||||
docker(t, "stop", "-t", "10", name)
|
||||
}
|
||||
step("new")
|
||||
step("old")
|
||||
step("new")
|
||||
}
|
||||
+12
-4
@@ -1,8 +1,10 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
// The snapshot and the way back, against real servers: the bus's own release in throwaway containers.
|
||||
//
|
||||
// MESH_TEST_DOCKER=1 go test -race ./...
|
||||
// go test -tags integration -race ./... (run through mesh-controller.try, never by hand on a workstation)
|
||||
//
|
||||
// Every container and file it makes is removed when it ends. MESH_TEST_SNAPSHOT_MESSAGES sets how
|
||||
// many events the source holds (default 6000), to time a snapshot at the live bus's size.
|
||||
@@ -35,7 +37,7 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// image is the bus's release (the Dockerfile's `# upstream:` line names the same one).
|
||||
// image is the bus's release (the manifest's server artifact names the same release in its tag).
|
||||
const image = "nats:2.11-alpine"
|
||||
|
||||
// snapshotUser is the bus's own module on a machine, as the controller names it.
|
||||
@@ -57,8 +59,14 @@ var snapshotGrants = struct{ publish, subscribe []string }{
|
||||
|
||||
func needDocker(t *testing.T) {
|
||||
t.Helper()
|
||||
if os.Getenv("MESH_TEST_DOCKER") != "1" {
|
||||
t.Skip("MESH_TEST_DOCKER is not 1: these tests start throwaway nats containers")
|
||||
if _, err := exec.LookPath("docker"); err != nil {
|
||||
// The Go toolchain container of the merge check holds no docker. Where docker is expected
|
||||
// (MESH_REQUIRE_DOCKER=1, which merge-check.sh sets for its docker step) its absence is a failure.
|
||||
if os.Getenv("MESH_REQUIRE_DOCKER") == "1" {
|
||||
t.Fatalf("MESH_REQUIRE_DOCKER=1 and there is no docker here: %v", err)
|
||||
}
|
||||
t.Skipf("NOT RUN: this test starts a throwaway nats container and there is no docker in this toolchain (%v); "+
|
||||
"it is proof only where merge-check.sh's docker step runs it (MESH_CHECK_DOCKER=1)", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,9 +9,9 @@
|
||||
// transfer with flow control that the server answers from a consistent view of the stream while
|
||||
// it goes on taking messages. This program asks for it, one stream at a time, and nothing else.
|
||||
//
|
||||
// It lives in the bus's image because that is where it runs: the machine's backup holder runs the
|
||||
// nats module's declared dump — `docker exec` into the server's container — and the archive comes
|
||||
// out on stdout, so nothing is mounted into the server for it and the credential arrives on stdin.
|
||||
// It is mounted into the bus's container (at /opt/mesh) because that is where it runs: the machine's
|
||||
// backup holder runs the nats module's declared dump — `docker exec` into the server's container — and
|
||||
// the archive comes out on stdout, so the credential arrives on stdin.
|
||||
//
|
||||
// mesh-nats-snapshot snapshot [--server URL] [--credential FILE|-] > bus.tar
|
||||
// mesh-nats-snapshot verify [--from FILE]
|
||||
+1
-1
@@ -8,7 +8,7 @@ import (
|
||||
)
|
||||
|
||||
// The module's directory, where its manifest and its files/ are.
|
||||
const moduleDir = ".."
|
||||
const moduleDir = "../.."
|
||||
|
||||
// templateText is the text a file resource places on the machine: its template's, in the module's files/
|
||||
// (novox/hq ADR 0349 §2). A resource that names no template, or a template that is not there, fails the test.
|
||||
@@ -0,0 +1,115 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The shape ADR 0480 clauses 1, 2, 3 and 5 give the bus's manifest: the upstream image by digest, the mesh's two
|
||||
// programs mounted read-only under /opt, unpacked by archive resources the container names under restart-on
|
||||
// (the node-engine does not look inside a mounted directory), and the container the same one it was.
|
||||
func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m struct {
|
||||
Data struct {
|
||||
Own []struct {
|
||||
ID string `json:"id"`
|
||||
Backup struct {
|
||||
Dump string `json:"dump"`
|
||||
} `json:"backup"`
|
||||
} `json:"own"`
|
||||
} `json:"data"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
Build struct {
|
||||
On any `json:"on"`
|
||||
Artifacts []struct {
|
||||
Name, Kind, From, Language, System, Binary string
|
||||
} `json:"artifacts"`
|
||||
} `json:"build"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m.Build.On != nil {
|
||||
t.Error("build.on is declared: nothing is built on a base image any more")
|
||||
}
|
||||
artifacts := map[string]struct{ Name, Kind, From, Language, System, Binary string }{}
|
||||
for _, a := range m.Build.Artifacts {
|
||||
artifacts[a.Name] = a
|
||||
if a.Kind == "image" {
|
||||
t.Errorf("artifact %s is an image built from a Dockerfile (ADR 0452)", a.Name)
|
||||
}
|
||||
}
|
||||
if a := artifacts["server"]; a.Kind != "upstream" || !strings.Contains(a.From, "@sha256:") {
|
||||
t.Errorf("server is %+v: an upstream image pinned by digest", a)
|
||||
}
|
||||
if a := artifacts["snapshot"]; a.Kind != "bundle" || a.Language != "go" || a.System != "alpine" ||
|
||||
a.From != "cmd/mesh-nats-snapshot" || a.Binary != "mesh-nats-snapshot" {
|
||||
t.Errorf("snapshot is %+v: a Go bundle of the module's own package, stamped for alpine", a)
|
||||
}
|
||||
if a := artifacts["entrypoint"]; a.Kind != "archive" || a.From != "program" {
|
||||
t.Errorf("entrypoint is %+v: an archive of program/", a)
|
||||
}
|
||||
if _, err := os.Stat("../../program/entrypoint.sh"); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if _, err := os.Stat("../../Dockerfile"); err == nil {
|
||||
t.Error("the Dockerfile is still there")
|
||||
}
|
||||
|
||||
byID := map[string]map[string]any{}
|
||||
for _, r := range m.Resources {
|
||||
byID[r["id"].(string)] = r
|
||||
}
|
||||
server := byID["server"]
|
||||
if server["name"] != "mesh-broker-nats" || server["artifact"] != "server" {
|
||||
t.Errorf("the container is %v on artifact %v: it keeps its name, which the dump and the jails use", server["name"], server["artifact"])
|
||||
}
|
||||
volumes, _ := server["volumes"].([]any)
|
||||
restart, _ := server["restart-on"].([]any)
|
||||
for _, p := range []struct{ resource, artifact, mount string }{
|
||||
{"program-snapshot", "snapshot", ":/opt/mesh:ro"},
|
||||
{"program-entrypoint", "entrypoint", ":/opt/mesh-entrypoint:ro"},
|
||||
} {
|
||||
r := byID[p.resource]
|
||||
if r == nil || r["type"] != "archive" || r["artifact"] != p.artifact {
|
||||
t.Errorf("%s is %v: an archive resource unpacking %s", p.resource, r, p.artifact)
|
||||
continue
|
||||
}
|
||||
// Under a directory the module declares (ADR 0030: nothing is mounted from a path no resource covers).
|
||||
if dir, _, _ := strings.Cut(strings.TrimPrefix(r["path"].(string), "${dir:"), "}"); !strings.HasPrefix(r["path"].(string), "${dir:") ||
|
||||
byID[dir] == nil || byID[dir]["type"] != "directory" {
|
||||
t.Errorf("%s is at %v: not under a directory the module declares", p.resource, r["path"])
|
||||
}
|
||||
mounted := false
|
||||
for _, v := range volumes {
|
||||
if v == r["path"].(string)+p.mount {
|
||||
mounted = true
|
||||
}
|
||||
}
|
||||
if !mounted {
|
||||
t.Errorf("%s is not mounted read-only as %s", p.resource, p.mount)
|
||||
}
|
||||
named := false
|
||||
for _, v := range restart {
|
||||
named = named || v == p.resource
|
||||
}
|
||||
if !named {
|
||||
t.Errorf("%s is not in the container's restart-on: a new program would leave the old one running", p.resource)
|
||||
}
|
||||
}
|
||||
args, _ := server["args"].([]any)
|
||||
if len(args) != 2 || args[0] != "sh" || args[1] != "/opt/mesh-entrypoint/entrypoint.sh" {
|
||||
t.Errorf("args are %v: the entrypoint run through sh, so no execute bit is relied on", args)
|
||||
}
|
||||
for _, d := range m.Data.Own {
|
||||
if d.ID == "jetstream" && !strings.Contains(d.Backup.Dump, "docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot snapshot") {
|
||||
t.Errorf("the dump is %q: it calls the mounted program", d.Backup.Dump)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand"
|
||||
@@ -15,21 +16,43 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// The server these tests run is the server the image runs. The image is pinned by a digest, which
|
||||
// names no release, so the Dockerfile says which release it is, and this keeps the two equal: a
|
||||
// test of the server below is only worth something about the server the mesh actually runs.
|
||||
// The server these tests run is the server the mesh runs. The manifest's `server` artifact is the upstream
|
||||
// image, written `name:tag@sha256:<digest>` (novox/hq ADR 0480 clause 3): the digest names the image and the
|
||||
// tag says the release it is. This keeps the tag and the server compiled in here equal: a test of the server
|
||||
// below is only worth something about the server the mesh actually runs.
|
||||
func TestTheImageIsTheServerTestedHere(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../Dockerfile")
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
said := regexp.MustCompile(`(?m)^# upstream: nats (\d+\.\d+\.\d+)-alpine$`).FindSubmatch(raw)
|
||||
var m struct {
|
||||
Build struct {
|
||||
Artifacts []struct {
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
From string `json:"from"`
|
||||
} `json:"artifacts"`
|
||||
} `json:"build"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
from := ""
|
||||
for _, a := range m.Build.Artifacts {
|
||||
if a.Name == "server" {
|
||||
if a.Kind != "upstream" {
|
||||
t.Fatalf("the server artifact is a %q, not an upstream image", a.Kind)
|
||||
}
|
||||
from = a.From
|
||||
}
|
||||
}
|
||||
said := regexp.MustCompile(`^nats:(\d+\.\d+\.\d+)-alpine@sha256:[0-9a-f]{64}$`).FindStringSubmatch(from)
|
||||
if said == nil {
|
||||
t.Fatal("the Dockerfile does not say which nats release its digest is (`# upstream: nats X.Y.Z-alpine`)")
|
||||
t.Fatalf("the server artifact is %q: it must read nats:X.Y.Z-alpine@sha256:<digest>, the tag saying the release", from)
|
||||
}
|
||||
// The server's own word for its release, compiled into it: a test binary's build information may list
|
||||
// no dependencies (the build seat's did not, so this read "these tests run no nats-server").
|
||||
if server.VERSION != string(said[1]) {
|
||||
if server.VERSION != said[1] {
|
||||
t.Fatalf("the image runs nats %s and these tests run %s: move them together", said[1], server.VERSION)
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -4,6 +4,7 @@ go 1.25.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.7
|
||||
github.com/klauspost/compress v1.18.5
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.51.0
|
||||
)
|
||||
@@ -11,7 +12,6 @@ require (
|
||||
require (
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
github.com/klauspost/compress v1.18.5 // indirect
|
||||
github.com/minio/highwayhash v1.0.4 // indirect
|
||||
github.com/nats-io/jwt/v2 v2.8.1 // indirect
|
||||
github.com/nats-io/nkeys v0.4.15 // indirect
|
||||
|
||||
+49
-15
@@ -57,7 +57,7 @@
|
||||
"class": "valuable",
|
||||
"active": "1d",
|
||||
"backup": {
|
||||
"dump": "docker exec -i mesh-broker-nats mesh-nats-snapshot snapshot < ${dir:mesh-state}/broker > ${dir:snapshots}/bus.tar.partial && mv ${dir:snapshots}/bus.tar.partial ${dir:snapshots}/bus.tar || { rm -f ${dir:snapshots}/bus.tar.partial; exit 1; }",
|
||||
"dump": "docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot snapshot < ${dir:mesh-state}/broker > ${dir:snapshots}/bus.tar.partial && mv ${dir:snapshots}/bus.tar.partial ${dir:snapshots}/bus.tar || { rm -f ${dir:snapshots}/bus.tar.partial; exit 1; }",
|
||||
"into": "snapshots"
|
||||
},
|
||||
"why": "the bus's streams and key-value buckets: the hand-act log, conditions, every module's state; written all the time, so copied by the server's own snapshot of each stream (novox/hq ADR 0235), never as live files"
|
||||
@@ -105,6 +105,27 @@
|
||||
"template": "files/nats.conf",
|
||||
"mode": "0644"
|
||||
},
|
||||
{
|
||||
"id": "state",
|
||||
"type": "directory",
|
||||
"mode": "0700",
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "program-snapshot",
|
||||
"type": "archive",
|
||||
"path": "${dir:state}/snapshot",
|
||||
"artifact": "snapshot",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "program-entrypoint",
|
||||
"type": "archive",
|
||||
"path": "${dir:state}/entrypoint",
|
||||
"artifact": "entrypoint",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "server",
|
||||
"type": "container",
|
||||
@@ -116,9 +137,19 @@
|
||||
"volumes": [
|
||||
"/var/lib/mesh-broker-nats:/data",
|
||||
"/var/lib/nats-module/conf:/etc/nats:ro",
|
||||
"${access:tls}:/tls:ro"
|
||||
"${access:tls}:/tls:ro",
|
||||
"${dir:state}/snapshot:/opt/mesh:ro",
|
||||
"${dir:state}/entrypoint:/opt/mesh-entrypoint:ro"
|
||||
],
|
||||
"artifact": "server"
|
||||
"args": [
|
||||
"sh",
|
||||
"/opt/mesh-entrypoint/entrypoint.sh"
|
||||
],
|
||||
"artifact": "server",
|
||||
"restart-on": [
|
||||
"program-snapshot",
|
||||
"program-entrypoint"
|
||||
]
|
||||
}
|
||||
],
|
||||
"accesses": [
|
||||
@@ -129,21 +160,24 @@
|
||||
}
|
||||
],
|
||||
"build": {
|
||||
"on": [
|
||||
{
|
||||
"arg": "NATS_BASE",
|
||||
"image": "nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00"
|
||||
},
|
||||
{
|
||||
"arg": "GO_BASE",
|
||||
"image": "golang@sha256:8ac98ca534ac3f51e1f420a1dd2c15e74c75cfa0f23f3ad27eb5d7236c349a0c"
|
||||
}
|
||||
],
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "server",
|
||||
"kind": "image",
|
||||
"from": "Dockerfile"
|
||||
"kind": "upstream",
|
||||
"from": "nats:2.11.17-alpine@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00"
|
||||
},
|
||||
{
|
||||
"name": "snapshot",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "alpine",
|
||||
"from": "cmd/mesh-nats-snapshot",
|
||||
"binary": "mesh-nats-snapshot"
|
||||
},
|
||||
{
|
||||
"name": "entrypoint",
|
||||
"kind": "archive",
|
||||
"from": "program"
|
||||
},
|
||||
{
|
||||
"name": "tools",
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
module nats-snapshot
|
||||
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
github.com/klauspost/compress v1.18.5
|
||||
github.com/nats-io/nats.go v1.51.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/nats-io/nkeys v0.4.15 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.49.0 // indirect
|
||||
golang.org/x/sys v0.42.0 // indirect
|
||||
)
|
||||
@@ -1,12 +0,0 @@
|
||||
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
|
||||
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI=
|
||||
github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
|
||||
github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
|
||||
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
|
||||
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
|
||||
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
|
||||
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
@@ -1,152 +0,0 @@
|
||||
package main
|
||||
|
||||
// The declared dump, exactly as the backup holder runs it, against the bus's own image.
|
||||
//
|
||||
// docker build --build-arg NATS_BASE=<digest> --build-arg GO_BASE=<digest> -t mesh-nats:check ..
|
||||
// MESH_TEST_NATS_IMAGE=mesh-nats:check go test -run TestTheDeclaredDump ./...
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// manifestOf is the nats module's own definition, beside this program.
|
||||
func manifestOf(t *testing.T) map[string]any {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile("../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func declaredDump(t *testing.T, m map[string]any) string {
|
||||
t.Helper()
|
||||
for _, item := range m["data"].(map[string]any)["own"].([]any) {
|
||||
it := item.(map[string]any)
|
||||
if it["id"] != "jetstream" {
|
||||
continue
|
||||
}
|
||||
backup, _ := it["backup"].(map[string]any)
|
||||
if backup == nil || backup["into"] != "snapshots" {
|
||||
t.Fatalf("the bus's streams are not protected by a dump into its snapshots: %v", it["backup"])
|
||||
}
|
||||
return backup["dump"].(string)
|
||||
}
|
||||
t.Fatal("the nats module declares no jetstream data item")
|
||||
return ""
|
||||
}
|
||||
|
||||
func serverConf(t *testing.T, m map[string]any) string {
|
||||
t.Helper()
|
||||
for _, r := range m["resources"].([]any) {
|
||||
res := r.(map[string]any)
|
||||
if res["id"] == "server-conf" {
|
||||
return res["content"].(string)
|
||||
}
|
||||
}
|
||||
t.Fatal("the nats module declares no server configuration")
|
||||
return ""
|
||||
}
|
||||
|
||||
// The dump is the one the manifest declares, run by `sh -c` as the holder runs it, with the module's
|
||||
// directories filled in; the server is the module's image with the module's own configuration and a
|
||||
// user list holding the bus's own module with the controller's grants for it.
|
||||
func TestTheDeclaredDumpRunsInTheBussImage(t *testing.T) {
|
||||
image := os.Getenv("MESH_TEST_NATS_IMAGE")
|
||||
if image == "" {
|
||||
t.Skip("MESH_TEST_NATS_IMAGE unset: the bus's image, built from ../Dockerfile")
|
||||
}
|
||||
const name = "mesh-broker-nats" // the container the dump names
|
||||
if out, _ := exec.Command("docker", "ps", "-a", "-q", "--filter", "name=^"+name+"$").Output(); len(strings.TrimSpace(string(out))) > 0 {
|
||||
t.Skipf("a container called %s already runs here; this test will not touch it", name)
|
||||
}
|
||||
m := manifestOf(t)
|
||||
dump := declaredDump(t, m)
|
||||
|
||||
conf, tlsDir, data, state, snapshots := t.TempDir(), t.TempDir(), t.TempDir(), t.TempDir(), t.TempDir()
|
||||
pin := selfSigned(t, tlsDir)
|
||||
_ = os.WriteFile(filepath.Join(conf, "nats.conf"), []byte(serverConf(t, m)), 0o644)
|
||||
accounts := liveBusConf()
|
||||
accounts = accounts[strings.Index(accounts, "accounts {"):]
|
||||
_ = os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(accounts), 0o644)
|
||||
|
||||
docker(t, "run", "-d", "--rm", "--name", name, "--user", owner(), "-p", "127.0.0.1::4222",
|
||||
"-v", data+":/data", "-v", conf+":/etc/nats:ro", "-v", tlsDir+":/tls:ro", image)
|
||||
t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", name).Run() })
|
||||
_, port, _ := strings.Cut(docker(t, "port", name, "4222/tcp"), ":")
|
||||
|
||||
admin := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
|
||||
js, _ := admin.JetStream()
|
||||
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_hand-acts"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, k := range []string{"a", "b", "c"} {
|
||||
if _, err := kv.Put(k, []byte("act "+k)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
run := func(cred Credential) (string, error) {
|
||||
raw, _ := json.Marshal(cred)
|
||||
if err := os.WriteFile(filepath.Join(state, "broker"), raw, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
command := strings.NewReplacer("${dir:mesh-state}", state, "${dir:snapshots}", snapshots).Replace(dump)
|
||||
out, err := exec.Command("sh", "-c", command).CombinedOutput()
|
||||
return string(out), err
|
||||
}
|
||||
|
||||
began := time.Now()
|
||||
out, err := run(Credential{URL: "nats://bus.example:4222", Fingerprint: pin, User: snapshotUser, Password: "snap"})
|
||||
if err != nil {
|
||||
t.Fatalf("the declared dump failed: %v\n%s", err, out)
|
||||
}
|
||||
t.Logf("the declared dump took %s:\n%s", time.Since(began).Round(time.Millisecond), out)
|
||||
tarPath := filepath.Join(snapshots, "bus.tar")
|
||||
f, err := os.Open(tarPath)
|
||||
if err != nil {
|
||||
t.Fatalf("the dump left no bus.tar: %v", err)
|
||||
}
|
||||
a, err := ReadArchive(f, "")
|
||||
f.Close()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer a.Close()
|
||||
if err := a.Verify(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(a.Manifest.Streams) != 1 || a.Manifest.Streams[0].Name != "KV_mesh-controller_hand-acts" || a.Manifest.Streams[0].Messages != 3 {
|
||||
t.Fatalf("the snapshot holds %+v", a.Manifest.Streams)
|
||||
}
|
||||
before, _ := os.ReadFile(tarPath)
|
||||
|
||||
// A night that cannot snapshot fails, says why, and leaves last night's snapshot as it was.
|
||||
out, err = run(Credential{Fingerprint: pin, User: snapshotUser, Password: "wrong"})
|
||||
if err == nil {
|
||||
t.Fatalf("a dump the bus refused succeeded:\n%s", out)
|
||||
}
|
||||
if !strings.Contains(out, "Authorization Violation") && !strings.Contains(strings.ToLower(out), "authorization") {
|
||||
t.Errorf("the failure does not say the bus refused the credential: %s", out)
|
||||
}
|
||||
after, _ := os.ReadFile(tarPath)
|
||||
if string(after) != string(before) {
|
||||
t.Error("a failed night replaced the last good snapshot")
|
||||
}
|
||||
if _, err := os.Stat(tarPath + ".partial"); !os.IsNotExist(err) {
|
||||
t.Error("a failed night left its partial file behind")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user