wip: add crd storage migration to safely deprecate old storage versions

Author: exdx

pkg/controller/operators/catalog/step.go

@@ -133,13 +133,32 @@ func (b *builder) NewCRDV1Step(client apiextensionsv1client.ApiextensionsV1Inter
 					logger.Debugf("Found one owner for CRD %v", crd)
 				} else if len(matchedCSV) > 1 {
 					logger.Debugf("Found multiple owners for CRD %v", crd)
-
 					if err = validateV1CRDCompatibility(b.dynamicClient, currentCRD, crd); err != nil {
 						return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error validating existing CRs agains new CRD's schema: %s", step.Resource.Name)
 					}
 				}
 
-				// TODO ensure stored version compatibility
+				// (optional) run migration to new stored version to ensure no data loss during CRD upgrade
+				if crdlib.RunStorageMigration(currentCRD, crd) {
+					// add new storage version to current CRD
+					currentCRD.Status.StoredVersions = append(currentCRD.Status.StoredVersions, crdlib.GetNewStorageVersion(crd))
+					_, err := client.CustomResourceDefinitions().UpdateStatus(context.TODO(), crd, metav1.UpdateOptions{})
+					if err != nil {
+						return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error updating CRD status: %s", step.Resource.Name)
+					}
+					// list all CRs corresponding to the CRD
+					kind := fmt.Sprintf("kind:%s", currentCRD.Spec.Names.Kind)
+					// TODO get all CRs via rest client
+					b.logger.Infof("updating %s", kind)
+					// remove old storage version from CRD
+					newVersions := currentCRD.Status.StoredVersions[1:] //remove first (oldest) version
+					currentCRD.Status.StoredVersions = newVersions
+					_, err = client.CustomResourceDefinitions().UpdateStatus(context.TODO(), crd, metav1.UpdateOptions{})
+					if err != nil {
+						return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error updating CRD status: %s", step.Resource.Name)
+					}
+				}
+				
 				// Update CRD to new version
 				_, err = client.CustomResourceDefinitions().Update(context.TODO(), crd, metav1.UpdateOptions{})
 				if err != nil {

pkg/lib/crd/version.go

@@ -76,9 +76,70 @@ func StoredVersions(obj runtime.Object) []string {
 	return nil
 }
 
-// NewStoredVersions returns all the versions that are stored in both the new and old CRD.
-// If some versions are no longer stored in the new CRD then OLM must migrate existing CRs at the old storage version.
-func NewStoredVersions(oldCRD runtime.Object, newCRD runtime.Object) ([]string, error) {
+// RunStorageMigration determines whether the new CRD changes the storage version of the existing CRD.
+// If true, OLM must run a migration process to ensure all CRs can be stored at the new version.
+// See https://kubernetes.io/docs/tasks/access-kubernetes-api/custom-resources/custom-resource-definition-versioning/#upgrade-existing-objects-to-a-new-stored-version
+func RunStorageMigration(oldCRD runtime.Object, newCRD runtime.Object) bool {
+	newStoredVersion := make(map[string]struct{})
+	switch crd := newCRD.(type) {
+	case *apiextensionsv1.CustomResourceDefinition:
+		for _, version := range crd.Spec.Versions {
+			if version.Storage {
+				newStoredVersion[version.Name] = struct{}{}
+			}
+		}
+	case *apiextensionsv1beta1.CustomResourceDefinition:
+		for _, version := range crd.Spec.Versions {
+			if version.Storage {
+				newStoredVersion[version.Name] = struct{}{}
+			}
+		}
+	}
+
+	// find old storage versions by inspect the status field of the existing on-cluster CRD
+	oldStoredVersions := make(map[string]struct{})
+	switch crd := oldCRD.(type) {
+	case *apiextensionsv1.CustomResourceDefinition:
+		for _, version := range crd.Status.StoredVersions {
+			oldStoredVersions[version] = struct{}{}
+		}
+	case *apiextensionsv1beta1.CustomResourceDefinition:
+		for _, version := range crd.Status.StoredVersions {
+			oldStoredVersions[version] = struct{}{}
+		}
+	}
+
+	for name := range oldStoredVersions {
+		if _, ok := newStoredVersion[name]; ok {
+			// new storage version exists in old CRD present on the cluster
+			// no need to run migration
+			return false
+		}
+	}
+	return true
+}
+
+func GetNewStorageVersion(crd runtime.Object) string {
+	switch crd := crd.(type) {
+	case *apiextensionsv1.CustomResourceDefinition:
+		for _, version := range crd.Spec.Versions {
+			if version.Storage {
+				return version.Name
+			}
+		}
+	case *apiextensionsv1beta1.CustomResourceDefinition:
+		for _, version := range crd.Spec.Versions {
+			if version.Storage {
+				return version.Name
+			}
+		}
+	}
+	return ""
+}
+
+// DeprecatedStorageVersions returns all the versions that are in the old CRD but no longer in the new CRD.
+// These versions should be removed after the storage upgrade procedure.
+func DeprecatedStorageVersions(oldCRD runtime.Object, newCRD runtime.Object) ([]string, error) {
 	var newStoredVersions []string
 
 	newCRDVersions, err := ResourceVersions(newCRD)
@@ -93,6 +154,9 @@ func NewStoredVersions(oldCRD runtime.Object, newCRD runtime.Object) ([]string,
 		}
 	}
 
-	return newStoredVersions, nil
+	var deprecatedVersions []string
+
+
+	return deprecatedVersions, nil
 }