feat(v2/auto-salvage): failed usable replica candidate selection

longhorn/longhorn-8430

Signed-off-by: Chin-Ya Huang <chin-ya.huang@suse.com>

controller/volume_controller.go

@@ -1347,6 +1347,11 @@ func (c *VolumeController) ReconcileVolumeState(v *longhorn.Volume, es map[strin
 			failedUsableReplicas := map[string]*longhorn.Replica{}
 			dataExists := false
 
+			snapshots, err := c.ds.ListVolumeSnapshotsRO(v.Name)
+			if err != nil {
+				return err
+			}
+
 			for _, r := range rs {
 				if r.Spec.HealthyAt == "" {
 					continue
@@ -1377,6 +1382,22 @@ func (c *VolumeController) ReconcileVolumeState(v *longhorn.Volume, es map[strin
 				if !diskSchedulable {
 					continue
 				}
+
+				isReplicaHasAllSnapshots := true
+				// The v1 volume candidate selection will be handled in the engine instance.
+				if types.IsDataEngineV2(v.Spec.DataEngine) {
+					for snapshotName := range snapshots {
+						if _, exist := r.Status.Snapshots[snapshotName]; !exist {
+							isReplicaHasAllSnapshots = false
+							continue
+						}
+					}
+				}
+				if !isReplicaHasAllSnapshots {
+					log.WithField("replica", r.Name).Debug("Cannot reuse failed replica because it does not contain all snapshots")
+					continue
+				}
+
 				failedAt, err := util.ParseTime(r.Spec.FailedAt)
 				if err != nil {
 					log.WithField("replica", r.Name).WithError(err).Warn("Failed to parse FailedAt timestamp for replica")