Adding a TransitionMode to the CRD/API for the future (currently only BASIC is supported).Tests to confirm the deployments are deleted accurately after the specified deletion delay.

Author: schongloo

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/TransitionMode.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.api.bluegreen;
+
+/** Possible transition modes supported by the `FlinkBlueGreenDeploymentController`. */
+public enum TransitionMode {
+    /**
+     * FLIP-503: simple transition that deletes the previous deployment as soon as the new one is
+     * RUNNING/STABLE.
+     */
+    BASIC,
+
+    /**
+     * FLIP-504: advanced coordination between deployment stages during transition. Not supported
+     * until FLIP-504 is implemented.
+     */
+    ADVANCED;
+}

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkDeploymentTemplateSpec.java

@@ -17,6 +17,8 @@
 
 package org.apache.flink.kubernetes.operator.api.spec;
 
+import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionMode;
+
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
@@ -38,15 +40,18 @@ public class FlinkDeploymentTemplateSpec {
     @JsonProperty("metadata")
     private ObjectMeta metadata;
 
-    @JsonProperty("deploymentDeletionDelaySec")
-    private int deploymentDeletionDelaySec;
+    @JsonProperty("deploymentDeletionDelayMs")
+    private int deploymentDeletionDelayMs;
 
     @JsonProperty("maxNumRetries")
     private int maxNumRetries;
 
     @JsonProperty("reconciliationReschedulingIntervalMs")
     private int reconciliationReschedulingIntervalMs;
 
+    @JsonProperty("transitionMode")
+    private TransitionMode transitionMode;
+
     @JsonProperty("spec")
     private FlinkDeploymentSpec spec;
 

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java

@@ -50,6 +50,9 @@ public class FlinkBlueGreenDeploymentStatus {
     /** Current number of retries. */
     private int numRetries;
 
+    /** Timestamp when the deployment became READY/STABLE. Used to determine when to delete it. */
+    private long deploymentReadyTimestamp;
+
     /** Information about the TaskManagers for the scale subresource. */
     private TaskManagerInfo taskManager;
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java

@@ -61,14 +61,16 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static io.javaoperatorsdk.operator.api.reconciler.UpdateControl.noUpdate;
+
 /** Controller that runs the main reconcile loop for Flink Blue/Green deployments. */
 @ControllerConfiguration
 public class FlinkBlueGreenDeploymentController
         implements Reconciler<FlinkBlueGreenDeployment>,
                 EventSourceInitializer<FlinkBlueGreenDeployment> {
 
     private static final Logger LOG = LoggerFactory.getLogger(FlinkDeploymentController.class);
-    private static final int DEFAULT_MAX_NUM_RETRIES = 5;
+    private static final int DEFAULT_MAX_NUM_RETRIES = 10;
     private static final int DEFAULT_RECONCILIATION_RESCHEDULING_INTERVAL_MS = 15000;
 
     private final FlinkResourceContextFactory ctxFactory;
@@ -144,7 +146,7 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
                             DeploymentType.BLUE,
                             josdkContext);
                 default:
-                    return UpdateControl.noUpdate();
+                    return noUpdate();
             }
         }
     }
@@ -191,39 +193,82 @@ private UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
                         + currentDeploymentType);
 
         if (isDeploymentReady(nextDeployment, josdkContext, deploymentStatus)) {
-            return deleteAndFinalize(
+            return canDelete(
                     bgDeployment, deploymentStatus, josdkContext, currentDeployment, nextState);
         } else {
-            int maxNumRetries = bgDeployment.getSpec().getTemplate().getMaxNumRetries();
-            if (maxNumRetries <= 0) {
-                maxNumRetries = DEFAULT_MAX_NUM_RETRIES;
-            }
+            return retryOrAbort(
+                    bgDeployment, deploymentStatus, josdkContext, nextDeployment, nextState);
+        }
+    }
 
