refactor(soak): split the complex functions of the soak tester and of the cleanup command (#319)

Tests pin the behavior before the refactor: the error texts of the config validation, the defaults, the metric descriptions, the order of the API calls of Delete, fetch, rotateOffline and collect, golden plans of newPlan and planAutoscale, the window of execute, and the output of the cleanup command. The golden files were written on the base code.

Mutations.validate, Config.setDefaults, Config.validate, Mutator.execute, Mutator.fetch, newPlan, planResources, planAutoscale, stuckChange, rotateOffline, Observer.apply, Observer.collect, Instance.Delete, metrics.New, run and runCleanup are split into helpers, so that each has a cyclomatic complexity of at most 15, a cognitive complexity of at most 20 and at most 80 lines. The reads of the instance move to mutator/fetch.go and the reads of the pods to observer/pods.go (moved code unchanged), so that each file has fewer than 800 lines. TestDeleteOrder ends its waits after a number of calls, so that it does not depend on the speed of the machine.

No change of logic, messages or the order of API calls. Left over: validateVersions, Chaos.validate, validateChain, Observer.observeMaster and validateAutoscale are still above 15. Not tested: the rest of run, which needs a cluster.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01G4mmyo3sqLwf23m5FE2nBE

Author: Saremox

cmd/instance/cleanup/cmd.go

@@ -67,40 +67,10 @@ func runCleanup(_ *cobra.Command, _ []string) error {
 	var totalSize int64
 
 	for _, entry := range entries {
-		if entry.IsDir() {
-			continue
+		if size, ok := cleanEntry(entry); ok {
+			cleaned++
+			totalSize += size
 		}
-
-		name := entry.Name()
-
-		if !strings.HasSuffix(name, ".rdb") {
-			continue
-		}
-
-		if name == dbFilename {
-			continue
-		}
-
-		filePath := filepath.Join(dataDir, name)
-
-		fileInfo, err := entry.Info()
-		if err != nil {
-			fmt.Printf("Warning: failed to get info for %s: %v\n", name, err)
-			continue
-		}
-
-		if dryRun {
-			fmt.Printf("Would delete: %s (%d bytes)\n", filePath, fileInfo.Size())
-		} else {
-			if err := os.Remove(filePath); err != nil {
-				fmt.Printf("Warning: failed to remove %s: %v\n", filePath, err)
-				continue
-			}
-			fmt.Printf("Deleted: %s (%d bytes)\n", filePath, fileInfo.Size())
-		}
-
-		cleaned++
-		totalSize += fileInfo.Size()
 	}
 
 	if cleaned > 0 {
@@ -115,3 +85,42 @@ func runCleanup(_ *cobra.Command, _ []string) error {
 
 	return nil
 }
+
+// cleanEntry removes the entry if it is a stale RDB file, or prints it in a
+// dry run. It returns the size of the file, and whether the file counts as
+// cleaned.
+func cleanEntry(entry os.DirEntry) (int64, bool) {
+	if entry.IsDir() {
+		return 0, false
+	}
+
+	name := entry.Name()
+
+	if !strings.HasSuffix(name, ".rdb") {
+		return 0, false
+	}
+
+	if name == dbFilename {
+		return 0, false
+	}
+
+	filePath := filepath.Join(dataDir, name)
+
+	fileInfo, err := entry.Info()
+	if err != nil {
+		fmt.Printf("Warning: failed to get info for %s: %v\n", name, err)
+		return 0, false
+	}
+
+	if dryRun {
+		fmt.Printf("Would delete: %s (%d bytes)\n", filePath, fileInfo.Size())
+	} else {
+		if err := os.Remove(filePath); err != nil {
+			fmt.Printf("Warning: failed to remove %s: %v\n", filePath, err)
+			return 0, false
+		}
+		fmt.Printf("Deleted: %s (%d bytes)\n", filePath, fileInfo.Size())
+	}
+
+	return fileInfo.Size(), true
+}

cmd/instance/cleanup/output_test.go

@@ -0,0 +1,71 @@
+package cleanup
+
+import (
+	"errors"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stdout returns what f prints on standard output.
+func stdout(t *testing.T, f func() error) string {
+	t.Helper()
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	prev := os.Stdout
+	os.Stdout = w
+	out := make(chan string)
+	go func() {
+		b, _ := io.ReadAll(r)
+		out <- string(b)
+	}()
+	ferr := f()
+	os.Stdout = prev
+	require.NoError(t, w.Close())
+	require.NoError(t, ferr)
+	return <-out
+}
+
+// TestRunCleanupOutput pins the lines that cleanup prints for each entry and
+// for the summary.
+func TestRunCleanupOutput(t *testing.T) {
+	dir := t.TempDir()
+	stale := filepath.Join(dir, "temp-1.rdb")
+	require.NoError(t, os.WriteFile(stale, []byte("stale"), 0o600))
+	staleInfo, err := os.Stat(stale)
+	require.NoError(t, err)
+	entries := func(string) ([]fs.DirEntry, error) {
+		return []fs.DirEntry{
+			fakeDirEntry{name: "dump.rdb", info: staleInfo},
+			fakeDirEntry{name: "notes.txt", info: staleInfo},
+			fakeDirEntry{name: "no-info.rdb", infoErr: errors.New("injected info error")},
+			fakeDirEntry{name: "gone.rdb", info: staleInfo},
+			fakeDirEntry{name: "temp-1.rdb", info: staleInfo},
+		}, nil
+	}
+	gone := filepath.Join(dir, "gone.rdb")
+
+	setCleanupFlags(t, dir, "dump.rdb", true)
+	readDir = entries
+	assert.Equal(t, "Warning: failed to get info for no-info.rdb: injected info error\n"+
+		"Would delete: "+gone+" (5 bytes)\n"+
+		"Would delete: "+stale+" (5 bytes)\n"+
+		"Would delete 2 stale RDB file(s), freed 10 bytes\n", stdout(t, func() error { return runCleanup(nil, nil) }))
+
+	dryRun = false
+	assert.Equal(t, "Warning: failed to get info for no-info.rdb: injected info error\n"+
+		"Warning: failed to remove "+gone+": remove "+gone+": no such file or directory\n"+
+		"Deleted: "+stale+" (5 bytes)\n"+
+		"Deleted 1 stale RDB file(s), freed 5 bytes\n", stdout(t, func() error { return runCleanup(nil, nil) }))
+
+	readDir = os.ReadDir
+	assert.Equal(t, "No stale RDB files found\n", stdout(t, func() error { return runCleanup(nil, nil) }))
+
+	dataDir = filepath.Join(dir, "missing")
+	assert.Equal(t, "Data directory "+dataDir+" does not exist, skipping cleanup\n", stdout(t, func() error { return runCleanup(nil, nil) }))
+}

test/soak/cmd/soak/main.go

@@ -56,19 +56,7 @@ func run(configPath, listen string, log *slog.Logger) error {
 	if err != nil {
 		return err
 	}
-	restCfg, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
-		clientcmd.NewDefaultClientConfigLoadingRules(), nil).ClientConfig()
-	if err != nil {
-		return err
-	}
-	// The observers list pods, EndpointSlices and RedisFailovers for every
-	// instance every few seconds.
-	restCfg.QPS, restCfg.Burst = 50, 100
-	kube, err := kubernetes.NewForConfig(restCfg)
-	if err != nil {
-		return err
-	}
-	rfs, err := versioned.NewForConfig(restCfg)
+	kube, rfs, err := clients()
 	if err != nil {
 		return err
 	}
@@ -90,15 +78,7 @@ func run(configPath, listen string, log *slog.Logger) error {
 		return err
 	}
 	lock := &global.Lock{}
-	observers := map[string]*observer.Observer{}
-	datas := map[string]*data.Data{}
-	for _, in := range cfg.Instances {
-		o := observer.New(in, cfg, kube, rfs, sources[in.Name], lock, m, log)
-		observers[in.Name] = o
-		if in.Data != nil {
-			datas[in.Name] = data.New(in, cfg, o, sources[in.Name], m, log)
-		}
-	}
+	observers, datas := newObservers(cfg, kube, rfs, sources, lock, m, log)
 
 	var wg sync.WaitGroup
 	wg.Go(func() { reportBuildInfo(ctx, kube, cfg.Operator, m, log) })
@@ -125,14 +105,7 @@ func run(configPath, listen string, log *slog.Logger) error {
 			wg.Go(func() { mu.Run(ctx) })
 		}
 		wg.Go(func() { o.Run(ctx) })
-		// An outage counts for the chaos action, or else the mutation, that
-		// ran when it started.
-		r.SetEvent(func() string {
-			if k := lock.Chaos(); k != "" || mu == nil {
-				return k
-			}
-			return mu.Current()
-		})
+		r.SetEvent(outageEvent(lock, mu))
 		wg.Go(func() { r.Run(ctx, func() []string { return prober.Names(in, o.SentinelPath()) }) })
 		lane = append(lane, chaos.Instance{Name: in.Name, Namespace: in.Namespace, Observer: o, Data: d})
 	}
@@ -157,6 +130,54 @@ func run(configPath, listen string, log *slog.Logger) error {
 	return err
 }
 
+// outageEvent returns the event of an outage: the chaos action, or else the
+// mutation of mu, that ran when it started.
+func outageEvent(lock *global.Lock, mu *mutator.Mutator) func() string {
+	return func() string {
+		if k := lock.Chaos(); k != "" || mu == nil {
+			return k
+		}
+		return mu.Current()
+	}
+}
+
+// clients returns the clients of the cluster that the kubeconfig or the
+// in-cluster config names.
+func clients() (*kubernetes.Clientset, *versioned.Clientset, error) {
+	restCfg, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
+		clientcmd.NewDefaultClientConfigLoadingRules(), nil).ClientConfig()
+	if err != nil {
+		return nil, nil, err
+	}
+	// The observers list pods, EndpointSlices and RedisFailovers for every
+	// instance every few seconds.
+	restCfg.QPS, restCfg.Burst = 50, 100
+	kube, err := kubernetes.NewForConfig(restCfg)
+	if err != nil {
+		return nil, nil, err
+	}
+	rfs, err := versioned.NewForConfig(restCfg)
+	if err != nil {
+		return nil, nil, err
+	}
+	return kube, rfs, nil
+}
+
+// newObservers returns the observer of each instance, and the data of each
+// instance with data, by instance name.
+func newObservers(cfg *config.Config, kube kubernetes.Interface, rfs versioned.Interface, sources map[string]*auth.Source, lock *global.Lock, m *metrics.Metrics, log *slog.Logger) (map[string]*observer.Observer, map[string]*data.Data) {
+	observers := map[string]*observer.Observer{}
+	datas := map[string]*data.Data{}
+	for _, in := range cfg.Instances {
+		o := observer.New(in, cfg, kube, rfs, sources[in.Name], lock, m, log)
+		observers[in.Name] = o
+		if in.Data != nil {
+			datas[in.Name] = data.New(in, cfg, o, sources[in.Name], m, log)
+		}
+	}
+	return observers, datas
+}
+
 // ensureInstances creates each instance with a template that does not exist
 // yet, on its configured versions.
 func ensureInstances(ctx context.Context, kube kubernetes.Interface, rfs versioned.Interface, cfg *config.Config, log *slog.Logger) (map[string]*instances.Instance, error) {

test/soak/cmd/soak/main_test.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/client-go/kubernetes/fake"
+
+	rffake "github.com/saremox/redis-operator/client/k8s/clientset/versioned/fake"
+	"github.com/saremox/redis-operator/test/soak/internal/auth"
+	"github.com/saremox/redis-operator/test/soak/internal/config"
+	"github.com/saremox/redis-operator/test/soak/internal/global"
+	"github.com/saremox/redis-operator/test/soak/internal/metrics"
+	"github.com/saremox/redis-operator/test/soak/internal/mutator"
+)
+
+// run returns the error of the config, and of the client config, before it
+// starts anything.
+func TestRunEarlyErrors(t *testing.T) {
+	dir := t.TempDir()
+	log := slog.New(slog.DiscardHandler)
+	if err := run(filepath.Join(dir, "missing.yaml"), ":0", log); err == nil || !strings.Contains(err.Error(), "no such file or directory") {
+		t.Errorf("missing config: %v", err)
+	}
+	cfg := filepath.Join(dir, "config.yaml")
+	if err := os.WriteFile(cfg, []byte("instances: [{name: a}]"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	kubeconfig := filepath.Join(dir, "kubeconfig")
+	if err := os.WriteFile(kubeconfig, []byte("{"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("KUBECONFIG", kubeconfig)
+	t.Setenv("KUBERNETES_SERVICE_HOST", "")
+	if err := run(cfg, ":0", log); err == nil || !strings.Contains(err.Error(), kubeconfig) {
+		t.Errorf("bad kubeconfig: %v", err)
+	}
+}
+
+// clients reads the kubeconfig and raises the rate limit of the observers.
+func TestClients(t *testing.T) {
+	kubeconfig := filepath.Join(t.TempDir(), "kubeconfig")
+	const body = `apiVersion: v1
+kind: Config
+clusters: [{name: c, cluster: {server: "https://127.0.0.1:1"}}]
+users: [{name: u, user: {token: t}}]
+contexts: [{name: x, context: {cluster: c, user: u}}]
+current-context: x
+`
+	if err := os.WriteFile(kubeconfig, []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("KUBECONFIG", kubeconfig)
+	kube, rfs, err := clients()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if qps := kube.CoreV1().RESTClient().GetRateLimiter().QPS(); qps != 50 || rfs == nil {
+		t.Errorf("qps %v, rfs %v", qps, rfs)
+	}
+}
+
+// An outage counts for the chaos action, or else the mutation, that ran when
+// it started.
+func TestOutageEvent(t *testing.T) {
+	lock := &global.Lock{}
+	if got := outageEvent(lock, nil)(); got != "" {
+		t.Errorf("no event: %q", got)
+	}
+	if got := outageEvent(lock, &mutator.Mutator{})(); got != "" {
+		t.Errorf("no mutation: %q", got)
+	}
+	lock.SetChaos("operator_restart")
+	for _, mu := range []*mutator.Mutator{nil, {}} {
+		if got := outageEvent(lock, mu)(); got != "operator_restart" {
+			t.Errorf("chaos: %q", got)
+		}
+	}
+}
+
+// newObservers makes an observer of each instance, and the data of the
+// instances with data.
+func TestNewObservers(t *testing.T) {
+	cfg, err := config.Parse([]byte("instances: [{name: a, data: {fill: {}}}, {name: b}]"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	sources := map[string]*auth.Source{"a": auth.New(nil), "b": auth.New(nil)}
+	m := metrics.New(prometheus.NewRegistry(), time.Minute)
+	observers, datas := newObservers(cfg, fake.NewClientset(), rffake.NewSimpleClientset(), sources, &global.Lock{}, m, slog.New(slog.DiscardHandler))
+	if len(observers) != 2 || observers["a"] == nil || observers["b"] == nil {
+		t.Errorf("observers %v", observers)
+	}
+	if len(datas) != 1 || datas["a"] == nil {
+		t.Errorf("data %v", datas)
+	}
+}