mirror of
https://github.com/hanzokv/go.git
synced 2026-09-24 00:31:51 +00:00
693 lines
21 KiB
Go
693 lines
21 KiB
Go
package redis_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"os/exec"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/hanzoai/kv-go/v9"
|
|
)
|
|
|
|
// multiNodeProxy represents a cae-resp-proxy instance that can mimic multiple cluster nodes
|
|
type multiNodeProxy struct {
|
|
apiPort int
|
|
apiBaseURL string
|
|
cmd *exec.Cmd
|
|
httpClient *http.Client
|
|
nodes []proxyNode
|
|
}
|
|
|
|
// proxyNode represents a single node in the multi-node proxy
|
|
type proxyNode struct {
|
|
listenPort int
|
|
targetHost string
|
|
targetPort int
|
|
proxyAddr string
|
|
nodeID string
|
|
}
|
|
|
|
// newMultiNodeProxy creates a new multi-node proxy instance
|
|
func newMultiNodeProxy(apiPort int) *multiNodeProxy {
|
|
return &multiNodeProxy{
|
|
apiPort: apiPort,
|
|
apiBaseURL: fmt.Sprintf("http://localhost:%d", apiPort),
|
|
httpClient: &http.Client{
|
|
Timeout: 5 * time.Second,
|
|
},
|
|
nodes: make([]proxyNode, 0),
|
|
}
|
|
}
|
|
|
|
// start starts the proxy server with initial node
|
|
func (mp *multiNodeProxy) start(initialListenPort, targetPort int, targetHost string) error {
|
|
// Start cae-resp-proxy with just the API port
|
|
// We'll add nodes dynamically via the API
|
|
mp.cmd = exec.Command("cae-resp-proxy",
|
|
"--api-port", fmt.Sprintf("%d", mp.apiPort),
|
|
"--listen-port", fmt.Sprintf("%d", initialListenPort),
|
|
"--target-host", targetHost,
|
|
"--target-port", fmt.Sprintf("%d", targetPort),
|
|
)
|
|
|
|
if err := mp.cmd.Start(); err != nil {
|
|
return fmt.Errorf("failed to start proxy: %w", err)
|
|
}
|
|
|
|
// Wait for proxy to be ready
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Verify proxy is responding
|
|
for i := 0; i < 10; i++ {
|
|
resp, err := mp.httpClient.Get(mp.apiBaseURL + "/stats")
|
|
if err == nil {
|
|
resp.Body.Close()
|
|
if resp.StatusCode == 200 {
|
|
// Add the initial node to our tracking
|
|
mp.nodes = append(mp.nodes, proxyNode{
|
|
listenPort: initialListenPort,
|
|
targetHost: targetHost,
|
|
targetPort: targetPort,
|
|
proxyAddr: fmt.Sprintf("localhost:%d", initialListenPort),
|
|
nodeID: fmt.Sprintf("localhost:%d:%d", initialListenPort, targetPort),
|
|
})
|
|
return nil
|
|
}
|
|
}
|
|
time.Sleep(200 * time.Millisecond)
|
|
}
|
|
|
|
return fmt.Errorf("proxy did not become ready")
|
|
}
|
|
|
|
// addNode adds a new proxy node dynamically
|
|
func (mp *multiNodeProxy) addNode(listenPort, targetPort int, targetHost string) (*proxyNode, error) {
|
|
// Use the /nodes API to add a new proxy node
|
|
nodeConfig := map[string]interface{}{
|
|
"listenPort": listenPort,
|
|
"targetHost": targetHost,
|
|
"targetPort": targetPort,
|
|
}
|
|
|
|
jsonData, err := json.Marshal(nodeConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to marshal node config: %w", err)
|
|
}
|
|
|
|
resp, err := mp.httpClient.Post(
|
|
mp.apiBaseURL+"/nodes",
|
|
"application/json",
|
|
bytes.NewReader(jsonData),
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to add node: %w", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, _ := io.ReadAll(resp.Body)
|
|
return nil, fmt.Errorf("failed to add node, status %d: %s", resp.StatusCode, string(body))
|
|
}
|
|
|
|
// Create node tracking
|
|
node := proxyNode{
|
|
listenPort: listenPort,
|
|
targetHost: targetHost,
|
|
targetPort: targetPort,
|
|
proxyAddr: fmt.Sprintf("localhost:%d", listenPort),
|
|
nodeID: fmt.Sprintf("localhost:%d:%d", listenPort, targetPort),
|
|
}
|
|
|
|
mp.nodes = append(mp.nodes, node)
|
|
|
|
// Wait a bit for the node to be ready
|
|
time.Sleep(200 * time.Millisecond)
|
|
|
|
return &node, nil
|
|
}
|
|
|
|
// getNodes returns all proxy nodes
|
|
func (mp *multiNodeProxy) getNodes() []proxyNode {
|
|
return mp.nodes
|
|
}
|
|
|
|
// getNodeAddrs returns all proxy node addresses for cluster client
|
|
func (mp *multiNodeProxy) getNodeAddrs() []string {
|
|
addrs := make([]string, len(mp.nodes))
|
|
for i, node := range mp.nodes {
|
|
addrs[i] = node.proxyAddr
|
|
}
|
|
return addrs
|
|
}
|
|
|
|
// stop stops the proxy server
|
|
func (mp *multiNodeProxy) stop() error {
|
|
if mp.cmd != nil && mp.cmd.Process != nil {
|
|
return mp.cmd.Process.Kill()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// injectNotification injects a RESP3 push notification to all connected clients
|
|
func (mp *multiNodeProxy) injectNotification(notification string) error {
|
|
url := mp.apiBaseURL + "/send-to-all-clients?encoding=raw"
|
|
resp, err := mp.httpClient.Post(url, "application/octet-stream", strings.NewReader(notification))
|
|
if err != nil {
|
|
return fmt.Errorf("failed to inject notification: %w", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, _ := io.ReadAll(resp.Body)
|
|
return fmt.Errorf("injection failed with status %d: %s", resp.StatusCode, string(body))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// injectNotificationToNode injects a notification to clients connected to a specific node
|
|
func (mp *multiNodeProxy) injectNotificationToNode(nodeAddr string, notification string) error {
|
|
// Get all connections
|
|
resp, err := mp.httpClient.Get(mp.apiBaseURL + "/connections")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get connections: %w", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
var connResp struct {
|
|
ConnectionIDs []string `json:"connectionIds"`
|
|
}
|
|
if err := json.NewDecoder(resp.Body).Decode(&connResp); err != nil {
|
|
return fmt.Errorf("failed to decode connections: %w", err)
|
|
}
|
|
|
|
// For simplicity, inject to all clients
|
|
// In a real scenario, you'd filter by node
|
|
return mp.injectNotification(notification)
|
|
}
|
|
|
|
// formatSMigratingNotification creates a SMIGRATING push notification in RESP3 format
|
|
// Format: ["SMIGRATING", SeqID, slot/range1-range2, ...]
|
|
func formatSMigratingNotification(seqID int64, slots ...string) string {
|
|
// >N\r\n where N is the number of elements
|
|
parts := []string{fmt.Sprintf(">%d\r\n", 2+len(slots))}
|
|
|
|
// $10\r\nSMIGRATING\r\n
|
|
parts = append(parts, "$10\r\nSMIGRATING\r\n")
|
|
|
|
// :seqID\r\n
|
|
parts = append(parts, fmt.Sprintf(":%d\r\n", seqID))
|
|
|
|
// Add each slot/range as bulk string
|
|
for _, slot := range slots {
|
|
parts = append(parts, fmt.Sprintf("$%d\r\n%s\r\n", len(slot), slot))
|
|
}
|
|
|
|
return strings.Join(parts, "")
|
|
}
|
|
|
|
// formatSMigratedNotification creates a SMIGRATED push notification in RESP3 format
|
|
// RESP3 wire format:
|
|
//
|
|
// >3 <- push frame with 3 top-level elements
|
|
// +SMIGRATED <- message name
|
|
// :SeqID <- sequence id integer
|
|
// *<num_entries> <- array of triplet arrays
|
|
// *3 <- each triplet is a 3-element array
|
|
// +<source> <- node from which slots are migrating FROM
|
|
// +<destination> <- node to which slots are migrating TO
|
|
// +<slots> <- comma-separated slots and/or ranges
|
|
//
|
|
// Each triplet is formatted as: "source target slots"
|
|
// Example: "abc.com:6789 abc.com:6790 123,789-1000"
|
|
func formatSMigratedNotification(seqID int64, triplets ...string) string {
|
|
parts := []string{">3\r\n"}
|
|
|
|
// +SMIGRATED\r\n
|
|
parts = append(parts, "+SMIGRATED\r\n")
|
|
|
|
// :seqID\r\n
|
|
parts = append(parts, fmt.Sprintf(":%d\r\n", seqID))
|
|
|
|
// Outer array containing all triplets
|
|
parts = append(parts, fmt.Sprintf("*%d\r\n", len(triplets)))
|
|
|
|
for _, triplet := range triplets {
|
|
// Split triplet into source, target, and slots
|
|
tripletParts := strings.SplitN(triplet, " ", 3)
|
|
if len(tripletParts) != 3 {
|
|
continue
|
|
}
|
|
source := tripletParts[0]
|
|
target := tripletParts[1]
|
|
slots := tripletParts[2]
|
|
|
|
// Each triplet is a 3-element array
|
|
parts = append(parts, "*3\r\n")
|
|
parts = append(parts, fmt.Sprintf("+%s\r\n", source))
|
|
parts = append(parts, fmt.Sprintf("+%s\r\n", target))
|
|
parts = append(parts, fmt.Sprintf("+%s\r\n", slots))
|
|
}
|
|
|
|
return strings.Join(parts, "")
|
|
}
|
|
|
|
// TestClusterMaintNotifications_SMIGRATING tests SMIGRATING notification handling
|
|
func TestClusterMaintNotifications_SMIGRATING(t *testing.T) {
|
|
if os.Getenv("CLUSTER_MAINT_INTEGRATION_TEST") != "true" {
|
|
t.Skip("Skipping cluster maintnotifications integration test. Set CLUSTER_MAINT_INTEGRATION_TEST=true to run")
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create multi-node proxy that mimics a 3-node cluster
|
|
proxy := newMultiNodeProxy(8000)
|
|
if err := proxy.start(7000, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to start proxy: %v", err)
|
|
}
|
|
defer proxy.stop()
|
|
|
|
// Add two more nodes to mimic a 3-node cluster
|
|
if _, err := proxy.addNode(7001, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 2: %v", err)
|
|
}
|
|
if _, err := proxy.addNode(7002, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 3: %v", err)
|
|
}
|
|
|
|
t.Logf("Started proxy with %d nodes: %v", len(proxy.getNodes()), proxy.getNodeAddrs())
|
|
|
|
// Create cluster client pointing to all proxy nodes
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: proxy.getNodeAddrs(),
|
|
Protocol: 3, // RESP3 required for push notifications
|
|
})
|
|
defer client.Close()
|
|
|
|
// Verify connection works
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
t.Fatalf("Failed to connect to cluster via proxy: %v", err)
|
|
}
|
|
|
|
// Perform some operations to establish connections
|
|
for i := 0; i < 5; i++ {
|
|
if err := client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0).Err(); err != nil {
|
|
t.Logf("Warning: Failed to set key: %v", err)
|
|
}
|
|
}
|
|
|
|
// Inject SMIGRATING notification to all nodes
|
|
notification := formatSMigratingNotification(12345, "1000", "2000-3000")
|
|
if err := proxy.injectNotification(notification); err != nil {
|
|
t.Fatalf("Failed to inject SMIGRATING notification: %v", err)
|
|
}
|
|
|
|
// Wait for notification processing
|
|
time.Sleep(200 * time.Millisecond)
|
|
|
|
// Verify operations still work (timeouts should be relaxed)
|
|
if err := client.Set(ctx, "test-key-during-migration", "value", 0).Err(); err != nil {
|
|
t.Errorf("Expected operations to work during migration, got error: %v", err)
|
|
}
|
|
|
|
t.Log("SMIGRATING notification test passed")
|
|
}
|
|
|
|
// TestClusterMaintNotifications_SMIGRATED tests SMIGRATED notification handling and cluster state reload
|
|
func TestClusterMaintNotifications_SMIGRATED(t *testing.T) {
|
|
if os.Getenv("CLUSTER_MAINT_INTEGRATION_TEST") != "true" {
|
|
t.Skip("Skipping cluster maintnotifications integration test. Set CLUSTER_MAINT_INTEGRATION_TEST=true to run")
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create multi-node proxy
|
|
proxy := newMultiNodeProxy(8001)
|
|
if err := proxy.start(7010, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to start proxy: %v", err)
|
|
}
|
|
defer proxy.stop()
|
|
|
|
// Add more nodes
|
|
if _, err := proxy.addNode(7011, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 2: %v", err)
|
|
}
|
|
if _, err := proxy.addNode(7012, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 3: %v", err)
|
|
}
|
|
|
|
t.Logf("Started proxy with %d nodes: %v", len(proxy.getNodes()), proxy.getNodeAddrs())
|
|
|
|
// Track cluster state reloads
|
|
var reloadCount atomic.Int32
|
|
|
|
// Create cluster client
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: proxy.getNodeAddrs(),
|
|
Protocol: 3,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Hook to track state reloads via callback
|
|
client.OnNewNode(func(nodeClient *redis.Client) {
|
|
manager := nodeClient.GetMaintNotificationsManager()
|
|
if manager != nil {
|
|
manager.SetClusterStateReloadCallback(func(ctx context.Context, hostPort string, slotRanges []string) {
|
|
reloadCount.Add(1)
|
|
})
|
|
}
|
|
})
|
|
|
|
// Verify connection
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
|
|
// Perform operations to establish connections
|
|
for i := 0; i < 5; i++ {
|
|
client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0)
|
|
}
|
|
|
|
initialReloads := reloadCount.Load()
|
|
|
|
// Inject SMIGRATED notification with new format
|
|
// Simulate migration from node 1 to node 2
|
|
notification := formatSMigratedNotification(12346, "127.0.0.1:7011 1000,2000-3000")
|
|
if err := proxy.injectNotification(notification); err != nil {
|
|
t.Fatalf("Failed to inject SMIGRATED notification: %v", err)
|
|
}
|
|
|
|
// Wait for notification processing and state reload
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Verify cluster state was reloaded
|
|
finalReloads := reloadCount.Load()
|
|
if finalReloads <= initialReloads {
|
|
t.Errorf("Expected cluster state reload after SMIGRATED, reloads: initial=%d, final=%d",
|
|
initialReloads, finalReloads)
|
|
}
|
|
|
|
// Verify operations still work
|
|
if err := client.Set(ctx, "test-key-after-smigrated", "value", 0).Err(); err != nil {
|
|
t.Errorf("Expected operations to work after SMIGRATED, got error: %v", err)
|
|
}
|
|
|
|
t.Log("SMIGRATED notification test passed")
|
|
}
|
|
|
|
// TestClusterMaintNotifications_Deduplication tests that SMIGRATED notifications are deduplicated by SeqID
|
|
func TestClusterMaintNotifications_Deduplication(t *testing.T) {
|
|
if os.Getenv("CLUSTER_MAINT_INTEGRATION_TEST") != "true" {
|
|
t.Skip("Skipping cluster maintnotifications integration test. Set CLUSTER_MAINT_INTEGRATION_TEST=true to run")
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create multi-node proxy
|
|
proxy := newMultiNodeProxy(8002)
|
|
if err := proxy.start(7020, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to start proxy: %v", err)
|
|
}
|
|
defer proxy.stop()
|
|
|
|
// Add more nodes
|
|
if _, err := proxy.addNode(7021, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 2: %v", err)
|
|
}
|
|
if _, err := proxy.addNode(7022, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 3: %v", err)
|
|
}
|
|
|
|
t.Logf("Started proxy with %d nodes: %v", len(proxy.getNodes()), proxy.getNodeAddrs())
|
|
|
|
var reloadCount atomic.Int32
|
|
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: proxy.getNodeAddrs(),
|
|
Protocol: 3,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Track reloads via callback
|
|
client.OnNewNode(func(nodeClient *redis.Client) {
|
|
manager := nodeClient.GetMaintNotificationsManager()
|
|
if manager != nil {
|
|
manager.SetClusterStateReloadCallback(func(ctx context.Context, hostPort string, slotRanges []string) {
|
|
reloadCount.Add(1)
|
|
})
|
|
}
|
|
})
|
|
|
|
// Verify connection
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
|
|
// Perform operations
|
|
for i := 0; i < 5; i++ {
|
|
client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0)
|
|
}
|
|
|
|
initialReloads := reloadCount.Load()
|
|
|
|
// Inject the same SMIGRATED notification multiple times (same SeqID)
|
|
// This simulates receiving the same notification from multiple nodes
|
|
seqID := int64(99999)
|
|
notification := formatSMigratedNotification(seqID, "127.0.0.1:7021 5000-6000")
|
|
|
|
// Inject 5 times to all nodes
|
|
for i := 0; i < 5; i++ {
|
|
if err := proxy.injectNotification(notification); err != nil {
|
|
t.Fatalf("Failed to inject notification: %v", err)
|
|
}
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
|
|
// Wait for processing
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Verify only ONE reload happened (deduplication by SeqID)
|
|
finalReloads := reloadCount.Load()
|
|
reloadDiff := finalReloads - initialReloads
|
|
if reloadDiff != 1 {
|
|
t.Errorf("Expected exactly 1 reload due to deduplication, got %d reloads", reloadDiff)
|
|
}
|
|
|
|
t.Log("Deduplication test passed")
|
|
}
|
|
|
|
// TestClusterMaintNotifications_MultiNode tests notifications across multiple cluster nodes
|
|
func TestClusterMaintNotifications_MultiNode(t *testing.T) {
|
|
if os.Getenv("CLUSTER_MAINT_INTEGRATION_TEST") != "true" {
|
|
t.Skip("Skipping cluster maintnotifications integration test. Set CLUSTER_MAINT_INTEGRATION_TEST=true to run")
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create multi-node proxy with 5 nodes to mimic a real cluster
|
|
proxy := newMultiNodeProxy(8003)
|
|
if err := proxy.start(7030, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to start proxy: %v", err)
|
|
}
|
|
defer proxy.stop()
|
|
|
|
// Add 4 more nodes for a 5-node cluster
|
|
for i := 1; i < 5; i++ {
|
|
if _, err := proxy.addNode(7030+i, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node %d: %v", i+1, err)
|
|
}
|
|
}
|
|
|
|
nodes := proxy.getNodes()
|
|
t.Logf("Started proxy with %d nodes: %v", len(nodes), proxy.getNodeAddrs())
|
|
|
|
// Track notifications received
|
|
var migratingCount atomic.Int32
|
|
var migratedCount atomic.Int32
|
|
var reloadCount atomic.Int32
|
|
|
|
// Create cluster client
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: proxy.getNodeAddrs(),
|
|
Protocol: 3,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Set up tracking
|
|
client.OnNewNode(func(nodeClient *redis.Client) {
|
|
manager := nodeClient.GetMaintNotificationsManager()
|
|
if manager != nil {
|
|
manager.SetClusterStateReloadCallback(func(ctx context.Context, hostPort string, slotRanges []string) {
|
|
reloadCount.Add(1)
|
|
t.Logf("Cluster state reload triggered for %s, slots: %v", hostPort, slotRanges)
|
|
})
|
|
}
|
|
})
|
|
|
|
// Verify connection
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
|
|
// Perform operations to establish connections to all nodes
|
|
for i := 0; i < 20; i++ {
|
|
client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0)
|
|
}
|
|
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Simulate slot migration scenario:
|
|
// 1. SMIGRATING: Slots 0-5000 are migrating from node 1 to node 2
|
|
t.Log("Injecting SMIGRATING notification...")
|
|
migratingNotif := formatSMigratingNotification(10001, "0-5000")
|
|
if err := proxy.injectNotification(migratingNotif); err != nil {
|
|
t.Fatalf("Failed to inject SMIGRATING: %v", err)
|
|
}
|
|
migratingCount.Add(1)
|
|
|
|
time.Sleep(300 * time.Millisecond)
|
|
|
|
// 2. SMIGRATED: Migration completed, slots now on node 2
|
|
t.Log("Injecting SMIGRATED notification...")
|
|
migratedNotif := formatSMigratedNotification(10002,
|
|
fmt.Sprintf("127.0.0.1:%d 0-5000", nodes[1].listenPort))
|
|
if err := proxy.injectNotification(migratedNotif); err != nil {
|
|
t.Fatalf("Failed to inject SMIGRATED: %v", err)
|
|
}
|
|
migratedCount.Add(1)
|
|
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// 3. Another migration: Slots 5001-10000 from node 2 to node 3
|
|
t.Log("Injecting second SMIGRATING notification...")
|
|
migratingNotif2 := formatSMigratingNotification(10003, "5001-10000")
|
|
if err := proxy.injectNotification(migratingNotif2); err != nil {
|
|
t.Fatalf("Failed to inject second SMIGRATING: %v", err)
|
|
}
|
|
migratingCount.Add(1)
|
|
|
|
time.Sleep(300 * time.Millisecond)
|
|
|
|
// 4. Second migration completed
|
|
t.Log("Injecting second SMIGRATED notification...")
|
|
migratedNotif2 := formatSMigratedNotification(10004,
|
|
fmt.Sprintf("127.0.0.1:%d 5001-10000", nodes[2].listenPort))
|
|
if err := proxy.injectNotification(migratedNotif2); err != nil {
|
|
t.Fatalf("Failed to inject second SMIGRATED: %v", err)
|
|
}
|
|
migratedCount.Add(1)
|
|
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Verify cluster state was reloaded for each SMIGRATED
|
|
finalReloads := reloadCount.Load()
|
|
if finalReloads < 2 {
|
|
t.Errorf("Expected at least 2 cluster state reloads, got %d", finalReloads)
|
|
}
|
|
|
|
// Verify operations still work after migrations
|
|
for i := 0; i < 10; i++ {
|
|
if err := client.Set(ctx, fmt.Sprintf("post-migration-key%d", i), "value", 0).Err(); err != nil {
|
|
t.Errorf("Expected operations to work after migrations, got error: %v", err)
|
|
}
|
|
}
|
|
|
|
t.Logf("Multi-node test passed: SMIGRATING=%d, SMIGRATED=%d, Reloads=%d",
|
|
migratingCount.Load(), migratedCount.Load(), finalReloads)
|
|
}
|
|
|
|
// TestClusterMaintNotifications_ComplexMigration tests complex multi-endpoint migration
|
|
func TestClusterMaintNotifications_ComplexMigration(t *testing.T) {
|
|
if os.Getenv("CLUSTER_MAINT_INTEGRATION_TEST") != "true" {
|
|
t.Skip("Skipping cluster maintnotifications integration test. Set CLUSTER_MAINT_INTEGRATION_TEST=true to run")
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// Create multi-node proxy
|
|
proxy := newMultiNodeProxy(8004)
|
|
if err := proxy.start(7040, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to start proxy: %v", err)
|
|
}
|
|
defer proxy.stop()
|
|
|
|
// Add 2 more nodes
|
|
if _, err := proxy.addNode(7041, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 2: %v", err)
|
|
}
|
|
if _, err := proxy.addNode(7042, 6379, "localhost"); err != nil {
|
|
t.Fatalf("Failed to add node 3: %v", err)
|
|
}
|
|
|
|
nodes := proxy.getNodes()
|
|
t.Logf("Started proxy with %d nodes: %v", len(nodes), proxy.getNodeAddrs())
|
|
|
|
var reloadCount atomic.Int32
|
|
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: proxy.getNodeAddrs(),
|
|
Protocol: 3,
|
|
})
|
|
defer client.Close()
|
|
|
|
client.OnNewNode(func(nodeClient *redis.Client) {
|
|
manager := nodeClient.GetMaintNotificationsManager()
|
|
if manager != nil {
|
|
manager.SetClusterStateReloadCallback(func(ctx context.Context, hostPort string, slotRanges []string) {
|
|
reloadCount.Add(1)
|
|
t.Logf("Reload for %s, slots: %v", hostPort, slotRanges)
|
|
})
|
|
}
|
|
})
|
|
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
|
|
// Perform operations
|
|
for i := 0; i < 10; i++ {
|
|
client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0)
|
|
}
|
|
|
|
initialReloads := reloadCount.Load()
|
|
|
|
// Test new SMIGRATED format with multiple endpoints
|
|
// Simulate a complex resharding where slots are distributed to multiple nodes
|
|
t.Log("Injecting complex SMIGRATED notification with multiple endpoints...")
|
|
notification := formatSMigratedNotification(20001,
|
|
fmt.Sprintf("127.0.0.1:%d 0-5000,10000-12000", nodes[0].listenPort),
|
|
fmt.Sprintf("127.0.0.1:%d 5001-9999", nodes[1].listenPort),
|
|
fmt.Sprintf("127.0.0.1:%d 12001-16383", nodes[2].listenPort),
|
|
)
|
|
|
|
if err := proxy.injectNotification(notification); err != nil {
|
|
t.Fatalf("Failed to inject complex SMIGRATED: %v", err)
|
|
}
|
|
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Verify reload happened
|
|
finalReloads := reloadCount.Load()
|
|
if finalReloads <= initialReloads {
|
|
t.Errorf("Expected cluster state reload, reloads: initial=%d, final=%d",
|
|
initialReloads, finalReloads)
|
|
}
|
|
|
|
// Verify operations work
|
|
for i := 0; i < 10; i++ {
|
|
if err := client.Set(ctx, fmt.Sprintf("complex-key%d", i), "value", 0).Err(); err != nil {
|
|
t.Errorf("Operations failed after complex migration: %v", err)
|
|
}
|
|
}
|
|
|
|
t.Log("Complex migration test passed")
|
|
}
|