Bug 1620608: Fix dc adoption

pkg/apps/apis/apps/validation/validation.go

@@ -160,8 +160,6 @@ func ValidateDeploymentConfigStatusUpdate(newConfig *appsapi.DeploymentConfig, o
 	statusPath := field.NewPath("status")
 	if newConfig.Status.LatestVersion < oldConfig.Status.LatestVersion {
 		allErrs = append(allErrs, field.Invalid(statusPath.Child("latestVersion"), newConfig.Status.LatestVersion, "latestVersion cannot be decremented"))
-	} else if newConfig.Status.LatestVersion > (oldConfig.Status.LatestVersion + 1) {
-		allErrs = append(allErrs, field.Invalid(statusPath.Child("latestVersion"), newConfig.Status.LatestVersion, "latestVersion can only be incremented by 1"))
 	}
 	if newConfig.Status.ObservedGeneration < oldConfig.Status.ObservedGeneration {
 		allErrs = append(allErrs, field.Invalid(statusPath.Child("observedGeneration"), newConfig.Status.ObservedGeneration, "observedGeneration cannot be decremented"))

pkg/apps/apis/apps/validation/validation_test.go

@@ -713,7 +713,6 @@ func TestValidateDeploymentConfigUpdate(t *testing.T) {
 		newLatestVersion int64
 	}{
 		{5, 3},
-		{5, 7},
 		{0, -1},
 	}
 

pkg/apps/controller/deploymentconfig/deploymentconfig_controller.go

@@ -133,7 +133,16 @@ func (c *DeploymentConfigController) Handle(config *appsv1.DeploymentConfig) err
 		return c.updateStatus(config, existingDeployments, true)
 	}
 
+	configCopy := config.DeepCopy()
+
 	latestExists, latestDeployment := appsutil.LatestDeploymentInfo(config, existingDeployments)
+	candidateVersion := appsutil.DeploymentVersionFor(latestDeployment)
+	if candidateVersion > config.Status.LatestVersion {
+		// FIXME: update LatestVersion to candidateVersion directly when validation allows it in all supported skews
+		configCopy.Status.LatestVersion++
+		_, err := c.appsClient.DeploymentConfigs(configCopy.Namespace).UpdateStatus(configCopy)
+		return err
+	}
 
 	if !latestExists {
 		if err := c.cancelRunningRollouts(config, existingDeployments, cm); err != nil {
@@ -149,7 +158,6 @@ func (c *DeploymentConfigController) Handle(config *appsv1.DeploymentConfig) err
 		}
 	}
 
-	configCopy := config.DeepCopy()
 	// Process triggers and start an initial rollouts
 	shouldTrigger, shouldSkip, err := triggerActivated(configCopy, latestExists, latestDeployment)
 	if err != nil {
@@ -186,30 +194,29 @@ func (c *DeploymentConfigController) Handle(config *appsv1.DeploymentConfig) err
 	}
 	created, err := c.kubeClient.ReplicationControllers(config.Namespace).Create(deployment)
 	if err != nil {
-		// We need to find out if our controller owns that deployment and report error if not
 		if kapierrors.IsAlreadyExists(err) {
-			rc, err := c.rcLister.ReplicationControllers(deployment.Namespace).Get(deployment.Name)
+			rc, err := c.kubeClient.ReplicationControllers(deployment.Namespace).Get(deployment.Name, metav1.GetOptions{})
 			if err != nil {
-				return fmt.Errorf("error while deploymentConfigController getting the replication controller %s/%s: %v", deployment.Namespace, deployment.Name, err)
+				return fmt.Errorf("error while getting replication controller %s/%s: %v", deployment.Namespace, deployment.Name, err)
 			}
-			// We need to make sure we own that RC or adopt it if possible
 			isOurs, err := cm.ClaimReplicationController(rc)
-			if err != nil {
-				return fmt.Errorf("error while deploymentConfigController claiming the replication controller: %v", err)
-			}
-			if isOurs {
-				// If the deployment was already created, just move on. The cache could be
-				// stale, or another process could have already handled this update.
-				return c.updateStatus(config, existingDeployments, true)
+			if err == nil {
+				if isOurs {
+					// Caches are stalled, wait for them to sync (errors cause requeue)
+					return fmt.Errorf("caches for RC %s/%s are stale, waiting to catch up", deployment.Namespace, deployment.Name)
+				} else {
+					err = fmt.Errorf("deployment %s/%s already exists and it couldn't be adopted",
+						deployment.Namespace, deployment.Name)
+				}
 			} else {
-				err = fmt.Errorf("replication controller %s already exists and deployment config is not allowed to claim it", deployment.Name)
-				c.recorder.Eventf(config, v1.EventTypeWarning, "DeploymentCreationFailed", "Couldn't deploy version %d: %v", config.Status.LatestVersion, err)
-				return c.updateStatus(config, existingDeployments, true)
+				err = fmt.Errorf("deployment %s/%s already exists and it couldn't be adopted: %v",
+					deployment.Namespace, deployment.Name, err)
 			}
 		}
+
 		c.recorder.Eventf(config, v1.EventTypeWarning, "DeploymentCreationFailed", "Couldn't deploy version %d: %s", config.Status.LatestVersion, err)
-		// We don't care about this error since we need to report the create failure.
 		cond := appsutil.NewDeploymentCondition(appsv1.DeploymentProgressing, v1.ConditionFalse, appsutil.FailedRcCreateReason, err.Error())
+		// We don't care about this error since we need to report the create failure.
 		_ = c.updateStatus(config, existingDeployments, true, *cond)
 		return fmt.Errorf("couldn't create deployment for deployment config %s: %v", appsutil.LabelForDeploymentConfig(config), err)
 	}

test/extended/deployments/deployments.go

@@ -1545,4 +1545,125 @@ var _ = g.Describe("[Feature:DeploymentConfig] deploymentconfigs", func() {
 			o.Expect(strings.TrimSpace(rcs.Items[0].Spec.Template.Spec.Containers[0].Image)).NotTo(o.BeEmpty())
 		})
 	})
