dpeacock commented on code in PR #1043: URL: https://github.com/apache/flink-kubernetes-operator/pull/1043#discussion_r4414353454
########## flink-kubernetes-operator-bluegreen-agent/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/agent/GateInjectionAgent.java: ########## @@ -0,0 +1,95 @@ +/* + * 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.bluegreen.agent; + +import net.bytebuddy.agent.builder.AgentBuilder; +import net.bytebuddy.implementation.MethodDelegation; + +import java.io.File; +import java.lang.instrument.Instrumentation; +import java.util.jar.JarFile; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +/** + * Java agent that transparently injects BlueGreen gate operators into every Flink Application-mode + * job, without requiring any user code changes. + * + * <p>Intercept point: {@code StreamExecutionEnvironment.execute(StreamGraph)}. This is a stable + * public API method present from Flink 1.14 through 2.x. It is called regardless of whether the + * user invokes {@code env.execute()}, {@code env.execute(String)}, or {@code + * env.execute(StreamGraph)} directly. + * + * <p>In production the agent JAR is injected into the JobManager pod by the BlueGreen controller + * via an init container — users do not need to modify their Dockerfile or set {@code + * env.java.opts.jobmanager} manually. + */ +public class GateInjectionAgent { + + public static void premain(String agentArgs, Instrumentation inst) throws Exception { + // The instrumented StreamExecutionEnvironment bytecode will reference + // GateInjectionInterceptor by name. Make the agent JAR visible to the + // system classloader so that reference resolves at runtime. + String jarPath = + GateInjectionAgent.class + .getProtectionDomain() + .getCodeSource() + .getLocation() + .getPath(); + inst.appendToSystemClassLoaderSearch(new JarFile(new File(jarPath))); + + System.err.println("[BlueGreen Agent] premain started, registering transformer"); + + new AgentBuilder.Default() + // Instrument both the base class and its Application-mode subclass. + // + // StreamContextEnvironment (flink-clients) overrides execute(StreamGraph) + // and is the concrete type used in Application mode and client-side execution. + // StreamExecutionEnvironment.execute(StreamGraph) is never reached via virtual + // dispatch when the runtime type is StreamContextEnvironment, so we must + // intercept the override directly. + // + // StreamExecutionEnvironment is kept as a fallback for local/test execution + // where LocalStreamEnvironment does NOT override execute(StreamGraph). + .type( + named( + "org.apache.flink.streaming.api.environment.StreamExecutionEnvironment") + .or( + named( + "org.apache.flink.client.program.StreamContextEnvironment"))) + .transform( + (builder, typeDescription, classLoader, module, protectionDomain) -> + builder.method( + named("execute") Review Comment: This misses `executeAsync`. With the same application and injected agent on Minikube, the execute job’s REST plan contained the gate, but the executeAsync plan did not. Should we cover the async submission path and add a graph-level regression test? ########## flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/WatermarkGateProcessFunction.java: ########## @@ -0,0 +1,357 @@ +/* + * 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.bluegreen.client; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.functions.OpenContext; +import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateMetrics; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateOutputMode; +import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionStage; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.streaming.api.TimerService; +import org.apache.flink.streaming.api.functions.ProcessFunction; +import org.apache.flink.util.Collector; +import org.apache.flink.util.Preconditions; + +import java.io.Serializable; +import java.util.Map; +import java.util.Objects; +import java.util.function.Function; + +/** Watermark based GateProcessFunction (streaming). */ +public class WatermarkGateProcessFunction<I> extends GateProcessFunction<I> + implements Serializable { + + // A standby subtask without a record for this long since it learned the toggle owes nothing to + // the hand-over: there is nothing it could still emit before its watermark passes the toggle. + @VisibleForTesting static final long IDLE_DONE_MS = 30_000L; + + private final Function<I, Long> watermarkExtractor; + + private WatermarkGateContext currentWatermarkGateContext; + + // Hand-over progress, read by the HAND_OVER_DONE gauge from the metric reporter's thread + private transient volatile boolean pastToggle; + private transient volatile long toggleKnownSince = -1L; + private transient volatile long lastRecordAt = -1L; + + // What this subtask does with its records, and for which toggle, as last logged + private transient Decision decision; + private transient Long decisionToggle; + private transient boolean noWatermarkLogged; + + WatermarkGateProcessFunction( + BlueGreenDeploymentType blueGreenDeploymentType, + String namespace, + String configMapName, + Function<I, Long> watermarkExtractor) { + super(blueGreenDeploymentType, namespace, configMapName); + + Preconditions.checkNotNull(watermarkExtractor); + + this.watermarkExtractor = watermarkExtractor; + } + + public static <I> WatermarkGateProcessFunction<I> create( + Map<String, String> flinkConfig, Function<I, Long> watermarkExtractor) { + requireConfigKey(flinkConfig, "bluegreen.active-deployment-type"); + requireConfigKey(flinkConfig, "kubernetes.namespace"); + requireConfigKey(flinkConfig, "bluegreen.configmap.name"); + return new WatermarkGateProcessFunction<I>( + BlueGreenDeploymentType.valueOf( + flinkConfig.get("bluegreen.active-deployment-type")), + flinkConfig.get("kubernetes.namespace"), + flinkConfig.get("bluegreen.configmap.name"), + watermarkExtractor); + } + + private static void requireConfigKey(Map<String, String> config, String key) { + if (config.get(key) == null) { + throw new IllegalArgumentException( + "WatermarkGateProcessFunction requires config key '" + + key + + "' to be set. If using automatic injection this is set by the" + + " operator; if instantiating directly, provide it in the" + + " flinkConfig map."); + } + } + + @Override + public void open(OpenContext openContext) throws Exception { + // Deserialization does not run field initializers: set the progress before super.open() + // reads the ConfigMap for the first time + pastToggle = false; + toggleKnownSince = -1L; + lastRecordAt = -1L; + decision = null; + decisionToggle = null; + noWatermarkLogged = false; + super.open(openContext); + getRuntimeContext() + .getMetricGroup() + .gauge( + GateMetrics.HAND_OVER_DONE, + (Gauge<Integer>) () -> isHandOverDone(System.currentTimeMillis()) ? 1 : 0); + } + + /** + * Whether this subtask, as the standby, has nothing left to emit for the hand-over: its + * watermark has passed the toggle, or it has had no record for {@link #IDLE_DONE_MS} since it + * learned the toggle. A subtask that receives no records never sees its watermark pass the + * toggle, but it owes nothing either. + */ + @VisibleForTesting + boolean isHandOverDone(long now) { + if (pastToggle) { + return true; + } + long since = toggleKnownSince; + return since >= 0 && now - Math.max(since, lastRecordAt) >= IDLE_DONE_MS; + } + + @Override + protected void onContextUpdate(GateContext baseContext, Map<String, String> data) { + var fetchedWatermarkContext = WatermarkGateContext.create(baseContext, data); + logDebug("Refreshing WatermarkGateContext with data: " + data); + + if (currentWatermarkGateContext == null) { + logInfo("currentWatermarkGateContext INITIALIZED: " + fetchedWatermarkContext); + currentWatermarkGateContext = fetchedWatermarkContext; + } else if (!currentWatermarkGateContext.equals(fetchedWatermarkContext)) { + logInfo("currentWatermarkGateContext UPDATED: " + fetchedWatermarkContext); + currentWatermarkGateContext = fetchedWatermarkContext; + } + trackToggle(currentWatermarkGateContext.getWatermarkToggleValue()); + } + + /** A new transition starts without a toggle, which resets the hand-over progress. */ + private void trackToggle(Long toggle) { + if (toggle == null) { + pastToggle = false; + toggleKnownSince = -1L; + } else { + noWatermarkLogged = false; + if (toggleKnownSince < 0) { + toggleKnownSince = System.currentTimeMillis(); + } + } + } + + @Override + protected void processElementActive( + I value, ProcessFunction<I, I>.Context ctx, Collector<I> out) + throws IllegalAccessException { + Long wmToggleValue = currentWatermarkGateContext.getWatermarkToggleValue(); + if (wmToggleValue != null) { + if (isFullyOpen(ctx.timerService(), wmToggleValue)) { + decide( + currentWatermarkGateContext.getBaseContext().isFirstDeployment() + ? Decision.PASS_ALL + : Decision.PASS_ALL_PAST_TOGGLE, + wmToggleValue); + out.collect(value); + return; + } + decide(Decision.PASS_FROM_TOGGLE, wmToggleValue); + if (wmToggleValue <= timestampOf(value)) { + out.collect(value); + } + } else { + // Transitioning to Active + var currentGateStage = currentWatermarkGateContext.getBaseContext().getGateStage(); + if (currentGateStage == TransitionStage.TRANSITIONING) { + decide(Decision.HOLD_UNTIL_TOGGLE, null); + notifyWaitingForWatermark(ctx); + } else { + decide(Decision.HOLD_UNTIL_TRANSITION, null); + } + } + } + + @Override + protected void processElementStandby( + I value, ProcessFunction<I, I>.Context ctx, Collector<I> out) + throws IllegalAccessException { + lastRecordAt = System.currentTimeMillis(); + if (currentWatermarkGateContext.getWatermarkToggleValue() != null) { + var watermarkToggleValue = currentWatermarkGateContext.getWatermarkToggleValue(); + + if (ctx.timerService().currentWatermark() <= watermarkToggleValue) { + decide(Decision.PASS_BEFORE_TOGGLE, watermarkToggleValue); + if (watermarkToggleValue > timestampOf(value)) { + out.collect(value); + } + } else { + decide(Decision.HOLD_ALL_PAST_TOGGLE, watermarkToggleValue); + pastToggle = true; + notifyClearToTeardown(ctx); + } + } else { + // This ACTIVE job is transitioning to STANDBY, output elements + decide(Decision.PASS_ALL_UNTIL_TOGGLE, null); + out.collect(value); + // Set the watermark when the other new job is ready + updateWatermarkInConfigMap(ctx); + } + } + + /** + * The active gate passes every record when there is no standby deployment to share the stream + * with (a first deployment, or the deployment kept after an abort), or once this subtask's own + * watermark has passed the toggle. The standby's gate stops at that same point, so it never + * emits the records that arrive after it, including late ones older than the toggle. Only the + * record-time watermark counts: until the first one arrives, currentWatermark() is + * Long.MIN_VALUE and records are compared with the toggle one by one. + */ + private boolean isFullyOpen(TimerService timerService, long wmToggleValue) { + return currentWatermarkGateContext.getBaseContext().isFirstDeployment() + || timerService.currentWatermark() > wmToggleValue; + } + + /** + * The record's timestamp from the job's extractor. A record without one (null) counts as older + * than any toggle, like a null field-path field or field-index column: the standby emits it + * until it stops, the active once it is fully open. + */ + private long timestampOf(I value) { + Long timestamp = watermarkExtractor.apply(value); + return timestamp == null ? Long.MIN_VALUE : timestamp; + } + + /** + * The toggle the standby proposes: its own record-time watermark plus the deletion delay, or + * null while no watermark has reached this subtask. It is never taken from the wall clock. + */ + @VisibleForTesting + static Long nextWatermarkToggleValue(long currentWatermark, long deploymentTeardownDelayMs) { + return currentWatermark > 0 ? currentWatermark + deploymentTeardownDelayMs : null; Review Comment: The example notes that the delay must exceed its watermark bound, but ADVANCED accepts the default 0s. Should ADVANCED require explicit configuration and make this constraint part of validation/documentation? ########## flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenTransitionUtils.java: ########## @@ -0,0 +1,580 @@ +/* + * 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.utils.bluegreen; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.kubernetes.operator.api.FlinkDeployment; +import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateMetrics; +import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionMode; +import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionStage; +import org.apache.flink.kubernetes.operator.api.spec.ConfigObjectNode; +import org.apache.flink.kubernetes.operator.api.spec.JobManagerSpec; +import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode; +import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; +import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext; +import org.apache.flink.kubernetes.operator.utils.FlinkUtils; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.VolumeBuilder; +import io.fabric8.kubernetes.api.model.VolumeMount; +import io.fabric8.kubernetes.api.model.VolumeMountBuilder; +import lombok.SneakyThrows; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.naming.OperationNotSupportedException; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +import static org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions.ACTIVE_DEPLOYMENT_TYPE; +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; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.BLUEGREEN_DEPLOYMENT_DELETION_DELAY; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.BLUEGREEN_GATE_TIMEOUT; +import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.getConfigMap; +import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.updateConfigMapEntry; +import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.upsertConfigMap; +import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getDeploymentDeletionDelay; +import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getGateTimeout; + +/** Utility class for Blue/Green transition stage operations. */ +public class BlueGreenTransitionUtils { + + private static final Logger LOG = LoggerFactory.getLogger(BlueGreenTransitionUtils.class); + + private static final String GATE_INJECTION_ENABLED = "bluegreen.gate.injection.enabled"; + + // The WATERMARK gate's cutover point (WatermarkGateContext.WATERMARK_TOGGLE_VALUE) + private static final String WATERMARK_TOGGLE_VALUE = "watermark-toggle-value"; + + /** The WATERMARK gate's extraction strategies. A job sets exactly one. */ + @VisibleForTesting + static final List<String> WATERMARK_EXTRACTOR_KEYS = + List.of( + "bluegreen.gate.watermark.extractor-class", + "bluegreen.gate.watermark.field-index", + "bluegreen.gate.watermark.field-path"); + + /** + * Test override for the {@code OPERATOR_IMAGE} env lookup used by {@link + * #prepareTransitionMetadata}. {@code null} means read the environment (the production path). + */ + private static String operatorImageOverride = null; + + /** + * Overrides the {@code OPERATOR_IMAGE} env lookup for tests (including tests in other + * packages). Pass {@code null} to restore the production path of reading the environment + * variable. + */ + @VisibleForTesting + public static void setOperatorImageOverride(String operatorImage) { + operatorImageOverride = operatorImage; + } + + @SneakyThrows + public static TransitionMode getTransitionMode(BlueGreenContext context) { + TransitionMode transitionMode = context.getBgDeployment().getSpec().getTransitionMode(); + + if (transitionMode == null) { + throw new OperationNotSupportedException("Please specify the TransitionMode"); + } + + return transitionMode; + } + + public static void prepareTransitionMetadata( + BlueGreenContext context, + BlueGreenDeploymentType blueGreenDeploymentType, + FlinkDeployment flinkDeployment, + boolean isFirstDeployment) { + if (TransitionMode.ADVANCED != getTransitionMode(context)) { + return; + } + + var transitionDefaultMetadata = + Map.of( + IS_FIRST_DEPLOYMENT.getLabel(), isFirstDeployment ? "true" : "false", + GateContextOptions.DEPLOYMENT_DELETION_DELAY.getLabel(), + String.valueOf(getDeploymentDeletionDelay(context)), + ACTIVE_DEPLOYMENT_TYPE.getLabel(), blueGreenDeploymentType.toString(), + TRANSITION_STAGE.getLabel(), TransitionStage.INITIALIZING.toString()); + + upsertConfigMap(context, transitionDefaultMetadata); + + // Preparing the FlinkConfiguration for the OutputDecider + flinkDeployment + .getSpec() + .getFlinkConfiguration() + .put( + "bluegreen." + ACTIVE_DEPLOYMENT_TYPE.getLabel(), + blueGreenDeploymentType.toString()); + flinkDeployment + .getSpec() + .getFlinkConfiguration() + .put("bluegreen.configmap.name", context.getConfigMapName()); + + // Auto-inject agent config and init container when a gate strategy is declared + ConfigObjectNode flinkConfig = flinkDeployment.getSpec().getFlinkConfiguration(); + if (flinkConfig.has("bluegreen.gate.strategy")) { + String operatorImage = + operatorImageOverride != null + ? operatorImageOverride + : System.getenv("OPERATOR_IMAGE"); + injectGateAgent( + flinkDeployment, + flinkConfig, + operatorImage, + mergesPodTemplateArraysByName(context, flinkDeployment)); + } + } + + /** + * The {@code kubernetes.operator.pod-template.merge-arrays-by-name} the deployment is deployed + * with: its own flinkConfiguration over the operator's defaults. + */ + private static boolean mergesPodTemplateArraysByName( + BlueGreenContext context, FlinkDeployment flinkDeployment) { + var spec = flinkDeployment.getSpec(); + Configuration conf = + context.getCtxFactory() + .getConfigManager() + .getDefaultConfig( + flinkDeployment.getMetadata().getNamespace(), + spec.getFlinkVersion()); + conf.addAll(spec.getFlinkConfiguration().asConfiguration()); + return conf.get(KubernetesOperatorConfigOptions.POD_TEMPLATE_MERGE_BY_NAME); + } + + /** + * Moves an ADVANCED transition out of INITIALIZING once the new deployment is ready. + * + * @return {@code true} if the gate phase started, i.e. the stage moved to TRANSITIONING + */ + public static boolean moveToFirstTransitionStage(BlueGreenContext context) { + if (TransitionMode.ADVANCED != getTransitionMode(context)) { + return false; + } + + ConfigMap configMap = getConfigMap(context); + String stage = configMap.getData().get(TRANSITION_STAGE.getLabel()); + + if (stage.equals(TransitionStage.INITIALIZING.toString())) { + TransitionStage nextStage; + + if (context.getDeployments().getNumberOfDeployments() == 2) { + LOG.info("Stage INITIALIZING to TRANSITIONING"); + nextStage = TransitionStage.TRANSITIONING; + } else { + LOG.info( + "No transition between deployments detected, Stage INITIALIZING -> CLEAR_TO_TEARDOWN"); + nextStage = TransitionStage.CLEAR_TO_TEARDOWN; + } + + updateTransitionStage(context, nextStage); + return nextStage == TransitionStage.TRANSITIONING; + } + + return false; + } + + public static boolean isClearToTeardown(BlueGreenContext context) { + + if (TransitionMode.ADVANCED == getTransitionMode(context)) { + ConfigMap configMap = getConfigMap(context); + String stage = configMap.getData().get(TRANSITION_STAGE.getLabel()); + + if (!stage.equals(TransitionStage.CLEAR_TO_TEARDOWN.toString())) { + LOG.info("Waiting for CLEAR_TO_TEARDOWN, current stage: " + stage); + return false; + } + } + + return true; + } + + /** + * Hands the gate back to the deployment that keeps running after an aborted transition. The + * ConfigMap is rewritten whole, which drops the strategy entries (e.g. the watermark toggle) of + * the aborted hand-over, and it is marked as a first deployment: with no counterpart left to + * hand over to, the remaining gate must pass every record, as it does before any transition. + * + * <p>Call it after the reconciliation's other ConfigMap writes: those patch single entries, + * while this write replaces the whole ConfigMap. + */ + public static void resetGateOnAbort( + BlueGreenContext context, FlinkBlueGreenDeploymentState previousState) { + + if (TransitionMode.ADVANCED != getTransitionMode(context)) { + return; + } + + var previousDeploymentType = + previousState.name().contains("BLUE") + ? BlueGreenDeploymentType.BLUE + : BlueGreenDeploymentType.GREEN; + + upsertConfigMap( + context, + Map.of( + IS_FIRST_DEPLOYMENT.getLabel(), "true", + GateContextOptions.DEPLOYMENT_DELETION_DELAY.getLabel(), + String.valueOf(getDeploymentDeletionDelay(context)), + ACTIVE_DEPLOYMENT_TYPE.getLabel(), previousDeploymentType.toString(), + TRANSITION_STAGE.getLabel(), TransitionStage.FAILING.toString())); + } + + public static void updateTransitionStageFromJobStatus( + BlueGreenContext context, JobStatus jobStatus) { + + if (TransitionMode.ADVANCED != getTransitionMode(context)) { + return; + } + + // ConfigMap may not exist yet if validation failed before prepareTransitionMetadata ran + var secondaryConfigMaps = context.getJosdkContext().getSecondaryResources(ConfigMap.class); + boolean configMapPresent = + secondaryConfigMaps != null + && secondaryConfigMaps.stream() + .anyMatch( + cm -> + cm.getMetadata() + .getName() + .equals(context.getConfigMapName())); + if (!configMapPresent) { + return; + } + + TransitionStage transitionStage; + switch (jobStatus) { + case RUNNING: + transitionStage = TransitionStage.RUNNING; + break; + case FAILING: + transitionStage = TransitionStage.FAILING; + break; + case RECONCILING: + transitionStage = TransitionStage.INITIALIZING; + break; + default: + throw new RuntimeException("Unsupported JobStatus: " + jobStatus); Review Comment: Normal suspension reaches this branch with SUSPENDED. The supplied e2e passed, but the operator logged Unsupported JobStatus: SUSPENDED and left the parent reconciling after the child stopped. Should we handle that state and assert parent completion in the test? ########## flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/WatermarkGateProcessFunction.java: ########## @@ -0,0 +1,357 @@ +/* + * 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.bluegreen.client; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.functions.OpenContext; +import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateMetrics; +import org.apache.flink.kubernetes.operator.api.bluegreen.GateOutputMode; +import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionStage; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.streaming.api.TimerService; +import org.apache.flink.streaming.api.functions.ProcessFunction; +import org.apache.flink.util.Collector; +import org.apache.flink.util.Preconditions; + +import java.io.Serializable; +import java.util.Map; +import java.util.Objects; +import java.util.function.Function; + +/** Watermark based GateProcessFunction (streaming). */ +public class WatermarkGateProcessFunction<I> extends GateProcessFunction<I> + implements Serializable { + + // A standby subtask without a record for this long since it learned the toggle owes nothing to + // the hand-over: there is nothing it could still emit before its watermark passes the toggle. + @VisibleForTesting static final long IDLE_DONE_MS = 30_000L; + + private final Function<I, Long> watermarkExtractor; + + private WatermarkGateContext currentWatermarkGateContext; + + // Hand-over progress, read by the HAND_OVER_DONE gauge from the metric reporter's thread + private transient volatile boolean pastToggle; + private transient volatile long toggleKnownSince = -1L; + private transient volatile long lastRecordAt = -1L; + + // What this subtask does with its records, and for which toggle, as last logged + private transient Decision decision; + private transient Long decisionToggle; + private transient boolean noWatermarkLogged; + + WatermarkGateProcessFunction( + BlueGreenDeploymentType blueGreenDeploymentType, + String namespace, + String configMapName, + Function<I, Long> watermarkExtractor) { + super(blueGreenDeploymentType, namespace, configMapName); + + Preconditions.checkNotNull(watermarkExtractor); + + this.watermarkExtractor = watermarkExtractor; + } + + public static <I> WatermarkGateProcessFunction<I> create( + Map<String, String> flinkConfig, Function<I, Long> watermarkExtractor) { + requireConfigKey(flinkConfig, "bluegreen.active-deployment-type"); + requireConfigKey(flinkConfig, "kubernetes.namespace"); + requireConfigKey(flinkConfig, "bluegreen.configmap.name"); + return new WatermarkGateProcessFunction<I>( + BlueGreenDeploymentType.valueOf( + flinkConfig.get("bluegreen.active-deployment-type")), + flinkConfig.get("kubernetes.namespace"), + flinkConfig.get("bluegreen.configmap.name"), + watermarkExtractor); + } + + private static void requireConfigKey(Map<String, String> config, String key) { + if (config.get(key) == null) { + throw new IllegalArgumentException( + "WatermarkGateProcessFunction requires config key '" + + key + + "' to be set. If using automatic injection this is set by the" + + " operator; if instantiating directly, provide it in the" + + " flinkConfig map."); + } + } + + @Override + public void open(OpenContext openContext) throws Exception { + // Deserialization does not run field initializers: set the progress before super.open() + // reads the ConfigMap for the first time + pastToggle = false; + toggleKnownSince = -1L; + lastRecordAt = -1L; + decision = null; + decisionToggle = null; + noWatermarkLogged = false; + super.open(openContext); + getRuntimeContext() + .getMetricGroup() + .gauge( + GateMetrics.HAND_OVER_DONE, + (Gauge<Integer>) () -> isHandOverDone(System.currentTimeMillis()) ? 1 : 0); + } + + /** + * Whether this subtask, as the standby, has nothing left to emit for the hand-over: its + * watermark has passed the toggle, or it has had no record for {@link #IDLE_DONE_MS} since it + * learned the toggle. A subtask that receives no records never sees its watermark pass the + * toggle, but it owes nothing either. + */ + @VisibleForTesting + boolean isHandOverDone(long now) { + if (pastToggle) { + return true; + } + long since = toggleKnownSince; + return since >= 0 && now - Math.max(since, lastRecordAt) >= IDLE_DONE_MS; + } + + @Override + protected void onContextUpdate(GateContext baseContext, Map<String, String> data) { + var fetchedWatermarkContext = WatermarkGateContext.create(baseContext, data); + logDebug("Refreshing WatermarkGateContext with data: " + data); + + if (currentWatermarkGateContext == null) { + logInfo("currentWatermarkGateContext INITIALIZED: " + fetchedWatermarkContext); + currentWatermarkGateContext = fetchedWatermarkContext; + } else if (!currentWatermarkGateContext.equals(fetchedWatermarkContext)) { + logInfo("currentWatermarkGateContext UPDATED: " + fetchedWatermarkContext); + currentWatermarkGateContext = fetchedWatermarkContext; + } + trackToggle(currentWatermarkGateContext.getWatermarkToggleValue()); + } + + /** A new transition starts without a toggle, which resets the hand-over progress. */ + private void trackToggle(Long toggle) { + if (toggle == null) { + pastToggle = false; + toggleKnownSince = -1L; + } else { + noWatermarkLogged = false; + if (toggleKnownSince < 0) { + toggleKnownSince = System.currentTimeMillis(); + } + } + } + + @Override + protected void processElementActive( + I value, ProcessFunction<I, I>.Context ctx, Collector<I> out) + throws IllegalAccessException { + Long wmToggleValue = currentWatermarkGateContext.getWatermarkToggleValue(); + if (wmToggleValue != null) { + if (isFullyOpen(ctx.timerService(), wmToggleValue)) { + decide( + currentWatermarkGateContext.getBaseContext().isFirstDeployment() + ? Decision.PASS_ALL + : Decision.PASS_ALL_PAST_TOGGLE, + wmToggleValue); + out.collect(value); + return; + } + decide(Decision.PASS_FROM_TOGGLE, wmToggleValue); + if (wmToggleValue <= timestampOf(value)) { + out.collect(value); + } + } else { + // Transitioning to Active + var currentGateStage = currentWatermarkGateContext.getBaseContext().getGateStage(); + if (currentGateStage == TransitionStage.TRANSITIONING) { + decide(Decision.HOLD_UNTIL_TOGGLE, null); + notifyWaitingForWatermark(ctx); + } else { + decide(Decision.HOLD_UNTIL_TRANSITION, null); + } + } + } + + @Override + protected void processElementStandby( + I value, ProcessFunction<I, I>.Context ctx, Collector<I> out) + throws IllegalAccessException { + lastRecordAt = System.currentTimeMillis(); + if (currentWatermarkGateContext.getWatermarkToggleValue() != null) { + var watermarkToggleValue = currentWatermarkGateContext.getWatermarkToggleValue(); + + if (ctx.timerService().currentWatermark() <= watermarkToggleValue) { + decide(Decision.PASS_BEFORE_TOGGLE, watermarkToggleValue); + if (watermarkToggleValue > timestampOf(value)) { + out.collect(value); + } + } else { + decide(Decision.HOLD_ALL_PAST_TOGGLE, watermarkToggleValue); + pastToggle = true; + notifyClearToTeardown(ctx); + } + } else { + // This ACTIVE job is transitioning to STANDBY, output elements + decide(Decision.PASS_ALL_UNTIL_TOGGLE, null); + out.collect(value); + // Set the watermark when the other new job is ready + updateWatermarkInConfigMap(ctx); + } + } + + /** + * The active gate passes every record when there is no standby deployment to share the stream + * with (a first deployment, or the deployment kept after an abort), or once this subtask's own + * watermark has passed the toggle. The standby's gate stops at that same point, so it never + * emits the records that arrive after it, including late ones older than the toggle. Only the + * record-time watermark counts: until the first one arrives, currentWatermark() is + * Long.MIN_VALUE and records are compared with the toggle one by one. + */ + private boolean isFullyOpen(TimerService timerService, long wmToggleValue) { + return currentWatermarkGateContext.getBaseContext().isFirstDeployment() + || timerService.currentWatermark() > wmToggleValue; Review Comment: After successful handover, a TaskManager restart resets the watermark and closes this gate again. Should we checkpoint the fully-open state so recovery doesn’t discard late records after handover? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
