Fix engine migration crash

controller/replica_controller.go

@@ -303,11 +303,18 @@ func (rc *ReplicaController) syncReplica(key string) (err error) {
 			return errors.Wrapf(err, "failed to cleanup the related replica instance before deleting replica %v", replica.Name)
 		}
 
+		rs, err := rc.ds.ListReplicasByNodeRO(replica.Spec.NodeID)
+		if err != nil {
+			return errors.Wrapf(err, "failed to list replicas by node before deleting replica %v", replica.Name)
+		}
+
 		if replica.Spec.NodeID != "" && replica.Spec.NodeID != rc.controllerID {
 			log.Warn("Failed to cleanup replica's data because the replica's data is not on this node")
 		} else if replica.Spec.NodeID != "" {
 			if replica.Spec.BackendStoreDriver == longhorn.BackendStoreDriverTypeV1 {
-				if replica.Spec.Active && dataPath != "" {
+				// Clean up the data directory if this is the active replica or if this inactive replica is the only one
+				// using it.
+				if (replica.Spec.Active || !hasMatchingReplica(replica, rs)) && dataPath != "" {
 					// prevent accidentally deletion
 					if !strings.Contains(filepath.Base(filepath.Clean(dataPath)), "-") {
 						return fmt.Errorf("%v doesn't look like a replica data path", dataPath)
@@ -937,3 +944,12 @@ func (rc *ReplicaController) enqueueAllRebuildingReplicaOnCurrentNode() {
 func (rc *ReplicaController) isResponsibleFor(r *longhorn.Replica) bool {
 	return isControllerResponsibleFor(rc.controllerID, rc.ds, r.Name, r.Spec.NodeID, r.Status.OwnerID)
 }
+
+func hasMatchingReplica(replica *longhorn.Replica, replicas []*longhorn.Replica) bool {
+	for _, r := range replicas {
+		if r.Name != replica.Name && r.Spec.DataDirectoryName == replica.Spec.DataDirectoryName {
+			return true
+		}
+	}
+	return false
+}

controller/volume_controller.go

@@ -1757,6 +1757,12 @@ func (c *VolumeController) openVolumeDependentResources(v *longhorn.Volume, e *l
 			log.WithField("replica", r.Name).Warn("Replica is running but Port is empty")
 			continue
 		}
+		if _, ok := e.Spec.ReplicaAddressMap[r.Name]; !ok && isVolumeMigrating(v) && e.Spec.NodeID == v.Spec.NodeID {
+			// The volume is migrating from this engine. Don't allow new replicas to be added until migration is
+			// complete per https://github.com/longhorn/longhorn/issues/6961.
+			log.WithField("replica", r.Name).Warn("Replica is running, but can't be added while migration is ongoing")
+			continue
+		}
 		replicaAddressMap[r.Name] = imutil.GetURL(r.Status.StorageIP, r.Status.Port)
 	}
 	if len(replicaAddressMap) == 0 {
@@ -2845,11 +2851,9 @@ func (c *VolumeController) upgradeEngineForVolume(v *longhorn.Volume, es map[str
 		return nil
 	}
 
-	if err := c.switchActiveReplicas(rs, func(r *longhorn.Replica, image string) bool {
+	c.switchActiveReplicas(rs, func(r *longhorn.Replica, image string) bool {
 		return r.Spec.Image == image && r.DeletionTimestamp.IsZero()
-	}, v.Spec.Image); err != nil {
-		return err
-	}
+	}, v.Spec.Image)
 
 	e.Spec.ReplicaAddressMap = e.Spec.UpgradedReplicaAddressMap
 	e.Spec.UpgradedReplicaAddressMap = map[string]string{}
@@ -3769,7 +3773,7 @@ func (c *VolumeController) deleteInvalidMigrationReplicas(rs, pathToOldRs, pathT
 }
 
 func (c *VolumeController) switchActiveReplicas(rs map[string]*longhorn.Replica,
-	activeCondFunc func(r *longhorn.Replica, obj string) bool, obj string) error {
+	activeCondFunc func(r *longhorn.Replica, obj string) bool, obj string) {
 
 	// Deletion of an active replica will trigger the cleanup process to
 	// delete the volume data on the disk.
@@ -3780,7 +3784,6 @@ func (c *VolumeController) switchActiveReplicas(rs map[string]*longhorn.Replica,
 			r.Spec.Active = !r.Spec.Active
 		}
 	}
-	return nil
 }
 
 func (c *VolumeController) processMigration(v *longhorn.Volume, es map[string]*longhorn.Engine, rs map[string]*longhorn.Replica) (err error) {
@@ -3848,11 +3851,9 @@ func (c *VolumeController) processMigration(v *longhorn.Volume, es map[string]*l
 		currentEngine.Spec.Active = true
 
 		// cleanupCorruptedOrStaleReplicas() will take care of old replicas
-		if err := c.switchActiveReplicas(rs, func(r *longhorn.Replica, engineName string) bool {
-			return r.Spec.EngineName == engineName
-		}, currentEngine.Name); err != nil {
-			return err
-		}
+		c.switchActiveReplicas(rs, func(r *longhorn.Replica, engineName string) bool {
+			return r.Spec.EngineName == engineName && r.Spec.HealthyAt != ""
+		}, currentEngine.Name)
 
 		// migration rollback or confirmation finished
 		v.Status.CurrentMigrationNodeID = ""
@@ -3995,9 +3996,10 @@ func (c *VolumeController) prepareReplicasAndEngineForMigration(v *longhorn.Volu
 				return false, true, nil
 			case longhorn.ReplicaModeRW:
 				currentAvailableReplicas[dataPath] = r
+			case "":
+				log.Warnf("Running replica %v wasn't added to engine, will ignore it and continue migration", r.Name)
 			default:
 				log.Warnf("Unexpected mode %v for the current replica %v, will ignore it and continue migration", currentEngine.Status.ReplicaModeMap[r.Name], r.Name)
-				continue
 			}
 		} else if r.Spec.EngineName == migrationEngine.Name {
 			migrationReplicas[dataPath] = r