+
+	g.Describe("adoption [Conformance]", func() {
+		dcName := "deployment-simple"
+		g.AfterEach(func() {
+			failureTrap(oc, dcName, g.CurrentGinkgoTestDescription().Failed)
+		})
+
+		g.It("will orphan all RCs and adopt them back when recreated", func() {
+			namespace := oc.Namespace()
+
+			g.By("creating DC")
+			dc, err := readDCFixture(simpleDeploymentFixture)
+			o.Expect(err).NotTo(o.HaveOccurred())
+			o.Expect(dc.Name).To(o.Equal(dcName))
+
+			dc, err = oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Create(dc)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			o.Expect(dc.Status.LatestVersion).To(o.BeEquivalentTo(0))
+
+			g.By("waiting for initial deployment to complete")
+			o.Expect(waitForLatestCondition(oc, dcName, deploymentRunTimeout, deploymentReachedCompletion)).NotTo(o.HaveOccurred())
+
+			g.By("modifying the template and triggering new deployment")
+			dc, err = oc.AppsClient().AppsV1().DeploymentConfigs(oc.Namespace()).Patch(dcName, types.StrategicMergePatchType, []byte(`{"spec": {"template": {"metadata": {"labels": {"rev": "2"}}}}}`))
+			o.Expect(err).NotTo(o.HaveOccurred())
+			// LatestVersion is always 1 behind on api calls before the controller detects the change and raises it
+			o.Expect(dc.Status.LatestVersion).To(o.BeEquivalentTo(1))
+
+			g.By("waiting for the second deployment to complete")
+			o.Expect(waitForLatestCondition(oc, dcName, deploymentRunTimeout, deploymentReachedCompletion)).NotTo(o.HaveOccurred())
+
+			g.By("verifying the second deployment")
+			dc, err = oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Get(dc.Name, metav1.GetOptions{})
+			o.Expect(err).NotTo(o.HaveOccurred())
+			o.Expect(dc.Status.LatestVersion).To(o.BeEquivalentTo(2))
+
+			g.By("deleting the DC and orphaning RCs")
+			deletePropagationOrphan := metav1.DeletePropagationOrphan
+			err = oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Delete(dc.Name, &metav1.DeleteOptions{
+				PropagationPolicy: &deletePropagationOrphan,
+			})
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			// Wait for deletion
+			w, err := oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Watch(metav1.SingleObject(dc.ObjectMeta))
+			o.Expect(err).NotTo(o.HaveOccurred())
+			_, err = watch.Until(deploymentChangeTimeout, w, func(e watch.Event) (bool, error) {
+				switch e.Type {
+				case watch.Added, watch.Modified:
+					e2e.Logf("delete: LatestVersion: %d", e.Object.(*appsv1.DeploymentConfig).Status.LatestVersion)
+					return false, nil
+				case watch.Deleted:
+					return true, nil
+				case watch.Error:
+					return true, kerrors.FromObject(e.Object)
+				default:
+					return true, fmt.Errorf("unexpected event %#v", e)
+				}
+			})
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			g.By("recreating the DC")
+			dc.ResourceVersion = ""
+			dc, err = oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Create(dc)
+			o.Expect(err).NotTo(o.HaveOccurred())
+			// When a DC is recreated it has LatestVersion 0, it will get updated after adopting the Rcs
+			o.Expect(dc.Status.LatestVersion).To(o.BeEquivalentTo(0))
+
+			g.By("waiting for DC.status.latestVersion to be raised after adopting RCs and availableReplicas to match replicas")
+			w, err = oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Watch(metav1.SingleObject(dc.ObjectMeta))
+			o.Expect(err).NotTo(o.HaveOccurred())
+			event, err := watch.Until(deploymentChangeTimeout, w, func(e watch.Event) (bool, error) {
+				switch e.Type {
+				case watch.Added, watch.Modified:
+					evDC := e.Object.(*appsv1.DeploymentConfig)
+					e2e.Logf("wait: LatestVersion: %d", e.Object.(*appsv1.DeploymentConfig).Status.LatestVersion)
+					return evDC.Status.LatestVersion == 2 && evDC.Status.AvailableReplicas == evDC.Spec.Replicas, nil
+				case watch.Deleted:
+					return true, fmt.Errorf("dc deleted while waiting for latestVersion to be raised")
+				case watch.Error:
+					return true, kerrors.FromObject(e.Object)
+				default:
+					return true, fmt.Errorf("unexpected event %#v", e)
+				}
+			})
+			o.Expect(err).NotTo(o.HaveOccurred())
+			dc = event.Object.(*appsv1.DeploymentConfig)
+
+			g.By("making sure DC can be scaled")
+			newScale := dc.Spec.Replicas + 2
+			dc, err = oc.AppsClient().AppsV1().DeploymentConfigs(oc.Namespace()).Patch(dcName, types.StrategicMergePatchType, []byte(fmt.Sprintf(`{"spec": {"replicas": %d}}`, newScale)))
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			w, err = oc.AppsClient().AppsV1().DeploymentConfigs(namespace).Watch(metav1.SingleObject(dc.ObjectMeta))
+			o.Expect(err).NotTo(o.HaveOccurred())
+			event, err = watch.Until(deploymentChangeTimeout, w, func(e watch.Event) (bool, error) {
+				switch e.Type {
+				case watch.Added, watch.Modified:
+					evDC := e.Object.(*appsv1.DeploymentConfig)
+					return evDC.Status.AvailableReplicas == evDC.Spec.Replicas, nil
+				case watch.Deleted:
+					return true, fmt.Errorf("dc deleted while waiting for latestVersion to be raised")
+				case watch.Error:
+					return true, kerrors.FromObject(e.Object)
+				default:
+					return true, fmt.Errorf("unexpected event %#v", e)
+				}
+			})
+			o.Expect(err).NotTo(o.HaveOccurred())
+			dc = event.Object.(*appsv1.DeploymentConfig)
+
+			g.By("rolling out new version")
+			o.Expect(dc.Status.LatestVersion).To(o.BeEquivalentTo(2))
+
+			dc, err = oc.AppsClient().AppsV1().DeploymentConfigs(oc.Namespace()).Patch(dcName, types.StrategicMergePatchType, []byte(fmt.Sprintf(`{"spec": {"template": {"metadata": {"labels": {"rev": "%d"}}}}}`, dc.Status.LatestVersion+1)))
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			o.Expect(waitForLatestCondition(oc, dcName, deploymentRunTimeout, deploymentReachedCompletion)).NotTo(o.HaveOccurred())
+		})
+	})
 })

test/extended/deployments/util.go

@@ -205,6 +205,9 @@ func GetDeploymentCondition(status appsv1.DeploymentConfigStatus, condType appsv
 }
 
 func deploymentReachedCompletion(dc *appsv1.DeploymentConfig, rcs []*corev1.ReplicationController, pods []corev1.Pod) (bool, error) {
+	if dc.Status.ObservedGeneration != dc.Generation {
+		return false, nil
+	}
 	if len(rcs) == 0 {
 		return false, nil
 	}

test/extended/testdata/deployments/deployment-trigger.yaml

@@ -16,7 +16,7 @@ spec:
         name: test
         command:
         - /bin/sleep
-        - "100"
+        - "infinity"
   test: false
   triggers:
   - imageChangeParams: