Adding extra savepoint "carry over" feature

Author: schongloo

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

@@ -36,6 +36,8 @@
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -129,6 +131,9 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                             context, currentBlueGreenDeploymentType, currentFlinkDeployment);
                 } else {
                     setLastReconciledSpec(context);
+                    LOG.info(
+                            "Patching FlinkDeployment '{}' during checkAndInitiateDeployment",
+                            currentFlinkDeployment.getMetadata().getName());
                     return patchFlinkDeployment(context, currentBlueGreenDeploymentType);
                 }
             } else {
@@ -150,25 +155,65 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
     }
 
     private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
-            BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType) {
+            BlueGreenContext context, BlueGreenDeploymentType blueGreenDeploymentTypeToPatch) {
 
         String childDeploymentName =
                 context.getBgDeployment().getMetadata().getName()
                         + "-"
-                        + currentBlueGreenDeploymentType.toString().toLowerCase();
-        LOG.info("Patching FlinkDeployment '{}'", childDeploymentName);
+                        + blueGreenDeploymentTypeToPatch.toString().toLowerCase();
 
         // We want to patch, therefore the transition should point to the existing deployment
         // details
-        var patchingState = calculatePatchingState(currentBlueGreenDeploymentType);
+        var patchingState = calculatePatchingState(blueGreenDeploymentTypeToPatch);
 
         // If we're not transitioning between deployments, mark as a single deployment to have it
         // not wait for synchronization
         var isFirstDeployment = context.getDeployments().getNumberOfDeployments() != 2;
 
-        // No checkpoint will be used when patching
+        // TODO: if the resource failed right after being deployed with an initialSavepointPath,
+        //  will it be used by this patching? otherwise this is unnecessary, keep lastSavepoint =
+        // null.
+        Savepoint lastSavepoint =
+                carryOverSavepoint(context, blueGreenDeploymentTypeToPatch, childDeploymentName);
+
         return initiateDeployment(
-                context, currentBlueGreenDeploymentType, patchingState, null, isFirstDeployment);
+                context,
+                blueGreenDeploymentTypeToPatch,
+                patchingState,
+                lastSavepoint,
+                isFirstDeployment);
+    }
+
+    @Nullable
+    private static Savepoint carryOverSavepoint(
+            BlueGreenContext context,
+            BlueGreenDeploymentType blueGreenDeploymentTypeToPatch,
+            String childDeploymentName) {
+        var deploymentToPatch = context.getDeploymentByType(blueGreenDeploymentTypeToPatch);
+        var initialSavepointPath = deploymentToPatch.getSpec().getJob().getInitialSavepointPath();
+
+        if (initialSavepointPath == null || initialSavepointPath.isEmpty()) {
+            initialSavepointPath =
+                    deploymentToPatch.getStatus().getJobStatus().getUpgradeSavepointPath();
+        }
+
+        Savepoint lastSavepoint = null;
+        if (initialSavepointPath != null && !initialSavepointPath.isEmpty()) {
+            var ctx =
+                    context.getCtxFactory()
+                            .getResourceContext(deploymentToPatch, context.getJosdkContext());
+
+            lastSavepoint = getSavepointObject(ctx, initialSavepointPath);
+
+            LOG.info(
+                    "Patching FlinkDeployment '{}', carrying over Savepoint at: '{}'",
+                    childDeploymentName,
+                    initialSavepointPath);
+        } else {
+            LOG.info("Patching FlinkDeployment '{}'", childDeploymentName);
+        }
+
+        return lastSavepoint;
     }
 
     private UpdateControl<FlinkBlueGreenDeployment> startTransition(
@@ -227,28 +272,17 @@ public boolean monitorSavepoint(
 
     private Savepoint configureInitialSavepoint(
             BlueGreenContext context, FlinkDeployment currentFlinkDeployment) {
-
-        FlinkResourceContext<FlinkDeployment> ctx =
-                context.getCtxFactory()
-                        .getResourceContext(currentFlinkDeployment, context.getJosdkContext());
-
         // Create savepoint for all upgrade modes except STATELESS
         // (originally only SAVEPOINT mode required savepoints)
         if (isSavepointRequired(context)) {
+            FlinkResourceContext<FlinkDeployment> ctx =
+                    context.getCtxFactory()
+                            .getResourceContext(currentFlinkDeployment, context.getJosdkContext());
+
             String savepointTriggerId = context.getDeploymentStatus().getSavepointTriggerId();
             var savepointFetchResult = fetchSavepointInfo(ctx, savepointTriggerId);
 
-            org.apache.flink.core.execution.SavepointFormatType coreSavepointFormatType =
-                    ctx.getObserveConfig()
-                            .get(KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE);
-
-            var savepointFormatType =
-                    SavepointFormatType.valueOf(coreSavepointFormatType.toString());
-
-            return Savepoint.of(
-                    savepointFetchResult.getLocation(),
-                    SnapshotTriggerType.MANUAL,
-                    savepointFormatType);
+            return getSavepointObject(ctx, savepointFetchResult.getLocation());
         }
 
         // Currently not using last checkpoint recovery for LAST_STATE upgrade mode
@@ -262,6 +296,18 @@ private Savepoint configureInitialSavepoint(
         //        return getLastCheckpoint(ctx);
     }
 
+    @NotNull
+    private static Savepoint getSavepointObject(
+            FlinkResourceContext<FlinkDeployment> ctx, String savepointLocation) {
+        org.apache.flink.core.execution.SavepointFormatType coreSavepointFormatType =
+                ctx.getObserveConfig()
+                        .get(KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE);
+
+        var savepointFormatType = SavepointFormatType.valueOf(coreSavepointFormatType.toString());
+
+        return Savepoint.of(savepointLocation, SnapshotTriggerType.MANUAL, savepointFormatType);
+    }
+
     private boolean handleSavepoint(
             BlueGreenContext context, FlinkDeployment currentFlinkDeployment) {
 
@@ -338,6 +384,11 @@ private UpdateControl<FlinkBlueGreenDeployment> handleSpecChangesDuringTransitio
                 setLastReconciledSpec(context);
                 var oppositeDeploymentType =
                         context.getOppositeDeploymentType(currentBlueGreenDeploymentType);
+                LOG.info(
+                        "Patching FlinkDeployment '{}' during handleSpecChangesDuringTransition",
+                        context.getDeploymentByType(oppositeDeploymentType)
+                                .getMetadata()
+                                .getName());
                 return patchFlinkDeployment(context, oppositeDeploymentType);
             }
         }