[FLINK-37515] Basic support for Blue/Green deployments

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

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import io.fabric8.kubernetes.api.model.Namespaced;
+import io.fabric8.kubernetes.client.CustomResource;
+import io.fabric8.kubernetes.model.annotation.Group;
+import io.fabric8.kubernetes.model.annotation.ShortNames;
+import io.fabric8.kubernetes.model.annotation.Version;
+
+/** Custom resource definition that represents a deployments with Blue/Green rollout capability. */
+@Experimental
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonDeserialize()
+@Group(CrdConstants.API_GROUP)
+@Version(CrdConstants.API_VERSION)
+@ShortNames({"flinkbgdep"})
+public class FlinkBlueGreenDeployment
+        extends CustomResource<FlinkBlueGreenDeploymentSpec, FlinkBlueGreenDeploymentStatus>
+        implements Namespaced {}

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

@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+
+/** Enumeration of the two possible Flink Blue/Green deployment types. */
+public enum DeploymentType {
+    BLUE,
+    GREEN;
+
+    public static DeploymentType fromDeployment(FlinkDeployment flinkDeployment) {
+        String typeAnnotation =
+                flinkDeployment.getMetadata().getLabels().get(DeploymentType.class.getSimpleName());
+        return DeploymentType.valueOf(typeAnnotation);
+    }
+}

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkBlueGreenDeploymentSpec.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.spec;
+
+import org.apache.flink.annotation.Experimental;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Spec that describes a Flink application with blue/green deployment capabilities. */
+@Experimental
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class FlinkBlueGreenDeploymentSpec {
+
+    private FlinkDeploymentTemplateSpec template;
+}

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

@@ -0,0 +1,55 @@
+/*
+ * 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.spec;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** Template Spec that describes a Flink application managed by the blue/green controller. */
+@AllArgsConstructor
+@NoArgsConstructor
+@Data
+@SuperBuilder
+public class FlinkDeploymentTemplateSpec {
+
+    @JsonProperty("metadata")
+    private ObjectMeta metadata;
+
+    @JsonProperty("deploymentDeletionDelaySec")
+    private int deploymentDeletionDelaySec;
+
+    @JsonProperty("maxNumRetries")
+    private int maxNumRetries;
+
+    @JsonProperty("reconciliationReschedulingIntervalMs")
+    private int reconciliationReschedulingIntervalMs;
+
+    @JsonProperty("spec")
+    private FlinkDeploymentSpec spec;
+
+    @JsonIgnore
+    private Map<String, Object> additionalProperties = new LinkedHashMap<String, Object>();
+}

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

@@ -0,0 +1,26 @@
+/*
+ * 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.status;
+
+/** Enumeration of the possible states of the blue/green transition. */
+public enum FlinkBlueGreenDeploymentState {
+    ACTIVE_BLUE,
+    ACTIVE_GREEN,
+    TRANSITIONING_TO_BLUE,
+    TRANSITIONING_TO_GREEN,
+}

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

@@ -0,0 +1,55 @@
+/*
+ * 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.status;
+
+import org.apache.flink.annotation.Experimental;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+/** Last observed status of the Flink Blue/Green deployment. */
+@Experimental
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@ToString(callSuper = true)
+@SuperBuilder
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class FlinkBlueGreenDeploymentStatus {
+
+    private JobStatus jobStatus = new JobStatus();
+
+    /** The state of the blue/green transition. */
+    private FlinkBlueGreenDeploymentState blueGreenState;
+
+    /** Last reconciled (serialized) deployment spec. */
+    private String lastReconciledSpec;
+
+    /** Timestamp of last reconciliation. */
+    private Long lastReconciledTimestamp;
+
+    /** Current number of retries. */
+    private int numRetries;
+
+    /** Information about the TaskManagers for the scale subresource. */
+    private TaskManagerInfo taskManager;
+}

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/SpecUtils.java

@@ -99,6 +99,28 @@ public static String writeSpecWithMeta(
         }
     }
 
+    public static String serializeObject(Object object, String wrapperKey) {
+        ObjectNode wrapper = objectMapper.createObjectNode();
+        wrapper.set(wrapperKey, objectMapper.valueToTree(checkNotNull(object)));
+
+        try {
+            return objectMapper.writeValueAsString(wrapper);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(
+                    "Could not serialize " + wrapperKey + ", this indicates a bug...", e);
+        }
+    }
+
+    public static <T> T deserializeObject(String serialized, String wrapperKey, Class<T> valueType)
+            throws JsonProcessingException {
+        try {
+            ObjectNode wrapper = (ObjectNode) objectMapper.readTree(serialized);
+            return objectMapper.treeToValue(wrapper.get(wrapperKey), valueType);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException("Could not deserialize spec, this indicates a bug...", e);
+        }
+    }
+
     // We do not have access to  Flink's Preconditions from here
     private static <T> T checkNotNull(T object) {
         if (object == null) {

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java

@@ -30,6 +30,7 @@
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
 import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeploymentController;
 import org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController;
 import org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController;
 import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotController;
@@ -242,6 +243,12 @@ void registerSnapshotController() {
         registeredControllers.add(operator.register(controller, this::overrideControllerConfigs));
     }
 
+    @VisibleForTesting
+    void registerBlueGreenController() {
+        var controller = new FlinkBlueGreenDeploymentController(ctxFactory);
+        registeredControllers.add(operator.register(controller, this::overrideControllerConfigs));
+    }
+
     private void overrideControllerConfigs(ControllerConfigurationOverrider<?> overrider) {
         var operatorConf = configManager.getOperatorConfiguration();
         var watchNamespaces = operatorConf.getWatchedNamespaces();
@@ -262,6 +269,7 @@ public void run() {
         registerDeploymentController();
         registerSessionJobController();
         registerSnapshotController();
+        registerBlueGreenController();
         operator.installShutdownHook(
                 baseConfig.get(KubernetesOperatorConfigOptions.OPERATOR_TERMINATION_TIMEOUT));
         operator.start();