diff --git a/.gitignore b/.gitignore index 00703d4369482..75ca7aaa234e1 100644 --- a/.gitignore +++ b/.gitignore @@ -294,3 +294,4 @@ src/go/rpk/cmd/rpk/rpk # antithesis /.antithesis +tests/antithesis/ct_stress/workload/workload diff --git a/src/v/cluster/rm_stm.cc b/src/v/cluster/rm_stm.cc index ec6a985b2c675..eab83cb631123 100644 --- a/src/v/cluster/rm_stm.cc +++ b/src/v/cluster/rm_stm.cc @@ -2124,17 +2124,17 @@ ss::future rm_stm::do_take_local_snapshot( } stm_snapshot.abort_indexes = final_abort_indexes.copy(); - kafka::offset start_kafka_offset = from_log_offset(start_offset); + // kafka::offset start_kafka_offset = from_log_offset(start_offset); stm_snapshot.highest_producer_id = _highest_producer_id; // producers state (includes idempotent and transactional producers) for (const auto& [_, state] : _producers) { auto snap = state->snapshot(); // Discard finished requests below the local snapshot start offset, // they are no longer relevant. - std::erase_if( - snap.finished_requests, [start_kafka_offset](const auto& req) { - return req.last_offset < start_kafka_offset; - }); + // std::erase_if( + // snap.finished_requests, [start_kafka_offset](const auto& req) { + // return req.last_offset < start_kafka_offset; + // }); if ( !snap.finished_requests.empty() || state->has_transaction_in_progress()) { diff --git a/tests/antithesis/ct_stress/docker-compose.yaml.j2 b/tests/antithesis/ct_stress/docker-compose.yaml.j2 new file mode 100644 index 0000000000000..0594b8dace6b4 --- /dev/null +++ b/tests/antithesis/ct_stress/docker-compose.yaml.j2 @@ -0,0 +1,109 @@ +name: {{ test_slug }} + +services: + minio: + image: {{ minio_image }} + container_name: minio + hostname: minio + platform: linux/amd64 + init: true + restart: "no" + command: server /data + environment: + MINIO_REGION_NAME: {{ s3_region }} + MINIO_ROOT_USER: {{ s3_access_key }} + MINIO_ROOT_PASSWORD: {{ s3_secret_key }} + MINIO_BROWSER: "off" + expose: + - "9000" + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] + interval: 5s + timeout: 5s + retries: 60 + + minio-init: + image: {{ minio_image }} + container_name: minio-init + hostname: minio-init + platform: linux/amd64 + init: true + restart: "no" + entrypoint: ["/bin/bash", "/scripts/minio-init.sh"] + environment: + MINIO_URL: http://minio:9000 + BUCKET: {{ bucket }} + ACCESS_KEY: {{ s3_access_key }} + SECRET_KEY: {{ s3_secret_key }} + volumes: + - ./scripts:/scripts:ro + depends_on: + minio: + condition: service_healthy +{% for i in range(nodes) %} + + redpanda-{{ i }}: + image: {{ redpanda_image }} + container_name: redpanda-{{ i }} + hostname: redpanda-{{ i }} + platform: linux/amd64 + init: true + # No restart policy: docker-compose restarting a killed broker would + # nullify Antithesis node-termination faults. Antithesis restarts the + # container itself after the fault window. See + # https://antithesis.com/docs/concepts/fault_injection/fault_types/#node-termination + restart: "no" + entrypoint: ["/scripts/redpanda-entrypoint.sh"] + command: + - redpanda + - start + # Pin node_id to the container index so redpanda-N is node N. Auto-assigned + # ids follow registration order, not container names, which makes log + # correlation error-prone. + - --node-id={{ i }} + - --smp={{ smp }} + - --memory={{ memory }} + - --reserve-memory=0M + - --overprovisioned + - --kafka-addr=PLAINTEXT://0.0.0.0:9092 + - --advertise-kafka-addr=PLAINTEXT://redpanda-{{ i }}:9092 + - --rpc-addr=0.0.0.0:33145 + - --advertise-rpc-addr=redpanda-{{ i }}:33145 +{% if i > 0 %} + - --seeds=redpanda-0:33145 +{% endif %} + volumes: + - ./scripts:/scripts:ro + ulimits: + nofile: + soft: 131072 + hard: 131072 + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9644/v1/status/ready"] + interval: 5s + timeout: 5s + retries: 120 + depends_on: + minio-init: + condition: service_completed_successfully +{% endfor %} + + # franz-go workload. Its entrypoint (workload setup) waits for the cluster, + # emits setup_complete, then idles so Antithesis runs the baked-in test + # commands (/opt/antithesis/test/v1/main/) in this container. + workload: + image: {{ workload_image }} + container_name: workload + hostname: workload + platform: linux/amd64 + init: true + restart: "no" + environment: + KAFKA_BROKERS: "{% for i in range(nodes) %}{% if not loop.first %},{% endif %}redpanda-{{ i }}:9092{% endfor %}" + ADMIN_HOSTS: "{% for i in range(nodes) %}{% if not loop.first %},{% endif %}redpanda-{{ i }}:9644{% endfor %}" + EXPECTED_BROKERS: "{{ nodes }}" + depends_on: +{% for i in range(nodes) %} + redpanda-{{ i }}: + condition: service_healthy +{% endfor %} diff --git a/tests/antithesis/ct_stress/scripts/bootstrap.yaml.j2 b/tests/antithesis/ct_stress/scripts/bootstrap.yaml.j2 new file mode 100644 index 0000000000000..593812aba4f86 --- /dev/null +++ b/tests/antithesis/ct_stress/scripts/bootstrap.yaml.j2 @@ -0,0 +1,39 @@ +# Initial cluster configuration applied on first cluster bootstrap (staged to +# /etc/redpanda/.bootstrap.yaml on every node by redpanda-entrypoint.sh). +cloud_storage_enabled: true +cloud_storage_access_key: {{ s3_access_key }} +cloud_storage_secret_key: {{ s3_secret_key }} +cloud_storage_region: {{ s3_region }} +cloud_storage_api_endpoint: minio +cloud_storage_api_endpoint_port: 9000 +cloud_storage_disable_tls: true +cloud_storage_url_style: path +cloud_storage_bucket: {{ bucket }} + +auto_create_topics_enabled: false + +default_redpanda_storage_mode_tiered_impl: tiered_v2 + +# One metastore/domain partition (kafka_internal/ct_l1_domain) to have +# more concentrated activity. Since they don't interact between each other, +# more partitions don't add more coverage. +cloud_topics_num_metastore_partitions: 1 + +# Shrink cloud-topics object sizes so a light workload still produces many +# L0/L1 objects and dense indexes. This drives reconciliation, leveling and +# the offset-mapping paths across many object boundaries in a short run, +# where the checker's ordering/contiguity assertions do their work. +cloud_topics_produce_batching_size_threshold: 65536 # 64 KiB (default 4 MiB) +cloud_topics_reconciliation_max_object_size: 1048576 # 1 MiB (default 80 MiB) +# NB: config key is cloud_topics_indexing_interval (not ..._l1_...). +cloud_topics_indexing_interval: 16384 # 16 KiB (default 4 MiB) + +# Compact cloud topics more often than the 30s default so the compacted +# topic (ctc) goes through many compaction rounds in a short run. +cloud_topics_compaction_interval_ms: 10000 + +# Smallest metastore LSM memtable and SST read chunk the config allows +# (both floor at 1 MiB), to get as much SST churn / chunked object-storage +# reads as the low metastore load permits. +cloud_topics_metastore_write_buffer_size: 1048576 # 1 MiB (min; default 16 MiB) +cloud_topics_metastore_sst_chunk_size: 1048576 # 1 MiB (min; default 24 MiB) diff --git a/tests/antithesis/ct_stress/scripts/minio-init.sh b/tests/antithesis/ct_stress/scripts/minio-init.sh new file mode 100755 index 0000000000000..e6333734f1b49 --- /dev/null +++ b/tests/antithesis/ct_stress/scripts/minio-init.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash +# Copyright 2026 Redpanda Data, Inc. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.md +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0 +# ================================================================== +# +# One-shot init: creates the object storage bucket Redpanda is +# configured with. Services wait on it via +# depends_on: service_completed_successfully. + +set -euo pipefail + +: "${MINIO_URL:?}" +: "${BUCKET:?}" +: "${ACCESS_KEY:?}" +: "${SECRET_KEY:?}" + +mc alias set local "$MINIO_URL" "$ACCESS_KEY" "$SECRET_KEY" +mc mb --ignore-existing "local/$BUCKET" + +echo "bucket $BUCKET ready" diff --git a/tests/antithesis/ct_stress/scripts/redpanda-entrypoint.sh b/tests/antithesis/ct_stress/scripts/redpanda-entrypoint.sh new file mode 100755 index 0000000000000..afa1ca1a518ac --- /dev/null +++ b/tests/antithesis/ct_stress/scripts/redpanda-entrypoint.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +# Copyright 2026 Redpanda Data, Inc. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.md +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0 +# ================================================================== + +set -euo pipefail + +# Copy the bootstrap config off its read-only bind mount onto the container's +# writable layer, then start Redpanda. Under Antithesis a bind mount is backed +# by overlay-over-squashfs, which mishandles O_DIRECT (close() returns EINVAL); +# Redpanda opens config files with O_DIRECT and would crash reading it there. +# The writable layer handles O_DIRECT, so a plain (buffered) copy sidesteps it. +# See seastar issue 3518. +cp /scripts/bootstrap.yaml /etc/redpanda/.bootstrap.yaml + +exec rpk "$@" diff --git a/tests/antithesis/ct_stress/test.yml b/tests/antithesis/ct_stress/test.yml new file mode 100644 index 0000000000000..cd09b04db066f --- /dev/null +++ b/tests/antithesis/ct_stress/test.yml @@ -0,0 +1,56 @@ +# Antithesis test manifest, consumed by tools/antithesis/template_test_package.py. +# +# vars provides the default values for the Jinja2 templates (*.j2) in this +# directory; override individual values at packaging time with --var KEY=VALUE. +# +# images declares companion images built from this test's own sources. Each +# is built, tagged with a hash of its build context, exposed to the templates +# as the named variable, and (for real runs) pushed and registered with +# Antithesis so libvoidstar is injected into it. +name: cloud topics stress testing + +images: + workload_image: + context: workload + +# Antithesis fault scoping, applied at launch (--submit). List entries name +# container roles: `redpanda` expands to every broker (redpanda-0..nodes-1), +# other entries are literal container names. Every entry is sent explicitly +# (empty lists as ""), so this manifest -- not the webhook defaults -- is the +# source of truth for the fault posture. +faults: + # Global toggles (pinned here so the fault posture is self-documenting). + # clock_skew: on by the webhook default anyway. + # cpu_mod: simulates different hardware by varying CPU/instruction speed, + # reordering concurrent execution to surface races. Runs in addition to the + # thread pausing induced by the Antithesis instrumentation in our + # --config=antithesis build. Not recorded in the fault log. + clock_skew: true + cpu_mod: true + node_hang: [redpanda, workload] + node_throttle: [redpanda, workload] + # Explicitly empty: no containers included, so node termination is off. Enable + # it by adding brokers -- never minio (single instance, no S3 durability). + node_termination: [] + # Containers excluded from network faults. Explicitly empty: nothing excluded, + # so every container -- minio too -- sees network faults, exercising + # object-storage disruption. + exclude_from_faults: [] + +vars: + # Any image with the stock rpk entrypoint works here. Override with an + # instrumented build for real Antithesis runs (--redpanda-image). + redpanda_image: docker.redpanda.com/redpandadata/redpanda-nightly:latest + minio_image: minio/minio:RELEASE.2025-07-23T15-54-02Z + nodes: 5 + # 2 shards (threads) per broker. Memory must fit Antithesis' ~10G total: + # 5 brokers x 1.2G = 6G leaves room for MinIO, the workload, the OS page + # cache and hypervisor overhead. + # This is below Redpanda's 1 GiB/shard floor (unavoidable: 5x2 shards at + # 1 GiB would need the whole 10G), so expect a cosmetic memory warning. + smp: 2 + memory: 1200M + bucket: ct-stress + s3_access_key: panda-user + s3_secret_key: panda-secret + s3_region: panda-region diff --git a/tests/antithesis/ct_stress/workload/Dockerfile b/tests/antithesis/ct_stress/workload/Dockerfile new file mode 100644 index 0000000000000..8d679118219f5 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/Dockerfile @@ -0,0 +1,50 @@ +# syntax=docker/dockerfile:1 +# +# ================================================================== +# Copyright 2026 Redpanda Data, Inc. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.md +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0 +# ================================================================== +# +# ct_stress Antithesis workload image. Builds the franz-go workload with +# Antithesis assertion cataloging, then ships it on a glibc base so the SDK +# can dlopen /usr/lib/libvoidstar.so (injected by Antithesis) at runtime. +# CGO must stay enabled for that path; without libvoidstar the SDK falls +# back to local behavior, so the same image also runs locally. + +FROM golang:1.26-bookworm AS build +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . +# Generate the assertion catalog (antithesis_catalog.go) so every assertion +# is pre-registered with Antithesis, then build with cgo for libvoidstar. +RUN go run github.com/antithesishq/antithesis-sdk-go/tools/antithesis-go-instrumentor@v0.7.2 \ + -assert_only -skip_test_files . \ + && CGO_ENABLED=1 go build -o /out/workload . + +FROM debian:bookworm-slim +ENV NO_COLOR=1 + +# The binary lives inside the test template as a helper_ file (Antithesis +# ignores helper_-prefixed entries) and each command is a RELATIVE symlink to +# it. Relative, in-directory links stay valid when the template is copied out +# of the container (e.g. by `snouty validate` or Antithesis discovery); an +# absolute symlink to a path outside the template would dangle. The binary +# reports its own commands, so the list lives only in the Go source. +COPY --from=build /out/workload /opt/antithesis/test/v1/main/helper_workload +RUN cd /opt/antithesis/test/v1/main \ + && for c in $(./helper_workload list-commands); do \ + ln -s helper_workload "$c"; \ + done + +# The container idles here (setup waits for health, emits setup_complete, +# then blocks) so Antithesis can execute the test commands inside it. +ENTRYPOINT ["/opt/antithesis/test/v1/main/helper_workload", "setup"] diff --git a/tests/antithesis/ct_stress/workload/check.go b/tests/antithesis/ct_stress/workload/check.go new file mode 100644 index 0000000000000..8c531b08aaaed --- /dev/null +++ b/tests/antithesis/ct_stress/workload/check.go @@ -0,0 +1,343 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/antithesishq/antithesis-sdk-go/assert" + "github.com/twmb/franz-go/pkg/kadm" + "github.com/twmb/franz-go/pkg/kgo" +) + +// partitionBounds returns a partition's current start and high-watermark +// offsets. Records occupy [lo, hi); the partition is empty when hi <= lo. +func partitionBounds(t string, part int32) (lo, hi int64, err error) { + cl, err := newClient() + if err != nil { + return 0, 0, err + } + defer cl.Close() + adm := kadm.NewClient(cl) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + starts, err := adm.ListStartOffsets(ctx, t) + if err != nil { + return 0, 0, err + } + ends, err := adm.ListEndOffsets(ctx, t) + if err != nil { + return 0, 0, err + } + + s, ok := starts.Lookup(t, part) + if !ok || s.Err != nil { + return 0, 0, s.Err + } + e, ok := ends.Lookup(t, part) + if !ok || e.Err != nil { + return 0, 0, e.Err + } + return s.Offset, e.Offset, nil +} + +// randomFooPartitionRange picks a random foo partition and returns its current +// start and high-watermark offsets. +func randomFooPartitionRange() (part int32, lo, hi int64, err error) { + part = int32(randN(fooPartitions)) + lo, hi, err = partitionBounds(fooTopic, part) + return part, lo, hi, err +} + +// readRange consumes offsets [o1, o2) from a single partition and returns the +// records in the order the broker served them. Bounded by a timeout so a fault +// that stalls the range cannot hang the command. +func readRange(t string, part int32, o1, o2 int64) ([]*kgo.Record, error) { + cl, err := newClient(kgo.ConsumePartitions(map[string]map[int32]kgo.Offset{ + t: {part: kgo.NewOffset().At(o1)}, + })) + if err != nil { + return nil, err + } + defer cl.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + var recs []*kgo.Record + for { + fs := cl.PollFetches(ctx) + if ctx.Err() != nil { + break // timed out; return whatever we have + } + if errs := fs.Errors(); len(errs) > 0 { + break // transient under faults + } + done := false + iter := fs.RecordIter() + for !iter.Done() { + r := iter.Next() + if r.Offset >= o2 { + done = true + break + } + recs = append(recs, r) + if r.Offset == o2-1 { + done = true + } + } + if done { + break + } + } + return recs, nil +} + +// offsets extracts the offsets of recs in order. +func offsets(recs []*kgo.Record) []int64 { + offs := make([]int64, len(recs)) + for i, r := range recs { + offs[i] = r.Offset + } + return offs +} + +// checkFoo backs both anytime_check_range_foo and parallel_driver_consume_foo: +// pick a random sub-range of a random foo partition and validate the records +// it returns. +func checkFoo() error { + part, lo, hi, err := randomFooPartitionRange() + if err != nil || hi <= lo { + return nil // empty or unreadable under faults; nothing to check + } + o1 := lo + int64(randN(int(hi-lo))) + o2 := o1 + 1 + int64(randN(int(hi-o1))) + + recs, err := readRange(fooTopic, part, o1, o2) + if err != nil { + return nil + } + validateFooRange(part, lo, hi, o1, o2, recs) + return nil +} + +// validateFooRange asserts that recs — the result of reading [o1, o2) from part — +// form a well-formed slice of the log: offsets in order, contiguous, within the +// requested bounds, and records that are intact, ours, and in per-producer +// order. lo/hi are the partition's bounds, carried only for context in the +// assertion details. An empty read is a no-op: the invariants hold on whatever +// prefix a fault leaves behind, so they never false-positive on truncation, and +// completeness is only a liveness (Sometimes) property. +// +// Contiguity assumes non-transactional produce with cleanup.policy=delete +// (our setup); markers or compaction would create legal gaps. +func validateFooRange(part int32, lo, hi, o1, o2 int64, recs []*kgo.Record) { + if len(recs) == 0 { + return + } + offs := offsets(recs) + + first, last := offs[0], offs[len(offs)-1] + inOrder, contiguous := true, true + for i := 1; i < len(offs); i++ { + if offs[i] <= offs[i-1] { + inOrder = false + } + if offs[i] != offs[i-1]+1 { + contiguous = false + } + } + withinBounds := first >= o1 && last < o2 + full := first == o1 && int64(len(offs)) == o2-o1 + v := verifyFooRecords(part, recs) + + details := map[string]any{ + "command": cmdName, "partition": part, "o1": o1, "o2": o2, + "count": len(offs), "first": first, "last": last, "lo": lo, "hi": hi, + "bad_data": v.bad, "reordered": v.reordered, + "bad_offset": v.firstOff, "bad_reason": v.firstReason, + } + + assert.Always(inOrder, "delete-policy cloud topic range read returns in-order offsets", details) + assert.Always(contiguous, "delete-policy cloud topic range read returns contiguous offsets", details) + assert.Always(withinBounds, "delete-policy cloud topic range read stays within requested bounds", details) + assert.Always(v.bad == 0, "delete-policy cloud topic records are intact and self-consistent", details) + assert.Always(v.reordered == 0, "delete-policy cloud topic per-producer produce order is preserved", details) + if !quiescedPhase() { + assert.Reachable("checker read a non-empty delete-policy cloud topic range", details) + assert.Sometimes(full, "delete-policy cloud topic range read returns the full requested range", details) + } + + fmt.Printf("checked foo/%d offsets %d:%d -> %d records (in_order=%v contiguous=%v bounds=%v full=%v bad_data=%d reordered=%d)\n", + part, o1, o2, len(offs), inOrder, contiguous, withinBounds, full, v.bad, v.reordered) +} + +// checkFooOffsets is a manual command (no test-composer prefix, so it gets no +// symlink and Antithesis never schedules it): read and validate an explicit +// [o1, o2) range on a given partition. Unlike the random anytime_check_range_foo, +// the range is fixed by its arguments, so the read replays verbatim. Under the +// multiverse debugger this lets you roll back to different points in a timeline +// and repeat the exact same read to pin down when a range first goes bad. +// +// Usage: helper_workload check_offsets_foo +func checkFooOffsets() error { + if len(cmdArgs) != 3 { + return fmt.Errorf("usage: check_offsets_foo ") + } + part, err := strconv.ParseInt(cmdArgs[0], 10, 32) + if err != nil { + return fmt.Errorf("invalid partition %q: %w", cmdArgs[0], err) + } + o1, err := strconv.ParseInt(cmdArgs[1], 10, 64) + if err != nil { + return fmt.Errorf("invalid o1 %q: %w", cmdArgs[1], err) + } + o2, err := strconv.ParseInt(cmdArgs[2], 10, 64) + if err != nil { + return fmt.Errorf("invalid o2 %q: %w", cmdArgs[2], err) + } + if o2 <= o1 { + return fmt.Errorf("empty range: o2 (%d) must be > o1 (%d)", o2, o1) + } + + lo, hi, err := partitionBounds(fooTopic, int32(part)) + if err != nil { + fmt.Printf("check_offsets_foo: could not read bounds for foo/%d: %v\n", part, err) + lo, hi = -1, -1 + } + recs, err := readRange(fooTopic, int32(part), o1, o2) + if err != nil { + return fmt.Errorf("read of foo/%d [%d,%d) failed: %w", part, o1, o2, err) + } + fmt.Printf("check_offsets_foo: foo/%d [%d,%d) bounds=[%d,%d) -> %d records\n", + part, o1, o2, lo, hi, len(recs)) + validateFooRange(int32(part), lo, hi, o1, o2, recs) + return nil +} + +// readPartitionToEnd re-reads a partition's bounds and consumes the whole +// [lo, hi) range, retrying until the read reaches the high watermark (a +// record at offset hi-1; on a compacted topic that is fewer than hi-lo +// records) or the deadline passes. The retries only cover the post-fault +// recovery window (an eventually command runs after faults stop but the +// cluster may still be settling and there is no concurrent produce, so hi is +// stable). +// Returns the last-seen bounds and whatever the final attempt read: the +// complete slice on success, or a prefix if it gave up. +func readPartitionToEnd(t string, part int32, deadline time.Time) (lo, hi int64, recs []*kgo.Record) { + for { + var err error + lo, hi, err = partitionBounds(t, part) + if err == nil { + if hi <= lo { + return lo, hi, nil // empty partition + } + recs, err = readRange(t, part, lo, hi) + if err == nil && len(recs) > 0 && recs[len(recs)-1].Offset == hi-1 { + return lo, hi, recs + } + } + if time.Now().After(deadline) { + return lo, hi, recs + } + time.Sleep(2 * time.Second) + } +} + +// eventually_check_complete: after Antithesis stops fault injection for the +// timeline, wait for the cluster to recover, then read every partition of +// every test topic from its start offset to its high watermark, assert the +// read is complete, and run the same shape validation the anytime checkers +// use: validateFooRange for the delete-policy topic (in-order, contiguous, +// intact, per-producer order), validateCtcRange for the compacted one (same +// minus contiguity, since compaction leaves legal gaps). +// +// Because this runs on a quiesced, healed cluster with no concurrent produce +// or faults, completeness is a hard Always here (one property per cleanup +// policy, since the predicate differs) — contrast the anytime checkers, where +// a fault can truncate a read so completeness is only Sometimes. Complete +// means the read reaches the high watermark; on the +// delete-policy topic it also means every offset in [lo, hi) is present, +// while compaction legally removes records anywhere, including at lo (the +// record at hi-1 is the newest and thus the latest for its key, so it always +// survives). +func checkComplete() error { + // Faults stop when an eventually command starts, but containers need time + // to come back. Refuse to validate until the cluster serves metadata + // again; a cluster that never recovers is itself a failure. + if err := waitClusterReady(expectedBrokers(), 5*time.Minute); err != nil { + assert.Unreachable("eventually: cluster did not recover after faults stopped", + map[string]any{"err": err.Error()}) + return err + } + + // No producers run in the eventually phase, so the acked summaries are + // final. A producer killed mid-round may have observed acks it never + // persisted; the summary only ever understates what was acked, which + // weakens the check without falsifying it. + ackedSnap, err := loadCtcAckedSnapshot() + if err != nil { + return err + } + + for _, t := range testTopics { + for part := range t.partitions { + fmt.Printf("eventually: checking %s/%d\n", t.name, part) + + lo, hi, recs := readPartitionToEnd(t.name, part, time.Now().Add(2*time.Minute)) + if hi <= lo { + // An empty partition is only innocent if nothing was ever + // acked on it; acked keys with no log at all are loss. + if t.compacted { + validateCtcAcked(part, hi, nil, ackedSnap, true) + } + continue // empty partition; nothing else to verify + } + + first, last := int64(-1), int64(-1) + if len(recs) > 0 { + first, last = recs[0].Offset, recs[len(recs)-1].Offset + } + complete := last == hi-1 + if !t.compacted { + complete = complete && first == lo && int64(len(recs)) == hi-lo + } + + details := map[string]any{ + "topic": t.name, "partition": part, "lo": lo, "hi": hi, + "count": len(recs), "first": first, "last": last, + } + fmt.Printf("eventually %s/%d [%d,%d) -> %d records (complete=%v)\n", + t.name, part, lo, hi, len(recs), complete) + + // Reachability and completeness are per cleanup policy: distinct + // names keep each topic's eventually coverage a separate obligation + // (foo validating must not mask ctc never getting validated), and + // the completeness predicate differs between the policies anyway. + if t.compacted { + assert.Reachable("eventually: validated a non-empty compacted cloud topic partition", details) + assert.Always(complete, "eventually: compacted cloud topic partition is readable to the high watermark", details) + latest, rangeComplete := validateCtcRange(part, lo, hi, recs) + // The cluster is quiesced, so hi cannot move mid-read; a + // complete read alone makes the loss check sound here. + validateCtcAcked(part, hi, latest, ackedSnap, rangeComplete) + } else { + assert.Reachable("eventually: validated a non-empty delete-policy cloud topic partition", details) + assert.Always(complete, "eventually: delete-policy cloud topic partition is fully readable to the high watermark", details) + validateFooRange(part, lo, hi, lo, hi, recs) + } + } + } + return nil +} diff --git a/tests/antithesis/ct_stress/workload/commands.go b/tests/antithesis/ct_stress/workload/commands.go new file mode 100644 index 0000000000000..1df4ffd1ffce0 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/commands.go @@ -0,0 +1,166 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "context" + "errors" + "fmt" + "math/rand" + "time" + + "github.com/antithesishq/antithesis-sdk-go/assert" + "github.com/antithesishq/antithesis-sdk-go/random" + "github.com/twmb/franz-go/pkg/kadm" + "github.com/twmb/franz-go/pkg/kerr" + "github.com/twmb/franz-go/pkg/kgo" +) + +// rng draws from Antithesis-controlled entropy so the platform both controls +// and can steer every choice; outside Antithesis it falls back to crypto/rand. +// Each Intn consults the source live, which is the per-decision use the SDK +// wants (no seeding, no caching draws for later). +var rng = rand.New(random.Source()) + +// randN returns a uniformly random int in [0, n). +func randN(n int) int { + if n <= 0 { + return 0 + } + return rng.Intn(n) +} + +const topicPropertyCleanupPolicy = "cleanup.policy" +const topicPropertyStorageMode = "redpanda.storage.mode" + +const storageModeCloud = "cloud" +const storageModeTiered = "tiered" + +var storageModes = []string{storageModeCloud, storageModeTiered} + +// first_create_topics: create the topics the workload exercises. Runs once +// per timeline after setup_complete; must not signal lifecycle itself. +// +// Each topic's storage mode is an independent choice between cloud and +// tiered_cloud, drawn with the SDK's RandomChoice so Antithesis knows a +// structured random decision happens here and can steer it to explore the +// mode combinations deliberately. +func createTestTopics() error { + cl, err := newClient() + if err != nil { + return err + } + defer cl.Close() + adm := kadm.NewClient(cl) + + for _, t := range testTopics { + mode := random.RandomChoice(storageModes) + fmt.Printf("randomly chosen storage mode: %s=%s\n", t.name, mode) + cfg := map[string]*string{topicPropertyStorageMode: &mode} + if t.compacted { + cfg[topicPropertyCleanupPolicy] = new("compact") + } + if err := createOneTopic(adm, t.name, t.partitions, t.replicas, cfg); err != nil { + return err + } + } + + return nil +} + +func createOneTopic(adm *kadm.Client, name string, partitions int32, replicas int16, cfg map[string]*string) error { + modeForLog := "unset" + if m := cfg[topicPropertyStorageMode]; m != nil { + modeForLog = *m + } + + ctx := context.Background() + for attempt := 1; attempt <= 30; attempt++ { + resp, err := adm.CreateTopics(ctx, partitions, replicas, cfg, name) + if err == nil { + if terr := resp[name].Err; terr == nil { + fmt.Printf("created topic %s (%d partitions, %d replicas, mode=%s)\n", + name, partitions, replicas, modeForLog) + return nil + } else if errors.Is(terr, kerr.TopicAlreadyExists) { + fmt.Printf("topic %s already exists\n", name) + return nil + } else { + err = terr + } + } + fmt.Printf("attempt %d: create %s failed, retrying: %v\n", attempt, name, err) + time.Sleep(2 * time.Second) + } + + return fmt.Errorf("failed to create topic %s", name) +} + +// parallel_driver_produce_foo: produce a bounded random batch. Best-effort under +// fault injection — transient failures are expected and not bugs. +func produceFoo() error { + // A per-invocation nonce keeps this producer's keys distinct from every + // other concurrent producer's, so (nonce, seq) uniquely identifies a + // record. It also rides in the ClientID so Redpanda's request logs can be + // correlated back to this batch. + nonce := rng.Uint64() + cl, err := newClient( + kgo.ClientID(fmt.Sprintf("ct_stress/produce/%016x", nonce)), + kgo.DefaultProduceTopic(fooTopic), + kgo.RequiredAcks(kgo.AllISRAcks()), + kgo.ProducerLinger(5*time.Millisecond), + // Assign partitions ourselves so each record can carry the partition + // it was written to; a reader then verifies it was served from there. + kgo.RecordPartitioner(kgo.ManualPartitioner()), + // Flag franz-go's data-loss detection: it otherwise resets the + // producer id and silently carries on, hiding an anomaly we want seen. + kgo.ProducerOnDataLossDetected(reportDataLoss), + ) + if err != nil { + return err + } + defer cl.Close() + + count := 1 + randN(50) + recs := make([]*kgo.Record, count) + for i := range recs { + recs[i] = makeFooRecord(nonce, i, int32(randN(fooPartitions))) + } + + fmt.Printf("producing %d records to foo (nonce=%016x)\n", count, nonce) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := cl.ProduceSync(ctx, recs...).FirstErr(); err != nil { + fmt.Printf("produce did not complete (expected under faults) (nonce=%016x): %v\n", nonce, err) + return nil + } + // Prove the workload actually writes data in some timeline; without this a + // run where produce never succeeds would pass every safety check vacuously, + // since the checkers only assert on non-empty reads. + assert.Reachable("workload produced a batch to foo", + map[string]any{"count": count, "nonce": fmt.Sprintf("%016x", nonce)}) + fmt.Printf("produced %d records to foo (nonce=%016x)\n", count, nonce) + return nil +} + +// reportDataLoss is the franz-go ProducerOnDataLossDetected hook. franz-go +// calls it when a produce response returns an out-of-order sequence number or +// unknown producer id that it cannot attribute to benign prefix truncation +// (the broker's log start offset moving past records we'd already had acked). +// franz-go treats that as data loss and, since we don't stop the producer, +// resets the producer id and sequence numbers and continues. It is a +// presumption, not proof — a producer-id expiry can trigger it too — but on an +// acks=all cloud-topic producer it is an anomaly worth surfacing, so record it +// as a reachability failure for Antithesis. +func reportDataLoss(topic string, part int32) { + details := map[string]any{"topic": topic, "partition": part} + assert.Unreachable("idempotent producer detected data loss (out-of-order sequence or unknown producer id)", details) + fmt.Printf("data loss detected on %s/%d (out-of-order sequence or unknown producer id)\n", topic, part) +} diff --git a/tests/antithesis/ct_stress/workload/ctc.go b/tests/antithesis/ct_stress/workload/ctc.go new file mode 100644 index 0000000000000..cb27caef9dad1 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/ctc.go @@ -0,0 +1,665 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +// The ctc workload fuzzes compaction on a compacted cloud topic with a +// convergent key/value model. A fixed key space is split into buckets, and +// a tracker file per bucket records, per key, the last committed counter +// value and the highest broker-acked write (offset plus record identity). +// A producer round exclusively locks one bucket, produces the next run of +// counter values for a subset of its keys, folds every ack it observes into +// the acked summary (on failed rounds too), and commits the new counters +// only once every record is acked. +// +// Committed counters drive value generation; they are not a safety bound on +// surviving values. Idempotent sessions do not fence each other, so a +// produce request from an abandoned producer session can sit in a paused +// broker and apply after a later round commits higher counters, legally +// leaving a lower value as a key's latest record. What can never legally happen is a +// key's newest surviving offset regressing below an acked offset: +// compaction drops a record only when the key has a newer one, and such +// stragglers only push the newest offset up. The checkers hold reads +// against the acked summary (validateCtcAcked) on top of the log-shape +// invariants that survive compaction (in-order offsets, record integrity, +// per-producer produce order) and the liveness property that compaction +// visibly runs. +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "hash/crc32" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/antithesishq/antithesis-sdk-go/assert" + "github.com/twmb/franz-go/pkg/kgo" + "golang.org/x/sys/unix" +) + +const ( + ctcTopic = "ctc" + ctcPartitions = 3 + ctcReplicas = 3 + ctcBuckets = 32 + ctcKeysPerBucket = 32 + // ctcMaxStep caps how many new values one round appends per key. + ctcMaxStep = 16 + // ctcMaxPadChunks caps the record padding (16 hex chars per chunk) that + // varies record sizes. + ctcMaxPadChunks = 32 + // ctcProduceBudget bounds one round's produce, applied as the produce + // context timeout; franz-go retries transient failures within it. A + // round that cannot deliver within the budget exits uncommitted and a + // later round for the bucket re-produces the range. + ctcProduceBudget = 3 * time.Minute + + ctcMagic = "CTC1" +) + +func ctcTrackerDir() string { + if d := os.Getenv("CTC_TRACKER_DIR"); d != "" { + return d + } + return "/var/lib/ct_stress" +} + +// ctcKey is the compaction identity of logical key id. Ids map to partitions +// statically so a key's whole history lives in one partition and "latest +// record" is well defined. +func ctcKey(id int) string { + return fmt.Sprintf("%s:%04d", ctcMagic, id) +} + +func ctcPartition(id int) int32 { + return int32(id % ctcPartitions) +} + +// makeCtcRecord builds the record carrying counter value val of key id: +// +// key = MAGIC: +// value = MAGIC:::::: +// +// following the foo record layout (see makeFooRecord): the CRC covers everything +// before it, the value echoes the key, and the embedded partition lets a +// reader verify the record was served from the partition it was written to. +// The pad varies record sizes but is derived from (nonce, id, val) rather +// than drawn from randomness, so a record is a pure function of its identity. +func makeCtcRecord(nonce uint64, id int, val int64) *kgo.Record { + key := ctcKey(id) + part := ctcPartition(id) + h := nonce ^ uint64(id)*2654435761 ^ uint64(val)*0x9e3779b97f4a7c15 + pad := strings.Repeat(fmt.Sprintf("%016x", h), 1+int(h%ctcMaxPadChunks)) + body := fmt.Sprintf("%s:%d:%016x:%d:%s", key, val, nonce, part, pad) + crc := crc32.ChecksumIEEE([]byte(body)) + return &kgo.Record{ + Key: []byte(key), + Value: fmt.Appendf(nil, "%s:%08x", body, crc), + Partition: part, + } +} + +// parsedCtcRecord is the decoded identity of a ctc record. +type parsedCtcRecord struct { + id int + val int64 + nonce uint64 +} + +// parseCtcRecord checks that r, read from partition readPart, is a +// well-formed, intact ctc record and returns its decoded identity. The +// reason is empty when the record checks out, otherwise it describes the +// first problem found and the parsedCtcRecord is zero. +func parseCtcRecord(r *kgo.Record, readPart int32) (parsedCtcRecord, string) { + s := string(r.Value) + i := strings.LastIndexByte(s, ':') + if i < 0 { + return parsedCtcRecord{}, "value has no crc field" + } + body, crcStr := s[:i], s[i+1:] + crc, err := strconv.ParseUint(crcStr, 16, 32) + if err != nil { + return parsedCtcRecord{}, "crc is not hex" + } + if uint32(crc) != crc32.ChecksumIEEE([]byte(body)) { + return parsedCtcRecord{}, "crc mismatch (corrupt bytes)" + } + + // body = MAGIC::::: + f := strings.SplitN(body, ":", 6) + if len(f) != 6 || f[0] != ctcMagic { + return parsedCtcRecord{}, "bad magic or layout" + } + id, err := strconv.Atoi(f[1]) + if err != nil { + return parsedCtcRecord{}, "key id is not an int" + } + val, err := strconv.ParseInt(f[2], 10, 64) + if err != nil { + return parsedCtcRecord{}, "value counter is not an int" + } + nonce, err := strconv.ParseUint(f[3], 16, 64) + if err != nil { + return parsedCtcRecord{}, "nonce is not hex" + } + part, err := strconv.ParseInt(f[4], 10, 32) + if err != nil { + return parsedCtcRecord{}, "partition is not an int" + } + if int32(part) != readPart { + return parsedCtcRecord{}, fmt.Sprintf("partition mismatch: value claims %d, read from %d", part, readPart) + } + wantKey := f[0] + ":" + f[1] + if string(r.Key) != wantKey { + return parsedCtcRecord{}, fmt.Sprintf("key/value mismatch: key=%q, value embeds %q", r.Key, wantKey) + } + return parsedCtcRecord{id: id, val: val, nonce: nonce}, "" +} + +// ctcTracker is a bucket's per-key progress, indexed by the key's position +// within the bucket: the committed counter (zero, the seed state, means +// nothing committed; produced values start at 1) and the highest acked +// write. Committed only moves once a whole round is acked; Acked advances +// on every observed ack, including in rounds that end uncommitted. +type ctcTracker struct { + Committed []int64 `json:"committed"` + Acked []ctcAcked `json:"acked"` +} + +// ctcAcked is a key's highest broker-acked write: the offset the ack +// reported and the identity of the record at it, for content comparison +// when that exact offset is the key's surviving latest. Offset -1 (the seed +// state) means no ack observed yet. Only observed acks are recorded, never +// intent, so the summary can lag reality but never overstate it. +type ctcAcked struct { + Offset int64 `json:"offset"` + Val int64 `json:"val"` + Nonce uint64 `json:"nonce"` +} + +func newCtcTracker() *ctcTracker { + t := &ctcTracker{ + Committed: make([]int64, ctcKeysPerBucket), + Acked: make([]ctcAcked, ctcKeysPerBucket), + } + for i := range t.Acked { + t.Acked[i].Offset = -1 + } + return t +} + +func ctcBucketPaths(b int) (lock, data string) { + dir := ctcTrackerDir() + return filepath.Join(dir, fmt.Sprintf("bucket-%02d.lock", b)), + filepath.Join(dir, fmt.Sprintf("bucket-%02d.json", b)) +} + +// lockCtcBucket takes an exclusive flock on a random free bucket, probing +// each bucket at most once, and returns a nil file when all are busy. The +// lock lives on a file separate from the tracker data because the data file +// is replaced by rename on commit, which would silently detach a lock held +// on it. Closing the returned file (or the process dying) releases the lock. +func lockCtcBucket() (int, *os.File, error) { + if err := os.MkdirAll(ctcTrackerDir(), 0o755); err != nil { + return 0, nil, err + } + start := randN(ctcBuckets) + for i := range ctcBuckets { + b := (start + i) % ctcBuckets + lockPath, _ := ctcBucketPaths(b) + f, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return 0, nil, err + } + err = unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB) + if err == nil { + return b, f, nil + } + f.Close() + if !errors.Is(err, unix.EWOULDBLOCK) { + return 0, nil, err + } + } + return 0, nil, nil +} + +// loadCtcTracker reads the bucket's committed counters; a missing file is +// the seed state. A tracker that exists but does not decode is an error, not +// a reset: silently reseeding would lower committed values and invalidate +// everything produced so far. +func loadCtcTracker(b int) (*ctcTracker, error) { + _, dataPath := ctcBucketPaths(b) + raw, err := os.ReadFile(dataPath) + if errors.Is(err, os.ErrNotExist) { + return newCtcTracker(), nil + } + if err != nil { + return nil, err + } + var t ctcTracker + if err := json.Unmarshal(raw, &t); err != nil { + return nil, fmt.Errorf("tracker %s corrupt: %w", dataPath, err) + } + if len(t.Committed) != ctcKeysPerBucket || len(t.Acked) != ctcKeysPerBucket { + return nil, fmt.Errorf("tracker %s has %d committed and %d acked keys, want %d", + dataPath, len(t.Committed), len(t.Acked), ctcKeysPerBucket) + } + return &t, nil +} + +// loadCtcAckedSnapshot reads every bucket's acked summary into one slice +// indexed by key id. No locks: saveCtcTracker replaces the data file by +// rename, so each read sees a complete tracker. The summary is monotone +// (offsets only advance), which is what makes checking a log read against +// it race-free: take the snapshot before reading the log, and whatever the +// log shows is at least as new as the snapshot. +func loadCtcAckedSnapshot() ([]ctcAcked, error) { + snap := make([]ctcAcked, ctcBuckets*ctcKeysPerBucket) + for b := range ctcBuckets { + t, err := loadCtcTracker(b) + if err != nil { + return nil, err + } + copy(snap[b*ctcKeysPerBucket:], t.Acked) + } + return snap, nil +} + +// recordCtcAcks folds one produce attempt's per-record results into the +// bucket's acked summary, keeping the highest acked offset per key. It runs +// on failed attempts too: a timed-out round may have acked a subset, and +// those acks are facts the checkers hold Redpanda to. +func recordCtcAcks(t *ctcTracker, b int, res kgo.ProduceResults) { + for _, pr := range res { + if pr.Err != nil { + continue + } + p, reason := parseCtcRecord(pr.Record, pr.Record.Partition) + if reason != "" { + continue + } + i := p.id - b*ctcKeysPerBucket + if i < 0 || i >= ctcKeysPerBucket { + continue + } + if a := &t.Acked[i]; pr.Record.Offset > a.Offset { + *a = ctcAcked{Offset: pr.Record.Offset, Val: p.val, Nonce: p.nonce} + } + } +} + +// saveCtcTracker atomically replaces the bucket's tracker file. Must be +// called with the bucket lock held. +func saveCtcTracker(b int, t *ctcTracker) error { + _, dataPath := ctcBucketPaths(b) + raw, err := json.Marshal(t) + if err != nil { + return err + } + tmp := dataPath + ".tmp" + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return err + } + if _, err := f.Write(raw); err != nil { + f.Close() + return err + } + if err := f.Sync(); err != nil { + f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + return os.Rename(tmp, dataPath) +} + +// parallel_driver_produce_ctc: advance one tracker bucket. Locks a random +// free bucket, produces the next run of counter values for a random subset +// of its keys, and commits the new counters only after every record is +// acked. The produce is a single attempt on one idempotent session, bounded +// by the round context, within which franz-go retries transient failures +// itself by resending the same batches under the same sequence numbers. If +// the budget expires or franz-go gives up on any record, the round exits +// with the counters untouched (see the comment on the error path) and a +// later round for the bucket re-produces the range. Either way, every ack +// observed is folded into the tracker's acked summary (recordCtcAcks) and +// persisted before exit. +func produceCtc() error { + b, lock, err := lockCtcBucket() + if err != nil { + return err + } + if lock == nil { + fmt.Println("ctc: all buckets locked by other producers; skipping") + return nil + } + defer lock.Close() + + tracker, err := loadCtcTracker(b) + if err != nil { + return err + } + + // Advance a random subset of the bucket's keys (at least one) by a + // random step each, so keys age unevenly and compaction sees a mix of + // hot and cold keys. + steps := make([]int64, ctcKeysPerBucket) + picked := 0 + for i := range steps { + if randN(2) == 1 { + steps[i] = 1 + int64(randN(ctcMaxStep)) + picked++ + } + } + if picked == 0 { + steps[randN(ctcKeysPerBucket)] = 1 + int64(randN(ctcMaxStep)) + picked = 1 + } + + // The round's records are emitted in ascending (key id, value) order, + // which per partition is also the produce order. Within one nonce that + // order is preserved end to end: the idempotent session keeps it on the + // wire, and compaction only removes records, never reorders them — the + // invariant the sweeper asserts per nonce. + nonce := rng.Uint64() + var recs []*kgo.Record + for i, step := range steps { + id := b*ctcKeysPerBucket + i + for v := tracker.Committed[i] + 1; v <= tracker.Committed[i]+step; v++ { + recs = append(recs, makeCtcRecord(nonce, id, v)) + } + } + + cl, err := newClient( + kgo.ClientID(fmt.Sprintf("ct_stress/produce_ctc/%016x", nonce)), + kgo.DefaultProduceTopic(ctcTopic), + kgo.RequiredAcks(kgo.AllISRAcks()), + kgo.ProducerLinger(5*time.Millisecond), + kgo.RecordPartitioner(kgo.ManualPartitioner()), + kgo.ProducerOnDataLossDetected(reportDataLoss), + // Lets the round context below actually end the round. By default + // franz-go refuses to fail an in-flight record — it cannot tell + // "never received" from "written but the reply was lost" — and waits + // for the outcome, unbounded under a long fault. Records that land + // despite being failed to us become stragglers under this nonce, + // which the acked summary (they are never recorded) and the + // acked-offset checks tolerate by design. The session is never + // produced on again after a failure, so its then-inconsistent + // sequence window does not matter. + kgo.AllowIdempotentProduceCancellation(), + ) + if err != nil { + return err + } + defer cl.Close() + + fmt.Printf("ctc bucket %d: producing %d records for %d keys (nonce=%016x)\n", + b, len(recs), picked, nonce) + // The round's single bound: when it fires, franz-go fails whatever is + // still unresolved (in-flight included, see the client option above) and + // the round exits uncommitted. + ctx, cancel := context.WithTimeout(context.Background(), ctcProduceBudget) + defer cancel() + res := cl.ProduceSync(ctx, recs...) + recordCtcAcks(tracker, b, res) + if err := res.FirstErr(); err != nil { + // No in-process retry: how franz-go leaves the session after giving + // up is failure-specific — some paths keep the producer id and rewind + // the sequence numbers, others reload the id — so a resend on it + // risks mis-attributed acks (rewound sequences dedup whatever bytes + // come next, acking them at offsets holding the old records), and a + // correct retry needs a fresh session producing the range under a + // fresh nonce. That is exactly what the next round for this bucket + // is. Exit uncommitted, keeping the acks that were observed; this + // session's applied-but-unacked batches may surface later as + // stragglers, which the acked-offset checks tolerate by design. + fmt.Printf("ctc bucket %d: round uncommitted (expected under faults) (nonce=%016x): %v\n", + b, nonce, err) + return saveCtcTracker(b, tracker) + } + + for i, step := range steps { + tracker.Committed[i] += step + } + if err := saveCtcTracker(b, tracker); err != nil { + return err + } + assert.Reachable("workload committed a compacted cloud topic round", + map[string]any{"bucket": b, "records": len(recs), "nonce": fmt.Sprintf("%016x", nonce)}) + fmt.Printf("ctc bucket %d: committed %d records (nonce=%016x)\n", b, len(recs), nonce) + return nil +} + +// parallel_driver_sweep_ctc: read one random ctc partition end to end, +// assert the shape of a compacted log (validateCtcRange), and hold the read +// against the trackers' acked summaries (validateCtcAcked). The summary is +// snapshotted before the read so the comparison is race-free. The loss half +// of the acked check is only sound when the read reached the high watermark +// and the high watermark held still across the read — a record appended +// mid-read beyond hi legally lets compaction drop everything the [lo, hi) +// window held for that key — so the bounds are re-fetched afterwards and +// loss is only asserted when hi is unchanged. +func sweepCtc() error { + acked, err := loadCtcAckedSnapshot() + if err != nil { + return err + } + part := int32(randN(ctcPartitions)) + lo, hi, err := partitionBounds(ctcTopic, part) + if err != nil || hi <= lo { + return nil // empty or unreadable under faults; nothing to sweep + } + recs, err := readRange(ctcTopic, part, lo, hi) + if err != nil { + return nil + } + assert.Sometimes(len(recs) > 0, "compacted cloud topic sweep consumes a non-empty range", + map[string]any{"partition": part, "lo": lo, "hi": hi}) + latest, complete := validateCtcRange(part, lo, hi, recs) + stable := false + if complete { + _, hi2, err := partitionBounds(ctcTopic, part) + stable = err == nil && hi2 == hi + } + validateCtcAcked(part, hi, latest, acked, complete && stable) + return nil +} + +// ctcLatest is a key's newest record within one read: the record at the +// highest offset, with its decoded identity. +type ctcLatest struct { + off int64 + val int64 + nonce uint64 +} + +// validateCtcRange asserts that recs — the result of reading [lo, hi) from a +// ctc partition — have the shape of a compacted log; it backs both the +// anytime sweep and eventually_check_complete. Compaction removes records but +// never reorders survivors, so offsets must still be strictly increasing +// (though gapped, unlike foo, so no contiguity here) and each producer +// nonce's records must still appear in its produce order — ascending (key +// id, value), the order build() emits and the idempotent session preserves. +// Records must be intact and ours. Offset gaps are compaction's fingerprint +// in an otherwise gapless non-transactional log, so seeing one sometimes +// proves compaction actually runs and the workload has not degraded to +// plain produce/consume. An empty read is a no-op: the invariants hold on +// whatever a fault leaves readable, so they never false-positive on +// truncation. +// +// Returns each key's newest record in the read and whether the read reached +// the high watermark — the inputs validateCtcAcked needs. +func validateCtcRange(part int32, lo, hi int64, recs []*kgo.Record) (map[int]ctcLatest, bool) { + if len(recs) == 0 { + return nil, false + } + + type lastPos struct { + id int + val int64 + } + lastByNonce := map[uint64]lastPos{} + latest := make(map[int]ctcLatest) + var bad, reordered, gaps int + var missing int64 + firstOff, firstReason := int64(-1), "" + inOrder := true + prev := int64(-1) + for _, r := range recs { + if prev >= 0 { + if r.Offset <= prev { + inOrder = false + } else if r.Offset > prev+1 { + gaps++ + missing += r.Offset - prev - 1 + } + } + prev = r.Offset + p, reason := parseCtcRecord(r, part) + isReorder := false + if reason == "" { + // latest tracks each key's newest record by offset regardless of + // the per-nonce order check below: a mis-ordered record still + // survives in the log, and the acked-write check compares against + // what survives. + if l, ok := latest[p.id]; !ok || r.Offset > l.off { + latest[p.id] = ctcLatest{off: r.Offset, val: p.val, nonce: p.nonce} + } + lp, seen := lastByNonce[p.nonce] + if seen && (p.id < lp.id || (p.id == lp.id && p.val <= lp.val)) { + isReorder = true + reason = fmt.Sprintf("producer %016x out of order: key %d val %d at or after key %d val %d", + p.nonce, p.id, p.val, lp.id, lp.val) + } else { + lastByNonce[p.nonce] = lastPos{id: p.id, val: p.val} + } + } + if reason != "" { + if isReorder { + reordered++ + } else { + bad++ + } + if firstReason == "" { + firstOff, firstReason = r.Offset, reason + } + } + } + superseded := len(recs) - bad - reordered - len(latest) + partial := prev != hi-1 + + details := map[string]any{ + "command": cmdName, "partition": part, "lo": lo, "hi": hi, + "count": len(recs), "keys": len(latest), "superseded": superseded, + "gaps": gaps, "missing": missing, + "bad_data": bad, "reordered": reordered, + "bad_offset": firstOff, "bad_reason": firstReason, + } + assert.Always(inOrder, "compacted cloud topic read returns in-order offsets", details) + assert.Always(bad == 0, "compacted cloud topic records are intact and self-consistent", details) + assert.Always(reordered == 0, "compacted cloud topic per-producer produce order is preserved", details) + if !quiescedPhase() { + assert.Sometimes(gaps > 0, "compacted cloud topic sweep observes compaction gaps", details) + } + + fmt.Printf("ctc sweep %d [%d,%d) -> %d records (keys=%d superseded=%d gaps=%d missing=%d bad=%d reordered=%d partial=%v)\n", + part, lo, hi, len(recs), len(latest), superseded, gaps, missing, bad, reordered, partial) + if firstReason != "" { + fmt.Printf("ctc sweep %d first bad record at offset %d: %s\n", part, firstOff, firstReason) + } + return latest, !partial +} + +// validateCtcAcked holds one partition read against the trackers' acked +// summaries — latest is the read's newest record per key, acked a snapshot +// taken before the read (both sides are monotone, so the comparison is +// race-free). Two properties: +// +// - immutability: when a key's newest surviving record sits exactly at +// its acked offset, it must be the acked record. A log never rewrites +// an offset, so any read may assert this, partial or not. +// - no acked write lost: every key acked on this partition must have a +// surviving record at or beyond its acked offset. Compaction drops a +// record only when the key has a newer one, and stragglers from +// abandoned rounds only push the newest offset up, so the newest +// surviving offset never legally regresses below an acked one. This is +// only sound when the read reached the high watermark and hi held still +// across it; assertLoss says the caller vouches for that. +// +// Keys whose acked offset is at or beyond hi prove nothing in the anytime +// phase (a stale leader can serve an old hi) and are skipped. In the +// eventually phase the cluster is quiesced, hi is the true end of the log, +// and an acked offset beyond it means the high watermark regressed, so there +// they are a hard failure. +func validateCtcAcked(part int32, hi int64, latest map[int]ctcLatest, acked []ctcAcked, assertLoss bool) { + checked, lost, mismatched, beyondHi := 0, 0, 0, 0 + firstKey, firstReason := -1, "" + for id, a := range acked { + if a.Offset < 0 || ctcPartition(id) != part { + continue + } + if a.Offset >= hi { + beyondHi++ + if quiescedPhase() && firstReason == "" { + firstKey, firstReason = id, fmt.Sprintf("acked offset %d at or beyond quiesced hi %d", a.Offset, hi) + } + continue + } + checked++ + reason := "" + l, ok := latest[id] + switch { + case !ok: + if assertLoss { + lost++ + reason = fmt.Sprintf("no surviving record; acked offset %d val %d", a.Offset, a.Val) + } + case l.off < a.Offset: + if assertLoss { + lost++ + reason = fmt.Sprintf("newest surviving offset %d below acked offset %d (val %d)", l.off, a.Offset, a.Val) + } + case l.off == a.Offset && (l.val != a.Val || l.nonce != a.Nonce): + mismatched++ + reason = fmt.Sprintf("record at acked offset %d is not the acked one: got val %d nonce %016x, acked val %d nonce %016x", + a.Offset, l.val, l.nonce, a.Val, a.Nonce) + } + if reason != "" && firstReason == "" { + firstKey, firstReason = id, reason + } + } + + details := map[string]any{ + "command": cmdName, "partition": part, "hi": hi, + "acked_keys": checked, "beyond_hi": beyondHi, + "lost": lost, "mismatched": mismatched, "loss_checked": assertLoss, + "bad_key": firstKey, "bad_reason": firstReason, + } + assert.Always(mismatched == 0, "compacted cloud topic record at an acked offset is the acked record", details) + if assertLoss { + assert.Always(lost == 0, "compacted cloud topic keeps a record at or beyond every acked offset", details) + } + if quiescedPhase() { + assert.Always(beyondHi == 0, "eventually: compacted cloud topic high watermark covers every acked offset", details) + } else if assertLoss { + assert.Sometimes(checked > 0, "compacted cloud topic sweep checks acked writes against a complete read", details) + } + + fmt.Printf("ctc acked check %d: %d keys checked (loss_checked=%v lost=%d mismatched=%d beyond_hi=%d)\n", + part, checked, assertLoss, lost, mismatched, beyondHi) + if firstReason != "" { + fmt.Printf("ctc acked check %d first bad key %d: %s\n", part, firstKey, firstReason) + } +} diff --git a/tests/antithesis/ct_stress/workload/flip.go b/tests/antithesis/ct_stress/workload/flip.go new file mode 100644 index 0000000000000..adb95e3d8b776 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/flip.go @@ -0,0 +1,124 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "context" + "fmt" + "time" + + "github.com/antithesishq/antithesis-sdk-go/assert" + "github.com/antithesishq/antithesis-sdk-go/random" + "github.com/twmb/franz-go/pkg/kadm" +) + +// parallel_driver_flip_storage_mode: toggle a random test topic between the +// two cloud-backed storage modes (cloud <-> tiered_v2) while the rest of the +// workload keeps producing, consuming and moving replicas. Antithesis +// scheduling this driver at arbitrary points is what exercises mid-flight +// transitions. +// +// The flip reads the topic's current mode from redpanda.storage.mode (which +// reports plain "cloud" or "tiered") and alters it to the other one. Altering +// to "tiered" selects tiered_v2 only because the cluster bootstraps with +// default_redpanda_storage_mode_tiered_impl=tiered_v2 (cloud -> tiered_v1 is +// a forbidden transition). Best-effort under fault injection: a failed +// describe or a rejected alter is expected and not a bug. +func flipStorageMode() error { + t := random.RandomChoice(testTopics).name + + cl, err := newClient() + if err != nil { + return err + } + defer cl.Close() + adm := kadm.NewClient(cl) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + rcs, err := adm.DescribeTopicConfigs(ctx, t) + if err != nil { + fmt.Printf("flip %s skipped: describe configs failed (expected under faults): %v\n", t, err) + return nil + } + rc, err := rcs.On(t, nil) + if err == nil && rc.Err != nil { + err = rc.Err + } + if err != nil { + fmt.Printf("flip %s skipped: describe configs failed (expected under faults): %v\n", t, err) + return nil + } + + mode := "" + for _, c := range rc.Configs { + if c.Key == topicPropertyStorageMode { + mode = c.MaybeValue() + } + } + var target string + switch mode { + case storageModeCloud: + target = storageModeTiered + case storageModeTiered: + target = storageModeCloud + default: + // The topics are created in one of the two modes above and only this + // driver alters the mode, so anything else is a real anomaly. + assert.Unreachable("cloud topic reports an unexpected storage mode", + map[string]any{"topic": t, "mode": mode}) + return nil + } + + resps, err := adm.AlterTopicConfigs(ctx, []kadm.AlterConfig{{ + Op: kadm.SetConfig, + Name: topicPropertyStorageMode, + Value: &target, + }}, t) + accepted := false + if err == nil { + resp, rerr := resps.On(t, nil) + if rerr == nil && resp.Err == nil { + accepted = true + } else if rerr != nil { + err = rerr + } else { + err = fmt.Errorf("%w (%s)", resp.Err, resp.ErrMessage) + } + } + + // One liveness property per topic and direction, so a run only passes + // this set once every transition has been accepted on every topic + // somewhere. + details := map[string]any{"topic": t, "from": mode, "to": target} + toTiered := target == storageModeTiered + switch t { + case fooTopic: + if toTiered { + assert.Sometimes(accepted, "cloud topic foo storage mode flip cloud -> tiered accepted", details) + } else { + assert.Sometimes(accepted, "cloud topic foo storage mode flip tiered -> cloud accepted", details) + } + case ctcTopic: + if toTiered { + assert.Sometimes(accepted, "cloud topic ctc storage mode flip cloud -> tiered accepted", details) + } else { + assert.Sometimes(accepted, "cloud topic ctc storage mode flip tiered -> cloud accepted", details) + } + } + + if accepted { + fmt.Printf("flipped %s storage mode %s -> %s\n", t, mode, target) + } else { + fmt.Printf("flip %s %s -> %s not accepted (expected under faults): %v\n", t, mode, target, err) + } + return nil +} diff --git a/tests/antithesis/ct_stress/workload/go.mod b/tests/antithesis/ct_stress/workload/go.mod new file mode 100644 index 0000000000000..149549f3bb359 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/go.mod @@ -0,0 +1,17 @@ +module github.com/redpanda-data/redpanda/tests/antithesis/ct_stress/workload + +go 1.26 + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.2 + github.com/twmb/franz-go v1.21.2 + github.com/twmb/franz-go/pkg/kadm v1.18.0 + golang.org/x/sys v0.47.0 +) + +require ( + github.com/klauspost/compress v1.18.6 // indirect + github.com/pierrec/lz4/v4 v4.1.26 // indirect + github.com/twmb/franz-go/pkg/kmsg v1.13.1 // indirect + golang.org/x/crypto v0.51.0 // indirect +) diff --git a/tests/antithesis/ct_stress/workload/go.sum b/tests/antithesis/ct_stress/workload/go.sum new file mode 100644 index 0000000000000..9537399ed1fa6 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/go.sum @@ -0,0 +1,16 @@ +github.com/antithesishq/antithesis-sdk-go v0.7.2 h1:oEEedg1Xgi8drRjqB0f9tfjhLoInE0IYZfZ6zAhQUbY= +github.com/antithesishq/antithesis-sdk-go v0.7.2/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= +github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/twmb/franz-go v1.21.2 h1:WrvV/spF48JzcRylqDQy02Vm6V6W4lhtD9Y4BOYNMu4= +github.com/twmb/franz-go v1.21.2/go.mod h1:rfoMTnVk7107fhTGxfEKIHP/e7tPe6oyij/ywzO0czk= +github.com/twmb/franz-go/pkg/kadm v1.18.0 h1:WRf/LZmDdcDXwX7WMbtDU++v+b3NzYh2bCGoPMmzirw= +github.com/twmb/franz-go/pkg/kadm v1.18.0/go.mod h1:XeLhGoLXLFzK8/ryv5FfpxPxGwj4oFEGpPJMB/x6KDE= +github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI= +github.com/twmb/franz-go/pkg/kmsg v1.13.1/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= +golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= +golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= diff --git a/tests/antithesis/ct_stress/workload/main.go b/tests/antithesis/ct_stress/workload/main.go new file mode 100644 index 0000000000000..c35487d15f2bc --- /dev/null +++ b/tests/antithesis/ct_stress/workload/main.go @@ -0,0 +1,172 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +// Command workload is the ct_stress Antithesis workload. A single binary +// dispatches on its invocation name (argv[0]) so the test-composer command +// files (first_create_topics, parallel_driver_produce_foo, ...) are just links +// to it. It talks to Redpanda with franz-go and reports properties and +// draws randomness through the Antithesis Go SDK. +// +// Randomness for every test-affecting choice goes through random.GetRandom +// so Antithesis both controls and can steer it; outside Antithesis the SDK +// falls back to crypto/rand. +package main + +import ( + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + "github.com/twmb/franz-go/pkg/kgo" +) + +const ( + fooTopic = "foo" + fooPartitions = 3 + fooReplicas = 3 +) + +// commands maps every invocation name to its handler. Entries whose name +// carries an Antithesis test-composer prefix are the test commands, exposed +// as symlinks under /opt/antithesis/test/v1/main/ (see `list-commands`, which +// the image build uses to create those links). `setup` is the container +// entrypoint, not a test command. +var commands = map[string]func() error{ + "setup": setup, + "first_create_topics": createTestTopics, + "parallel_driver_produce_foo": produceFoo, + // consume reads and validates the same way the anytime checker does, but + // as a parallel driver Antithesis may run several concurrent copies of it, + // applying real read pressure while still asserting the invariants. + "parallel_driver_consume_foo": checkFoo, + "parallel_driver_move_metastore": moveMetastore, + "parallel_driver_move_kafka_topic": moveKafkaTopic, + "parallel_driver_flip_storage_mode": flipStorageMode, + // The ctc commands fuzz compaction on a compacted cloud topic: the + // sweeper asserts log-shape invariants and holds surviving records + // against the trackers' acked-write summaries. See ctc.go. + "parallel_driver_produce_ctc": produceCtc, + "parallel_driver_sweep_ctc": sweepCtc, + "anytime_check_range_foo": checkFoo, + "anytime_check_cloud_io": checkCloudIO, + // eventually rather than finally: finally commands only run on timelines + // where every started command completed and none was killed by a fault, + // which excludes the most hostile histories — exactly the ones this data + // completeness check exists for. An eventually command runs after any + // driver has started; Antithesis kills the stragglers, which the acked + // summaries tolerate by design (only observed acks, atomic replace). + "eventually_check_complete": checkComplete, + // check_offsets_foo has no test-composer prefix, so it gets no symlink and + // Antithesis never schedules it. It is a manual replay tool for the + // multiverse debugger; see checkFooOffsets. + "check_offsets_foo": checkFooOffsets, +} + +// cmdArgs holds the positional arguments that follow the command token, for +// the few manual commands (e.g. check_offsets_foo) that take parameters. +var cmdArgs []string + +// cmdName is the command this invocation resolved to. The checkers shared +// across commands (validateFooRange, validateCtcRange) carry it in assertion +// details so a tripped property names the phase it fired in — e.g. +// parallel_driver_consume_foo vs anytime_check_range_foo vs eventually_check_complete. +var cmdName string + +// quiescedPhase reports whether this invocation is an eventually command, +// i.e. fault injection has stopped and the cluster has healed. The shared +// validators skip their Sometimes/Reachable assertions in this phase: those +// liveness properties exist to prove reads make progress while faults are +// possible, and a quiesced-cluster read satisfying them would mask timelines +// where no such read ever succeeded. The eventually checks assert their own +// liveness under distinct names instead. +func quiescedPhase() bool { + return strings.HasPrefix(cmdName, "eventually_") +} + +// testCommandPrefixes are the Antithesis test-composer command prefixes; a +// command with one of these is scheduled by Antithesis and needs a symlink. +var testCommandPrefixes = []string{ + "first_", "parallel_driver_", "serial_driver_", + "singleton_driver_", "anytime_", "eventually_", "finally_", +} + +func isTestCommand(name string) bool { + for _, p := range testCommandPrefixes { + if strings.HasPrefix(name, p) { + return true + } + } + return false +} + +// listCommands prints the test-composer command names, one per line, for the +// image build to turn into symlinks. Keeping the list here means the command +// set is defined once, in `commands`. It is dispatched specially (not via +// `commands`) to avoid an initialization cycle with the map it reads. +func listCommands() error { + names := make([]string, 0, len(commands)) + for name := range commands { + if isTestCommand(name) { + names = append(names, name) + } + } + sort.Strings(names) + for _, n := range names { + fmt.Println(n) + } + return nil +} + +func brokers() []string { + return strings.Split(os.Getenv("KAFKA_BROKERS"), ",") +} + +func newClient(opts ...kgo.Opt) (*kgo.Client, error) { + base := []kgo.Opt{ + kgo.SeedBrokers(brokers()...), + kgo.WithLogger(kgo.BasicLogger(os.Stderr, kgo.LogLevelInfo, nil)), + } + return kgo.NewClient(append(base, opts...)...) +} + +func main() { + cmd := filepath.Base(os.Args[0]) + cmdArgs = os.Args[1:] + // When invoked by the binary's own name (e.g. the entrypoint's + // `helper_workload setup`, or `workload parallel_driver_produce_foo` by + // hand) rather than through a command symlink, take the command from + // argv[1]. + if _, known := commands[cmd]; !known && len(os.Args) > 1 { + cmd = os.Args[1] + cmdArgs = os.Args[2:] + } + cmdName = cmd + + // Dispatched outside `commands` to avoid an init cycle (it reads the map). + if cmd == "list-commands" { + if err := listCommands(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + return + } + + fn, ok := commands[cmd] + if !ok { + fmt.Fprintf(os.Stderr, "unknown command %q\n", cmd) + os.Exit(2) + } + fmt.Printf("workload: running command %q\n", cmd) + if err := fn(); err != nil { + fmt.Fprintf(os.Stderr, "%s: %v\n", cmd, err) + os.Exit(1) + } +} diff --git a/tests/antithesis/ct_stress/workload/metrics.go b/tests/antithesis/ct_stress/workload/metrics.go new file mode 100644 index 0000000000000..8a7edc24d3135 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/metrics.go @@ -0,0 +1,123 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "fmt" + "io" + "net/http" + "strconv" + "strings" + + "github.com/antithesishq/antithesis-sdk-go/assert" +) + +// scrapeMetrics sums the named seastar counters across the cluster, reading +// each broker's internal /metrics endpoint once and adding every matching +// series (one per shard). Returns name -> cluster-wide total, plus how many +// hosts answered the scrape, so a caller asserting on a zero total can tell +// "genuinely zero" from "a broker was unreachable and its counters missing". +func scrapeMetrics(names []string) (map[string]float64, int) { + sums := make(map[string]float64, len(names)) + scraped := 0 + + hosts := adminHosts() + fmt.Printf("scrapeMetrics: scraping %d hosts %v for %v\n", len(hosts), hosts, names) + + for _, h := range hosts { + resp, err := httpClient.Get("http://" + h + "/metrics") + if err != nil { + fmt.Printf("scrapeMetrics: host %s GET /metrics failed: %v\n", h, err) + continue + } + body, err := io.ReadAll(resp.Body) + resp.Body.Close() + if err != nil { + fmt.Printf("scrapeMetrics: host %s read /metrics body failed after %d bytes: %v\n", h, len(body), err) + continue + } + if resp.StatusCode != http.StatusOK { + snippet := string(body) + if len(snippet) > 200 { + snippet = snippet[:200] + } + fmt.Printf("scrapeMetrics: host %s GET /metrics -> %d: %s\n", h, resp.StatusCode, snippet) + continue + } + scraped++ + matched := 0 + hostSums := make(map[string]float64, len(names)) + for line := range strings.SplitSeq(string(body), "\n") { + if line == "" || line[0] == '#' { + continue + } + // Prometheus text: `name{labels} value` or `name value`. + for _, name := range names { + if !strings.HasPrefix(line, name+"{") && !strings.HasPrefix(line, name+" ") { + continue + } + fields := strings.Fields(line) + if v, err := strconv.ParseFloat(fields[len(fields)-1], 64); err == nil { + sums[name] += v + hostSums[name] += v + matched++ + } + break + } + } + fmt.Printf("scrapeMetrics: host %s status=%d bytes=%d matched_series=%d sums=%v\n", + h, resp.StatusCode, len(body), matched, hostSums) + } + return sums, scraped +} + +// anytime_check_cloud_io: assert the cloud-topics I/O paths are exercised at +// least sometimes across the run — the L0 batcher uploads to object storage, +// and the L1 read path (file-io reads, plus index-driven byte skipping) runs. +// +// Note: the batch-cache miss / pipeline cloud-read counters stay ~0 here — the +// batcher populates caches write-through and the workload reads recent, local +// data, so reads are served warm. Genuinely cold object-storage reads would +// need cache eviction (larger backlog) or reads of data no broker cached; see +// the read path via level_one_file_io instead. +func checkCloudIO() error { + const ( + readMetric = "vectorized_cloud_topics_level_one_reader_read_bytes" + uploadMetric = "vectorized_cloud_topics_batcher_bytes_uploaded" + fileReadMetric = "vectorized_cloud_topics_level_one_file_io_reads" + skippedMetric = "vectorized_cloud_topics_level_one_reader_skipped_bytes" + ) + // Sampled before the scrape so the zero-upload assertion below is not + // raced by a produce landing in between: an ack recorded before a scrape + // that still reads zero means no acked cloud-mode produce can hide in + // the gap. Both topics must have acked, or a mixed-mode timeline whose + // cloud topic simply has not produced yet would satisfy the assertion. + sums, _ := scrapeMetrics([]string{ + readMetric, uploadMetric, fileReadMetric, skippedMetric, + }) + + assert.Sometimes(sums[readMetric] > 0, + "cloud topics L1 reader reads bytes", + map[string]any{"read_bytes": sums[readMetric]}) + assert.Sometimes(sums[uploadMetric] > 0, + "cloud topics batcher uploads bytes", + map[string]any{"bytes_uploaded": sums[uploadMetric]}) + assert.Sometimes(sums[fileReadMetric] > 0, + "cloud topics L1 file-io performs reads", + map[string]any{"reads": sums[fileReadMetric]}) + assert.Sometimes(sums[skippedMetric] > 0, + "cloud topics L1 reader skips bytes via index", + map[string]any{"skipped_bytes": sums[skippedMetric]}) + + fmt.Printf("L1 read_bytes=%.0f uploaded=%.0f file_reads=%.0f skipped_bytes=%.0f\n", + sums[readMetric], sums[uploadMetric], sums[fileReadMetric], sums[skippedMetric]) + + return nil +} diff --git a/tests/antithesis/ct_stress/workload/move.go b/tests/antithesis/ct_stress/workload/move.go new file mode 100644 index 0000000000000..cefa8821d4920 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/move.go @@ -0,0 +1,178 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" + + "github.com/antithesishq/antithesis-sdk-go/assert" +) + +// The cloud-topics L1 metastore / domain topic is an internal topic, so its +// replicas are moved through Redpanda's admin API rather than the Kafka +// reassignment API. With cloud_topics_num_metastore_partitions=1 there is a +// single partition to target. +const ( + metastoreNS = "kafka_internal" + metastoreTopic = "ct_l1_domain" + metastorePart = 0 +) + +type replica struct { + NodeID int `json:"node_id"` + Core int `json:"core"` +} + +type partitionInfo struct { + Status string `json:"status"` + LeaderID int `json:"leader_id"` + Replicas []replica `json:"replicas"` +} + +type brokerInfo struct { + NodeID int `json:"node_id"` + NumCores int `json:"num_cores"` +} + +var httpClient = &http.Client{Timeout: 10 * time.Second} + +func adminHosts() []string { + return strings.Split(os.Getenv("ADMIN_HOSTS"), ",") +} + +// adminGET issues a GET against the first admin host that answers. +func adminGET(path string, out any) error { + var lastErr error + for _, h := range adminHosts() { + resp, err := httpClient.Get("http://" + h + path) + if err != nil { + lastErr = err + continue + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + lastErr = fmt.Errorf("GET %s -> %d: %s", path, resp.StatusCode, body) + continue + } + return json.Unmarshal(body, out) + } + return lastErr +} + +// submitMove reassigns ns/topic/part to a fresh random replica set of the same +// replication factor. attempted is false when the move was skipped (partition +// mid-reconfiguration, too few brokers, or the admin API was unreachable); +// accepted reports whether the POST was taken. Best-effort under fault +// injection: skips and rejections are expected and not bugs. +func submitMove(ns, topic string, part int) (from, to []replica, attempted, accepted bool) { + var brokers []brokerInfo + if err := adminGET("/v1/brokers", &brokers); err != nil { + return nil, nil, false, false + } + path := fmt.Sprintf("/v1/partitions/%s/%s/%d", ns, topic, part) + var info partitionInfo + if err := adminGET(path, &info); err != nil { + return nil, nil, false, false + } + + rf := len(info.Replicas) + // Only move once the previous reconfiguration has settled, and only if + // there are enough brokers for the replication factor. + if info.Status != "done" || rf == 0 || len(brokers) < rf { + fmt.Printf("%s/%s/%d move skipped: status=%s rf=%d brokers=%d\n", + ns, topic, part, info.Status, rf, len(brokers)) + return info.Replicas, nil, false, false + } + + // Pick rf distinct random brokers (Fisher-Yates via Antithesis rng). + shuffled := make([]brokerInfo, len(brokers)) + copy(shuffled, brokers) + for i := len(shuffled) - 1; i > 0; i-- { + j := randN(i + 1) + shuffled[i], shuffled[j] = shuffled[j], shuffled[i] + } + target := make([]replica, rf) + for i := range rf { + core := 0 + if shuffled[i].NumCores > 0 { + core = randN(shuffled[i].NumCores) + } + target[i] = replica{NodeID: shuffled[i].NodeID, Core: core} + } + + body, _ := json.Marshal(target) + for _, h := range adminHosts() { + resp, err := httpClient.Post( + "http://"+h+path+"/replicas", "application/json", bytes.NewReader(body)) + if err != nil { + continue + } + respBody, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode == http.StatusOK { + fmt.Printf("%s/%s/%d replica move %v -> %v\n", ns, topic, part, info.Replicas, target) + return info.Replicas, target, true, true + } + fmt.Printf("%s/%s/%d move POST %d: %s\n", ns, topic, part, resp.StatusCode, respBody) + } + return info.Replicas, target, true, false +} + +func haveAdminHosts() bool { + h := adminHosts() + return len(h) > 0 && h[0] != "" +} + +// parallel_driver_move_metastore: reassign the single metastore/domain +// partition, exercising reconfiguration of cloud-topics metadata alongside the +// produce/consume workload. +func moveMetastore() error { + if !haveAdminHosts() { + return fmt.Errorf("ADMIN_HOSTS not set") + } + from, to, attempted, accepted := submitMove(metastoreNS, metastoreTopic, metastorePart) + if attempted { + assert.Sometimes(accepted, "cloud topics metastore replica move accepted", + map[string]any{"from": from, "to": to}) + } + return nil +} + +// parallel_driver_move_kafka_topic: reassign a random partition of a random +// kafka test topic, exercising reconfiguration of user data alongside the +// workload. +func moveKafkaTopic() error { + if !haveAdminHosts() { + return fmt.Errorf("ADMIN_HOSTS not set") + } + t := testTopics[randN(len(testTopics))] + part := randN(int(t.partitions)) + from, to, attempted, accepted := submitMove("kafka", t.name, part) + if attempted { + details := map[string]any{"topic": t.name, "partition": part, "from": from, "to": to} + // One liveness property per topic, so a run only proves the mover + // works once a move has been accepted on each topic somewhere. + switch t.name { + case fooTopic: + assert.Sometimes(accepted, "cloud topic foo replica move accepted", details) + case ctcTopic: + assert.Sometimes(accepted, "cloud topic ctc replica move accepted", details) + } + } + return nil +} diff --git a/tests/antithesis/ct_stress/workload/record.go b/tests/antithesis/ct_stress/workload/record.go new file mode 100644 index 0000000000000..53171ba6a2a7b --- /dev/null +++ b/tests/antithesis/ct_stress/workload/record.go @@ -0,0 +1,150 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "fmt" + "hash/crc32" + "strconv" + "strings" + + "github.com/twmb/franz-go/pkg/kgo" +) + +// fooMagic tags every value produced to foo. Concurrent producers +// land records at offsets none of them can predict, so the offset checks alone +// cannot tell whether a record's contents are the ones we wrote. A reader +// recovers that guarantee from the record itself: the magic proves the record +// is one of ours (not garbage, a foreign topic, or a torn read), the trailing +// CRC proves the bytes survived the round trip through cloud storage intact, +// the key echoes the value's identity so a torn-apart key/value pair is caught, +// and the embedded partition proves the record was served from the partition it +// was written to. +const fooMagic = "CTS1" + +// makeFooRecord builds a self-describing record for the given producer nonce, +// per-producer sequence, and target partition. The nonce keeps keys from +// concurrent producers distinct. The layout is: +// +// key = MAGIC:: +// value = MAGIC::::: +// +// where the CRC covers everything before it. Partition is set so a manual +// partitioner honours it and the value's claim matches where it lands. +func makeFooRecord(nonce uint64, seq int, part int32) *kgo.Record { + key := fmt.Sprintf("%s:%016x:%d", fooMagic, nonce, seq) + body := fmt.Sprintf("%s:%d:%x", key, part, randN(1<<30)) + crc := crc32.ChecksumIEEE([]byte(body)) + return &kgo.Record{ + Key: []byte(key), + Value: fmt.Appendf(nil, "%s:%08x", body, crc), + Partition: part, + } +} + +// parsedFooRecord is the decoded identity of a foo record. +type parsedFooRecord struct { + nonce uint64 + seq int64 + part int32 +} + +// parseFooRecord checks that r, read from partition readPart, is a well-formed, +// intact record this workload produced, and returns its decoded identity. The +// reason is empty when the record checks out, otherwise it describes the first +// problem found and the parsedFooRecord is zero. +func parseFooRecord(r *kgo.Record, readPart int32) (parsedFooRecord, string) { + s := string(r.Value) + // The CRC is the final colon-separated field; it is hex, so the last + // colon in the value always separates body from CRC even though the + // payload may itself contain colons. + i := strings.LastIndexByte(s, ':') + if i < 0 { + return parsedFooRecord{}, "value has no crc field" + } + body, crcStr := s[:i], s[i+1:] + crc, err := strconv.ParseUint(crcStr, 16, 32) + if err != nil { + return parsedFooRecord{}, "crc is not hex" + } + if uint32(crc) != crc32.ChecksumIEEE([]byte(body)) { + return parsedFooRecord{}, "crc mismatch (corrupt bytes)" + } + + // body = MAGIC:::: + f := strings.SplitN(body, ":", 5) + if len(f) != 5 || f[0] != fooMagic { + return parsedFooRecord{}, "bad magic or layout" + } + nonce, err := strconv.ParseUint(f[1], 16, 64) + if err != nil { + return parsedFooRecord{}, "nonce is not hex" + } + seq, err := strconv.ParseInt(f[2], 10, 64) + if err != nil { + return parsedFooRecord{}, "seq is not an int" + } + part, err := strconv.ParseInt(f[3], 10, 32) + if err != nil { + return parsedFooRecord{}, "partition is not an int" + } + if int32(part) != readPart { + return parsedFooRecord{}, fmt.Sprintf("partition mismatch: value claims %d, read from %d", part, readPart) + } + wantKey := f[0] + ":" + f[1] + ":" + f[2] + if string(r.Key) != wantKey { + return parsedFooRecord{}, fmt.Sprintf("key/value mismatch: key=%q, value embeds %q", r.Key, wantKey) + } + return parsedFooRecord{nonce: nonce, seq: seq, part: int32(part)}, "" +} + +// verifyResult summarises a scan of a partition's records. +type verifyResult struct { + bad int // records that failed an integrity/provenance check + reordered int // records out of their producer's sequence order + firstOff int64 // offset of the first bad/reordered record, -1 if none + firstReason string +} + +// verifyFooRecords scans recs (in the offset order the broker served them) and +// checks two things: every record is intact and belongs to us (parseFooRecord), +// and each producer's records appear in strictly increasing sequence order. +// The order check holds because a producer writes its batch in seq order and +// the idempotent producer preserves per-partition order, so the seqs of any one +// nonce form an increasing subsequence within a partition; a non-increasing +// step means reordering or a duplicate. Records this reader cannot parse are +// left out of the order check. +func verifyFooRecords(part int32, recs []*kgo.Record) verifyResult { + res := verifyResult{firstOff: -1} + lastSeq := map[uint64]int64{} + for _, r := range recs { + p, reason := parseFooRecord(r, part) + reordered := false + if reason == "" { + if prev, ok := lastSeq[p.nonce]; ok && p.seq <= prev { + reordered = true + reason = fmt.Sprintf("producer %016x out of order: seq %d at or after %d", p.nonce, p.seq, prev) + } else { + lastSeq[p.nonce] = p.seq + } + } + if reason != "" { + if reordered { + res.reordered++ + } else { + res.bad++ + } + if res.firstReason == "" { + res.firstOff, res.firstReason = r.Offset, reason + } + } + } + return res +} diff --git a/tests/antithesis/ct_stress/workload/setup.go b/tests/antithesis/ct_stress/workload/setup.go new file mode 100644 index 0000000000000..f980056568b18 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/setup.go @@ -0,0 +1,82 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "strconv" + "time" + + "github.com/antithesishq/antithesis-sdk-go/lifecycle" + "github.com/twmb/franz-go/pkg/kadm" + "golang.org/x/sys/unix" +) + +// expectedBrokers is the broker count the workload waits for, from the +// EXPECTED_BROKERS env var (set by the compose template), defaulting to 1. +func expectedBrokers() int { + if v := os.Getenv("EXPECTED_BROKERS"); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return 1 +} + +// waitClusterReady blocks until the cluster serves metadata with at least +// `expected` brokers and an elected controller, or `timeout` elapses. Used both +// at startup and by the eventually check while the cluster recovers after faults. +func waitClusterReady(expected int, timeout time.Duration) error { + cl, err := newClient() + if err != nil { + return err + } + defer cl.Close() + adm := kadm.NewClient(cl) + + deadline := time.Now().Add(timeout) + for { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + md, err := adm.Metadata(ctx) + cancel() + if err == nil && len(md.Brokers) >= expected && md.Controller >= 0 { + fmt.Printf("cluster ready: %d brokers, controller %d\n", len(md.Brokers), md.Controller) + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("cluster not ready after timeout (brokers=%d want=%d)", len(md.Brokers), expected) + } + fmt.Printf("waiting for cluster: %d/%d brokers\n", len(md.Brokers), expected) + time.Sleep(2 * time.Second) + } +} + +// setup is the workload container's entrypoint (not a test command): it waits +// until the cluster serves metadata with the expected brokers, emits the +// Antithesis setup_complete signal, then idles so Antithesis can run the test +// commands in this container. setup_complete must come from a long-lived +// process here, never from a test command. +func setup() error { + expected := expectedBrokers() + if err := waitClusterReady(expected, 5*time.Minute); err != nil { + return err + } + + lifecycle.SetupComplete(map[string]any{"brokers": expected}) + fmt.Println("emitted setup_complete; workload container idle") + + sig := make(chan os.Signal, 1) + signal.Notify(sig, unix.SIGINT, unix.SIGTERM) + <-sig + return nil +} diff --git a/tests/antithesis/ct_stress/workload/topics.go b/tests/antithesis/ct_stress/workload/topics.go new file mode 100644 index 0000000000000..134423b1384f9 --- /dev/null +++ b/tests/antithesis/ct_stress/workload/topics.go @@ -0,0 +1,25 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package main + +// testTopic describes one topic the workload creates and exercises. +type testTopic struct { + name string + partitions int32 + replicas int16 + compacted bool +} + +// testTopics is the single definition of the workload's topics; the +// create/flip/move drivers all derive from it. +var testTopics = []testTopic{ + {name: fooTopic, partitions: fooPartitions, replicas: fooReplicas}, + {name: ctcTopic, partitions: ctcPartitions, replicas: ctcReplicas, compacted: true}, +} diff --git a/tools/antithesis/at_common.py b/tools/antithesis/at_common.py index 5b47617ea6c41..51fd01a4d4460 100644 --- a/tools/antithesis/at_common.py +++ b/tools/antithesis/at_common.py @@ -116,26 +116,46 @@ def upload_images(registry: str, images: dict[str, str]) -> None: def submit_test_run( *, password: str, - description: str, duration_min: int, config_image: str, images: list[str], + test_name: str | None = None, + description: str | None = None, + source: str | None = None, + ephemeral: bool = False, recipients: list[str] | None = None, + extra_params: dict[str, str] | None = None, ) -> None: """Launch a basic_test run via the Antithesis API. - config_image and images are fully-qualified registry references (the - ones produced by upload_images). recipients is an optional list of - report email addresses. + test_name sets antithesis.test_name, the run's name in reports; without it + runs carry the launch endpoint's built-in name ("Basic Test"). description + sets antithesis.description, shown in report headers. source sets + antithesis.source (groups property history across runs); ephemeral sets + antithesis.is_ephemeral, keeping the run out of that history. config_image + and images are fully-qualified registry references (the ones produced by + upload_images). recipients is an optional list of report email addresses. + extra_params are additional launch parameters (e.g. tenant-custom `custom.*` + fault-scoping keys) merged into the body. """ params = { - "antithesis.description": description, "antithesis.duration": str(duration_min), "antithesis.config_image": config_image, - "antithesis.images": ";".join(images), } + if test_name: + params["antithesis.test_name"] = test_name + if description: + params["antithesis.description"] = description + if source: + params["antithesis.source"] = source + if ephemeral: + params["antithesis.is_ephemeral"] = "true" + if images: + params["antithesis.images"] = ";".join(images) if recipients: params["antithesis.report.recipients"] = ";".join(recipients) + if extra_params: + params.update(extra_params) body = json.dumps({"params": params}) diff --git a/tools/antithesis/template_deps/redpanda.Dockerfile.j2 b/tools/antithesis/template_deps/redpanda.Dockerfile.j2 new file mode 100644 index 0000000000000..e7dddeb6e2c9d --- /dev/null +++ b/tools/antithesis/template_deps/redpanda.Dockerfile.j2 @@ -0,0 +1,51 @@ +FROM {{ base_image }} +# Base must match the Bazel sysroot's glibc (Ubuntu 22.04 / glibc 2.35, per +# bazel/repositories.bzl). The package bundles that 2.35 glibc; the injected +# Antithesis libvoidstar pulls in the base image's libpthread, so if the base +# has a newer glibc (e.g. redpanda-nightly / Debian 2.36) the versions split +# and libvoidstar fails to dlopen ("Can not load the Antithesis native +# library"). Matching the base to the sysroot keeps a single, consistent glibc. + +# The package's ELF interpreter is baked to {{ install_root }}/lib, so it must +# be installed there. The rpk symlinks let `rpk redpanda start` resolve its +# install dir and launch the Bazel-built redpanda; /symbols exposes the +# unstripped binaries to Antithesis for symbolization. +# curl is used by the compose healthcheck (the stock nightly base shipped it; +# ubuntu:22.04 does not). The rest are for interactive debugging inside an +# Antithesis multiverse-debugger session, where you cannot install tools at +# inspection time: iproute2 (ss/ip/tc), iputils-ping and netcat-openbsd +# (reachability), iptables/nftables/conntrack (see injected network faults), +# procps/lsof (process and socket state), dnsutils and jq (name resolution and +# admin-API/rpk JSON), tcpdump (confirm one-directional drops), gdb (thread +# dumps of a wedged broker; /symbols has the unstripped binaries). +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + curl ca-certificates \ + iproute2 iputils-ping netcat-openbsd \ + iptables nftables conntrack \ + procps lsof dnsutils jq \ + tcpdump gdb \ + && rm -rf /var/lib/apt/lists/* + +COPY --from=packages {{ package_dir }}/ {{ install_root }}/ +RUN mkdir -p /opt/redpanda/bin /symbols /var/lib/redpanda/data /etc/redpanda \ + && ln -sf {{ install_root }}/bin/rpk /usr/bin/rpk \ + && ln -sf {{ install_root }}/bin/rpk /opt/redpanda/bin/rpk \ + && ln -sf {{ install_root }}/bin/redpanda /opt/redpanda/bin/redpanda \ + && ln -s {{ install_root }}/bin/redpanda /symbols/redpanda \ + && find {{ install_root }}/lib/ -name '*.so*' \ + -exec sh -c 'ln -s "$1" /symbols/"$(basename "$1")"' _ {} \; + +COPY < +# vars: +# redpanda_image: +# ... +# +# The directory is rendered into an Antithesis config directory: every +# *.j2 file is rendered with Jinja2 (using the manifest vars, overridable +# via --var/--redpanda-image), everything else is copied verbatim. The +# result must contain a docker-compose.yaml at its root; it is packaged +# into a FROM scratch config image tagged with a content hash of the +# rendered files, and also written to .antithesis// for local runs. +# +# Usage: +# ./tools/antithesis/template_test_package.py --list +# ./tools/antithesis/template_test_package.py ct_stress \ +# --redpanda-image my-instrumented-redpanda:abc123 +# ./tools/antithesis/template_test_package.py ct_stress \ +# --var nodes=3 --bootstrap-set write_caching_default=true --submit +# +# With --build-redpanda-image, Redpanda is built with Bazel (instrumented +# via --config=antithesis unless --no-instrumented) and layered over the +# stock image, and the result is used as the redpanda image: +# ./tools/antithesis/template_test_package.py ct_stress --build-redpanda-image +# + +import argparse +import hashlib +import os +import shlex +import shutil +import sys +import tempfile +from pathlib import Path + +import yaml +from jinja2 import StrictUndefined, Template +from jinja2.exceptions import UndefinedError + +from at_common import ( + add_common_args, + run, + submit_test_run, + validate_common_args, +) + +TOOLS_DIR = Path(__file__).resolve().parent +DEPS_DIR = TOOLS_DIR / "template_deps" +REPO_ROOT = TOOLS_DIR.parent.parent +TESTS_ROOT = REPO_ROOT / "tests" / "antithesis" + +MANIFEST_NAME = "test.yml" +COMPOSE_NAME = "docker-compose.yaml" +# Relative to the config root. The bootstrap config lives under scripts/ so it +# rides the container's scripts bind mount; the entrypoint stages it into place. +BOOTSTRAP_NAME = "scripts/bootstrap.yaml" + +PACKAGE_TARGET = "//bazel/packaging:ducktape" +PACKAGE_DIR = "redpanda_ducktape" +# Baked into the package binaries' ELF interpreter path (install_path on +# the packaging target); the image must install the package there. +INSTALL_ROOT = "/opt/redpanda_installs/redpanda" + + +def render_template(path: Path, **kwargs) -> str: + template = Template( + path.read_text(), + undefined=StrictUndefined, + keep_trailing_newline=True, + ) + return template.render(**kwargs) + + +def git_tag() -> str: + sha = run( + ["git", "rev-parse", "--short=12", "HEAD"], cwd=REPO_ROOT, capture=True + ).stdout.strip() + dirty = run( + ["git", "status", "--porcelain", "--untracked-files=no"], + cwd=REPO_ROOT, + capture=True, + ).stdout.strip() + return f"{sha}-dirty" if dirty else sha + + +def build_redpanda_image(args: argparse.Namespace) -> str: + """Build Redpanda with Bazel and layer it over the base image, keeping + the stock image conventions (entrypoint, rpk, default config) that the + compose templates rely on. Returns the built image reference.""" + if not args.skip_bazel_build: + print("==> Building Redpanda with Bazel") + cmd = ["bazel", "build", PACKAGE_TARGET] + if args.instrumented: + cmd.append("--config=antithesis") + cmd.extend(shlex.split(args.bazel_args)) + run(cmd, cwd=REPO_ROOT) + + pkg_root = REPO_ROOT / "bazel-bin" / "bazel" / "packaging" + if not (pkg_root / PACKAGE_DIR).exists(): + sys.exit( + f"Error: {pkg_root / PACKAGE_DIR} not found. " + f"Run without --skip-bazel-build." + ) + + ref = f"redpanda-antithesis:{git_tag()}" + print(f"==> Building redpanda image: {ref}") + with tempfile.TemporaryDirectory() as tmpdir: + dockerfile = Path(tmpdir) / "Dockerfile" + dockerfile.write_text( + render_template( + DEPS_DIR / "redpanda.Dockerfile.j2", + base_image=args.rp_base_image, + package_dir=PACKAGE_DIR, + install_root=INSTALL_ROOT, + ) + ) + run( + [ + "docker", + "build", + "-f", + str(dockerfile), + "--build-context", + f"packages={pkg_root}", + "--tag", + ref, + tmpdir, + ] + ) + return ref + + +def discover_tests() -> dict[str, Path]: + return { + p.parent.name: p.parent for p in sorted(TESTS_ROOT.glob(f"*/{MANIFEST_NAME}")) + } + + +def parse_assignments(pairs: list[str], flag: str) -> dict: + """Parse repeated KEY=VALUE flags; values are parsed as YAML so that + numbers and booleans keep their type.""" + out = {} + for pair in pairs: + key, sep, value = pair.partition("=") + if not sep or not key: + sys.exit(f"Error: {flag} expects KEY=VALUE, got {pair!r}") + out[key] = yaml.safe_load(value) if value else "" + return out + + +FAULT_PARAM = { + "clock_skew": "custom.clock_skew", + "cpu_mod": "custom.cpu_mod", + "exclude_from_faults": "custom.exclude_from_faults", + "node_hang": "custom.include_for_node_hang", + "node_termination": "custom.include_for_node_termination", + "node_throttle": "custom.include_for_node_throttle", +} + + +def fault_params(manifest: dict, tmpl_vars: dict) -> dict[str, str]: + """Translate the manifest's `faults` block into custom.* launch params. + + Boolean-valued entries (clock_skew, cpu_mod) are global on/off toggles. + List-valued entries name container roles: `redpanda` expands to every broker + (redpanda-0..nodes-1), any other entry is a literal container name. Every + entry is sent explicitly, empty lists included (as ""), so the manifest is + the single source of truth for the fault posture rather than the webhook + defaults. + """ + faults = manifest.get("faults") or {} + + def expand(roles: list[str]) -> list[str]: + out: list[str] = [] + for role in roles: + if role == "redpanda": + nodes = tmpl_vars.get("nodes") + if nodes is None: + sys.exit("Error: faults reference 'redpanda' but 'nodes' is unset") + out.extend(f"redpanda-{i}" for i in range(int(nodes))) + else: + out.append(role) + return out + + params: dict[str, str] = {} + for fault, value in faults.items(): + param = FAULT_PARAM.get(fault) + if param is None: + sys.exit( + f"Error: unknown fault {fault!r} in {MANIFEST_NAME} faults " + f"(known: {', '.join(FAULT_PARAM)})" + ) + if isinstance(value, bool): + params[param] = "true" if value else "false" + else: + params[param] = " ".join(expand(value)) + return params + + +def render_tree(src: Path, dst: Path, tmpl_vars: dict) -> None: + """Render src into dst: *.j2 files are rendered with tmpl_vars (and + lose the suffix), everything else is copied verbatim. The manifest + itself is not part of the output.""" + for path in sorted(src.rglob("*")): + rel = path.relative_to(src) + if rel == Path(MANIFEST_NAME): + continue + target = dst / rel + if path.is_dir(): + target.mkdir(parents=True, exist_ok=True) + continue + target.parent.mkdir(parents=True, exist_ok=True) + if path.suffix == ".j2": + try: + rendered = render_template(path, **tmpl_vars) + except UndefinedError as e: + sys.exit( + f"Error: rendering {path} failed: {e} " + f"(set it in {MANIFEST_NAME} vars or pass --var)" + ) + target = target.with_suffix("") + target.write_text(rendered) + shutil.copymode(path, target) + else: + shutil.copy2(path, target) + + +def content_tag(root: Path) -> str: + """Deterministic tag derived from the rendered config contents, so a + given tag always identifies exactly one configuration.""" + digest = hashlib.sha256() + for path in sorted(p for p in root.rglob("*") if p.is_file()): + digest.update(str(path.relative_to(root)).encode()) + digest.update(b"\0") + digest.update(path.read_bytes()) + digest.update(b"\0") + return digest.hexdigest()[:12] + + +def is_registry_qualified(image: str) -> bool: + """True if the image reference names a registry host (docker's own + heuristic), i.e. it is pullable and does not need to be pushed to the + Antithesis registry by us.""" + if "/" not in image: + return False + first = image.split("/", 1)[0] + return "." in first or ":" in first or first == "localhost" + + +def build_config_image(tag: str, config_dir: Path) -> None: + print(f"==> Building config image: {tag}") + with tempfile.TemporaryDirectory() as tmpdir: + dockerfile = Path(tmpdir) / "Dockerfile" + dockerfile.write_text("FROM scratch\nCOPY . /\n") + run( + [ + "docker", + "build", + "-f", + str(dockerfile), + "--tag", + tag, + str(config_dir), + ] + ) + + +def push_image(registry: str, local_ref: str) -> str: + ref = f"{registry}/{local_ref}" + run(["docker", "tag", local_ref, ref]) + run(["docker", "push", ref]) + return ref + + +def build_companion_images( + name: str, test_dir: Path, manifest: dict, slug: str +) -> dict[str, str]: + """Build the images declared in the manifest's `images:` map. + + Returns {template_var: image_ref}. Each image is a local build tagged + with a content hash of its build context, so a given tag identifies + exactly one build. The refs are set as template variables and, for real + runs, pushed and submitted so Antithesis injects libvoidstar into them. + """ + built: dict[str, str] = {} + for var_name, spec in (manifest.get("images") or {}).items(): + context = test_dir / spec["context"] + if not context.is_dir(): + sys.exit( + f"Error: test {name} image {var_name}: context {context} does not exist" + ) + dockerfile = test_dir / spec.get("dockerfile", f"{spec['context']}/Dockerfile") + short = var_name.removesuffix("_image").replace("_", "-") + ref = f"{slug}-{short}:{content_tag(context)}" + print(f"==> Building image {var_name}: {ref}") + run(["docker", "build", "-f", str(dockerfile), "-t", ref, str(context)]) + built[var_name] = ref + return built + + +def package_test(name: str, test_dir: Path, args: argparse.Namespace) -> None: + print(f"==> Packaging test: {name}") + + manifest = yaml.safe_load((test_dir / MANIFEST_NAME).read_text()) or {} + tmpl_vars: dict = {"test_name": name, "test_slug": name.replace("_", "-")} + tmpl_vars.update(manifest.get("vars") or {}) + tmpl_vars.update(parse_assignments(args.var, "--var")) + if args.redpanda_image: + tmpl_vars["redpanda_image"] = args.redpanda_image + bootstrap_overrides = parse_assignments(args.bootstrap_set, "--bootstrap-set") + + companion = build_companion_images(name, test_dir, manifest, tmpl_vars["test_slug"]) + tmpl_vars.update(companion) + + with tempfile.TemporaryDirectory() as tmpdir: + staging = Path(tmpdir) / "config" + staging.mkdir() + render_tree(test_dir, staging, tmpl_vars) + + if not (staging / COMPOSE_NAME).exists(): + sys.exit(f"Error: test {name} does not produce a {COMPOSE_NAME}") + + if bootstrap_overrides: + bootstrap_path = staging / BOOTSTRAP_NAME + if not bootstrap_path.exists(): + sys.exit( + f"Error: --bootstrap-set given but test {name} " + f"has no {BOOTSTRAP_NAME}" + ) + # Note: rewriting drops the comments from the rendered file. + bootstrap = yaml.safe_load(bootstrap_path.read_text()) or {} + bootstrap.update(bootstrap_overrides) + bootstrap_path.write_text(yaml.safe_dump(bootstrap, sort_keys=False)) + + tag = args.tag or content_tag(staging) + config_ref = f"{tmpl_vars['test_slug']}-config:{tag}" + build_config_image(config_ref, staging) + + # Keep a rendered copy for local runs. + out_dir = REPO_ROOT / ".antithesis" / name + if out_dir.exists(): + shutil.rmtree(out_dir) + out_dir.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(staging, out_dir) + + # The config image and companion images are always local builds that need + # pushing; the redpanda image only if it is a local build (registry- + # qualified references are pullable as-is). + local_refs = [config_ref, *companion.values()] + rp_image = tmpl_vars.get("redpanda_image", "") + if rp_image and not is_registry_qualified(rp_image): + local_refs.append(rp_image) + + pushed: dict[str, str] = {} + if not args.skip_registry_upload: + for ref in local_refs: + pushed[ref] = push_image(args.registry, ref) + + if args.submit: + submit_test_run( + password=os.environ["AT_PASSWORD"], + test_name=manifest.get("name") or name, + description=args.description or None, + source=args.source or None, + ephemeral=args.ephemeral, + duration_min=args.duration, + config_image=pushed[config_ref], + images=[v for k, v in pushed.items() if k != config_ref], + recipients=[r for r in args.recipients.split(";") if r] or None, + extra_params=fault_params(manifest, tmpl_vars), + ) + + compose_path = out_dir / COMPOSE_NAME + lines = [ + "", + f"Test {name} packaged:", + f" config image: {config_ref}", + f" redpanda image: {rp_image}", + ] + if pushed: + lines.append("Pushed:") + lines.extend(f" {ref}" for ref in pushed.values()) + else: + lines.append("Push skipped (--skip-registry-upload); to push manually:") + for ref in local_refs: + lines.append(f" docker tag {ref} {args.registry}/{ref}") + lines.append(f" docker push {args.registry}/{ref}") + lines += [ + "Run locally:", + f" docker compose -f {compose_path} up -d", + f" docker compose -f {compose_path} down -v", + "", + ] + print("\n".join(lines)) + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Package template-based Antithesis tests " + f"(directories with a {MANIFEST_NAME} under {TESTS_ROOT})." + ) + parser.add_argument( + "tests", + nargs="*", + help="Names of the tests to package (default: all)", + ) + parser.add_argument( + "--list", + action="store_true", + help="List the available tests and exit", + ) + parser.add_argument( + "--redpanda-image", + default="", + help="Override the redpanda_image template variable", + ) + parser.add_argument( + "--build-redpanda-image", + action="store_true", + help="Build Redpanda with Bazel and package it as the redpanda " + "image (used as the redpanda_image template variable)", + ) + parser.add_argument( + "--rp-base-image", + default="ubuntu:22.04", + help="Base image the Bazel-built Redpanda is layered onto. Must match " + "the Bazel sysroot glibc (Ubuntu 22.04) so the Antithesis-injected " + "libvoidstar loads; a newer base splits glibc versions and breaks it.", + ) + parser.add_argument( + "--instrumented", + action=argparse.BooleanOptionalAction, + default=True, + help="Build with --config=antithesis (default: enabled)", + ) + parser.add_argument( + "--bazel-args", + default="", + help="Extra arguments passed to bazel build", + ) + parser.add_argument( + "--skip-bazel-build", + action="store_true", + help="Skip building Redpanda (use existing Bazel artifacts)", + ) + parser.add_argument( + "--var", + action="append", + default=[], + metavar="KEY=VALUE", + help="Override a template variable (repeatable; values are parsed as YAML)", + ) + parser.add_argument( + "--bootstrap-set", + action="append", + default=[], + metavar="KEY=VALUE", + help=f"Override a cluster property in the rendered {BOOTSTRAP_NAME} " + "(repeatable; values are parsed as YAML)", + ) + parser.add_argument( + "--tag", + default="", + help="Image tag (default: content hash of the rendered config)", + ) + parser.add_argument( + "--source", + default="", + help="antithesis.source: groups property history across runs. Use a " + "stable key such as the git branch; runs sharing a source share " + "history. Required for --no-ephemeral runs.", + ) + parser.add_argument( + "--ephemeral", + action=argparse.BooleanOptionalAction, + default=True, + help="Keep the run out of the findings history (antithesis.is_ephemeral). " + "Default: ephemeral, for ad-hoc runs. Pass --no-ephemeral for a " + "persistent run recorded to history (requires --source).", + ) + add_common_args(parser) + args = parser.parse_args() + + validate_common_args(parser, args) + + if args.submit and not args.ephemeral and not args.source: + parser.error( + "--no-ephemeral runs must set --source; use a stable grouping key " + 'such as the git branch, e.g. --source "$(git branch --show-current)"' + ) + + available = discover_tests() + if args.list: + for name, path in available.items(): + manifest = yaml.safe_load((path / MANIFEST_NAME).read_text()) or {} + print(f"{name}: {manifest.get('name', '')}") + return + if not available: + sys.exit(f"Error: no tests found under {TESTS_ROOT}") + + selected = args.tests or list(available) + unknown = [t for t in selected if t not in available] + if unknown: + parser.error( + f"unknown tests: {', '.join(unknown)} (available: {', '.join(available)})" + ) + + if args.submit and not os.environ.get("AT_PASSWORD"): + sys.exit("Error: --submit requires the AT_PASSWORD environment variable") + + if args.build_redpanda_image: + if args.redpanda_image: + parser.error( + "--redpanda-image and --build-redpanda-image are mutually exclusive" + ) + args.redpanda_image = build_redpanda_image(args) + + for name in selected: + package_test(name, available[name], args) + + +if __name__ == "__main__": + main()