-            if (deploymentStatus.getNumRetries() >= maxNumRetries) {
-                // ABORT
-                // Suspend the nextDeployment (FlinkDeployment)
-                nextDeployment.getStatus().getJobStatus().setState(JobStatus.SUSPENDED);
-                josdkContext.getClient().resource(nextDeployment).update();
+    private UpdateControl<FlinkBlueGreenDeployment> canDelete(
+            FlinkBlueGreenDeployment bgDeployment,
+            FlinkBlueGreenDeploymentStatus deploymentStatus,
+            Context<FlinkBlueGreenDeployment> josdkContext,
+            FlinkDeployment currentDeployment,
+            FlinkBlueGreenDeploymentState nextState) {
+        int deploymentDeletionDelayMs =
+                Math.max(bgDeployment.getSpec().getTemplate().getDeploymentDeletionDelayMs(), 0);
+
+        if (deploymentStatus.getDeploymentReadyTimestamp() == 0) {
+            LOG.info(
+                    "Deployment marked ready on "
+                            + System.currentTimeMillis()
+                            + ", rescheduling reconciliation in "
+                            + deploymentDeletionDelayMs
+                            + " ms.");
+            deploymentStatus.setDeploymentReadyTimestamp(System.currentTimeMillis());
+            return patchStatusUpdateControl(bgDeployment, deploymentStatus, null, null)
+                    .rescheduleAfter(deploymentDeletionDelayMs);
+        }
 
-                // We indicate this Blue/Green deployment is no longer Transitioning
-                //  and rollback the state value
-                deploymentStatus.setBlueGreenState(
-                        nextState == FlinkBlueGreenDeploymentState.ACTIVE_BLUE
-                                ? FlinkBlueGreenDeploymentState.ACTIVE_GREEN
-                                : FlinkBlueGreenDeploymentState.ACTIVE_BLUE);
+        var deletionTs = deploymentStatus.getDeploymentReadyTimestamp() + deploymentDeletionDelayMs;
 
-                // If the current running FlinkDeployment is not in RUNNING/STABLE,
-                // we flag this Blue/Green as FAILING
-                return patchStatusUpdateControl(
-                        bgDeployment, deploymentStatus, null, JobStatus.FAILING, false);
-            } else {
-                // RETRY
-                deploymentStatus.setNumRetries(deploymentStatus.getNumRetries() + 1);
+        if (deletionTs < System.currentTimeMillis()) {
+            return deleteAndFinalize(
+                    bgDeployment, deploymentStatus, josdkContext, currentDeployment, nextState);
+        } else {
+            long delay = deletionTs - System.currentTimeMillis();
+            LOG.info("Rescheduling reconciliation (to delete) in " + delay + " ms.");
+            return UpdateControl.<FlinkBlueGreenDeployment>noUpdate().rescheduleAfter(delay);
+        }
+    }
 
-                LOG.info("Deployment " + nextDeployment.getMetadata().getName() + " not ready yet");
-                return patchStatusUpdateControl(bgDeployment, deploymentStatus, null, null, false)
-                        .rescheduleAfter(getReconciliationReschedInterval(bgDeployment));
-            }
+    private UpdateControl<FlinkBlueGreenDeployment> retryOrAbort(
+            FlinkBlueGreenDeployment bgDeployment,
+            FlinkBlueGreenDeploymentStatus deploymentStatus,
+            Context<FlinkBlueGreenDeployment> josdkContext,
+            FlinkDeployment nextDeployment,
+            FlinkBlueGreenDeploymentState nextState) {
+        int maxNumRetries = bgDeployment.getSpec().getTemplate().getMaxNumRetries();
+        if (maxNumRetries <= 0) {
+            maxNumRetries = DEFAULT_MAX_NUM_RETRIES;
+        }
+
+        if (deploymentStatus.getNumRetries() >= maxNumRetries) {
+            // ABORT
+            // Suspend the nextDeployment (FlinkDeployment)
+            nextDeployment.getStatus().getJobStatus().setState(JobStatus.SUSPENDED);
+            josdkContext.getClient().resource(nextDeployment).replace();
+
+            // We indicate this Blue/Green deployment is no longer Transitioning
+            //  and rollback the state value
+            deploymentStatus.setBlueGreenState(
+                    nextState == FlinkBlueGreenDeploymentState.ACTIVE_BLUE
+                            ? FlinkBlueGreenDeploymentState.ACTIVE_GREEN
+                            : FlinkBlueGreenDeploymentState.ACTIVE_BLUE);
+
+            // If the current running FlinkDeployment is not in RUNNING/STABLE,
+            // we flag this Blue/Green as FAILING
+            return patchStatusUpdateControl(
+                    bgDeployment, deploymentStatus, null, JobStatus.FAILING);
+        } else {
+            // RETRY
+            deploymentStatus.setNumRetries(deploymentStatus.getNumRetries() + 1);
+
+            LOG.info("Deployment " + nextDeployment.getMetadata().getName() + " not ready yet");
+            return patchStatusUpdateControl(bgDeployment, deploymentStatus, null, null)
+                    .rescheduleAfter(getReconciliationReschedInterval(bgDeployment));
         }
     }
 
