fix(controller): don't lose the client disconnect when it fails

internal/controller/dragonfly_instance.go

@@ -478,6 +478,12 @@ func (dfi *DragonflyInstance) replicaOf(ctx context.Context, pod *corev1.Pod, ma
 		return fmt.Errorf("failed to determine the current role of the instance: %w", err)
 	}
 
+	if wasMaster {
+		if err := dfi.setPendingClientDisconnect(ctx, pod, true); err != nil {
+			return fmt.Errorf("could not mark pod for client disconnect: %w", err)
+		}
+	}
+
 	// Sanitize masterIp in case ipv6
 	masterIp = sanitizeIp(masterIp)
 
@@ -518,7 +524,16 @@ func (dfi *DragonflyInstance) replicaOf(ctx context.Context, pod *corev1.Pod, ma
 
 	if wasMaster {
 		// Prevent clients from sending commands to this old master
-		dfi.disconnectClients(ctx, redisClient, pod)
+		if err := dfi.disconnectClients(ctx, pod); err != nil {
+			dfi.log.Error(err, "failed to disconnect clients, will retry", "pod", pod.Name)
+			// Returning an error here would abort the caller's loop over the remaining pods.
+			// The annotation stays set, the next reconcile retries.
+			return nil
+		}
+
+		if err := dfi.setPendingClientDisconnect(ctx, pod, false); err != nil {
+			return fmt.Errorf("could not clear the client disconnect marker: %w", err)
+		}
 	}
 
 	return nil
@@ -557,6 +572,7 @@ func (dfi *DragonflyInstance) replicaOfNoOne(ctx context.Context, pod *corev1.Po
 		pod.Annotations = make(map[string]string)
 	}
 	pod.Annotations[resources.MasterIpAnnotationKey] = masterIp
+	delete(pod.Annotations, resources.PendingClientDisconnectAnnotationKey)
 
 	if err := dfi.client.Patch(ctx, pod, patch); err != nil {
 		return err
@@ -566,44 +582,38 @@ func (dfi *DragonflyInstance) replicaOfNoOne(ctx context.Context, pod *corev1.Po
 }
 
 // disconnectClients disconnects all non-replication clients from a pod.
-func (dfi *DragonflyInstance) disconnectClients(ctx context.Context, redisClient *redis.Client, pod *corev1.Pod) {
-	dfi.log.Info("disconnecting clients from replica", "pod", pod.Name)
-	clientList, err := redisClient.ClientList(ctx).Result()
+func (dfi *DragonflyInstance) disconnectClients(ctx context.Context, pod *corev1.Pod) error {
+	if pod.Status.PodIP == "" {
+		return fmt.Errorf("pod %s has no IP address", pod.Name)
+	}
+
+	addr := clientListenerAddress(pod.Status.PodIP)
+	dfi.log.Info("disconnecting clients", "pod", pod.Name, "addr", addr)
+
+	killed, err := dfi.getRedisClient(pod.Status.PodIP).ClientKillByFilter(ctx, "LADDR", addr).Result()
 	if err != nil {
-		dfi.log.Error(err, "failed to get client list from replica", "pod", pod.Name)
-		return
+		return fmt.Errorf("failed to kill clients on %s: %w", addr, err)
 	}
 
-	clients := []string{}
-	for _, clientInfo := range strings.Split(clientList, "\n") {
-		if clientInfo == "" {
-			continue
-		}
-		// Example clientInfo: "id=2 addr=10.42.1.123:50342 ... name=..."
-		// Avoid killing replication clients, internal clients, or this connection
-		if strings.Contains(clientInfo, "addr=127.0.0.1") ||
-			strings.Contains(clientInfo, "addr=::1") ||
-			strings.Contains(clientInfo, "addr=[::1]") ||
-			strings.Contains(clientInfo, "name=repl_") ||
-			strings.Contains(clientInfo, "name=dragonfly-operator") {
-			continue
-		}
+	dfi.log.Info("disconnected clients", "pod", pod.Name, "clients", killed)
+	return nil
+}
 
-		parts := strings.Split(clientInfo, " ")
-		for _, part := range parts {
-			if strings.HasPrefix(part, "addr=") {
-				addr := strings.TrimPrefix(part, "addr=")
-				if _, err := redisClient.ClientKill(ctx, addr).Result(); err != nil {
-					// Log and continue, don't block for a single failed kill
-					dfi.log.Error(err, "failed to kill client", "addr", addr)
-				} else {
-					clients = append(clients, addr)
-				}
-				break
-			}
+// setPendingClientDisconnect adds or removes the annotation that carries an unfinished
+// client disconnect over to the next reconcile.
+func (dfi *DragonflyInstance) setPendingClientDisconnect(ctx context.Context, pod *corev1.Pod, pending bool) error {
+	patch := client.MergeFrom(pod.DeepCopy())
+
+	if pending {
+		if pod.Annotations == nil {
+			pod.Annotations = make(map[string]string)
 		}
+		pod.Annotations[resources.PendingClientDisconnectAnnotationKey] = "true"
+	} else {
+		delete(pod.Annotations, resources.PendingClientDisconnectAnnotationKey)
 	}
-	dfi.log.Info("killed clients", "pod", pod.Name, "clients", clients)
+
+	return dfi.client.Patch(ctx, pod, patch)
 }
 
 // hasMasterRole returns true if the given pod is a master based on the replication info.

internal/controller/dragonfly_pod_lifecycle_controller.go

@@ -71,6 +71,17 @@ func (r *DfPodLifeCycleReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 	}
 	defer dfi.Close()
 
+	if needsClientDisconnect(&pod) {
+		if err := dfi.disconnectClients(ctx, &pod); err != nil {
+			log.Error(err, "failed to disconnect clients, will retry", "pod", pod.Name)
+			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+		}
+
+		if err := dfi.setPendingClientDisconnect(ctx, &pod, false); err != nil {
+			return ctrl.Result{}, fmt.Errorf("failed to clear the client disconnect marker: %w", err)
+		}
+	}
+
 	podReady, readinessErr := dfi.isPodReady(ctx, &pod)
 	if readinessErr != nil {
 		return ctrl.Result{}, fmt.Errorf("failed to verify pod readiness: %w", readinessErr)

internal/controller/util.go

@@ -195,6 +195,18 @@ func sanitizeIp(masterIp string) string {
 	return strings.Trim(masterIp, "[]")
 }
 
+func clientListenerAddress(podIp string) string {
+	return sanitizeIp(podIp) + ":" + strconv.Itoa(resources.DragonflyPort)
+}
+
+func needsClientDisconnect(pod *corev1.Pod) bool {
+	if pod.Annotations[resources.PendingClientDisconnectAnnotationKey] != "true" {
+		return false
+	}
+
+	return !isMaster(pod)
+}
+
 // getOrdinal returns the ordinal of the pod.
 func getOrdinal(podName string) int {
 	parts := strings.Split(podName, "-")

internal/controller/util_test.go

@@ -155,3 +155,88 @@ func TestSelectMasterCandidate(t *testing.T) {
 		})
 	}
 }
+
+func TestClientListenerAddress(t *testing.T) {
+	tests := []struct {
+		name  string
+		podIp string
+		want  string
+	}{
+		{
+			name:  "ipv4",
+			podIp: "10.42.1.7",
+			want:  "10.42.1.7:6379",
+		},
+		{
+			name:  "ipv6",
+			podIp: "fd00::1",
+			want:  "fd00::1:6379",
+		},
+		{
+			name:  "bracketed ipv6",
+			podIp: "[fd00::1]",
+			want:  "fd00::1:6379",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := clientListenerAddress(tc.podIp); got != tc.want {
+				t.Errorf("expected %q, got %q", tc.want, got)
+			}
+		})
+	}
+}
+
+func TestNeedsClientDisconnect(t *testing.T) {
+	tests := []struct {
+		name        string
+		annotations map[string]string
+		labels      map[string]string
+		want        bool
+	}{
+		{
+			name: "no annotations",
+			want: false,
+		},
+		{
+			name:        "marked and demoted",
+			annotations: map[string]string{resources.PendingClientDisconnectAnnotationKey: "true"},
+			labels:      map[string]string{resources.RoleLabelKey: resources.Replica},
+			want:        true,
+		},
+		{
+			name:        "marked without a role label",
+			annotations: map[string]string{resources.PendingClientDisconnectAnnotationKey: "true"},
+			want:        true,
+		},
+		{
+			name:        "marked but promoted again",
+			annotations: map[string]string{resources.PendingClientDisconnectAnnotationKey: "true"},
+			labels:      map[string]string{resources.RoleLabelKey: resources.Master},
+			want:        false,
+		},
+		{
+			name:        "unrelated annotation",
+			annotations: map[string]string{resources.MasterIpAnnotationKey: "10.42.1.7"},
+			labels:      map[string]string{resources.RoleLabelKey: resources.Replica},
+			want:        false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			pod := corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:        "df-0",
+					Labels:      tc.labels,
+					Annotations: tc.annotations,
+				},
+			}
+
+			if got := needsClientDisconnect(&pod); got != tc.want {
+				t.Errorf("expected %v, got %v", tc.want, got)
+			}
+		})
+	}
+}

internal/resources/const.go

@@ -89,6 +89,8 @@ const (
 
 	MasterIpAnnotationKey = "operator.dragonflydb.io/masterIP"
 
+	PendingClientDisconnectAnnotationKey = "operator.dragonflydb.io/pendingClientDisconnect"
+
 	RoleLabelKey = "role"
 
 	Master = "master"