Tanishq Dubey
58bdca5703
**Phase 1: State Management & Leader Election**
* **Goal**: A functional embedded etcd and leader election mechanism.
* **Tasks**:
1. Implement the `StateStore` interface (RFC 5.1) with an etcd backend (`internal/store/etcd.go`).
2. Integrate embedded etcd server into `kat-agent` (RFC 2.2, 5.2), configurable via `cluster.kat` parameters.
3. Implement leader election using `go.etcd.io/etcd/client/v3/concurrency` (RFC 5.3).
4. Basic `kat-agent init` functionality:
* Parse `cluster.kat`.
* Start single-node embedded etcd.
* Campaign for and become leader.
* Store initial cluster configuration (UID, CIDRs from `cluster.kat`) in etcd.
* **Milestone**:
* A single `kat-agent init --config cluster.kat` process starts, initializes etcd, and logs that it has become the leader.
* The cluster configuration from `cluster.kat` can be verified in etcd using an etcd client.
* `StateStore` interface methods (`Put`, `Get`, `Delete`, `List`) are testable against the embedded etcd.
Reviewed-on: #1
86 lines
2.3 KiB
Go
86 lines
2.3 KiB
Go
package testutil
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.dws.rip/dubey/kat/internal/store"
|
|
"github.com/stretchr/testify/require"
|
|
"go.etcd.io/etcd/server/v3/embed"
|
|
)
|
|
|
|
// SetupEmbeddedEtcd creates a temporary directory and starts an embedded etcd server for testing
|
|
func SetupEmbeddedEtcd(t *testing.T) (string, *embed.Etcd, string) {
|
|
// Create a temporary directory for etcd data
|
|
tempDir, err := os.MkdirTemp("", "etcd-test-*")
|
|
require.NoError(t, err)
|
|
|
|
// Configure and start embedded etcd
|
|
etcdConfig := store.EtcdEmbedConfig{
|
|
Name: "test-node",
|
|
DataDir: tempDir,
|
|
ClientURLs: []string{"http://localhost:0"}, // Use port 0 to get a random available port
|
|
PeerURLs: []string{"http://localhost:0"},
|
|
InitialCluster: "test-node=http://localhost:0",
|
|
}
|
|
|
|
etcdServer, err := store.StartEmbeddedEtcd(etcdConfig)
|
|
require.NoError(t, err)
|
|
|
|
// Get the actual client URL that was assigned
|
|
clientURL := etcdServer.Clients[0].Addr().String()
|
|
|
|
return tempDir, etcdServer, clientURL
|
|
}
|
|
|
|
// CreateTestClusterConfig creates a test cluster.kat file in the specified directory
|
|
func CreateTestClusterConfig(t *testing.T, dir string) string {
|
|
configContent := `apiVersion: kat.dws.rip/v1alpha1
|
|
kind: ClusterConfiguration
|
|
metadata:
|
|
name: test-cluster
|
|
spec:
|
|
clusterCidr: "10.100.0.0/16"
|
|
serviceCidr: "10.101.0.0/16"
|
|
nodeSubnetBits: 7
|
|
clusterDomain: "test.cluster.local"
|
|
agentPort: 9116
|
|
apiPort: 9115
|
|
etcdPeerPort: 2380
|
|
etcdClientPort: 2379
|
|
volumeBasePath: "/var/lib/kat/volumes"
|
|
backupPath: "/var/lib/kat/backups"
|
|
backupIntervalMinutes: 30
|
|
agentTickSeconds: 15
|
|
nodeLossTimeoutSeconds: 60
|
|
`
|
|
configPath := filepath.Join(dir, "cluster.kat")
|
|
err := os.WriteFile(configPath, []byte(configContent), 0644)
|
|
require.NoError(t, err)
|
|
return configPath
|
|
}
|
|
|
|
// WaitForCondition waits for the given condition function to return true or times out
|
|
func WaitForCondition(t *testing.T, condition func() bool, timeout time.Duration, message string) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
|
|
ticker := time.NewTicker(50 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
require.Fail(t, "Timed out waiting for condition: "+message)
|
|
return
|
|
case <-ticker.C:
|
|
if condition() {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|