handle storage state out of sync with storage migration

pkg/migrator/core.go

@@ -96,6 +96,7 @@ func (m *migrator) Run() error {
 			},
 		)
 		if errors.IsNotFound(listError) {
+			// fail this migration, we don't want to get stuck on a migration for a resource that does not exist)
 			return fmt.Errorf("failed to list resources: %v", listError)
 		}
 		if listError != nil && !errors.IsResourceExpired(listError) {

pkg/trigger/discovery_handler.go

@@ -185,10 +185,11 @@ func (mt *MigrationTrigger) processDiscoveryResource(r metav1.APIResource) {
 		utilruntime.HandleError(getErr)
 		return
 	}
-
-	stale := (getErr == nil && mt.staleStorageState(ss))
-	storageVersionChanged := (getErr == nil && ss.Status.CurrentStorageVersionHash != r.StorageVersionHash)
-	notFound := (getErr != nil && errors.IsNotFound(getErr))
+	found := getErr == nil || !errors.IsNotFound(getErr)
+	stale := found && mt.staleStorageState(ss)
+	storageVersionChanged := found && ss.Status.CurrentStorageVersionHash != r.StorageVersionHash
+	needsMigration := found && !mt.isMigrated(ss) && !mt.hasPendingOrRunningMigration(r)
+	relaunchMigration := stale || !found || storageVersionChanged || needsMigration
 
 	if stale {
 		if err := mt.client.MigrationV1alpha1().StorageStates().Delete(storageStateName(toGroupResource(r)), nil); err != nil {
@@ -197,7 +198,7 @@ func (mt *MigrationTrigger) processDiscoveryResource(r metav1.APIResource) {
 		}
 	}
 
-	if stale || storageVersionChanged || notFound {
+	if relaunchMigration {
 		// Note that this means historical migration objects are deleted.
 		if err := mt.relaunchMigration(r); err != nil {
 			utilruntime.HandleError(err)
@@ -207,3 +208,27 @@ func (mt *MigrationTrigger) processDiscoveryResource(r metav1.APIResource) {
 	// always update status.heartbeat, sometimes update the version hashes.
 	mt.updateStorageState(r.StorageVersionHash, r)
 }
+func (mt *MigrationTrigger) isMigrated(ss *migrationv1alpha1.StorageState) bool {
+	if len(ss.Status.PersistedStorageVersionHashes) != 1 {
+		return false
+	}
+	return ss.Status.CurrentStorageVersionHash == ss.Status.PersistedStorageVersionHashes[0]
+}
+
+func (mt *MigrationTrigger) hasPendingOrRunningMigration(r metav1.APIResource) bool {
+	// get the corresponding StorageVersionMigration resource
+	migrations, err := mt.migrationInformer.GetIndexer().ByIndex(controller.ResourceIndex, controller.ToIndex(toGroupResource(r)))
+	if err != nil {
+		utilruntime.HandleError(err)
+		return false
+	}
+	for _, migration := range migrations {
+		m := migration.(*migrationv1alpha1.StorageVersionMigration)
+		if controller.HasCondition(m, migrationv1alpha1.MigrationSucceeded) || controller.HasCondition(m, migrationv1alpha1.MigrationFailed) {
+			continue
+		}
+		// migration is running or pending
+		return true
+	}
+	return false
+}