@@ -245,10 +290,11 @@ private UpdateControl<FlinkBlueGreenDeployment> deleteAndFinalize(
 
         if (currentDeployment != null) {
             deleteDeployment(currentDeployment, josdkContext);
-            return UpdateControl.noUpdate();
+            return noUpdate();
         } else {
+            deploymentStatus.setDeploymentReadyTimestamp(0);
             return patchStatusUpdateControl(
-                    bgDeployment, deploymentStatus, nextState, JobStatus.RUNNING, false);
+                    bgDeployment, deploymentStatus, nextState, JobStatus.RUNNING);
         }
     }
 
@@ -301,16 +347,12 @@ private UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                 // we flag this Blue/Green as FAILING
                 if (deploymentStatus.getJobStatus().getState() != JobStatus.FAILING) {
                     return patchStatusUpdateControl(
-                            flinkBlueGreenDeployment,
-                            deploymentStatus,
-                            null,
-                            JobStatus.FAILING,
-                            false);
+                            flinkBlueGreenDeployment, deploymentStatus, null, JobStatus.FAILING);
                 }
             }
         }
 
-        return UpdateControl.noUpdate();
+        return noUpdate();
     }
 
     private static void setLastReconciledSpec(
@@ -418,12 +460,7 @@ private UpdateControl<FlinkBlueGreenDeployment> initiateDeployment(
                 josdkContext,
                 isFirstDeployment);
 
-        return patchStatusUpdateControl(
-                        flinkBlueGreenDeployment,
-                        deploymentStatus,
-                        nextState,
-                        null,
-                        isFirstDeployment)
+        return patchStatusUpdateControl(flinkBlueGreenDeployment, deploymentStatus, nextState, null)
                 .rescheduleAfter(getReconciliationReschedInterval(flinkBlueGreenDeployment));
     }
 
@@ -476,16 +513,15 @@ private boolean hasSpecChanged(
             DeploymentType deploymentType) {
 
         String lastReconciledSpec = deploymentStatus.getLastReconciledSpec();
-
-        return !lastReconciledSpec.equals(SpecUtils.serializeObject(newSpec, "spec"));
+        String newSpecSerialized = SpecUtils.serializeObject(newSpec, "spec");
+        return !lastReconciledSpec.equals(newSpecSerialized);
     }
 
     private UpdateControl<FlinkBlueGreenDeployment> patchStatusUpdateControl(
             FlinkBlueGreenDeployment flinkBlueGreenDeployment,
             FlinkBlueGreenDeploymentStatus deploymentStatus,
             FlinkBlueGreenDeploymentState deploymentState,
-            JobStatus jobState,
-            boolean isFirstDeployment) {
+            JobStatus jobState) {
         if (deploymentState != null) {
             deploymentStatus.setBlueGreenState(deploymentState);
         }
@@ -494,6 +530,7 @@ private UpdateControl<FlinkBlueGreenDeployment> patchStatusUpdateControl(
             deploymentStatus.getJobStatus().setState(jobState);
         }
 
+        deploymentStatus.setLastReconciledTimestamp(System.currentTimeMillis());
         flinkBlueGreenDeployment.setStatus(deploymentStatus);
         return UpdateControl.patchStatus(flinkBlueGreenDeployment);
     }