diff --git a/gitea/app.ini b/gitea/app.ini new file mode 100644 index 0000000..cbd9790 --- /dev/null +++ b/gitea/app.ini @@ -0,0 +1,56 @@ +APP_NAME = Gitea +RUN_USER = git +RUN_MODE = prod +WORK_PATH = /mnt/volume/gitea + +[server] +PROTOCOL = http +DOMAIN = %%GITEA_DOMAIN%% +ROOT_URL = %%GITEA_ROOT_URL%% +HTTP_ADDR = 0.0.0.0 +HTTP_PORT = 3000 +SSH_DOMAIN = %%GITEA_DOMAIN%% +START_SSH_SERVER = true +SSH_PORT = 2222 +SSH_LISTEN_PORT = 2222 +LFS_START_SERVER = true +APP_DATA_PATH = /mnt/volume/gitea/data +DISABLE_SSH = false +LFS_JWT_SECRET = %%LFS_JWT_SECRET%% + +[database] +DB_TYPE = postgres +HOST = %%DB_HOST%%:%%DB_PORT%% +NAME = %%DB_NAME%% +USER = %%DB_USER%% +PASSWD = %%DB_PASSWORD%% +SSL_MODE = disable +SCHEMA = public + +[repository] +ROOT = /mnt/volume/gitea/data/gitea-repositories + +[lfs] +PATH = /mnt/volume/gitea/data/lfs + +[log] +ROOT_PATH = /mnt/volume/gitea/log +MODE = console +LEVEL = info + +[service] +DISABLE_REGISTRATION = true +REQUIRE_SIGNIN_VIEW = false +DEFAULT_KEEP_EMAIL_PRIVATE = true +DEFAULT_ALLOW_CREATE_ORGANIZATION = true + +[security] +INSTALL_LOCK = true +SECRET_KEY = %%SECRET_KEY%% +INTERNAL_TOKEN = %%INTERNAL_TOKEN%% + +[oauth2] +JWT_SECRET = %%JWT_SECRET%% + +[session] +PROVIDER = file diff --git a/gitea/generate-secrets.sh b/gitea/generate-secrets.sh new file mode 100755 index 0000000..7af1c7b --- /dev/null +++ b/gitea/generate-secrets.sh @@ -0,0 +1,16 @@ +#!/usr/bin/env bash +# Generates JWT_SECRET, SECRET_KEY and INTERNAL_TOKEN in the exact format Gitea expects. +# Run once, then paste the three lines into Zerops as secret env vars for the gitea service. +set -euo pipefail + +GITEA_VERSION="${GITEA_VERSION:-1.26.1}" +BIN=$(mktemp) +trap 'rm -f "$BIN"' EXIT + +wget -qO "$BIN" "https://dl.gitea.com/gitea/${GITEA_VERSION}/gitea-${GITEA_VERSION}-linux-amd64" +chmod +x "$BIN" + +echo "JWT_SECRET=$("$BIN" generate secret JWT_SECRET)" +echo "SECRET_KEY=$("$BIN" generate secret SECRET_KEY)" +echo "INTERNAL_TOKEN=$("$BIN" generate secret INTERNAL_TOKEN)" +echo "LFS_JWT_SECRET=$("$BIN" generate secret LFS_JWT_SECRET)" diff --git a/gitea/zerops.yaml b/gitea/zerops.yaml new file mode 100644 index 0000000..bbd609d --- /dev/null +++ b/gitea/zerops.yaml @@ -0,0 +1,44 @@ +zerops: + - setup: gitea + build: + base: ubuntu@24.04 + deployFiles: app.ini + run: + base: ubuntu@24.04 + ports: + - port: 3000 + httpSupport: true + - port: 2222 + envVariables: + GITEA_VERSION: 1.26.1 + GITEA_WORK_DIR: /mnt/volume/gitea + GITEA_DOMAIN: git.matejpavlicek.cz + GITEA_ROOT_URL: https://git.matejpavlicek.cz + DB_HOST: db + DB_PORT: 5432 + DB_NAME: giteadb + DB_USER: gitea + envReplace: + delimiter: "%%" + target: + - app.ini + prepareCommands: + - sudo apt-get update + - sudo apt-get install -y --no-install-recommends git ca-certificates gettext-base wget gpg + - wget -q -O /tmp/gitea https://dl.gitea.com/gitea/${GITEA_VERSION}/gitea-${GITEA_VERSION}-linux-amd64 + - wget -q -O /tmp/gitea.asc https://dl.gitea.com/gitea/${GITEA_VERSION}/gitea-${GITEA_VERSION}-linux-amd64.asc + - gpg --keyserver hkps://keys.openpgp.org --recv 7C9E68152594688862D62AF62D9AE806EC1592E2 + - gpg --verify /tmp/gitea.asc /tmp/gitea + - sudo install -m 755 /tmp/gitea /usr/local/bin/gitea + - sudo groupadd --system git + - sudo useradd --system --gid git --shell /bin/bash --home-dir /home/git --create-home git + - sudo mkdir -p /etc/gitea + - sudo chown root:git /etc/gitea + - sudo chmod 770 /etc/gitea + initCommands: + - until mountpoint -q /mnt/volume; do sleep 1; done + - sudo mkdir -p /mnt/volume/gitea/{custom,data,indexers,public,log} + - sudo chown -R git:git /mnt/volume/gitea + - sudo chmod -R 750 /mnt/volume/gitea + - sudo install -m 660 -o root -g git /var/www/app.ini /etc/gitea/app.ini + start: sudo -u git -E /usr/local/bin/gitea web --config /etc/gitea/app.ini diff --git a/long-build/zerops-service-import.yml b/long-build/zerops-service-import.yml new file mode 100644 index 0000000..ca7275f --- /dev/null +++ b/long-build/zerops-service-import.yml @@ -0,0 +1,4 @@ +services: + - hostname: app + type: ubuntu@24.04 + maxContainers: 1 diff --git a/long-build/zerops.yml b/long-build/zerops.yml new file mode 100644 index 0000000..05da395 --- /dev/null +++ b/long-build/zerops.yml @@ -0,0 +1,23 @@ +zerops: + - setup: app + build: + os: ubuntu + base: ubuntu@latest + buildCommands: + - echo "build start at $(date -Iseconds)" + # 1h10m sleep exercises the build-side prolong chain: zbuilder heartbeat -> znode -> zbusiness. + # Old 1h cap would kill it at the 60min mark; with prolong the process should comfortably reach the echo below. + - sleep 4200 + - echo "build done at $(date -Iseconds)" + deployFiles: + - ./ + run: + os: ubuntu + base: ubuntu@latest + prepareCommands: + - echo "prepare start at $(date -Iseconds)" + # ~100 GiB of incompressible random data, written to the runtime image filesystem. + # status=progress prints throughput every second so we can watch from the dashboard. + - dd if=/dev/urandom of=/var/www/bigfile bs=1M count=20480 status=progress + - echo "image bloated at $(date -Iseconds)" + start: tail -f /dev/null diff --git a/valkey/.gitignore b/valkey/.gitignore new file mode 100644 index 0000000..2cd961c --- /dev/null +++ b/valkey/.gitignore @@ -0,0 +1 @@ +valkey-test diff --git a/valkey/ca.crt b/valkey/ca.crt new file mode 100644 index 0000000..dc8fbd5 --- /dev/null +++ b/valkey/ca.crt @@ -0,0 +1,13 @@ +-----BEGIN CERTIFICATE----- +MIIB8TCCAZagAwIBAgIQbwx3ZUCaU9IQr+ynltMGsTAKBggqhkjOPQQDAjA2MRUw +EwYDVQQKEwxaZXJvcHMgSW5mcmExHTAbBgNVBAMTFFplcm9wcyBJbmZyYSBSb290 +IENBMB4XDTI1MDkwMTA3NDMzOVoXDTI3MDkwMTA3NDMxM1owVjELMAkGA1UEBhMC +Q1oxDzANBgNVBAoTBlplcm9wczEQMA4GA1UECxMHU2VydmljZTEkMCIGA1UEAxMb +WmVyb3BzIFNlcnZpY2UgSW50ZXJtZWRpYXRlMFkwEwYHKoZIzj0CAQYIKoZIzj0D +AQcDQgAEeJwBKQfdfdRR7QJWVLegIXvn1k8SqUo8dp8bY2Fj9PUSNsTZ4yDsgpGo +Zummp6P92mOookbowR1nc0d8TyRYs6NmMGQwDgYDVR0PAQH/BAQDAgEGMBIGA1Ud +EwEB/wQIMAYBAf8CAQAwHQYDVR0OBBYEFLhO0kbsvwnGvuBp+gBy82mzsMbzMB8G +A1UdIwQYMBaAFJIVumuZisDXAdV9359xveeP3HkhMAoGCCqGSM49BAMCA0kAMEYC +IQCanJOFmOCQjnC/0oGQTHKMmmfPcCXp82XOYDnfY6YEGAIhAPYpIizvP5RpLymG +fn7wyg9VTA7W2wxF0ToJUXOKJO5O +-----END CERTIFICATE----- diff --git a/valkey/chaos.go b/valkey/chaos.go new file mode 100644 index 0000000..504fed0 --- /dev/null +++ b/valkey/chaos.go @@ -0,0 +1,1520 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "math/rand" + "net/http" + "os" + "os/signal" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/charmbracelet/bubbles/viewport" + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/lipgloss" + "github.com/redis/go-redis/v9" +) + +// --------------------------------------------------------------------------- +// Chaos / soak engine — shared core for local TUI mode and Zerops worker mode +// --------------------------------------------------------------------------- + +const ( + chaosCounterKey = "vk:chaos:counter" + chaosLedgerKeyFn = "vk:chaos:ledger:" + chaosBloatKeyFn = "vk:chaos:bloat:" + ledgerCapacity = 5000 + + tickInterval = time.Second + maxRecentLat = 200 + maxGlobalLat = 2000 + maxSparkSamples = 60 + maxSeriesPoints = 120 + maxEvents = 500 + + httpBatchSize = 200 + httpBatchInterval = 100 * time.Millisecond + httpBatchTimeout = 10 * time.Second + httpQueueSize = 10_000 +) + +// ---- ledger ------------------------------------------------------------- + +type ledgerEntry struct { + seq int64 + value string + ackedAt time.Time + writer string + // Which Valkey instance acked the write (role + short run_id). + // "role:master/abc12345" identifies the master at write time; lets us + // tell post-mortem whether a MISSING was acked by a node that later + // stopped being authoritative. + writeBackend string +} + +type ledger struct { + mu sync.RWMutex + entries []ledgerEntry + seq atomic.Int64 +} + +func newLedger() *ledger { return &ledger{} } + +func (l *ledger) next() int64 { return l.seq.Add(1) } + +func (l *ledger) append(e ledgerEntry) { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.entries) >= ledgerCapacity { + drop := ledgerCapacity / 10 + copy(l.entries, l.entries[drop:]) + l.entries = l.entries[:len(l.entries)-drop] + } + l.entries = append(l.entries, e) +} + +func (l *ledger) sample() (ledgerEntry, bool) { + l.mu.RLock() + defer l.mu.RUnlock() + if len(l.entries) == 0 { + return ledgerEntry{}, false + } + return l.entries[rand.Intn(len(l.entries))], true +} + +func (l *ledger) last() (ledgerEntry, bool) { + l.mu.RLock() + defer l.mu.RUnlock() + n := len(l.entries) + if n == 0 { + return ledgerEntry{}, false + } + return l.entries[n-1], true +} + +func (l *ledger) snapshot() []ledgerEntry { + l.mu.RLock() + defer l.mu.RUnlock() + out := make([]ledgerEntry, len(l.entries)) + copy(out, l.entries) + return out +} + +func ledgerKey(seq int64) string { return chaosLedgerKeyFn + strconv.FormatInt(seq, 10) } +func bloatKey(seq int64) string { return chaosBloatKeyFn + strconv.FormatInt(seq, 10) } + +// ---- shared chaos context (workers + sink) ------------------------------ + +type chaosCtx struct { + ledger *ledger + counter atomic.Int64 + sink EventSink + startAt time.Time +} + +type outageRec struct { + start, end time.Time + opsLost int64 +} + +// ---- event sink abstraction -------------------------------------------- +// +// Workers don't know whether their events go to an in-process Bubble Tea +// program (local TUI) or a remote collector over HTTP. Two implementations: +// +// teaSink wraps *tea.Program — used by --chaos +// httpSink batches and POSTs to a collector — used by --worker +// +// Both deliver the same opResultMsg / subRecvMsg / ledgerCheckMsg shapes. + +type EventSink interface { + OpResult(opResultMsg) + SubRecv(subRecvMsg) + LedgerCheck(ledgerCheckMsg) + Event(eventMsg) + Close() error +} + +// ---- tea sink (local TUI) ---------------------------------------------- + +type teaSink struct{ p *tea.Program } + +func (s *teaSink) OpResult(m opResultMsg) { s.p.Send(m) } +func (s *teaSink) SubRecv(m subRecvMsg) { s.p.Send(m) } +func (s *teaSink) LedgerCheck(m ledgerCheckMsg) { s.p.Send(m) } +func (s *teaSink) Event(m eventMsg) { s.p.Send(m) } +func (s *teaSink) Close() error { return nil } + +// ---- HTTP sink (Zerops worker) ----------------------------------------- + +type httpSink struct { + url string + containerID string + client *http.Client + + queue chan wireEvent + done chan struct{} + wg sync.WaitGroup +} + +func newHTTPSink(url, containerID string) *httpSink { + s := &httpSink{ + url: strings.TrimRight(url, "/") + "/events", + containerID: containerID, + client: &http.Client{Timeout: httpBatchTimeout}, + queue: make(chan wireEvent, httpQueueSize), + done: make(chan struct{}), + } + s.wg.Add(1) + go s.runBatcher() + return s +} + +func (s *httpSink) runBatcher() { + defer s.wg.Done() + defer close(s.done) + batch := make([]wireEvent, 0, httpBatchSize) + t := time.NewTicker(httpBatchInterval) + defer t.Stop() + flush := func() { + if len(batch) == 0 { + return + } + s.post(batch) + batch = batch[:0] + } + for { + select { + case ev, ok := <-s.queue: + if !ok { + flush() + return + } + batch = append(batch, ev) + if len(batch) >= httpBatchSize { + flush() + } + case <-t.C: + flush() + } + } +} + +func (s *httpSink) post(events []wireEvent) { + body, err := json.Marshal(wireBatch{ContainerID: s.containerID, Events: events}) + if err != nil { + log.Printf("httpSink: marshal: %v", err) + return + } + // minimal retry: best-effort, drops on persistent failure + for attempt := 0; attempt < 3; attempt++ { + req, _ := http.NewRequest("POST", s.url, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + resp, err := s.client.Do(req) + if err == nil { + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode/100 == 2 { + return + } + } + time.Sleep(time.Duration(100*(attempt+1)) * time.Millisecond) + } +} + +func (s *httpSink) enqueue(e wireEvent) { + if e.At.IsZero() { + e.At = time.Now() + } + select { + case s.queue <- e: + default: + // queue full — drop event rather than block the worker + } +} + +func (s *httpSink) OpResult(m opResultMsg) { + we := wireEvent{ + Type: "op", + Worker: m.worker, + Op: m.op, + At: m.at, + OK: m.err == nil, + DurUS: m.dur.Microseconds(), + } + if m.err != nil { + we.Err = m.err.Error() + } + s.enqueue(we) +} + +func (s *httpSink) SubRecv(m subRecvMsg) { + s.enqueue(wireEvent{Type: "sub", Worker: m.worker, Seq: m.seq, At: m.at}) +} + +func (s *httpSink) LedgerCheck(m ledgerCheckMsg) { + s.enqueue(wireEvent{Type: "ledger", Worker: m.worker, Seq: m.seq, Kind: m.kind, Detail: m.detail, At: time.Now()}) +} + +func (s *httpSink) Event(m eventMsg) { + s.enqueue(wireEvent{Type: "info", At: m.at, Level: m.level, Text: m.text}) +} + +func (s *httpSink) Close() error { + close(s.queue) + s.wg.Wait() + return nil +} + +// ---- wire format (worker → collector) ---------------------------------- + +type wireBatch struct { + ContainerID string `json:"container"` + Events []wireEvent `json:"events"` +} + +type wireEvent struct { + Type string `json:"type"` // op | sub | ledger | info + Worker string `json:"worker,omitempty"` + At time.Time `json:"at"` + + // op + Op string `json:"op,omitempty"` + OK bool `json:"ok,omitempty"` + DurUS int64 `json:"dur_us,omitempty"` + Err string `json:"err,omitempty"` + + // sub / ledger + Seq int64 `json:"seq,omitempty"` + Kind string `json:"kind,omitempty"` + Detail string `json:"detail,omitempty"` + + // info + Level string `json:"level,omitempty"` + Text string `json:"text,omitempty"` +} + +// ---- workers ------------------------------------------------------------ + +type worker struct { + cfg WorkerConfig + rdb *redis.Client + cc *chaosCtx + bloatBuf []byte // pre-allocated random payload for bloater role +} + +func validateWorker(w WorkerConfig) error { + if w.Name == "" { + return fmt.Errorf("missing name") + } + if w.Host == "" || w.Port == 0 { + return fmt.Errorf("missing host/port") + } + switch w.Role { + case "reader", "writer", "readwriter", "publisher", "subscriber", "bloater": + default: + return fmt.Errorf("unknown role %q", w.Role) + } + if (w.Role == "publisher" || w.Role == "subscriber") && w.Channel == "" { + return fmt.Errorf("role %s requires channel", w.Role) + } + if w.Role != "subscriber" && w.Interval <= 0 { + return fmt.Errorf("missing or invalid interval") + } + if w.Role == "bloater" && w.ValueBytes <= 0 { + return fmt.Errorf("role bloater requires valueBytes > 0") + } + return nil +} + +func (w *worker) run(ctx context.Context) { + switch w.cfg.Role { + case "writer": + w.runTicked(ctx, w.opWrite) + case "readwriter": + w.runTicked(ctx, w.opReadWrite) + case "reader": + w.runTicked(ctx, w.opRead) + case "publisher": + w.runTicked(ctx, w.opPublish) + case "subscriber": + w.runSubscriber(ctx) + case "bloater": + w.runTicked(ctx, w.opBloat) + } +} + +func (w *worker) runTicked(ctx context.Context, op func(context.Context) (string, error)) { + t := time.NewTicker(w.cfg.Interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + } + opCtx, cancel := context.WithTimeout(ctx, maxDur(w.cfg.Interval, 5*time.Second)) + start := time.Now() + name, err := op(opCtx) + cancel() + // If the worker's parent context has been cancelled (SIGTERM / + // Ctrl+C / TUI exit), this op's error is shutdown noise — don't + // flag it as a failure or trigger an outage. We're done. + if ctx.Err() != nil { + return + } + w.cc.sink.OpResult(opResultMsg{ + worker: w.cfg.Name, + op: name, + err: err, + dur: time.Since(start), + at: time.Now(), + }) + } +} + +// All read/write paths route through Client.Conn() so a single TCP socket +// carries the entire op + a probe INFO. HAProxy pins one frontend↔backend +// per TCP connection in tcp mode, so the probe reports the exact Valkey +// instance that handled the user command - critical for diagnosing whether +// a MISSING was answered by the new master, a lagging replica, or a node +// mid-shutdown. + +func (w *worker) opWrite(ctx context.Context) (string, error) { + conn := w.rdb.Conn() + defer conn.Close() + + if _, err := conn.Incr(ctx, chaosCounterKey).Result(); err != nil { + return "INCR", err + } + w.cc.counter.Add(1) + seq := w.cc.ledger.next() + val := fmt.Sprintf("%d:%d", seq, rand.Int63()) + if err := conn.Set(ctx, ledgerKey(seq), val, 0).Err(); err != nil { + return "SET", err + } + backend := backendIdent(ctx, conn) + w.cc.ledger.append(ledgerEntry{ + seq: seq, value: val, ackedAt: time.Now(), + writer: w.cfg.Name, writeBackend: backend, + }) + return "WRITE", nil +} + +func (w *worker) opReadWrite(ctx context.Context) (string, error) { + conn := w.rdb.Conn() + defer conn.Close() + + if _, err := conn.Incr(ctx, chaosCounterKey).Result(); err != nil { + return "INCR", err + } + w.cc.counter.Add(1) + seq := w.cc.ledger.next() + val := fmt.Sprintf("%d:%d", seq, rand.Int63()) + if err := conn.Set(ctx, ledgerKey(seq), val, 0).Err(); err != nil { + return "SET", err + } + writeBackend := backendIdent(ctx, conn) + w.cc.ledger.append(ledgerEntry{ + seq: seq, value: val, ackedAt: time.Now(), + writer: w.cfg.Name, writeBackend: writeBackend, + }) + // Read-after-write on the SAME conn → same backend (the master). Any + // mismatch here means the master we just acked from doesn't even agree + // with itself, which would be a real bug rather than replica lag. + got, err := conn.Get(ctx, ledgerKey(seq)).Result() + if err != nil { + return "GET", err + } + if got != val { + return "GET", fmt.Errorf("read-after-write mismatch seq=%d backend=%s", seq, writeBackend) + } + return "RW", nil +} + +func (w *worker) opRead(ctx context.Context) (string, error) { + e, ok := w.cc.ledger.sample() + if !ok { + if _, err := w.rdb.Ping(ctx).Result(); err != nil { + return "PING", err + } + return "PING", nil + } + conn := w.rdb.Conn() + defer conn.Close() + + got, err := conn.Get(ctx, ledgerKey(e.seq)).Result() + if err == redis.Nil { + readBackend := backendIdent(ctx, conn) + w.cc.sink.LedgerCheck(ledgerCheckMsg{ + worker: w.cfg.Name, seq: e.seq, kind: "missing", + detail: fmt.Sprintf("read-from=%s acked %s ago by %s on %s", + readBackend, + time.Since(e.ackedAt).Round(time.Millisecond), + e.writer, e.writeBackend), + }) + return "GET", nil + } + if err != nil { + return "GET", err + } + if got != e.value { + readBackend := backendIdent(ctx, conn) + w.cc.sink.LedgerCheck(ledgerCheckMsg{ + worker: w.cfg.Name, seq: e.seq, kind: "mismatch", + detail: fmt.Sprintf("read-from=%s expected=%s got=%s (acked by %s on %s)", + readBackend, trim(e.value, 30), trim(got, 30), + e.writer, e.writeBackend), + }) + } else { + w.cc.sink.LedgerCheck(ledgerCheckMsg{worker: w.cfg.Name, seq: e.seq, kind: "ok"}) + } + return "GET", nil +} + +// backendIdent returns "role:run_id_short" for the Valkey instance behind +// the given conn. INFO is best-effort: on probe failure (e.g. conn already +// broken after a NIL response from a node mid-shutdown), returns "?" so +// the missing-detail at least notes that the conn died on the way out. +func backendIdent(ctx context.Context, conn *redis.Conn) string { + info, err := conn.Info(ctx, "server", "replication").Result() + if err != nil { + return "?" + } + var role, runID string + for _, line := range strings.Split(info, "\n") { + line = strings.TrimRight(line, "\r") + switch { + case strings.HasPrefix(line, "run_id:"): + runID = strings.TrimPrefix(line, "run_id:") + case strings.HasPrefix(line, "role:"): + role = strings.TrimPrefix(line, "role:") + } + } + if len(runID) > 8 { + runID = runID[:8] + } + if role == "" && runID == "" { + return "?" + } + return role + "/" + runID +} + +func (w *worker) opPublish(ctx context.Context) (string, error) { + seq := w.cc.ledger.next() + payload := fmt.Sprintf("pub-%d", seq) + if err := w.rdb.Publish(ctx, w.cfg.Channel, payload).Err(); err != nil { + return "PUB", err + } + return "PUB", nil +} + +// opBloat writes a fixed-size random payload to a fresh key under a +// dedicated "bloat" namespace, no TTL. Each tick adds another key, so +// Valkey memory usage grows monotonically — useful for watching how +// failover, replication, and RAM scaling behave under increasing pressure. +// Bloat keys are intentionally NOT added to the durability ledger; they're +// memory load, not consistency probes. +func (w *worker) opBloat(ctx context.Context) (string, error) { + seq := w.cc.ledger.next() + if err := w.rdb.Set(ctx, bloatKey(seq), w.bloatBuf, 0).Err(); err != nil { + return "BLOAT", err + } + return "BLOAT", nil +} + +func (w *worker) runSubscriber(ctx context.Context) { + sub := w.rdb.Subscribe(ctx, w.cfg.Channel) + defer sub.Close() + if _, err := sub.Receive(ctx); err != nil { + // shutdown-time cancellation isn't a real subscribe failure + if ctx.Err() != nil { + return + } + w.cc.sink.OpResult(opResultMsg{ + worker: w.cfg.Name, op: "SUBSCRIBE", err: err, at: time.Now(), + }) + return + } + ch := sub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + seq, perr := strconv.ParseInt(strings.TrimPrefix(msg.Payload, "pub-"), 10, 64) + if perr != nil { + continue + } + w.cc.sink.SubRecv(subRecvMsg{worker: w.cfg.Name, seq: seq, at: time.Now()}) + } + } +} + +// ---- TUI message types -------------------------------------------------- + +type opResultMsg struct { + worker string + op string + err error + dur time.Duration + at time.Time +} + +type subRecvMsg struct { + worker string + seq int64 + at time.Time +} + +type ledgerCheckMsg struct { + worker string + seq int64 + kind string + detail string +} + +type eventMsg struct { + at time.Time + level string + text string +} + +type tickMsg time.Time + +// ---- TUI model ---------------------------------------------------------- + +type uiStats struct { + role string + addr string + + ops, ok, fail int64 + consecFail int + + inOutage bool + outageStart time.Time + outageOpsLost int64 + outages []outageRec + + recentLat []time.Duration + + prevOps int64 + opsHist []float64 + + verified, missing, mismatch int64 + + msgs int64 + gaps int64 + longestGap time.Duration + lastSeq int64 + lastSeqAt time.Time +} + +type eventLine struct { + at time.Time + level string + text string +} + +type chaosModel struct { + workers []WorkerConfig + stats map[string]*uiStats + + events []eventLine + eventVP viewport.Model + + p95Series []float64 + okPerSecSeries []float64 + failPerSecSeries []float64 + + globalLat []time.Duration + prevTotalOps int64 + prevTotalFail int64 + + width, height int + startAt time.Time + quitting bool +} + +func newChaosModel(ws []WorkerConfig, startAt time.Time) *chaosModel { + stats := make(map[string]*uiStats, len(ws)) + for _, w := range ws { + stats[w.Name] = &uiStats{ + role: w.Role, + addr: fmt.Sprintf("%s:%d", w.Host, w.Port), + } + } + vp := viewport.New(80, 8) + vp.SetContent(dimStyle.Render("(no events yet)")) + return &chaosModel{ + workers: ws, + stats: stats, + eventVP: vp, + startAt: startAt, + } +} + +func (m *chaosModel) Init() tea.Cmd { + return tea.Tick(tickInterval, func(t time.Time) tea.Msg { return tickMsg(t) }) +} + +func (m *chaosModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case tea.WindowSizeMsg: + m.width = msg.Width + m.height = msg.Height + m.recomputeLayout() + return m, nil + case tea.KeyMsg: + switch msg.String() { + case "q", "ctrl+c", "esc": + m.quitting = true + return m, tea.Quit + case "up", "k": + m.eventVP.LineUp(1) + case "down", "j": + m.eventVP.LineDown(1) + case "G", "end": + m.eventVP.GotoBottom() + case "g", "home": + m.eventVP.GotoTop() + case "pgup", "b": + m.eventVP.HalfViewUp() + case "pgdown", "f", " ": + m.eventVP.HalfViewDown() + } + return m, nil + case opResultMsg: + m.applyOpResult(msg) + return m, nil + case subRecvMsg: + m.applySubRecv(msg) + return m, nil + case ledgerCheckMsg: + m.applyLedgerCheck(msg) + return m, nil + case eventMsg: + m.appendEvent(msg.level, msg.at, msg.text) + return m, nil + case tickMsg: + m.onTick() + return m, tea.Tick(tickInterval, func(t time.Time) tea.Msg { return tickMsg(t) }) + } + return m, nil +} + +func (m *chaosModel) applyOpResult(msg opResultMsg) { + s, ok := m.stats[msg.worker] + if !ok { + return + } + s.ops++ + if msg.err != nil { + s.fail++ + s.consecFail++ + if !s.inOutage { + s.inOutage = true + s.outageStart = msg.at + s.outageOpsLost = 1 + m.appendEvent("err", msg.at, + fmt.Sprintf("[%s] OUTAGE START — %s: %s", msg.worker, msg.op, trim(msg.err.Error(), 80))) + } else { + s.outageOpsLost++ + } + return + } + s.ok++ + if s.inOutage { + d := msg.at.Sub(s.outageStart) + s.outages = append(s.outages, outageRec{ + start: s.outageStart, end: msg.at, opsLost: s.outageOpsLost, + }) + m.appendEvent("info", msg.at, + fmt.Sprintf("[%s] OUTAGE END — duration=%s ops_lost=%d", + msg.worker, d.Round(time.Millisecond), s.outageOpsLost)) + s.inOutage = false + s.consecFail = 0 + s.outageOpsLost = 0 + } + if msg.dur > 0 { + s.recentLat = appendBoundedDur(s.recentLat, msg.dur, maxRecentLat) + m.globalLat = appendBoundedDur(m.globalLat, msg.dur, maxGlobalLat) + } +} + +func (m *chaosModel) applySubRecv(msg subRecvMsg) { + s, ok := m.stats[msg.worker] + if !ok { + return + } + s.msgs++ + if s.lastSeq != 0 && msg.seq > s.lastSeq+1 { + gap := msg.seq - s.lastSeq - 1 + s.gaps += gap + m.appendEvent("warn", msg.at, + fmt.Sprintf("[%s] PUBSUB GAP last=%d now=%d missed=%d", + msg.worker, s.lastSeq, msg.seq, gap)) + } + if !s.lastSeqAt.IsZero() { + if d := msg.at.Sub(s.lastSeqAt); d > s.longestGap { + s.longestGap = d + } + } + s.lastSeq = msg.seq + s.lastSeqAt = msg.at +} + +func (m *chaosModel) applyLedgerCheck(msg ledgerCheckMsg) { + s, ok := m.stats[msg.worker] + if !ok { + return + } + switch msg.kind { + case "ok": + s.verified++ + case "missing": + s.missing++ + m.appendEvent("err", time.Now(), + fmt.Sprintf("[%s] LEDGER MISSING seq=%d %s", msg.worker, msg.seq, msg.detail)) + case "mismatch": + s.mismatch++ + m.appendEvent("err", time.Now(), + fmt.Sprintf("[%s] LEDGER MISMATCH seq=%d %s", msg.worker, msg.seq, msg.detail)) + } +} + +func (m *chaosModel) appendEvent(level string, at time.Time, text string) { + if len(m.events) >= maxEvents { + copy(m.events, m.events[1:]) + m.events = m.events[:len(m.events)-1] + } + m.events = append(m.events, eventLine{at: at, level: level, text: text}) + atBottom := m.eventVP.AtBottom() + m.eventVP.SetContent(m.renderEventContent()) + if atBottom { + m.eventVP.GotoBottom() + } +} + +func (m *chaosModel) onTick() { + if len(m.globalLat) > 0 { + cp := make([]time.Duration, len(m.globalLat)) + copy(cp, m.globalLat) + sort.Slice(cp, func(i, j int) bool { return cp[i] < cp[j] }) + idx := 95 * len(cp) / 100 + if idx >= len(cp) { + idx = len(cp) - 1 + } + p95ms := float64(cp[idx]) / float64(time.Millisecond) + m.p95Series = appendBoundedFloat(m.p95Series, p95ms, maxSeriesPoints) + } else { + m.p95Series = appendBoundedFloat(m.p95Series, 0, maxSeriesPoints) + } + + var totalOps, totalFail int64 + for _, s := range m.stats { + totalOps += s.ops + s.msgs + totalFail += s.fail + } + okDelta := (totalOps - totalFail) - (m.prevTotalOps - m.prevTotalFail) + if okDelta < 0 { + okDelta = 0 + } + failDelta := totalFail - m.prevTotalFail + if failDelta < 0 { + failDelta = 0 + } + m.prevTotalOps = totalOps + m.prevTotalFail = totalFail + m.okPerSecSeries = appendBoundedFloat(m.okPerSecSeries, float64(okDelta), maxSeriesPoints) + m.failPerSecSeries = appendBoundedFloat(m.failPerSecSeries, float64(failDelta), maxSeriesPoints) + + for _, s := range m.stats { + current := s.ops + s.msgs + delta := current - s.prevOps + if delta < 0 { + delta = 0 + } + s.prevOps = current + s.opsHist = appendBoundedFloat(s.opsHist, float64(delta), maxSparkSamples) + } +} + +func (m *chaosModel) recomputeLayout() { + headerH := 1 + tableTitleH := 1 + tableRows := len(m.workers) + 2 + chartTitleH := 1 + chartH := 8 + chartFooterH := 1 + eventTitleH := 1 + footerH := 1 + used := headerH + tableTitleH + tableRows + chartTitleH + chartH + chartFooterH + eventTitleH + footerH + 2 + eventH := m.height - used + if eventH < 4 { + eventH = 4 + } + w := m.width - 2 + if w < 20 { + w = 20 + } + m.eventVP.Width = w + m.eventVP.Height = eventH + m.eventVP.SetContent(m.renderEventContent()) +} + +// ---- TUI rendering ------------------------------------------------------ + +var ( + titleStyle = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("212")) + headStyle = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("39")) + dimStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("241")) + okStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("42")) + warnStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("214")) + errStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("196")) + chartStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("141")) +) + +func (m *chaosModel) View() string { + if m.quitting { + return "stopping…\n" + } + if m.width == 0 || m.height == 0 { + return "loading…\n" + } + if m.width < 90 || m.height < 25 { + return fmt.Sprintf("terminal too small: %dx%d (need 90x25 minimum). press q to quit.\n", + m.width, m.height) + } + + header := m.renderHeader() + table := m.renderTable() + charts := m.renderCharts() + events := m.renderEvents() + footer := dimStyle.Render(" q quit · ↑/↓ scroll · pgup/pgdn half-page · g/G top/bottom") + + return strings.Join([]string{header, "", table, "", charts, "", events, footer}, "\n") +} + +func (m *chaosModel) renderHeader() string { + elapsed := time.Since(m.startAt).Round(time.Second) + var inOutage int + for _, s := range m.stats { + if s.inOutage { + inOutage++ + } + } + tag := okStyle.Render("healthy") + if inOutage > 0 { + tag = errStyle.Render(fmt.Sprintf("%d worker(s) in outage", inOutage)) + } + return titleStyle.Render(" valkey-ha chaos") + + dimStyle.Render(fmt.Sprintf(" · %d workers · elapsed %s · ", len(m.workers), elapsed)) + + tag +} + +func (m *chaosModel) renderTable() string { + var b strings.Builder + b.WriteString(headStyle.Render(" workers") + "\n") + hdr := fmt.Sprintf(" %-14s %-11s %7s %7s %6s %7s %8s %8s %-12s %s", + "NAME", "ROLE", "OPS", "OK", "FAIL", "UPTIME", "P50", "P95", "OPS/SEC", "STATUS") + b.WriteString(dimStyle.Render(hdr) + "\n") + + for _, w := range m.workers { + s := m.stats[w.Name] + switch s.role { + case "subscriber": + line := fmt.Sprintf(" %-14s %-11s msgs=%-7d gaps=%-3d longest_gap=%-7s %s", + w.Name, "subscriber", s.msgs, s.gaps, + shortDur(s.longestGap), m.statusBadge(s)) + b.WriteString(line + "\n") + default: + uptime := 100.0 + if s.ops > 0 { + uptime = float64(s.ok) / float64(s.ops) * 100 + } + p50 := percentileLat(s.recentLat, 50) + p95 := percentileLat(s.recentLat, 95) + spark := chartStyle.Render(sparkline(s.opsHist, 12)) + extra := "" + if s.role == "reader" { + extra = dimStyle.Render(fmt.Sprintf(" v=%d miss=%d mis=%d", + s.verified, s.missing, s.mismatch)) + } + line := fmt.Sprintf(" %-14s %-11s %7d %7d %6d %6.2f%% %8s %8s %s %s%s", + w.Name, s.role, s.ops, s.ok, s.fail, uptime, + shortDur(p50), shortDur(p95), + spark, m.statusBadge(s), extra) + b.WriteString(line + "\n") + } + } + return strings.TrimRight(b.String(), "\n") +} + +func (m *chaosModel) statusBadge(s *uiStats) string { + if s.inOutage { + return errStyle.Render("OUTAGE") + } + if s.role == "subscriber" && s.msgs == 0 && time.Since(m.startAt) > 5*time.Second { + return warnStyle.Render("NO MSGS") + } + return okStyle.Render("OK") +} + +func (m *chaosModel) renderCharts() string { + chartH := 8 + chartW := (m.width - 4) / 3 // 3 charts with 2× " " separators between + if chartW < 20 { + chartW = 20 + } + + p95Max := maxFloat(m.p95Series) + if p95Max < 1 { + p95Max = 1 + } + p95Lines := renderBarChart(m.p95Series, chartW, chartH, p95Max) + p95Title := headStyle.Render(" p95 latency") + + dimStyle.Render(fmt.Sprintf(" peak %.1fms · last %s", + p95Max, lastFloatLabel(m.p95Series, "ms"))) + p95Box := p95Title + "\n" + chartStyle.Render(strings.Join(p95Lines, "\n")) + + tpMax := maxFloat(m.okPerSecSeries) + if tpMax < 5 { + tpMax = 5 + } + tpLines := renderBarChart(m.okPerSecSeries, chartW, chartH, tpMax) + tpTitle := headStyle.Render(" ok ops/sec") + + dimStyle.Render(fmt.Sprintf(" peak %.0f · last %s", + tpMax, lastFloatLabel(m.okPerSecSeries, ""))) + tpBox := tpTitle + "\n" + chartStyle.Render(strings.Join(tpLines, "\n")) + + failMax := maxFloat(m.failPerSecSeries) + if failMax < 1 { + failMax = 1 + } + failLines := renderBarChart(m.failPerSecSeries, chartW, chartH, failMax) + failTitle := headStyle.Render(" fail ops/sec") + + dimStyle.Render(fmt.Sprintf(" peak %.0f · last %s", + failMax, lastFloatLabel(m.failPerSecSeries, ""))) + failBox := failTitle + "\n" + errStyle.Render(strings.Join(failLines, "\n")) + + return lipgloss.JoinHorizontal(lipgloss.Top, p95Box, " ", tpBox, " ", failBox) +} + +func (m *chaosModel) renderEvents() string { + title := headStyle.Render(" events") + + dimStyle.Render(fmt.Sprintf(" (%d, scroll with ↑/↓)", len(m.events))) + return title + "\n" + m.eventVP.View() +} + +func (m *chaosModel) renderEventContent() string { + if len(m.events) == 0 { + return dimStyle.Render("(no events yet)") + } + var b strings.Builder + for _, e := range m.events { + ts := dimStyle.Render(e.at.Format("15:04:05.000")) + var styled string + switch e.level { + case "err": + styled = errStyle.Render(e.text) + case "warn": + styled = warnStyle.Render(e.text) + default: + styled = e.text + } + b.WriteString(ts + " " + styled + "\n") + } + return strings.TrimRight(b.String(), "\n") +} + +// ---- chart primitives --------------------------------------------------- + +var sparkChars = []rune(" ▁▂▃▄▅▆▇█") + +func sparkline(values []float64, width int) string { + if width <= 0 { + return "" + } + if len(values) == 0 { + return strings.Repeat(" ", width) + } + maxV := 0.0 + for _, v := range values { + if v > maxV { + maxV = v + } + } + var b strings.Builder + n := len(values) + for col := 0; col < width; col++ { + if maxV == 0 { + b.WriteRune(' ') + continue + } + idx := col * n / width + if idx >= n { + idx = n - 1 + } + v := values[idx] + if v < 0 { + v = 0 + } + if v > maxV { + v = maxV + } + i := int(v / maxV * float64(len(sparkChars)-1)) + if i < 0 { + i = 0 + } + if i >= len(sparkChars) { + i = len(sparkChars) - 1 + } + b.WriteRune(sparkChars[i]) + } + return b.String() +} + +func renderBarChart(values []float64, width, height int, maxV float64) []string { + rows := make([][]rune, height) + for i := range rows { + rows[i] = []rune(strings.Repeat(" ", width)) + } + flush := func() []string { + out := make([]string, height) + for i := range rows { + out[i] = string(rows[i]) + } + return out + } + if width <= 0 || height <= 0 || maxV <= 0 || len(values) == 0 { + return flush() + } + n := len(values) + subSteps := []rune(" ▁▂▃▄▅▆▇") + for col := 0; col < width; col++ { + idx := col * n / width + if idx >= n { + idx = n - 1 + } + v := values[idx] + if v < 0 { + v = 0 + } + if v > maxV { + v = maxV + } + cells := int(v / maxV * float64(height) * 8) + full := cells / 8 + rem := cells % 8 + for r := 0; r < full && r < height; r++ { + rows[height-1-r][col] = '█' + } + if full < height && rem > 0 { + rows[height-1-full][col] = subSteps[rem] + } + } + return flush() +} + +func lastFloatLabel(s []float64, unit string) string { + if len(s) == 0 { + return "-" + } + v := s[len(s)-1] + if unit == "ms" { + return fmt.Sprintf("%.1f%s", v, unit) + } + return fmt.Sprintf("%.0f%s", v, unit) +} + +// ---- runChaos (local TUI) ---------------------------------------------- + +func runChaos(ws []WorkerConfig) int { + for _, w := range ws { + if err := validateWorker(w); err != nil { + fmt.Fprintf(os.Stderr, "worker %q: %v\n", w.Name, err) + return 2 + } + } + + logFile, ferr := os.Create("chaos-debug.log") + if ferr == nil { + defer logFile.Close() + log.SetOutput(logFile) + } else { + log.SetOutput(io.Discard) + } + defer log.SetOutput(os.Stderr) + + workerCtx, cancel := context.WithCancel(context.Background()) + + cc := &chaosCtx{ + ledger: newLedger(), + startAt: time.Now(), + } + + workers := buildWorkers(ws, cc) + defer func() { + for _, w := range workers { + _ = w.rdb.Close() + } + }() + + model := newChaosModel(ws, cc.startAt) + p := tea.NewProgram(model, tea.WithAltScreen()) + cc.sink = &teaSink{p: p} + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) + go func() { + select { + case <-sigCh: + p.Quit() + case <-workerCtx.Done(): + } + }() + + var wg sync.WaitGroup + for _, w := range workers { + wg.Add(1) + go func(w *worker) { + defer wg.Done() + w.run(workerCtx) + }(w) + } + + if _, err := p.Run(); err != nil { + fmt.Fprintf(os.Stderr, "tui error: %v\n", err) + } + + cancel() + waitWorkers(&wg, 5*time.Second) + + fmt.Println("\nrunning final durability sweep…") + fd := finalSweep(workers, cc) + printFinalReport(model, time.Since(cc.startAt), fd) + + if fd.missing > 0 || fd.mismatch > 0 { + return 1 + } + return 0 +} + +// ---- runWorker (Zerops worker container) ------------------------------- + +func runWorker(ws []WorkerConfig, collectorURL string) int { + for _, w := range ws { + if err := validateWorker(w); err != nil { + fmt.Fprintf(os.Stderr, "worker %q: %v\n", w.Name, err) + return 2 + } + } + + containerID := containerIdent() + log.Printf("worker container_id=%s collector=%s workers=%d", containerID, collectorURL, len(ws)) + + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + + cc := &chaosCtx{ + ledger: newLedger(), + startAt: time.Now(), + } + cc.sink = newHTTPSink(collectorURL, containerID) + + workers := buildWorkers(ws, cc) + defer func() { + for _, w := range workers { + _ = w.rdb.Close() + } + }() + + // announce ourselves so the collector registers the worker rows immediately + for _, w := range workers { + cc.sink.Event(eventMsg{ + at: time.Now(), level: "info", + text: fmt.Sprintf("hello %s/%s role=%s addr=%s:%d db=%d", + containerID, w.cfg.Name, w.cfg.Role, w.cfg.Host, w.cfg.Port, w.cfg.DB), + }) + } + + var wg sync.WaitGroup + for _, w := range workers { + wg.Add(1) + go func(w *worker) { + defer wg.Done() + w.run(ctx) + }(w) + } + + <-ctx.Done() + log.Printf("worker shutting down, draining sink…") + waitWorkers(&wg, 5*time.Second) + + // Best-effort durability sweep against this container's own DB before exit. + fd := finalSweep(workers, cc) + cc.sink.Event(eventMsg{ + at: time.Now(), level: "info", + text: fmt.Sprintf("[%s] FINAL acked=%d verified=%d missing=%d mismatch=%d local_acked=%d server_counter=%d", + containerID, fd.acked, fd.verified, fd.missing, fd.mismatch, fd.localAcked, fd.serverCounter), + }) + _ = cc.sink.Close() + + if fd.missing > 0 || fd.mismatch > 0 { + return 1 + } + return 0 +} + +func buildWorkers(ws []WorkerConfig, cc *chaosCtx) []*worker { + workers := make([]*worker, 0, len(ws)) + for _, wc := range ws { + opts := optionsFromConn(wc.Host, wc.Port, wc.Password, wc.DB, wc.TLS) + w := &worker{cfg: wc, rdb: redis.NewClient(opts), cc: cc} + if wc.Role == "bloater" && wc.ValueBytes > 0 { + w.bloatBuf = make([]byte, wc.ValueBytes) + rand.Read(w.bloatBuf) + } + workers = append(workers, w) + } + return workers +} + +func waitWorkers(wg *sync.WaitGroup, max time.Duration) { + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(max): + } +} + +func containerIdent() string { + if v := os.Getenv("ZEROPS_Number"); v != "" { + return "vk-" + v + } + if v := os.Getenv("HOSTNAME"); v != "" { + return v + } + return "local-0" +} + +// ---- final report ------------------------------------------------------- + +type finalDur struct { + acked, verified, missing, mismatch int64 + missingSeqs []int64 + serverCounter, localAcked int64 +} + +func finalSweep(workers []*worker, cc *chaosCtx) finalDur { + var verifier *worker + for _, w := range workers { + if w.cfg.Role == "reader" || w.cfg.Role == "readwriter" { + verifier = w + break + } + } + if verifier == nil && len(workers) > 0 { + verifier = workers[0] + } + if verifier == nil { + return finalDur{} + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + entries := cc.ledger.snapshot() + var verified, missing, mismatch int64 + var missingSeqs []int64 + for _, e := range entries { + got, err := verifier.rdb.Get(ctx, ledgerKey(e.seq)).Result() + if err == redis.Nil || err != nil { + missing++ + missingSeqs = append(missingSeqs, e.seq) + continue + } + if got == e.value { + verified++ + } else { + mismatch++ + } + } + fd := finalDur{ + acked: int64(len(entries)), + verified: verified, + missing: missing, + mismatch: mismatch, + missingSeqs: missingSeqs, + localAcked: cc.counter.Load(), + } + if v, err := verifier.rdb.Get(ctx, chaosCounterKey).Int64(); err == nil { + fd.serverCounter = v + } + return fd +} + +func printFinalReport(m *chaosModel, dur time.Duration, fd finalDur) { + fmt.Println("=== valkey-ha chaos report ===") + fmt.Printf("duration: %s\n\n", dur.Round(time.Millisecond)) + + fmt.Println("per-worker:") + for _, w := range m.workers { + s := m.stats[w.Name] + switch s.role { + case "subscriber": + fmt.Printf(" %-14s [sub] msgs=%d gaps=%d longest_gap=%s\n", + w.Name, s.msgs, s.gaps, s.longestGap.Round(time.Millisecond)) + default: + uptime := 100.0 + if s.ops > 0 { + uptime = float64(s.ok) / float64(s.ops) * 100 + } + var longest time.Duration + for _, o := range s.outages { + if d := o.end.Sub(o.start); d > longest { + longest = d + } + } + extra := "" + if s.role == "reader" { + extra = fmt.Sprintf(" verified=%d missing=%d mismatch=%d", + s.verified, s.missing, s.mismatch) + } + fmt.Printf(" %-14s [%-10s] ops=%d ok=%d fail=%d uptime=%.2f%% p50=%s p95=%s outages=%d longest=%s%s\n", + w.Name, s.role, s.ops, s.ok, s.fail, uptime, + shortDur(percentileLat(s.recentLat, 50)), + shortDur(percentileLat(s.recentLat, 95)), + len(s.outages), longest.Round(time.Millisecond), + extra) + } + } + + fmt.Println("\ndurability:") + fmt.Printf(" ledger: acked=%d verified=%d missing=%d mismatch=%d\n", + fd.acked, fd.verified, fd.missing, fd.mismatch) + if len(fd.missingSeqs) > 0 { + n := 20 + if len(fd.missingSeqs) < n { + n = len(fd.missingSeqs) + } + fmt.Printf(" missing seqs (first %d): %v\n", n, fd.missingSeqs[:n]) + } + delta := fd.serverCounter - fd.localAcked + fmt.Printf(" counter: local_acked=%d server_value=%d delta=%d\n", + fd.localAcked, fd.serverCounter, delta) + + type tagged struct { + w string + rec outageRec + } + var all []tagged + for _, w := range m.workers { + for _, o := range m.stats[w.Name].outages { + all = append(all, tagged{w: w.Name, rec: o}) + } + } + sort.Slice(all, func(i, j int) bool { return all[i].rec.start.Before(all[j].rec.start) }) + if len(all) > 0 { + fmt.Println("\noutages (chronological):") + for _, t := range all { + fmt.Printf(" %s -> %s (%s) worker=%s ops_lost=%d\n", + t.rec.start.Format("15:04:05.000"), + t.rec.end.Format("15:04:05.000"), + t.rec.end.Sub(t.rec.start).Round(time.Millisecond), + t.w, t.rec.opsLost) + } + } + + fmt.Println("\nnotes:") + fmt.Println(" - Valkey 7.2 + Sentinel does NOT replicate pub/sub from master to replicas.") + fmt.Println(" A subscriber on the read-only VIP receiving 0 messages is expected.") + fmt.Println(" - debug log: ./chaos-debug.log") +} + +// ---- helpers ------------------------------------------------------------ + +func appendBoundedDur(s []time.Duration, v time.Duration, max int) []time.Duration { + s = append(s, v) + if len(s) > max { + s = s[len(s)-max:] + } + return s +} + +func appendBoundedFloat(s []float64, v float64, max int) []float64 { + s = append(s, v) + if len(s) > max { + s = s[len(s)-max:] + } + return s +} + +func percentileLat(values []time.Duration, pct int) time.Duration { + if len(values) == 0 { + return 0 + } + cp := make([]time.Duration, len(values)) + copy(cp, values) + sort.Slice(cp, func(i, j int) bool { return cp[i] < cp[j] }) + idx := pct * len(cp) / 100 + if idx >= len(cp) { + idx = len(cp) - 1 + } + return cp[idx] +} + +func maxFloat(s []float64) float64 { + m := 0.0 + for _, v := range s { + if v > m { + m = v + } + } + return m +} + +func maxDur(a, b time.Duration) time.Duration { + if a > b { + return a + } + return b +} + +func shortDur(d time.Duration) string { + if d <= 0 { + return "-" + } + if d < time.Microsecond { + return "0" + } + if d < time.Millisecond { + return fmt.Sprintf("%dµs", d.Microseconds()) + } + if d < time.Second { + return fmt.Sprintf("%.1fms", float64(d)/float64(time.Millisecond)) + } + return d.Round(time.Millisecond).String() +} + +func trim(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/valkey/collector.go b/valkey/collector.go new file mode 100644 index 0000000..3cdba0b --- /dev/null +++ b/valkey/collector.go @@ -0,0 +1,897 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "sort" + "strconv" + "sync" + "syscall" + "time" +) + +// --------------------------------------------------------------------------- +// Distributed collector — HTTP dashboard that ingests events from workers +// --------------------------------------------------------------------------- +// +// Endpoints: +// POST /events ingest a wireBatch from a worker container +// GET /api/state JSON snapshot for the live dashboard +// GET /api/events?cursor= JSON event log slice past the cursor +// GET / the dashboard HTML page +// +// Data model: +// workers keyed by "/" — same uiStats shape the +// TUI uses. A 1-second tick goroutine computes throughput deltas and p95 +// over a sliding window of recent latencies. + +type collectorWorker struct { + uiStats + containerID string + workerName string +} + +type collectorState struct { + mu sync.Mutex + + startAt time.Time + + // keyed by "/"; iteration order preserved via slice + keys []string + workers map[string]*collectorWorker + + events []eventLine + nextCursor int64 + + inconsistencies []eventLine + nextInconsCursor int64 + + // global series for the top-of-dashboard charts + p95Series []float64 + okPerSecSeries []float64 + failPerSecSeries []float64 + missingPerSecSeries []float64 + globalLat []time.Duration + prevTotalOps int64 + prevTotalFail int64 + prevTotalMissing int64 +} + +func newCollectorState() *collectorState { + return &collectorState{ + startAt: time.Now(), + workers: make(map[string]*collectorWorker), + } +} + +// ---- ingest ------------------------------------------------------------- + +func (c *collectorState) ingest(b wireBatch) { + c.mu.Lock() + defer c.mu.Unlock() + + for _, ev := range b.Events { + switch ev.Type { + case "op": + c.applyOp(b.ContainerID, ev) + case "sub": + c.applySub(b.ContainerID, ev) + case "ledger": + c.applyLedger(b.ContainerID, ev) + case "info": + c.appendEvent(ev.Level, ev.At, ev.Text) + } + } +} + +func (c *collectorState) workerByKey(containerID, workerName string) *collectorWorker { + key := containerID + "/" + workerName + w, ok := c.workers[key] + if !ok { + w = &collectorWorker{containerID: containerID, workerName: workerName} + c.workers[key] = w + c.keys = append(c.keys, key) + sort.Strings(c.keys) + } + return w +} + +func (c *collectorState) applyOp(containerID string, ev wireEvent) { + w := c.workerByKey(containerID, ev.Worker) + if w.role == "" { + w.role = roleFromOp(ev.Op) + } + w.ops++ + if !ev.OK { + // Surface every individual failure in the dashboard event log so + // the operator can see what's actually breaking, not just the + // summarized OUTAGE START/END boundaries. + c.appendEvent("err", ev.At, + fmt.Sprintf("[%s/%s] %s FAIL: %s", + containerID, ev.Worker, ev.Op, trim(ev.Err, 200))) + w.fail++ + w.consecFail++ + if !w.inOutage { + w.inOutage = true + w.outageStart = ev.At + w.outageOpsLost = 1 + c.appendEvent("err", ev.At, + fmt.Sprintf("[%s/%s] OUTAGE START", containerID, ev.Worker)) + } else { + w.outageOpsLost++ + } + return + } + w.ok++ + if w.inOutage { + w.outages = append(w.outages, outageRec{ + start: w.outageStart, end: ev.At, opsLost: w.outageOpsLost, + }) + w.inOutage = false + w.consecFail = 0 + w.outageOpsLost = 0 + } + if ev.DurUS > 0 { + dur := time.Duration(ev.DurUS) * time.Microsecond + w.recentLat = appendBoundedDur(w.recentLat, dur, maxRecentLat) + c.globalLat = appendBoundedDur(c.globalLat, dur, maxGlobalLat) + } +} + +func (c *collectorState) applySub(containerID string, ev wireEvent) { + w := c.workerByKey(containerID, ev.Worker) + if w.role == "" { + w.role = "subscriber" + } + w.msgs++ + if w.lastSeq != 0 && ev.Seq > w.lastSeq+1 { + gap := ev.Seq - w.lastSeq - 1 + w.gaps += gap + } + if !w.lastSeqAt.IsZero() { + if d := ev.At.Sub(w.lastSeqAt); d > w.longestGap { + w.longestGap = d + } + } + w.lastSeq = ev.Seq + w.lastSeqAt = ev.At +} + +func (c *collectorState) applyLedger(containerID string, ev wireEvent) { + w := c.workerByKey(containerID, ev.Worker) + if w.role == "" { + w.role = "reader" + } + switch ev.Kind { + case "ok": + w.verified++ + case "missing": + w.missing++ + c.appendInconsistency("err", ev.At, + fmt.Sprintf("[%s/%s] LEDGER MISSING seq=%d %s", + containerID, ev.Worker, ev.Seq, ev.Detail)) + case "mismatch": + w.mismatch++ + c.appendInconsistency("err", ev.At, + fmt.Sprintf("[%s/%s] LEDGER MISMATCH seq=%d %s", + containerID, ev.Worker, ev.Seq, ev.Detail)) + } +} + +// roleFromOp infers a worker's role from the first op name we see, since +// workers don't currently transmit role with each event. Best-effort only. +func roleFromOp(op string) string { + switch op { + case "WRITE": + return "writer" + case "RW": + return "readwriter" + case "GET", "PING": + return "reader" + case "PUB": + return "publisher" + case "SUBSCRIBE": + return "subscriber" + case "BLOAT": + return "bloater" + } + return "" +} + +func (c *collectorState) appendEvent(level string, at time.Time, text string) { + if at.IsZero() { + at = time.Now() + } + if level == "" { + level = "info" + } + if len(c.events) >= maxEvents { + copy(c.events, c.events[1:]) + c.events = c.events[:len(c.events)-1] + } + c.events = append(c.events, eventLine{at: at, level: level, text: text}) + c.nextCursor++ +} + +func (c *collectorState) appendInconsistency(level string, at time.Time, text string) { + if at.IsZero() { + at = time.Now() + } + if level == "" { + level = "info" + } + if len(c.inconsistencies) >= maxEvents { + copy(c.inconsistencies, c.inconsistencies[1:]) + c.inconsistencies = c.inconsistencies[:len(c.inconsistencies)-1] + } + c.inconsistencies = append(c.inconsistencies, eventLine{at: at, level: level, text: text}) + c.nextInconsCursor++ +} + +// ---- tick (1Hz) --------------------------------------------------------- + +func (c *collectorState) tick() { + c.mu.Lock() + defer c.mu.Unlock() + + if len(c.globalLat) > 0 { + cp := make([]time.Duration, len(c.globalLat)) + copy(cp, c.globalLat) + sort.Slice(cp, func(i, j int) bool { return cp[i] < cp[j] }) + idx := 95 * len(cp) / 100 + if idx >= len(cp) { + idx = len(cp) - 1 + } + p95ms := float64(cp[idx]) / float64(time.Millisecond) + c.p95Series = appendBoundedFloat(c.p95Series, p95ms, maxSeriesPoints) + } else { + c.p95Series = appendBoundedFloat(c.p95Series, 0, maxSeriesPoints) + } + + var totalOps, totalFail, totalMissing int64 + for _, w := range c.workers { + totalOps += w.ops + w.msgs + totalFail += w.fail + totalMissing += w.missing + } + okDelta := (totalOps - totalFail) - (c.prevTotalOps - c.prevTotalFail) + if okDelta < 0 { + okDelta = 0 + } + failDelta := totalFail - c.prevTotalFail + if failDelta < 0 { + failDelta = 0 + } + missingDelta := totalMissing - c.prevTotalMissing + if missingDelta < 0 { + missingDelta = 0 + } + c.prevTotalOps = totalOps + c.prevTotalFail = totalFail + c.prevTotalMissing = totalMissing + c.okPerSecSeries = appendBoundedFloat(c.okPerSecSeries, float64(okDelta), maxSeriesPoints) + c.failPerSecSeries = appendBoundedFloat(c.failPerSecSeries, float64(failDelta), maxSeriesPoints) + c.missingPerSecSeries = appendBoundedFloat(c.missingPerSecSeries, float64(missingDelta), maxSeriesPoints) + + for _, w := range c.workers { + current := w.ops + w.msgs + delta := current - w.prevOps + if delta < 0 { + delta = 0 + } + w.prevOps = current + w.opsHist = appendBoundedFloat(w.opsHist, float64(delta), maxSparkSamples) + } +} + +// ---- snapshot view ------------------------------------------------------ + +type stateView struct { + StartAt time.Time `json:"start_at"` + Elapsed string `json:"elapsed"` + Containers int `json:"containers"` + InOutage int `json:"in_outage"` + Workers []workerView `json:"workers"` + Charts chartsView `json:"charts"` +} + +type workerView struct { + Container string `json:"container"` + Name string `json:"name"` + Role string `json:"role"` + Ops int64 `json:"ops"` + OK int64 `json:"ok"` + Fail int64 `json:"fail"` + UptimePct float64 `json:"uptime_pct"` + P50MS float64 `json:"p50_ms"` + P95MS float64 `json:"p95_ms"` + OpsHist []float64 `json:"ops_hist"` + InOutage bool `json:"in_outage"` + Outages int `json:"outages"` + LongestMS float64 `json:"longest_outage_ms"` + // reader-only + Verified int64 `json:"verified,omitempty"` + Missing int64 `json:"missing,omitempty"` + Mismatch int64 `json:"mismatch,omitempty"` + // subscriber-only + Msgs int64 `json:"msgs,omitempty"` + Gaps int64 `json:"gaps,omitempty"` + LongestGapMS float64 `json:"longest_gap_ms,omitempty"` +} + +type chartsView struct { + P95 []float64 `json:"p95"` + OkPerSec []float64 `json:"ok_per_sec"` + FailPerSec []float64 `json:"fail_per_sec"` + MissingPerSec []float64 `json:"missing_per_sec"` + P95Peak float64 `json:"p95_peak"` + OkPeak float64 `json:"ok_peak"` + FailPeak float64 `json:"fail_peak"` + MissingPeak float64 `json:"missing_peak"` +} + +func (c *collectorState) snapshot() stateView { + c.mu.Lock() + defer c.mu.Unlock() + + containers := map[string]struct{}{} + for _, k := range c.keys { + w := c.workers[k] + containers[w.containerID] = struct{}{} + } + + view := stateView{ + StartAt: c.startAt, + Elapsed: time.Since(c.startAt).Round(time.Second).String(), + Containers: len(containers), + Charts: chartsView{ + P95: append([]float64(nil), c.p95Series...), + OkPerSec: append([]float64(nil), c.okPerSecSeries...), + FailPerSec: append([]float64(nil), c.failPerSecSeries...), + MissingPerSec: append([]float64(nil), c.missingPerSecSeries...), + P95Peak: maxFloat(c.p95Series), + OkPeak: maxFloat(c.okPerSecSeries), + FailPeak: maxFloat(c.failPerSecSeries), + MissingPeak: maxFloat(c.missingPerSecSeries), + }, + } + for _, k := range c.keys { + w := c.workers[k] + uptime := 100.0 + if w.ops > 0 { + uptime = float64(w.ok) / float64(w.ops) * 100 + } + var longest time.Duration + for _, o := range w.outages { + if d := o.end.Sub(o.start); d > longest { + longest = d + } + } + if w.inOutage { + view.InOutage++ + } + wv := workerView{ + Container: w.containerID, + Name: w.workerName, + Role: w.role, + Ops: w.ops, + OK: w.ok, + Fail: w.fail, + UptimePct: uptime, + P50MS: float64(percentileLat(w.recentLat, 50)) / float64(time.Millisecond), + P95MS: float64(percentileLat(w.recentLat, 95)) / float64(time.Millisecond), + OpsHist: append([]float64(nil), w.opsHist...), + InOutage: w.inOutage, + Outages: len(w.outages), + LongestMS: float64(longest) / float64(time.Millisecond), + } + switch w.role { + case "reader": + wv.Verified = w.verified + wv.Missing = w.missing + wv.Mismatch = w.mismatch + case "subscriber": + wv.Msgs = w.msgs + wv.Gaps = w.gaps + wv.LongestGapMS = float64(w.longestGap) / float64(time.Millisecond) + } + view.Workers = append(view.Workers, wv) + } + return view +} + +type eventOut struct { + ID int64 `json:"id"` + At string `json:"at"` + Level string `json:"level"` + Text string `json:"text"` +} + +func (c *collectorState) eventsSince(cursor int64) ([]eventOut, int64) { + c.mu.Lock() + defer c.mu.Unlock() + + // nextCursor counts events ever inserted; events slice may have dropped older. + // Compute base id of the first slot in the current slice: + baseID := c.nextCursor - int64(len(c.events)) + if cursor < baseID { + cursor = baseID + } + startIdx := int(cursor - baseID) + if startIdx < 0 { + startIdx = 0 + } + if startIdx > len(c.events) { + startIdx = len(c.events) + } + out := make([]eventOut, 0, len(c.events)-startIdx) + for i := startIdx; i < len(c.events); i++ { + e := c.events[i] + out = append(out, eventOut{ + ID: baseID + int64(i), + At: e.at.Format("15:04:05.000"), + Level: e.level, + Text: e.text, + }) + } + return out, c.nextCursor +} + +func (c *collectorState) inconsistenciesSince(cursor int64) ([]eventOut, int64) { + c.mu.Lock() + defer c.mu.Unlock() + + baseID := c.nextInconsCursor - int64(len(c.inconsistencies)) + if cursor < baseID { + cursor = baseID + } + startIdx := int(cursor - baseID) + if startIdx < 0 { + startIdx = 0 + } + if startIdx > len(c.inconsistencies) { + startIdx = len(c.inconsistencies) + } + out := make([]eventOut, 0, len(c.inconsistencies)-startIdx) + for i := startIdx; i < len(c.inconsistencies); i++ { + e := c.inconsistencies[i] + out = append(out, eventOut{ + ID: baseID + int64(i), + At: e.at.Format("15:04:05.000"), + Level: e.level, + Text: e.text, + }) + } + return out, c.nextInconsCursor +} + +// ---- HTTP server -------------------------------------------------------- + +func runCollector(port int) int { + state := newCollectorState() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + go func() { + t := time.NewTicker(tickInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + state.tick() + } + } + }() + + mux := http.NewServeMux() + mux.HandleFunc("/events", state.handleIngest) + mux.HandleFunc("/api/state", state.handleState) + mux.HandleFunc("/api/events", state.handleEvents) + mux.HandleFunc("/api/inconsistencies", state.handleInconsistencies) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write([]byte(dashboardHTML)) + }) + + srv := &http.Server{ + Addr: fmt.Sprintf(":%d", port), + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, + } + + go func() { + log.Printf("collector listening on :%d", port) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("collector: %v", err) + } + }() + + <-ctx.Done() + log.Printf("collector shutting down…") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = srv.Shutdown(shutdownCtx) + return 0 +} + +func (c *collectorState) handleIngest(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "POST only", http.StatusMethodNotAllowed) + return + } + var b wireBatch + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&b); err != nil { + http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest) + return + } + c.ingest(b) + w.WriteHeader(http.StatusNoContent) +} + +func (c *collectorState) handleState(w http.ResponseWriter, _ *http.Request) { + view := c.snapshot() + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(view) +} + +func (c *collectorState) handleEvents(w http.ResponseWriter, r *http.Request) { + cursor := int64(0) + if v := r.URL.Query().Get("cursor"); v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + cursor = n + } + } + events, next := c.eventsSince(cursor) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "events": events, + "next_cursor": next, + }) +} + +func (c *collectorState) handleInconsistencies(w http.ResponseWriter, r *http.Request) { + cursor := int64(0) + if v := r.URL.Query().Get("cursor"); v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + cursor = n + } + } + events, next := c.inconsistenciesSince(cursor) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "events": events, + "next_cursor": next, + }) +} + +// ---- embedded dashboard HTML ------------------------------------------- + +const dashboardHTML = ` + + + +valkey-ha chaos + + + + +

valkey-ha chaos

+ + +
+
+

p95 latency (rolling 2 min)

+
—
+ +
+
+

ok ops/sec (rolling 2 min)

+
—
+ +
+
+

fail ops/sec (rolling 2 min)

+
—
+ +
+
+

missing ledger/sec (rolling 2 min)

+
—
+ +
+
+ +
+

workers

+
+ +
+ + + + + + + + +
CONTAINERNAMEROLEOPSOKFAILUPTIMEP50P95SPARKOUTAGESSTATUSNOTES
+
+ +
+

client errors

+
+
+ +
+

ledger inconsistencies

+
+
+ + + + + + +` diff --git a/valkey/config-basic.yaml b/valkey/config-basic.yaml new file mode 100644 index 0000000..e47e0f5 --- /dev/null +++ b/valkey/config-basic.yaml @@ -0,0 +1,102 @@ +# Intermediate chaos workload: 1 readwriter + 1 TLS reader per valkey. +# +# Per target: +# - readwriter on plaintext :6379 at 10ms (~100 ops/sec) +# - reader on TLS :6380 at 5ms (~200 ops/sec) +# +# ⇒ ~300 ops/sec per target, ~1.8k ops/sec total per container, and both +# the plaintext and TLS ingress paths get exercised on every valkey. +# +# Per-worker passwords reference env vars that the deployment yaml +# (zerops-basic-import.yaml) wires in from each valkey service. +# +# NB: :6380 is the master TLS port; both valkey:single and valkey:ha +# masters serve GETs there, so a reader against :6380 works on both +# topologies. If a single-mode valkey ever stops exposing :6380, the +# corresponding ro-* worker will surface as a connect error — that's +# signal, not a bug here. + +workers: + - name: rw-single1 + role: readwriter + host: valkeysingle1.zerops + port: 6379 + password: ${VALKEY_VALKEYSINGLE1_PASSWORD} + interval: 10ms + - name: ro-single1 + role: reader + host: valkeysingle1.zerops + port: 6380 + tls: true + password: ${VALKEY_VALKEYSINGLE1_PASSWORD} + interval: 5ms + + - name: rw-single2 + role: readwriter + host: valkeysingle2.zerops + port: 6379 + password: ${VALKEY_VALKEYSINGLE2_PASSWORD} + interval: 10ms + - name: ro-single2 + role: reader + host: valkeysingle2.zerops + port: 6380 + tls: true + password: ${VALKEY_VALKEYSINGLE2_PASSWORD} + interval: 5ms + + - name: rw-single3 + role: readwriter + host: valkeysingle3.zerops + port: 6379 + password: ${VALKEY_VALKEYSINGLE3_PASSWORD} + interval: 10ms + - name: ro-single3 + role: reader + host: valkeysingle3.zerops + port: 6380 + tls: true + password: ${VALKEY_VALKEYSINGLE3_PASSWORD} + interval: 5ms + + - name: rw-ha1 + role: readwriter + host: valkeyha1.zerops + port: 6379 + password: ${VALKEY_VALKEYHA1_PASSWORD} + interval: 10ms + - name: ro-ha1 + role: reader + host: valkeyha1.zerops + port: 6380 + tls: true + password: ${VALKEY_VALKEYHA1_PASSWORD} + interval: 5ms + + - name: rw-ha2 + role: readwriter + host: valkeyha2.zerops + port: 6379 + password: ${VALKEY_VALKEYHA2_PASSWORD} + interval: 10ms + - name: ro-ha2 + role: reader + host: valkeyha2.zerops + port: 6380 + tls: true + password: ${VALKEY_VALKEYHA2_PASSWORD} + interval: 5ms + + - name: rw-ha3 + role: readwriter + host: valkeyha3.zerops + port: 6379 + password: ${VALKEY_VALKEYHA3_PASSWORD} + interval: 10ms + - name: ro-ha3 + role: reader + host: valkeyha3.zerops + port: 6380 + tls: true + password: ${VALKEY_VALKEYHA3_PASSWORD} + interval: 5ms diff --git a/valkey/config.yaml b/valkey/config.yaml new file mode 100644 index 0000000..55765b9 --- /dev/null +++ b/valkey/config.yaml @@ -0,0 +1,135 @@ +# Valkey HA chaos/soak config — spicy edition. +# +# 13 workers per container. With Zerops scaling to 10 replicas, that's +# ~130 concurrent connections × short intervals = serious load on +# HAProxy, Sentinel, and the Valkey master. +# +# Each $ZEROPS_Number replica gets its own DB and its own channel suffix +# (handled in main.go::applyZeropsOverrides), so the 10 containers don't +# step on each other's keyspace or pub/sub traffic. +# +# Topology: +# 6379 = read/write plaintext (master) +# 6380 = read/write TLS (master) +# 7000 = read-only plaintext (replicas via HAProxy) +# 7001 = read-only TLS (replicas via HAProxy) +# +# Roughly half the workers run over TLS so each ingress path gets +# exercised under load. +# +# Workload mix: +# - 1× readwriter (write + read-after-write on master) +# - 2× pure writer (separate TCP sessions — stress the master) +# - 1× master-side reader (does master serve GETs under load?) +# - 4× replica-side reader (stress replicas + verify replication catches up) +# - 2× publisher on two distinct channels (one slow, one tight) +# - 2× master-side subscriber (one per channel) +# - 1× replica-side subscriber (topology probe — should receive 0 msgs) + +workers: + # ---- write path (3 connections) ---------------------------------------- + - name: rw-main + role: readwriter + host: valkeyha.zerops + port: 6379 + db: 0 + interval: 20ms + + - name: rw-burst-1 + role: writer + host: valkeyha.zerops + port: 6380 + tls: true + interval: 3ms + + - name: rw-burst-2 + role: writer + host: valkeyha.zerops + port: 6379 + interval: 3ms + + # ---- master-side reader (sanity that master also serves reads) --------- + - name: ro-master + role: reader + host: valkeyha.zerops + port: 6380 + tls: true + interval: 5ms + + # ---- replica-side reader fan-out (4 connections) ----------------------- + - name: ro-1 + role: reader + host: valkeyha.zerops + port: 7000 + interval: 2ms + - name: ro-2 + role: reader + host: valkeyha.zerops + port: 7001 + tls: true + interval: 2ms + - name: ro-3 + role: reader + host: valkeyha.zerops + port: 7000 + interval: 3ms + - name: ro-4 + role: reader + host: valkeyha.zerops + port: 7001 + tls: true + interval: 3ms + + # ---- pub/sub: two channels at different cadences ----------------------- + - name: pub-bus + role: publisher + host: valkeyha.zerops + port: 6379 + channel: "vk:chaos:bus" + interval: 5ms + + - name: pub-fast + role: publisher + host: valkeyha.zerops + port: 6380 + tls: true + channel: "vk:chaos:fast" + interval: 2ms + + - name: sub-bus + role: subscriber + host: valkeyha.zerops + port: 6380 + tls: true + channel: "vk:chaos:bus" + + - name: sub-fast + role: subscriber + host: valkeyha.zerops + port: 6379 + channel: "vk:chaos:fast" + + # ---- memory bloater ---------------------------------------------------- + # Writes 8 KB of random bytes to a fresh key every 10ms with NO TTL. + # ~800 KB/sec per container × 10 containers = ~8 MB/sec growth on the + # master. Watch used_memory climb. Keys land under vk:chaos:bloat: + # so they don't pollute the durability ledger. To reset, FLUSHDB the + # affected DB (or each per-replica DB if running in Zerops). + # - name: bloat-1 + # role: bloater + # host: valkeyha.zerops + # port: 6379 + # interval: 10ms + # valueBytes: 8192 + + # ---- topology probe ---------------------------------------------------- + # Subscriber on the read-only VIP. Valkey 7.2 + Sentinel does not + # replicate pub/sub from master to replicas, so this should stay at + # msgs=0 the entire run. If it ever ticks up, replication semantics + # changed. + - name: sub-replica + role: subscriber + host: valkeyha.zerops + port: 7001 + tls: true + channel: "vk:chaos:bus" diff --git a/valkey/go.mod b/valkey/go.mod new file mode 100644 index 0000000..c71a613 --- /dev/null +++ b/valkey/go.mod @@ -0,0 +1,36 @@ +module valkey-test + +go 1.24.2 + +require ( + github.com/charmbracelet/bubbles v1.0.0 + github.com/charmbracelet/bubbletea v1.3.10 + github.com/charmbracelet/lipgloss v1.1.0 + github.com/redis/go-redis/v9 v9.7.0 + gopkg.in/yaml.v3 v3.0.1 +) + +require ( + github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect + github.com/cespare/xxhash/v2 v2.2.0 // indirect + github.com/charmbracelet/colorprofile v0.4.1 // indirect + github.com/charmbracelet/x/ansi v0.11.6 // indirect + github.com/charmbracelet/x/cellbuf v0.0.15 // indirect + github.com/charmbracelet/x/term v0.2.2 // indirect + github.com/clipperhouse/displaywidth v0.9.0 // indirect + github.com/clipperhouse/stringish v0.1.1 // indirect + github.com/clipperhouse/uax29/v2 v2.5.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect + github.com/lucasb-eyer/go-colorful v1.3.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/mattn/go-localereader v0.0.1 // indirect + github.com/mattn/go-runewidth v0.0.19 // indirect + github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 // indirect + github.com/muesli/cancelreader v0.2.2 // indirect + github.com/muesli/termenv v0.16.0 // indirect + github.com/rivo/uniseg v0.4.7 // indirect + github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect + golang.org/x/sys v0.38.0 // indirect + golang.org/x/text v0.3.8 // indirect +) diff --git a/valkey/go.sum b/valkey/go.sum new file mode 100644 index 0000000..93efbcc --- /dev/null +++ b/valkey/go.sum @@ -0,0 +1,64 @@ +github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k= +github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/charmbracelet/bubbles v1.0.0 h1:12J8/ak/uCZEMQ6KU7pcfwceyjLlWsDLAxB5fXonfvc= +github.com/charmbracelet/bubbles v1.0.0/go.mod h1:9d/Zd5GdnauMI5ivUIVisuEm3ave1XwXtD1ckyV6r3E= +github.com/charmbracelet/bubbletea v1.3.10 h1:otUDHWMMzQSB0Pkc87rm691KZ3SWa4KUlvF9nRvCICw= +github.com/charmbracelet/bubbletea v1.3.10/go.mod h1:ORQfo0fk8U+po9VaNvnV95UPWA1BitP1E0N6xJPlHr4= +github.com/charmbracelet/colorprofile v0.4.1 h1:a1lO03qTrSIRaK8c3JRxJDZOvhvIeSco3ej+ngLk1kk= +github.com/charmbracelet/colorprofile v0.4.1/go.mod h1:U1d9Dljmdf9DLegaJ0nGZNJvoXAhayhmidOdcBwAvKk= +github.com/charmbracelet/lipgloss v1.1.0 h1:vYXsiLHVkK7fp74RkV7b2kq9+zDLoEU4MZoFqR/noCY= +github.com/charmbracelet/lipgloss v1.1.0/go.mod h1:/6Q8FR2o+kj8rz4Dq0zQc3vYf7X+B0binUUBwA0aL30= +github.com/charmbracelet/x/ansi v0.11.6 h1:GhV21SiDz/45W9AnV2R61xZMRri5NlLnl6CVF7ihZW8= +github.com/charmbracelet/x/ansi v0.11.6/go.mod h1:2JNYLgQUsyqaiLovhU2Rv/pb8r6ydXKS3NIttu3VGZQ= +github.com/charmbracelet/x/cellbuf v0.0.15 h1:ur3pZy0o6z/R7EylET877CBxaiE1Sp1GMxoFPAIztPI= +github.com/charmbracelet/x/cellbuf v0.0.15/go.mod h1:J1YVbR7MUuEGIFPCaaZ96KDl5NoS0DAWkskup+mOY+Q= +github.com/charmbracelet/x/term v0.2.2 h1:xVRT/S2ZcKdhhOuSP4t5cLi5o+JxklsoEObBSgfgZRk= +github.com/charmbracelet/x/term v0.2.2/go.mod h1:kF8CY5RddLWrsgVwpw4kAa6TESp6EB5y3uxGLeCqzAI= +github.com/clipperhouse/displaywidth v0.9.0 h1:Qb4KOhYwRiN3viMv1v/3cTBlz3AcAZX3+y9OLhMtAtA= +github.com/clipperhouse/displaywidth v0.9.0/go.mod h1:aCAAqTlh4GIVkhQnJpbL0T/WfcrJXHcj8C0yjYcjOZA= +github.com/clipperhouse/stringish v0.1.1 h1:+NSqMOr3GR6k1FdRhhnXrLfztGzuG+VuFDfatpWHKCs= +github.com/clipperhouse/stringish v0.1.1/go.mod h1:v/WhFtE1q0ovMta2+m+UbpZ+2/HEXNWYXQgCt4hdOzA= +github.com/clipperhouse/uax29/v2 v2.5.0 h1:x7T0T4eTHDONxFJsL94uKNKPHrclyFI0lm7+w94cO8U= +github.com/clipperhouse/uax29/v2 v2.5.0/go.mod h1:Wn1g7MK6OoeDT0vL+Q0SQLDz/KpfsVRgg6W7ihQeh4g= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f h1:Y/CXytFA4m6baUTXGLOoWe4PQhGxaX0KpnayAqC48p4= +github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f/go.mod h1:vw97MGsxSvLiUE2X8qFplwetxpGLQrlU1Q9AUEIzCaM= +github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag= +github.com/lucasb-eyer/go-colorful v1.3.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-localereader v0.0.1 h1:ygSAOl7ZXTx4RdPYinUpg6W99U8jWvWi9Ye2JC/oIi4= +github.com/mattn/go-localereader v0.0.1/go.mod h1:8fBrzywKY7BI3czFoHkuzRoWE9C+EiG4R1k4Cjx5p88= +github.com/mattn/go-runewidth v0.0.19 h1:v++JhqYnZuu5jSKrk9RbgF5v4CGUjqRfBm05byFGLdw= +github.com/mattn/go-runewidth v0.0.19/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= +github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 h1:ZK8zHtRHOkbHy6Mmr5D264iyp3TiX5OmNcI5cIARiQI= +github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6/go.mod h1:CJlz5H+gyd6CUWT45Oy4q24RdLyn7Md9Vj2/ldJBSIo= +github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA= +github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo= +github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc= +github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= +github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= +github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= +github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= +github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= +github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no= +github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM= +golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI= +golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= +golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/valkey/main.go b/valkey/main.go new file mode 100644 index 0000000..993a8fc --- /dev/null +++ b/valkey/main.go @@ -0,0 +1,639 @@ +package main + +import ( + "context" + "crypto/tls" + "crypto/x509" + "flag" + "fmt" + "log" + "os" + "os/signal" + "strconv" + "syscall" + "time" + + "github.com/redis/go-redis/v9" + "gopkg.in/yaml.v3" +) + +// tlsRootCAs, when non-nil, is used as the RootCAs pool for every TLS +// connection. Set by --tls-ca for local testing against a Valkey whose +// cert is signed by a non-system CA. +var tlsRootCAs *x509.CertPool + +type WorkerConfig struct { + Name string `yaml:"name"` + Role string `yaml:"role"` + Host string `yaml:"host"` + Port int `yaml:"port"` + Password string `yaml:"password"` + DB int `yaml:"db"` + TLS bool `yaml:"tls"` + Interval time.Duration `yaml:"interval"` + Channel string `yaml:"channel"` + ValueBytes int `yaml:"valueBytes"` // bloater role only +} + +type Config struct { + Host string `yaml:"host,omitempty"` + Port int `yaml:"port,omitempty"` + Password string `yaml:"password,omitempty"` + DB int `yaml:"db,omitempty"` + TLS bool `yaml:"tls,omitempty"` + + Workers []WorkerConfig `yaml:"workers,omitempty"` +} + +func loadConfig(path string) (*Config, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read config: %w", err) + } + data = []byte(os.ExpandEnv(string(data))) + var cfg Config + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("parse config: %w", err) + } + applyZeropsOverrides(&cfg) + return &cfg, nil +} + +// applyZeropsOverrides applies env-var overrides on top of the parsed +// config so the same config.yaml works locally and in Zerops: +// +// $VALKEY_PASSWORD → overrides every worker's password (so the secret +// lives in the Zerops env, not in committed YAML). +// +// $ZEROPS_Number → ordinal injected per Zerops replica. Distributes +// 10 containers across distinct DBs and pub/sub +// channels so they don't collide on shared cluster: +// cfg.Workers[*].DB = ordinal mod 16 +// cfg.Workers[*].Channel = ":" +// The mod-16 wrap is needed because Valkey defaults +// to DB 0–15 and Zerops's ordinal grows across +// redeploys (a fresh deploy can start well above 16). +// +// Each override is independent: an unset env var is a no-op, leaving the +// YAML value in place. +func applyZeropsOverrides(cfg *Config) { + if pw := os.Getenv("VALKEY_PASSWORD"); pw != "" { + for i := range cfg.Workers { + cfg.Workers[i].Password = pw + } + } + if n := os.Getenv("ZEROPS_Number"); n != "" { + if raw, err := strconv.Atoi(n); err == nil { + dbN := raw % 16 + suffix := strconv.Itoa(dbN) + for i := range cfg.Workers { + cfg.Workers[i].DB = dbN + if cfg.Workers[i].Channel != "" { + cfg.Workers[i].Channel = cfg.Workers[i].Channel + ":" + suffix + } + } + } + } +} + +// applyFlagOverrides applies --host / --password / --tls-ca on top of the +// loaded config so the local binary can target a remote Valkey without +// editing config.yaml. --tls-ca loads the PEM into the package-level +// tlsRootCAs pool and forces TLS on every worker. +func applyFlagOverrides(cfg *Config, host string, port int, password string, tlsEnable bool, tlsCAPath string) error { + if host != "" { + cfg.Host = host + for i := range cfg.Workers { + cfg.Workers[i].Host = host + } + } + if port != 0 { + cfg.Port = port + for i := range cfg.Workers { + cfg.Workers[i].Port = port + } + } + if password != "" { + cfg.Password = password + for i := range cfg.Workers { + cfg.Workers[i].Password = password + } + } + if tlsCAPath != "" { + pem, err := os.ReadFile(tlsCAPath) + if err != nil { + return fmt.Errorf("read tls-ca: %w", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + return fmt.Errorf("tls-ca: no certificates parsed from %s", tlsCAPath) + } + tlsRootCAs = pool + } + if tlsEnable || tlsCAPath != "" { + cfg.TLS = true + for i := range cfg.Workers { + cfg.Workers[i].TLS = true + } + } + return nil +} + +func (c *Config) primary() (*redis.Options, error) { + if c.Host != "" && c.Port != 0 { + return optionsFromConn(c.Host, c.Port, c.Password, c.DB, c.TLS), nil + } + if len(c.Workers) > 0 { + w := c.Workers[0] + return optionsFromConn(w.Host, w.Port, w.Password, w.DB, w.TLS), nil + } + return nil, fmt.Errorf("config has neither top-level host nor workers") +} + +func optionsFromConn(host string, port int, password string, db int, useTLS bool) *redis.Options { + opts := &redis.Options{ + Addr: fmt.Sprintf("%s:%d", host, port), + Password: password, + DB: db, + } + if useTLS { + opts.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: tlsRootCAs} + } + return opts +} + +func redactPassword(pw string) string { + if pw == "" { + return "(none)" + } + return "(set)" +} + +func envInt(name string, def int) int { + if v := os.Getenv(name); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return def +} + +func main() { + configPath := flag.String("config", "config.yaml", "path to YAML config file") + continuous := flag.Bool("continuous", false, "single-worker rolling smoke loop") + chaosLocal := flag.Bool("chaos", false, "multi-worker chaos TUI (local, in-process)") + workerMode := flag.Bool("worker", false, "distributed worker — runs the chaos workload and ships events to --collector-url") + collectorMode := flag.Bool("collector", false, "distributed collector — HTTP dashboard that ingests events from workers") + collectorURL := flag.String("collector-url", os.Getenv("COLLECTOR_URL"), "collector base URL (worker mode); env: COLLECTOR_URL") + port := flag.Int("collector-port", envInt("PORT", 8080), "HTTP port (collector mode); env: PORT") + interval := flag.Duration("interval", time.Second, "tick interval for --continuous") + hostOverride := flag.String("host", "", "override host for all workers (local testing)") + portOverride := flag.Int("port", 0, "override port for all workers (local testing)") + passwordOverride := flag.String("password", "", "override password for all workers (local testing)") + tlsEnable := flag.Bool("tls", false, "enable TLS for all workers using system root CAs (local testing)") + tlsCAPath := flag.String("tls-ca", "", "path to TLS CA cert PEM; enables TLS for all workers (local testing)") + flag.Parse() + + switch { + case *collectorMode: + os.Exit(runCollector(*port)) + } + + cfg, err := loadConfig(*configPath) + if err != nil { + log.Fatalf("config error: %v", err) + } + if err := applyFlagOverrides(cfg, *hostOverride, *portOverride, *passwordOverride, *tlsEnable, *tlsCAPath); err != nil { + log.Fatalf("flag override error: %v", err) + } + + switch { + case *chaosLocal: + if len(cfg.Workers) == 0 { + log.Fatalf("--chaos requires workers: list in %s", *configPath) + } + os.Exit(runChaos(cfg.Workers)) + case *workerMode: + if len(cfg.Workers) == 0 { + log.Fatalf("--worker requires workers: list in %s", *configPath) + } + if *collectorURL == "" { + log.Fatalf("--worker requires --collector-url or $COLLECTOR_URL") + } + os.Exit(runWorker(cfg.Workers, *collectorURL)) + } + + primary, err := cfg.primary() + if err != nil { + log.Fatalf("config error: %v", err) + } + fmt.Printf("redis config: addr=%s db=%d tls=%t password=%s\n", + primary.Addr, primary.DB, primary.TLSConfig != nil, redactPassword(primary.Password)) + rdb := redis.NewClient(primary) + defer rdb.Close() + + if *continuous { + runContinuous(rdb, *interval) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + tests := []struct { + name string + fn func(context.Context, *redis.Client) error + }{ + {"PING", testPing}, + {"INFO version", testInfo}, + {"STRING SET/GET", testString}, + {"INCR/DECR", testCounter}, + {"EXPIRE/TTL", testExpire}, + {"LIST ops", testList}, + {"HASH ops", testHash}, + {"SET ops", testSet}, + {"SORTED SET ops", testZSet}, + {"TRANSACTION", testTxn}, + {"PIPELINE", testPipeline}, + {"PUB/SUB", testPubSub}, + {"FLUSHDB cleanup", testCleanup}, + } + + failed := 0 + for _, t := range tests { + fmt.Printf("[ .. ] %s\n", t.name) + if err := t.fn(ctx, rdb); err != nil { + fmt.Printf("[FAIL] %s: %v\n", t.name, err) + failed++ + continue + } + fmt.Printf("[ OK ] %s\n", t.name) + } + + fmt.Printf("\n%d/%d tests passed\n", len(tests)-failed, len(tests)) + if failed > 0 { + os.Exit(1) + } +} + +func testPing(ctx context.Context, r *redis.Client) error { + pong, err := r.Ping(ctx).Result() + if err != nil { + return err + } + if pong != "PONG" { + return fmt.Errorf("expected PONG, got %q", pong) + } + return nil +} + +func testInfo(ctx context.Context, r *redis.Client) error { + info, err := r.Info(ctx, "server").Result() + if err != nil { + return err + } + if len(info) == 0 { + return fmt.Errorf("empty INFO response") + } + return nil +} + +func testString(ctx context.Context, r *redis.Client) error { + if err := r.Set(ctx, "vk:string", "hello-valkey", 0).Err(); err != nil { + return err + } + got, err := r.Get(ctx, "vk:string").Result() + if err != nil { + return err + } + if got != "hello-valkey" { + return fmt.Errorf("expected hello-valkey, got %q", got) + } + return nil +} + +func testCounter(ctx context.Context, r *redis.Client) error { + r.Del(ctx, "vk:counter") + if _, err := r.Incr(ctx, "vk:counter").Result(); err != nil { + return err + } + if _, err := r.IncrBy(ctx, "vk:counter", 9).Result(); err != nil { + return err + } + v, err := r.Decr(ctx, "vk:counter").Result() + if err != nil { + return err + } + if v != 9 { + return fmt.Errorf("expected 9, got %d", v) + } + return nil +} + +func testExpire(ctx context.Context, r *redis.Client) error { + if err := r.Set(ctx, "vk:ttl", "x", 5*time.Second).Err(); err != nil { + return err + } + ttl, err := r.TTL(ctx, "vk:ttl").Result() + if err != nil { + return err + } + if ttl <= 0 || ttl > 5*time.Second { + return fmt.Errorf("unexpected TTL: %v", ttl) + } + return nil +} + +func testList(ctx context.Context, r *redis.Client) error { + r.Del(ctx, "vk:list") + if _, err := r.RPush(ctx, "vk:list", "a", "b", "c").Result(); err != nil { + return err + } + vals, err := r.LRange(ctx, "vk:list", 0, -1).Result() + if err != nil { + return err + } + if len(vals) != 3 || vals[0] != "a" || vals[2] != "c" { + return fmt.Errorf("unexpected list: %v", vals) + } + return nil +} + +func testHash(ctx context.Context, r *redis.Client) error { + r.Del(ctx, "vk:hash") + if err := r.HSet(ctx, "vk:hash", "name", "valkey", "version", "7.2").Err(); err != nil { + return err + } + got, err := r.HGetAll(ctx, "vk:hash").Result() + if err != nil { + return err + } + if got["name"] != "valkey" || got["version"] != "7.2" { + return fmt.Errorf("unexpected hash: %v", got) + } + return nil +} + +func testSet(ctx context.Context, r *redis.Client) error { + r.Del(ctx, "vk:set") + if _, err := r.SAdd(ctx, "vk:set", "x", "y", "z", "x").Result(); err != nil { + return err + } + n, err := r.SCard(ctx, "vk:set").Result() + if err != nil { + return err + } + if n != 3 { + return fmt.Errorf("expected 3 members, got %d", n) + } + return nil +} + +func testZSet(ctx context.Context, r *redis.Client) error { + r.Del(ctx, "vk:zset") + _, err := r.ZAdd(ctx, "vk:zset", + redis.Z{Score: 1, Member: "one"}, + redis.Z{Score: 2, Member: "two"}, + redis.Z{Score: 3, Member: "three"}, + ).Result() + if err != nil { + return err + } + vals, err := r.ZRangeByScore(ctx, "vk:zset", &redis.ZRangeBy{Min: "1", Max: "2"}).Result() + if err != nil { + return err + } + if len(vals) != 2 || vals[0] != "one" { + return fmt.Errorf("unexpected zrange: %v", vals) + } + return nil +} + +func testTxn(ctx context.Context, r *redis.Client) error { + r.Del(ctx, "vk:txn") + pipe := r.TxPipeline() + pipe.Set(ctx, "vk:txn", "1", 0) + pipe.Incr(ctx, "vk:txn") + pipe.Incr(ctx, "vk:txn") + if _, err := pipe.Exec(ctx); err != nil { + return err + } + v, err := r.Get(ctx, "vk:txn").Result() + if err != nil { + return err + } + if v != "3" { + return fmt.Errorf("expected 3, got %s", v) + } + return nil +} + +func testPipeline(ctx context.Context, r *redis.Client) error { + pipe := r.Pipeline() + for i := 0; i < 5; i++ { + pipe.Set(ctx, fmt.Sprintf("vk:pipe:%d", i), i, 0) + } + if _, err := pipe.Exec(ctx); err != nil { + return err + } + v, err := r.Get(ctx, "vk:pipe:3").Result() + if err != nil { + return err + } + if v != "3" { + return fmt.Errorf("expected 3, got %s", v) + } + return nil +} + +func testPubSub(ctx context.Context, r *redis.Client) error { + sub := r.Subscribe(ctx, "vk:channel") + defer sub.Close() + + if _, err := sub.Receive(ctx); err != nil { + return fmt.Errorf("subscribe: %w", err) + } + ch := sub.Channel() + + if err := r.Publish(ctx, "vk:channel", "ping").Err(); err != nil { + return err + } + + select { + case msg := <-ch: + if msg.Payload != "ping" { + return fmt.Errorf("expected ping, got %q", msg.Payload) + } + case <-time.After(3 * time.Second): + return fmt.Errorf("timeout waiting for message") + } + return nil +} + +func testCleanup(ctx context.Context, r *redis.Client) error { + keys, err := r.Keys(ctx, "vk:*").Result() + if err != nil { + return err + } + if len(keys) == 0 { + return nil + } + return r.Del(ctx, keys...).Err() +} + +func runContinuous(r *redis.Client, interval time.Duration) { + stopCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + subCtx, cancelSub := context.WithCancel(stopCtx) + defer cancelSub() + sub := r.Subscribe(subCtx, "vk:cont:channel") + defer sub.Close() + if _, err := sub.Receive(subCtx); err != nil { + log.Fatalf("subscribe: %v", err) + } + subCh := sub.Channel() + + ops := []struct { + name string + fn func(context.Context, *redis.Client, int64) error + }{ + {"PING", opPing}, + {"SET/GET", opSetGet}, + {"INCR", opIncr}, + {"LIST", opList}, + {"INFO", opInfo}, + {"PUB/SUB", func(ctx context.Context, r *redis.Client, i int64) error { + return opPubSub(ctx, r, i, subCh) + }}, + } + + var cycle, okCount, failCount int64 + ticker := time.NewTicker(interval) + defer ticker.Stop() + + fmt.Printf("continuous mode: addr=%s interval=%s (Ctrl+C to stop)\n", r.Options().Addr, interval) + + for { + select { + case <-stopCtx.Done(): + fmt.Printf("\nstopped: cycles=%d ok=%d failed=%d\n", cycle, okCount, failCount) + return + case <-ticker.C: + } + cycle++ + + results := make([]string, len(ops)) + var failures []string + cycleStart := time.Now() + for i, op := range ops { + ctx, cancel := context.WithTimeout(stopCtx, interval) + start := time.Now() + err := op.fn(ctx, r, cycle) + cancel() + dur := time.Since(start) + if err != nil { + results[i] = fmt.Sprintf("%s=FAIL", op.name) + failures = append(failures, fmt.Sprintf(" %s (%s): %v", op.name, dur, err)) + } else { + results[i] = fmt.Sprintf("%s=ok(%s)", op.name, dur.Round(time.Microsecond)) + } + } + + if len(failures) == 0 { + okCount++ + } else { + failCount++ + } + + fmt.Printf("[%s] #%d %s | total=%s\n", + time.Now().Format("15:04:05"), cycle, joinResults(results), time.Since(cycleStart).Round(time.Microsecond)) + for _, f := range failures { + fmt.Println(f) + } + } +} + +func joinResults(parts []string) string { + out := "" + for i, p := range parts { + if i > 0 { + out += " " + } + out += p + } + return out +} + +func opPing(ctx context.Context, r *redis.Client, _ int64) error { + pong, err := r.Ping(ctx).Result() + if err != nil { + return err + } + if pong != "PONG" { + return fmt.Errorf("expected PONG, got %q", pong) + } + return nil +} + +func opSetGet(ctx context.Context, r *redis.Client, i int64) error { + val := fmt.Sprintf("v-%d", i) + if err := r.Set(ctx, "vk:cont:str", val, 10*time.Second).Err(); err != nil { + return err + } + got, err := r.Get(ctx, "vk:cont:str").Result() + if err != nil { + return err + } + if got != val { + return fmt.Errorf("set %q, got %q", val, got) + } + return nil +} + +func opIncr(ctx context.Context, r *redis.Client, _ int64) error { + _, err := r.Incr(ctx, "vk:cont:counter").Result() + return err +} + +func opList(ctx context.Context, r *redis.Client, i int64) error { + if err := r.RPush(ctx, "vk:cont:list", fmt.Sprintf("item-%d", i)).Err(); err != nil { + return err + } + if err := r.LTrim(ctx, "vk:cont:list", -10, -1).Err(); err != nil { + return err + } + _, err := r.LRange(ctx, "vk:cont:list", 0, -1).Result() + return err +} + +func opInfo(ctx context.Context, r *redis.Client, _ int64) error { + info, err := r.Info(ctx, "server").Result() + if err != nil { + return err + } + if len(info) == 0 { + return fmt.Errorf("empty INFO") + } + return nil +} + +func opPubSub(ctx context.Context, r *redis.Client, i int64, ch <-chan *redis.Message) error { + payload := fmt.Sprintf("ping-%d", i) + if err := r.Publish(ctx, "vk:cont:channel", payload).Err(); err != nil { + return err + } + for { + select { + case msg := <-ch: + if msg == nil { + return fmt.Errorf("channel closed") + } + if msg.Payload == payload { + return nil + } + case <-ctx.Done(): + return fmt.Errorf("timeout waiting for %q", payload) + } + } +} diff --git a/valkey/test-all.sh b/valkey/test-all.sh new file mode 100755 index 0000000..2167bb4 --- /dev/null +++ b/valkey/test-all.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +set -euo pipefail + +PORT="${PORT:-6379}" + +services=$(zcli service list 2>/dev/null \ + | awk -F'│' '{gsub(/ /, "", $3); print $3}' \ + | grep -E '^valkey' || true) + +if [[ -z "$services" ]]; then + echo "no valkey* services found" >&2 + exit 1 +fi + +fail=0 +while IFS= read -r svc; do + echo "=== testing $svc ===" + password=$(zcli project env --service "$svc" 2>/dev/null \ + | grep -E '^password=' \ + | head -n1 \ + | sed -E 's/^password="?([^"]*)"?$/\1/') + if [[ -z "$password" ]]; then + echo "!!! $svc: could not fetch password" >&2 + fail=1 + continue + fi + if ! go run ./... -password "$password" -host "${svc}.zerops" -port "$PORT"; then + echo "!!! $svc FAILED" >&2 + fail=1 + fi +done <<< "$services" + +exit "$fail" diff --git a/valkey/zerops-basic-import.yaml b/valkey/zerops-basic-import.yaml new file mode 100644 index 0000000..4104e78 --- /dev/null +++ b/valkey/zerops-basic-import.yaml @@ -0,0 +1,26 @@ +# Basic stress import: one chaosbasic container that hits all 6 valkeys +# in the project with a single readwriter each, plaintext :6379, AND +# serves the collector dashboard on :8080 (subdomain access on). +# +# Build/run config (incl. the startCommands that boot both the collector +# and the worker in the same container) lives in zerops.yaml under +# setup: chaosbasic. +# +# Use with: zcli project service-import zerops-basic-import.yaml +# Then: zcli push --setup chaosbasic +# +# Prereqs in the same project (e.g. via zerops-mega-import.yaml): +# valkeysingle1 / valkeysingle2 / valkeysingle3 (valkey:single@7.2) +# valkeyha1 / valkeyha2 / valkeyha3 (valkey:ha@7.2) +# +# NB — config-basic.yaml uses literal ${VALKEY_..._PASSWORD} strings in +# each worker's password field. applyZeropsOverrides today only honors +# a single $VALKEY_PASSWORD env var; extend it to expand ${VAR} +# references in each worker.Password against os.Getenv at load time +# before this will authenticate against the per-host passwords below. + +services: + - hostname: chaosbasic + type: go@1 + minContainers: 1 + maxContainers: 1 diff --git a/valkey/zerops-chaos-import.yaml b/valkey/zerops-chaos-import.yaml new file mode 100644 index 0000000..1c55614 --- /dev/null +++ b/valkey/zerops-chaos-import.yaml @@ -0,0 +1,22 @@ +# Full project import for the chaos test infra: +# - Valkey HA cluster (the system under test) +# - chaosworker (10x runner replicas; user-defined Go service) +# - chaoscollector (1x dashboard; user-defined Go service, subdomain on) +# +# Use with: zcli project project-import zerops-chaos-import.yaml +# +# After import, push the runtime config from valkey/zerops.yaml to deploy +# the app code to chaosworker and chaoscollector. +# +# Hostnames are alphanumeric to match Zerops conventions. +services: + - hostname: chaoscollector + type: go@1 + enableSubdomainAccess: true + minContainers: 1 + maxContainers: 1 + + - hostname: chaosworker + type: go@1 + minContainers: 10 + maxContainers: 10 diff --git a/valkey/zerops-ha-import.yaml b/valkey/zerops-ha-import.yaml new file mode 100644 index 0000000..4f577ae --- /dev/null +++ b/valkey/zerops-ha-import.yaml @@ -0,0 +1,7 @@ +services: + - hostname: valkeyha + type: valkey@7.2 + mode: HA + verticalAutoscaling: + minFreeRamPercent: 40 + minRam: 4 diff --git a/valkey/zerops-mega-import.yaml b/valkey/zerops-mega-import.yaml new file mode 100644 index 0000000..93eef3f --- /dev/null +++ b/valkey/zerops-mega-import.yaml @@ -0,0 +1,13 @@ +services: + - hostname: valkeysingle1 + type: valkey:single@7.2 + - hostname: valkeysingle2 + type: valkey:single@7.2 + - hostname: valkeysingle3 + type: valkey:single@7.2 + - hostname: valkeyha1 + type: valkey:ha@7.2 + - hostname: valkeyha2 + type: valkey:ha@7.2 + - hostname: valkeyha3 + type: valkey:ha@7.2 diff --git a/valkey/zerops-single-import.yaml b/valkey/zerops-single-import.yaml new file mode 100644 index 0000000..0ac2cd4 --- /dev/null +++ b/valkey/zerops-single-import.yaml @@ -0,0 +1,4 @@ +services: + - hostname: valkeysingle + type: valkey@7.2 + mode: NON_HA diff --git a/valkey/zerops.yaml b/valkey/zerops.yaml new file mode 100644 index 0000000..f198cad --- /dev/null +++ b/valkey/zerops.yaml @@ -0,0 +1,76 @@ +# zerops.yaml — runtime build/run config for the chaos test services. +# +# Two setups, both built from the same Go source under valkey/. +# +# chaosworker 10 replicas, ships events to chaoscollector (no exposed port). +# Reads $ZEROPS_Number → maps to Redis DB N and pub/sub +# channel suffix ":N" so the 10 containers don't collide +# in the keyspace or on the same channel. +# +# chaoscollector 1 replica, exposes :8080 with subdomain access enabled, +# ingests events from the workers and serves the dashboard. +# +# NB: validate the exact schema (field names, autoscaling block, prepare +# commands) against the current Zerops docs before pushing. Field names +# below follow the typical zerops.yaml conventions but vary by runtime +# version. + +zerops: + - setup: chaosworker + build: + base: ubuntu/go@1 + buildCommands: + - go build -o app . + deployFiles: + - app + - config.yaml + cache: true + run: + base: ubuntu@latest + start: ./app --worker --config config.yaml + envVariables: + COLLECTOR_URL: http://chaoscollector.zerops:8080 + VALKEY_PASSWORD: ${valkeyha_password} + + - setup: chaoscollector + build: + base: ubuntu/go@1 + buildCommands: + - go build -o app . + deployFiles: + - app + cache: true + run: + base: ubuntu@latest + ports: + - port: 8080 + httpSupport: true + start: ./app --collector --collector-port 8080 + + - setup: chaosbasic + build: + base: ubuntu/go@1 + buildCommands: + - go build -o app . + deployFiles: + - app + - config-basic.yaml + cache: true + run: + base: ubuntu@latest + ports: + - port: 8080 + httpSupport: true + startCommands: + - name: collector + command: ./app --collector --collector-port 8080 + - name: worker + command: ./app --worker --config config-basic.yaml + envVariables: + COLLECTOR_URL: http://localhost:8080 + VALKEY_VALKEYSINGLE1_PASSWORD: ${valkeysingle1_password} + VALKEY_VALKEYSINGLE2_PASSWORD: ${valkeysingle2_password} + VALKEY_VALKEYSINGLE3_PASSWORD: ${valkeysingle3_password} + VALKEY_VALKEYHA1_PASSWORD: ${valkeyha1_password} + VALKEY_VALKEYHA2_PASSWORD: ${valkeyha2_password} + VALKEY_VALKEYHA3_PASSWORD: ${valkeyha3_password}