[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,39 @@
+/*
+ * 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. Only one of each type will be
+ * present at all times for a particular job.
+ */
+public enum DeploymentType {
+    /** Identifier for the first or "Blue" deployment type. */
+    BLUE,
+
+    /** Identifier for the second or "Green" deployment type. */
+    GREEN;
+
+    public static final String LABEL_KEY = "flink/blue-green-deployment-type";
+
+    public static DeploymentType fromDeployment(FlinkDeployment flinkDeployment) {
+        String typeAnnotation = flinkDeployment.getMetadata().getLabels().get(LABEL_KEY);
+        return DeploymentType.valueOf(typeAnnotation);
+    }
+}

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

@@ -0,0 +1,54 @@
+/*
+ * 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.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/** Configuration options to be used by the Flink Blue/Green Deployments. */
+public class FlinkBlueGreenDeploymentConfigOptions {
+
+    public static final String BLUE_GREEN_CONF_PREFIX = "bluegreen.";
+
+    public static final int MIN_ABORT_GRACE_PERIOD_MS = 120000; // 2 mins
+
+    public static ConfigOptions.OptionBuilder operatorConfig(String key) {
+        return ConfigOptions.key(BLUE_GREEN_CONF_PREFIX + key);
+    }
+
+    public static final ConfigOption<Integer> ABORT_GRACE_PERIOD_MS =
+            operatorConfig("abortGracePeriodMs")
+                    .intType()
+                    .defaultValue(0)
+                    .withDescription(
+                            "The max time to wait for a deployment to become ready before aborting it, in milliseconds. Cannot be smaller than 2 minutes.");
+
+    public static final ConfigOption<Integer> RECONCILIATION_RESCHEDULING_INTERVAL_MS =
+            operatorConfig("reconciliationReschedulingIntervalMs")
+                    .intType()
+                    .defaultValue(15000) // 15 seconds
+                    .withDescription(
+                            "Configurable delay in milliseconds to use when the operator reschedules a reconciliation.");
+
+    public static final ConfigOption<Integer> DEPLOYMENT_DELETION_DELAY_MS =
+            operatorConfig("deploymentDeletionDelayMs")
+                    .intType()
+                    .defaultValue(0)
+                    .withDescription(
+                            "Configurable delay before deleting a deployment after being marked done.");
+}

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,49 @@
+/*
+ * 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("configuration")
+    private Map<String, String> configuration;
+
+    @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,40 @@
+/*
+ * 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 {
+
+    /**
+     * We use this state while initializing for the first time, always with a "Blue" deployment
+     * type.
+     */
+    INITIALIZING_BLUE,
+
+    /** Identifies the system is running normally with a "Blue" deployment type. */
+    ACTIVE_BLUE,
+
+    /** Identifies the system is running normally with a "Green" deployment type. */
+    ACTIVE_GREEN,
+
+    /** Identifies the system is transitioning from "Green" to "Blue". */
+    TRANSITIONING_TO_BLUE,
+
+    /** Identifies the system is transitioning from "Blue" to "Green". */
+    TRANSITIONING_TO_GREEN,
+}

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

@@ -0,0 +1,58 @@
+/*
+ * 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 String lastReconciledTimestamp;
+
+    /** Computed from abortGracePeriodMs, timestamp after which the deployment should be aborted. */
+    private String abortTimestamp;
+
+    /** Timestamp when the deployment became READY/STABLE. Used to determine when to delete it. */
+    private String deploymentReadyTimestamp;
+
+    /** 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,27 @@ public static String writeSpecWithMeta(
         }
     }
 
+    public static String writeSpecAsJSON(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 readSpecFromJSON(String serialized, String wrapperKey, Class<T> valueType) {
+        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();