Introducing a Blue/Green State Machine

Author: schongloo

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

@@ -0,0 +1,36 @@
+/*
+ * 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;
+
+/** Enum representing different types of differences in Blue/Green deployment specifications. */
+public enum BlueGreenDiffType {
+    /** No changes detected, should be ignored. */
+    IGNORE,
+
+    /** Changes that require a Blue/Green transition. */
+    TRANSITION,
+
+    /** Changes that only affect the top-level configuration. */
+    PATCH_TOP_LEVEL,
+
+    /** Changes that only affect the child FlinkDeploymentSpec. */
+    PATCH_CHILD,
+
+    /** Changes that affect both top-level and child specifications. */
+    PATCH_BOTH
+}

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

@@ -19,8 +19,12 @@
 
 import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
 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.BlueGreenStateMachine.BlueGreenTransitionContext;
+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.service.FlinkResourceContextFactory;
 
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
@@ -48,13 +52,13 @@ public class FlinkBlueGreenDeploymentController implements Reconciler<FlinkBlueG
     private static final Logger LOG = LoggerFactory.getLogger(FlinkDeploymentController.class);
 
     private final FlinkResourceContextFactory ctxFactory;
-    private final BlueGreenStateMachine stateMachine;
+    private final BlueGreenStateHandlerRegistry handlerRegistry;
 
     public static long minimumAbortGracePeriodMs = ABORT_GRACE_PERIOD.defaultValue().toMillis();
 
     public FlinkBlueGreenDeploymentController(FlinkResourceContextFactory ctxFactory) {
         this.ctxFactory = ctxFactory;
-        this.stateMachine = new BlueGreenStateMachine();
+        this.handlerRegistry = new BlueGreenStateHandlerRegistry();
     }
 
     @Override
@@ -85,27 +89,36 @@ public UpdateControl<FlinkBlueGreenDeployment> reconcile(
 
         if (deploymentStatus == null) {
             var context =
-                    new BlueGreenTransitionContext(
+                    new BlueGreenContext(
                             bgDeployment,
                             new FlinkBlueGreenDeploymentStatus(),
                             josdkContext,
                             null,
                             ctxFactory);
-            return stateMachine
+            return BlueGreenDeploymentService
                     .patchStatusUpdateControl(context, INITIALIZING_BLUE, null)
                     .rescheduleAfter(100);
         } else {
+            FlinkBlueGreenDeploymentState currentState = deploymentStatus.getBlueGreenState();
             var context =
-                    new BlueGreenTransitionContext(
+                    new BlueGreenContext(
                             bgDeployment,
                             deploymentStatus,
                             josdkContext,
-                            deploymentStatus.getBlueGreenState() == INITIALIZING_BLUE
+                            currentState == INITIALIZING_BLUE
                                     ? null
                                     : FlinkBlueGreenDeployments.fromSecondaryResources(
                                             josdkContext),
                             ctxFactory);
-            return stateMachine.processState(context);
+
+            LOG.debug(
+                    "Processing state: {} for deployment: {}",
+                    currentState,
+                    context.getDeploymentName());
+
+            BlueGreenStateHandler handler = handlerRegistry.getHandler(currentState);
+            return handler.handle(context);
+//            return stateMachine.processState(context);
         }
     }
 

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

@@ -0,0 +1,41 @@
+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;
+
+/**
+ * Simplified context object containing all the necessary state and dependencies for Blue/Green
+ * deployment state transitions.
+ */
+@Getter
+@RequiredArgsConstructor
+public class BlueGreenContext {
+    private final FlinkBlueGreenDeployment bgDeployment;
+    private final FlinkBlueGreenDeploymentStatus deploymentStatus;
+    private final Context<FlinkBlueGreenDeployment> josdkContext;
+    private final FlinkBlueGreenDeployments deployments;
+    private final FlinkResourceContextFactory ctxFactory;
+
+    public String getDeploymentName() {
+        return bgDeployment.getMetadata().getName();
+    }
+
+    public FlinkDeployment getBlueDeployment() {
+        return deployments != null ? deployments.getFlinkDeploymentBlue() : null;
+    }
+
+    public FlinkDeployment getGreenDeployment() {
+        return deployments != null ? deployments.getFlinkDeploymentGreen() : null;
+    }
+
+    public FlinkDeployment getDeploymentByType(DeploymentType type) {
+        return type == DeploymentType.BLUE ? getBlueDeployment() : getGreenDeployment();
+    }
+}