Optimizing the State Handling

Author: schongloo

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

@@ -21,10 +21,10 @@
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenStateHandlerRegistry;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.BlueGreenStateHandler;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService;
+import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenStateHandlerRegistry;
+import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.BlueGreenStateHandler;
 import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
 
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
@@ -95,8 +95,8 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
                             josdkContext,
                             null,
                             ctxFactory);
-            return BlueGreenDeploymentService
-                    .patchStatusUpdateControl(context, INITIALIZING_BLUE, null)
+            return BlueGreenDeploymentService.patchStatusUpdateControl(
+                            context, INITIALIZING_BLUE, null)
                     .rescheduleAfter(100);
         } else {
             FlinkBlueGreenDeploymentState currentState = deploymentStatus.getBlueGreenState();
@@ -118,7 +118,7 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
 
             BlueGreenStateHandler handler = handlerRegistry.getHandler(currentState);
             return handler.handle(context);
-//            return stateMachine.processState(context);
+            //            return stateMachine.processState(context);
         }
     }
 

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

@@ -1,15 +1,16 @@
 package org.apache.flink.kubernetes.operator.controller.bluegreen;
 
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.bluegreen.DeploymentType;
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus;
 import org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeployments;
 import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
 
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
 /**
  * Simplified context object containing all the necessary state and dependencies for Blue/Green
  * deployment state transitions.

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

@@ -107,8 +107,7 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                         lastCheckpoint,
                         false);
             } else {
-                if (context.getDeploymentStatus().getJobStatus().getState()
-                        != JobStatus.FAILING) {
+                if (context.getDeploymentStatus().getJobStatus().getState() != JobStatus.FAILING) {
                     return patchStatusUpdateControl(context, null, JobStatus.FAILING);
                 }
             }
@@ -168,15 +167,17 @@ private TransitionState determineTransitionState(
         TransitionState transitionState;
 
         if (DeploymentType.BLUE == currentDeploymentType) {
-            transitionState = new TransitionState(
-                    context.getBlueDeployment(),                    // currentDeployment
-                    context.getGreenDeployment(),                   // nextDeployment
-                    FlinkBlueGreenDeploymentState.ACTIVE_GREEN);    // next State
+            transitionState =
+                    new TransitionState(
+                            context.getBlueDeployment(), // currentDeployment
+                            context.getGreenDeployment(), // nextDeployment
+                            FlinkBlueGreenDeploymentState.ACTIVE_GREEN); // next State
         } else {
-            transitionState = new TransitionState(
-                    context.getGreenDeployment(),               // currentDeployment
-                    context.getBlueDeployment(),                // nextDeployment
-                    FlinkBlueGreenDeploymentState.ACTIVE_BLUE); // next State
+            transitionState =
+                    new TransitionState(
+                            context.getGreenDeployment(), // currentDeployment
+                            context.getBlueDeployment(), // nextDeployment
+                            FlinkBlueGreenDeploymentState.ACTIVE_BLUE); // next State
         }
 
         Preconditions.checkNotNull(
@@ -263,8 +264,7 @@ private UpdateControl<FlinkBlueGreenDeployment> shouldWeAbort(
             FlinkBlueGreenDeploymentState nextState) {
 
         String deploymentName = nextDeployment.getMetadata().getName();
-        long abortTimestamp =
-                instantStrToMillis(context.getDeploymentStatus().getAbortTimestamp());
+        long abortTimestamp = instantStrToMillis(context.getDeploymentStatus().getAbortTimestamp());
 
         if (abortTimestamp == 0) {
             throw new IllegalStateException("Unexpected abortTimestamp == 0");
@@ -337,10 +337,7 @@ private static FlinkBlueGreenDeploymentState getPreviousState(
     public UpdateControl<FlinkBlueGreenDeployment> finalizeBlueGreenDeployment(
             BlueGreenContext context, FlinkBlueGreenDeploymentState nextState) {
 
-        LOG.info(
-                "Finalizing deployment '{}' to {} state",
-                context.getDeploymentName(),
-                nextState);
+        LOG.info("Finalizing deployment '{}' to {} state", context.getDeploymentName(), nextState);
 
         context.getDeploymentStatus().setDeploymentReadyTimestamp(millisToInstantStr(0));
         context.getDeploymentStatus().setAbortTimestamp(millisToInstantStr(0));
@@ -398,4 +395,4 @@ private static class TransitionState {
             this.nextState = nextState;
         }
     }
-}
+}

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

@@ -18,13 +18,10 @@
 package org.apache.flink.kubernetes.operator.controller.bluegreen;
 
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.ActiveBlueStateHandler;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.ActiveGreenStateHandler;
+import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.ActiveStateHandler;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.BlueGreenStateHandler;
 import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.InitializingBlueStateHandler;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.TransitioningToBlueStateHandler;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.TransitioningToGreenStateHandler;
-import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService;
+import org.apache.flink.kubernetes.operator.controller.bluegreen.handlers.TransitioningStateHandler;
 
 import java.util.Map;
 
@@ -47,12 +44,14 @@ public BlueGreenStateHandlerRegistry() {
         this.handlers =
                 Map.of(
                         INITIALIZING_BLUE, new InitializingBlueStateHandler(deploymentService),
-                        ACTIVE_BLUE, new ActiveBlueStateHandler(deploymentService),
-                        ACTIVE_GREEN, new ActiveGreenStateHandler(deploymentService),
+                        ACTIVE_BLUE, new ActiveStateHandler(ACTIVE_BLUE, deploymentService),
+                        ACTIVE_GREEN, new ActiveStateHandler(ACTIVE_GREEN, deploymentService),
                         TRANSITIONING_TO_BLUE,
-                                new TransitioningToBlueStateHandler(deploymentService),
+                                new TransitioningStateHandler(
+                                        TRANSITIONING_TO_BLUE, deploymentService),
                         TRANSITIONING_TO_GREEN,
-                                new TransitioningToGreenStateHandler(deploymentService));
+                                new TransitioningStateHandler(
+                                        TRANSITIONING_TO_GREEN, deploymentService));
     }
 
     /**

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

@@ -18,8 +18,8 @@
 package org.apache.flink.kubernetes.operator.controller.bluegreen.handlers;
 
 import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
-
 import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

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

@@ -25,15 +25,24 @@
 
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 
-/** State handler for the TRANSITIONING_TO_GREEN state. */
-public class TransitioningToGreenStateHandler extends AbstractBlueGreenStateHandler {
+/** Consolidated state handler for both ACTIVE_BLUE and ACTIVE_GREEN states. */
+public class ActiveStateHandler extends AbstractBlueGreenStateHandler {
 
-    public TransitioningToGreenStateHandler(BlueGreenDeploymentService deploymentService) {
-        super(FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, deploymentService);
+    public ActiveStateHandler(
+            FlinkBlueGreenDeploymentState supportedState,
+            BlueGreenDeploymentService deploymentService) {
+        super(supportedState, deploymentService);
     }
 
     @Override
     public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context) {
-        return deploymentService.monitorTransition(context, DeploymentType.BLUE);
+        DeploymentType currentType = getCurrentDeploymentType();
+        return deploymentService.checkAndInitiateDeployment(context, currentType);
+    }
+
+    private DeploymentType getCurrentDeploymentType() {
+        return getSupportedState() == FlinkBlueGreenDeploymentState.ACTIVE_BLUE
+                ? DeploymentType.BLUE
+                : DeploymentType.GREEN;
     }
 }

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

@@ -0,0 +1,48 @@
+/*
+ * 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.controller.bluegreen.handlers;
+
+import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
+import org.apache.flink.kubernetes.operator.api.bluegreen.DeploymentType;
+import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
+import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
+import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService;
+
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+
+/** Consolidated state handler for both TRANSITIONING_TO_BLUE and TRANSITIONING_TO_GREEN states. */
+public class TransitioningStateHandler extends AbstractBlueGreenStateHandler {
+
+    public TransitioningStateHandler(
+            FlinkBlueGreenDeploymentState supportedState,
+            BlueGreenDeploymentService deploymentService) {
+        super(supportedState, deploymentService);
+    }
+
+    @Override
+    public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context) {
+        DeploymentType currentType = getCurrentDeploymentType();
+        return deploymentService.monitorTransition(context, currentType);
+    }
+
+    private DeploymentType getCurrentDeploymentType() {
+        return getSupportedState() == FlinkBlueGreenDeploymentState.TRANSITIONING_TO_BLUE
+                ? DeploymentType.GREEN // Transitioning FROM green TO blue
+                : DeploymentType.BLUE; // Transitioning FROM blue TO green
+    }
+}