feat(olm): allow CRD updates with multiple owners

Currently CRD updates aren't done for existing CRDs, so this enables
that functionality in the cases of:
- New CRD is a different non-existing version
- New CRD is the same version without an openapi schema change

pkg/controller/operators/catalog/operator.go

@@ -1021,6 +1021,44 @@ func (o *Operator) ResolvePlan(plan *v1alpha1.InstallPlan) error {
 	return nil
 }
 
+// TODO: in the future this will be extended to verify more than just whether or not the schema changed
+func checkCRDSchemas(oldCRD *v1beta1ext.CustomResourceDefinition, newCRD *v1beta1ext.CustomResourceDefinition) bool {
+	if reflect.DeepEqual(oldCRD.Spec.Validation, newCRD.Spec.Validation) {
+		return true
+	}
+
+	return false
+}
+
+func safeToUpdate(oldCRD *v1beta1ext.CustomResourceDefinition, newCRD *v1beta1ext.CustomResourceDefinition) error {
+	shouldCheckSchema := len(oldCRD.Spec.Versions) == len(newCRD.Spec.Versions) // no version bump
+
+	// 1) if there's no version change, verify that schemas match 2) ensure all old versions are present
+	for _, oldVersion := range oldCRD.Spec.Versions {
+		var versionPresent bool
+		for _, newVersion := range newCRD.Spec.Versions {
+			if oldVersion.Name == newVersion.Name {
+				if shouldCheckSchema && !checkCRDSchemas(oldCRD, newCRD) {
+					return fmt.Errorf("not allowing CRD (%v) update with multiple owners with schema change on version %v", newCRD.GetName(), oldVersion)
+				}
+				versionPresent = true
+			}
+		}
+		if !versionPresent {
+			return fmt.Errorf("not allowing CRD (%v) update with unincluded version %v", newCRD.GetName(), oldVersion)
+		}
+	}
+
+	// ensure a CRD with multiple versions isn't checked
+	if newCRD.Spec.Version != "" {
+		if !checkCRDSchemas(oldCRD, newCRD) {
+			return fmt.Errorf("not allowing CRD (%v) update with multiple owners with schema change on (single) version %v", newCRD.GetName(), newCRD.Spec.Version)
+		}
+	}
+	return nil
+
+}
+
 // ExecutePlan applies a planned InstallPlan to a namespace.
 func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 	if plan.Status.Phase != v1alpha1.InstallPlanPhaseInstalling {
@@ -1067,9 +1105,19 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 						if err != nil {
 							return errorwrap.Wrapf(err, "error find matched CSV: %s", step.Resource.Name)
 						}
+						crd.SetResourceVersion(currentCRD.GetResourceVersion())
 						if len(matchedCSV) == 1 {
-							// Attempt to update CRD
-							crd.SetResourceVersion(currentCRD.GetResourceVersion())
+							o.logger.Debugf("Found one owner for CRD %v", crd)
+							_, err = o.opClient.ApiextensionsV1beta1Interface().ApiextensionsV1beta1().CustomResourceDefinitions().Update(&crd)
+							if err != nil {
+								return errorwrap.Wrapf(err, "error updating CRD: %s", step.Resource.Name)
+							}
+						} else if len(matchedCSV) > 1 {
+							o.logger.Debugf("Found multiple owners for CRD %v", crd)
+							if err := safeToUpdate(currentCRD, &crd); err != nil {
+								return err
+							}
+
 							_, err = o.opClient.ApiextensionsV1beta1Interface().ApiextensionsV1beta1().CustomResourceDefinitions().Update(&crd)
 							if err != nil {
 								return errorwrap.Wrapf(err, "error update CRD: %s", step.Resource.Name)

pkg/controller/operators/catalog/operator_test.go

@@ -121,6 +121,99 @@ func TestTransitionInstallPlan(t *testing.T) {
 	}
 }
 
+func TestSafeToUpdate(t *testing.T) {
+	mainCRDPlural := "ins-main-abcde"
+	differentSchema := &v1beta1.CustomResourceValidation{
+		OpenAPIV3Schema: &v1beta1.JSONSchemaProps{
+			Properties: map[string]v1beta1.JSONSchemaProps{
+				"spec": {
+					Type:        "object",
+					Description: "Spec of a test object.",
+					Properties: map[string]v1beta1.JSONSchemaProps{
+						"scalar": {
+							Type:        "number",
+							Description: "Scalar value that should have a min and max.",
+							Minimum: func() *float64 {
+								var min float64 = 2
+								return &min
+							}(),
+							Maximum: func() *float64 {
+								var max float64 = 256
+								return &max
+							}(),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	differentVersions := []v1beta1.CustomResourceDefinitionVersion{
+		{
+			Name:    "v1alpha1",
+			Served:  true,
+			Storage: false,
+		},
+		{
+			Name:    "v1alpha2",
+			Served:  true,
+			Storage: true,
+		},
+	}
+
+	tests := []struct {
+		name            string
+		oldCRD          v1beta1.CustomResourceDefinition
+		newCRD          v1beta1.CustomResourceDefinition
+		expectedFailure bool
+	}{
+		{
+			// in reality this would never happen within ExecutePlan, but fully exercises function
+			name:   "same version, same schema",
+			oldCRD: crd(mainCRDPlural),
+			newCRD: crd(mainCRDPlural),
+		},
+		{
+			name:   "same version, different schema",
+			oldCRD: crd(mainCRDPlural),
+			newCRD: func() v1beta1.CustomResourceDefinition {
+				newCRD := crd(mainCRDPlural)
+				newCRD.Spec.Validation = differentSchema
+				return newCRD
+			}(),
+			expectedFailure: true,
+		},
+		{
+			name:   "different version, different schema",
+			oldCRD: crd(mainCRDPlural),
+			newCRD: func() v1beta1.CustomResourceDefinition {
+				newCRD := crd(mainCRDPlural)
+				newCRD.Spec.Version = ""
+				newCRD.Spec.Versions = differentVersions
+				newCRD.Spec.Validation = differentSchema
+				return newCRD
+			}(),
+		},
+		{
+			name:   "different version, same schema",
+			oldCRD: crd(mainCRDPlural),
+			newCRD: func() v1beta1.CustomResourceDefinition {
+				newCRD := crd(mainCRDPlural)
+				newCRD.Spec.Version = ""
+				newCRD.Spec.Versions = differentVersions
+				return newCRD
+			}(),
+		},
+	}
+
+	for _, tt := range tests {
+		err := safeToUpdate(&tt.oldCRD, &tt.newCRD)
+		if tt.expectedFailure {
+			require.Error(t, err)
+		}
+	}
+}
+
 func TestExecutePlan(t *testing.T) {
 	namespace := "ns"