Files
seaweedfs/weed/storage/blockvol/testrunner/actions/system.go
T
pingqiuandClaude Opus 4.6 785a7d7efd feat: wire real pinner into flusher retention + real WAL scan executor (Phase 07 P1)
Pinner wired to real retention:
- NewPinner calls vol.SetV2RetentionFloor(p.MinWALRetentionFloor)
- Flusher.RetentionFloorFn() / SetRetentionFloorFn() exposed
- SetV2RetentionFloor chains with existing shipper retention floor
- Holds actually prevent WAL reclaim (not just tracked state)

Executor uses real WAL scan:
- BlockVol.ScanWALEntries(fromLSN, callback) wraps wal.ScanFrom
  with real fd, walOffset, checkpointLSN
- Executor.StreamWALEntries uses ScanWALEntries (not stub)
- Reads real WAL entries, tracks highest LSN scanned

CommittedLSN mapping:
- Explicitly documented as interim V1 model (committed = checkpointed)
- Will diverge when V2 distributed commit separates from local flush

Carry-forward:
- TransferSnapshot/TransferFullBase/TruncateWAL: stubs (need extent I/O)
- Control intent from confirmed failover: deferred

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-30 20:01:46 -07:00

239 lines
6.7 KiB
Go

package actions
import (
"context"
"fmt"
"strconv"
"strings"
"time"
tr "github.com/seaweedfs/seaweedfs/weed/storage/blockvol/testrunner"
)
// RegisterSystemActions registers system/assert actions.
func RegisterSystemActions(r *tr.Registry) {
r.RegisterFunc("exec", tr.TierCore, execAction)
r.RegisterFunc("sleep", tr.TierCore, sleepAction)
r.RegisterFunc("assert_equal", tr.TierCore, assertEqual)
r.RegisterFunc("assert_greater", tr.TierCore, assertGreater)
r.RegisterFunc("assert_status", tr.TierCore, assertStatus)
r.RegisterFunc("assert_contains", tr.TierCore, assertContains)
r.RegisterFunc("print", tr.TierCore, printAction)
r.RegisterFunc("fsck_ext4", tr.TierBlock, fsckExt4)
r.RegisterFunc("fsck_xfs", tr.TierBlock, fsckXfs)
r.RegisterFunc("grep_log", tr.TierCore, grepLog)
}
func execAction(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
cmd := act.Params["cmd"]
if cmd == "" {
return nil, fmt.Errorf("exec: cmd param required")
}
node, err := GetNode(actx, act.Node)
if err != nil {
return nil, err
}
root := act.Params["root"] == "true"
var stdout, stderr string
var code int
if root {
stdout, stderr, code, err = node.RunRoot(ctx, cmd)
} else {
stdout, stderr, code, err = node.Run(ctx, cmd)
}
if err != nil {
return nil, fmt.Errorf("exec: %w", err)
}
if code != 0 {
return nil, fmt.Errorf("exec: code=%d stderr=%s", code, stderr)
}
return map[string]string{"value": strings.TrimSpace(stdout)}, nil
}
func sleepAction(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
d := act.Params["duration"]
if d == "" {
d = "1s"
}
dur, err := time.ParseDuration(d)
if err != nil {
return nil, fmt.Errorf("sleep: invalid duration %q: %w", d, err)
}
select {
case <-time.After(dur):
return nil, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
func assertEqual(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
actual := act.Params["actual"]
expected := act.Params["expected"]
if actual != expected {
return nil, fmt.Errorf("assert_equal: %q != %q", actual, expected)
}
return nil, nil
}
func assertGreater(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
actualStr := act.Params["actual"]
threshStr := act.Params["threshold"]
if threshStr == "" {
threshStr = act.Params["expected"] // backward compat
}
actual, err := strconv.ParseFloat(actualStr, 64)
if err != nil {
return nil, fmt.Errorf("assert_greater: cannot parse actual %q as number: %w", actualStr, err)
}
threshold, err := strconv.ParseFloat(threshStr, 64)
if err != nil {
return nil, fmt.Errorf("assert_greater: cannot parse threshold %q as number: %w", threshStr, err)
}
if actual <= threshold {
return nil, fmt.Errorf("assert_greater: %.2f <= %.2f", actual, threshold)
}
return nil, nil
}
func assertStatus(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
tgt, err := getHATarget(actx, act.Target)
if err != nil {
return nil, err
}
st, err := tgt.Status(ctx)
if err != nil {
return nil, fmt.Errorf("assert_status: %w", err)
}
if role, ok := act.Params["role"]; ok {
if st.Role != role {
return nil, fmt.Errorf("assert_status: role %q != expected %q", st.Role, role)
}
}
if healthy, ok := act.Params["healthy"]; ok {
expectedHealthy := healthy == "true"
if st.Healthy != expectedHealthy {
return nil, fmt.Errorf("assert_status: healthy=%v != expected=%v", st.Healthy, expectedHealthy)
}
}
if hasLease, ok := act.Params["has_lease"]; ok {
expectedLease := hasLease == "true"
if st.HasLease != expectedLease {
return nil, fmt.Errorf("assert_status: has_lease=%v != expected=%v", st.HasLease, expectedLease)
}
}
return nil, nil
}
func assertContains(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
haystack := act.Params["value"]
needle := act.Params["contains"]
if !strings.Contains(haystack, needle) {
return nil, fmt.Errorf("assert_contains: %q not found in %q", needle, haystack)
}
return nil, nil
}
func printAction(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
msg := act.Params["msg"]
if msg == "" {
msg = act.Params["message"]
}
actx.Log(" [print] %s", msg)
return nil, nil
}
// fsckExt4 runs e2fsck -fn on an unmounted ext4 device. Fails if exit code >= 4.
// Params: device (required)
func fsckExt4(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
device := act.Params["device"]
if device == "" {
return nil, fmt.Errorf("fsck_ext4: device param required")
}
node, err := GetNode(actx, act.Node)
if err != nil {
return nil, err
}
stdout, stderr, code, err := node.RunRoot(ctx, fmt.Sprintf("e2fsck -fn %s 2>&1", device))
if err != nil {
return nil, fmt.Errorf("fsck_ext4: %w", err)
}
// e2fsck exit codes: 0=clean, 1=errors corrected, 2=reboot needed, 4+=serious error.
if code >= 4 {
return nil, fmt.Errorf("fsck_ext4: code=%d output=%s stderr=%s", code, stdout, stderr)
}
output := strings.TrimSpace(stdout)
return map[string]string{"value": output}, nil
}
// fsckXfs runs xfs_repair -n on an unmounted XFS device. Fails if non-zero exit.
// Params: device (required)
func fsckXfs(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
device := act.Params["device"]
if device == "" {
return nil, fmt.Errorf("fsck_xfs: device param required")
}
node, err := GetNode(actx, act.Node)
if err != nil {
return nil, err
}
stdout, stderr, code, err := node.RunRoot(ctx, fmt.Sprintf("xfs_repair -n %s 2>&1", device))
if err != nil {
return nil, fmt.Errorf("fsck_xfs: %w", err)
}
if code != 0 {
return nil, fmt.Errorf("fsck_xfs: code=%d output=%s stderr=%s", code, stdout, stderr)
}
output := strings.TrimSpace(stdout)
return map[string]string{"value": output}, nil
}
// grepLog counts occurrences of a pattern in a file. Returns count as value.
// Params: path (required), pattern (required)
func grepLog(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) {
path := act.Params["path"]
if path == "" {
return nil, fmt.Errorf("grep_log: path param required")
}
pattern := act.Params["pattern"]
if pattern == "" {
return nil, fmt.Errorf("grep_log: pattern param required")
}
node, err := GetNode(actx, act.Node)
if err != nil {
return nil, err
}
cmd := fmt.Sprintf("grep -c '%s' %s || true", pattern, path)
stdout, _, _, err := node.Run(ctx, cmd)
if err != nil {
return nil, fmt.Errorf("grep_log: %w", err)
}
count := strings.TrimSpace(stdout)
if count == "" {
count = "0"
}
return map[string]string{"value": count}, nil
}