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

Dockerfile

@@ -25,7 +25,7 @@ WORKDIR /app
 
 COPY . .
 
-RUN --mount=type=cache,target=/root/.m2 mvn -ntp clean install -pl flink-kubernetes-standalone,flink-kubernetes-operator-api,flink-kubernetes-operator,flink-autoscaler,flink-kubernetes-webhook -DskipTests=$SKIP_TESTS -Dfabric8.httpclient.impl="$HTTP_CLIENT"
+RUN --mount=type=cache,target=/root/.m2 mvn -ntp clean install -pl flink-kubernetes-standalone,flink-kubernetes-operator-api,flink-kubernetes-operator,flink-autoscaler,flink-kubernetes-webhook,flink-kubernetes-operator-bluegreen-agent -DskipTests=$SKIP_TESTS -Dfabric8.httpclient.impl="$HTTP_CLIENT"
 
 RUN cd /app/tools/license; mkdir jars; cd jars; \
     cp /app/flink-kubernetes-operator/target/flink-kubernetes-operator-*-shaded.jar . && \
@@ -42,9 +42,12 @@ ENV OPERATOR_VERSION=1.15-SNAPSHOT
 ENV OPERATOR_JAR=flink-kubernetes-operator-$OPERATOR_VERSION-shaded.jar
 ENV WEBHOOK_JAR=flink-kubernetes-webhook-$OPERATOR_VERSION-shaded.jar
 ENV KUBERNETES_STANDALONE_JAR=flink-kubernetes-standalone-$OPERATOR_VERSION.jar
+ENV AGENT_JAR=flink-kubernetes-operator-bluegreen-agent-$OPERATOR_VERSION.jar
+ENV ARTIFACTS_DIR=$FLINK_HOME/artifacts
 
 ENV OPERATOR_LIB=$FLINK_HOME/operator-lib
 RUN mkdir -p $OPERATOR_LIB
+RUN mkdir -p $ARTIFACTS_DIR
 
 WORKDIR /flink-kubernetes-operator
 RUN groupadd --system --gid=9999 flink && \
@@ -60,6 +63,7 @@ COPY --chown=flink:flink --from=build /app/tools/license/licenses-output/NOTICE
 COPY --chown=flink:flink --from=build /app/tools/license/licenses-output/licenses ./licenses
 COPY --chown=flink:flink --from=build /app/LICENSE ./LICENSE
 COPY --chown=flink:flink docker-entrypoint.sh /
+COPY --chown=flink:flink --from=build /app/flink-kubernetes-operator-bluegreen-agent/target/$AGENT_JAR $ARTIFACTS_DIR/bluegreen-agent.jar
 
 ARG SKIP_OS_UPDATE=true
 

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/FlinkBlueGreenDeploymentSpec.java

@@ -18,6 +18,7 @@
 package org.apache.flink.kubernetes.operator.api.spec;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionMode;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -42,5 +43,7 @@ public class FlinkBlueGreenDeploymentSpec {
 
     @Nullable private IngressSpec ingress;
 
+    private TransitionMode transitionMode;
+
     private FlinkDeploymentTemplateSpec template;
 }