FLIP-504: Blue/Green Deployments for Flink on Kubernetes - Phase 2 (WIP)

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

@@ -0,0 +1,66 @@
+/*
+ * 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 lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Map;
+
+import static org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions.ACTIVE_DEPLOYMENT_TYPE;
+import static org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions.DEPLOYMENT_DELETION_DELAY;
+import static org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions.IS_FIRST_DEPLOYMENT;
+import static org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions.TRANSITION_STAGE;
+
+/** Base functionality of the Context used for Gate implementations. */
+@Data
+@AllArgsConstructor
+public class GateContext implements Serializable {
+
+    /** GateContext enum. */
+    private final BlueGreenDeploymentType activeBlueGreenDeploymentType;
+
+    private final GateOutputMode outputMode;
+    private final int deploymentTeardownDelayMs;
+    private final TransitionStage gateStage;
+    private final boolean isFirstDeployment;
+
+    public static GateContext create(
+            Map<String, String> data, BlueGreenDeploymentType currentBlueGreenDeploymentType) {
+        var nextActiveDeploymentType =
+                BlueGreenDeploymentType.valueOf(data.get(ACTIVE_DEPLOYMENT_TYPE.getLabel()));
+
+        var deploymentDeletionDelaySec =
+                Integer.parseInt(data.get(DEPLOYMENT_DELETION_DELAY.getLabel()));
+
+        var outputMode =
+                currentBlueGreenDeploymentType == nextActiveDeploymentType
+                        ? GateOutputMode.ACTIVE
+                        : GateOutputMode.STANDBY;
+
+        var isFirstDeployment = Boolean.parseBoolean(data.get(IS_FIRST_DEPLOYMENT.getLabel()));
+
+        return new GateContext(
+                nextActiveDeploymentType,
+                outputMode,
+                deploymentDeletionDelaySec,
+                TransitionStage.valueOf(data.get(TRANSITION_STAGE.getLabel())),
+                isFirstDeployment);
+    }
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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 lombok.Getter;
+
+/** Options values for the GateContext. */
+public enum GateContextOptions {
+    IS_FIRST_DEPLOYMENT("is-first-deployment"),
+    ACTIVE_DEPLOYMENT_TYPE("active-deployment-type"),
+    DEPLOYMENT_DELETION_DELAY("deployment-deletion-delay-ms"),
+    TRANSITION_STAGE("stage");
+
+    @Getter private final String label;
+
+    private GateContextOptions(String label) {
+        this.label = label;
+    }
+}

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

@@ -0,0 +1,91 @@
+/*
+ * 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.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
+import lombok.Getter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Map;
+
+/** Simple Kubernetes service proxy for Gate operations. */
+public class GateKubernetesService implements Serializable {
+
+    private static final Logger logger = LoggerFactory.getLogger(GateKubernetesService.class);
+
+    @Getter private final KubernetesClient kubernetesClient;
+
+    private final String namespace;
+    private final String configMapName;
+
+    public GateKubernetesService(String namespace, String configMapName) {
+        Preconditions.checkNotNull(namespace);
+        Preconditions.checkNotNull(configMapName);
+
+        try {
+            kubernetesClient = new KubernetesClientBuilder().build();
+        } catch (Exception e) {
+            logger.error("Error instantiating Kubernetes Client", e);
+            throw e;
+        }
+
+        this.namespace = namespace;
+        this.configMapName = configMapName;
+    }
+
+    public void setInformers(ResourceEventHandler<ConfigMap> resourceEventHandler) {
+        kubernetesClient
+                .configMaps()
+                .inNamespace(namespace)
+                .withName(configMapName)
+                .inform(resourceEventHandler, 0);
+    }
+
+    public void updateConfigMapEntries(Map<String, String> kvps) {
+        var configMap = parseConfigMap();
+
+        kvps.forEach((key, value) -> configMap.getData().put(key, value));
+
+        try {
+            kubernetesClient.configMaps().inNamespace(namespace).resource(configMap).update();
+        } catch (Exception e) {
+            logger.error("Failed to UPDATE the ConfigMap", e);
+            throw e;
+        }
+    }
+
+    public ConfigMap parseConfigMap() {
+        try {
+            return kubernetesClient
+                    .configMaps()
+                    .inNamespace(namespace)
+                    .withName(configMapName)
+                    .get();
+        } catch (Exception e) {
+            logger.error("Failed to GET the ConfigMap", e);
+            throw e;
+        }
+    }
+}

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

@@ -0,0 +1,24 @@
+/*
+ * 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;
+
+/** Gate output enum values. */
+public enum GateOutputMode {
+    ACTIVE,
+    STANDBY
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+/** Possible transition modes supported by the `FlinkBlueGreenDeploymentController`. */
+public enum TransitionMode {
+    /**
+     * FLIP-503: simple transition that deletes the previous deployment as soon as the new one is
+     * RUNNING/STABLE.
+     */
+    BASIC,
+
+    /**
+     * FLIP-504: advanced coordination between deployment stages during transition. Not supported
+     * until FLIP-504 is implemented.
+     */
+    ADVANCED;
+}

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

@@ -0,0 +1,27 @@
+/*
+ * 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;
+
+/** Enumeration of the various stages for _ALL_ Blue/Green deployments. */
+public enum TransitionStage {
+    CLEAR_TO_TEARDOWN,
+    FAILING,
+    INITIALIZING,
+    RUNNING,
+    TRANSITIONING;
+}

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

@@ -17,6 +17,8 @@
 
 package org.apache.flink.kubernetes.operator.api.spec;
 
+import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionMode;
+
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
@@ -38,6 +40,9 @@ public class FlinkDeploymentTemplateSpec {
     @JsonProperty("metadata")
     private ObjectMeta metadata;
 
+    @JsonProperty("transitionMode")
+    private TransitionMode transitionMode;
+
     @JsonProperty("spec")
     private FlinkDeploymentSpec spec;
 

flink-kubernetes-operator-bluegreen-client/README.MD

@@ -0,0 +1,22 @@
+<!--
+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.
+-->
+
+# Flink Kubernetes Client Code Client
+
+TBD