diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md index 98d18e7d65..f67743cac8 100644 --- a/docs/content/docs/custom-resource/reference.md +++ b/docs/content/docs/custom-resource/reference.md @@ -361,6 +361,8 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r | ----------| ---- | ---- | | jobStatus | org.apache.flink.kubernetes.operator.api.status.JobStatus | | | blueGreenState | org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState | The state of the blue/green transition. | +| observedGeneration | java.lang.Long | Last observed generation of the FlinkBlueGreenDeployment. | +| lastStableGeneration | java.lang.Long | Last generation that reached a stable terminal state. | | lastReconciledSpec | java.lang.String | Last reconciled (serialized) deployment spec. | | lastReconciledTimestamp | java.lang.String | Timestamp of last reconciliation. | | abortTimestamp | java.lang.String | Computed from abortGracePeriodMs, timestamp after which the deployment should be aborted. | diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java index 104f195efa..4c86368f5b 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java @@ -41,6 +41,12 @@ public class FlinkBlueGreenDeploymentStatus { /** The state of the blue/green transition. */ private FlinkBlueGreenDeploymentState blueGreenState; + /** Last observed generation of the FlinkBlueGreenDeployment. */ + private Long observedGeneration; + + /** Last generation that reached a stable terminal state. */ + private Long lastStableGeneration; + /** Last reconciled (serialized) deployment spec. */ private String lastReconciledSpec; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java index a22574470d..1c0c18b20d 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java @@ -17,6 +17,7 @@ package org.apache.flink.kubernetes.operator.controller; +import org.apache.flink.api.common.JobStatus; import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment; import org.apache.flink.kubernetes.operator.api.FlinkDeployment; import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState; @@ -45,6 +46,8 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.ACTIVE_BLUE; +import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.ACTIVE_GREEN; import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.INITIALIZING_BLUE; /** @@ -151,11 +154,61 @@ public UpdateControl reconcile( BlueGreenStateHandler handler = handlerRegistry.getHandler(currentState); UpdateControl updateControl = handler.handle(context); + updateControl = + syncGenerationStatusIfNeeded( + bgDeployment, context, currentState, deploymentStatus, updateControl); + statusRecorder.patchAndCacheStatus(bgDeployment, josdkContext.getClient()); return updateControl; } } + private static UpdateControl syncGenerationStatusIfNeeded( + FlinkBlueGreenDeployment bgDeployment, + BlueGreenContext context, + FlinkBlueGreenDeploymentState currentState, + FlinkBlueGreenDeploymentStatus deploymentStatus, + UpdateControl updateControl) { + + // Migration/no-op path: kapp may reconcile an existing stable CR without rollout work when + // the rendered BlueGreen spec is unchanged, or when only non-spec resources/metadata + // changed. Handlers return noUpdate() in those cases, so sync generation status here. + if (updateControl.isNoUpdate()) { + var needsSync = !BlueGreenDeploymentService.isGenerationObserved(context); + + if (!BlueGreenDeploymentService.isGenerationStable(context) + && isStableBlueGreenJobStatus(currentState, deploymentStatus)) { + deploymentStatus.setLastStableGeneration( + bgDeployment.getMetadata().getGeneration()); + needsSync = true; + } + + if (needsSync) { + var statusUpdateControl = + BlueGreenDeploymentService.patchStatusUpdateControl( + context, null, null, null); + updateControl.getScheduleDelay().ifPresent(statusUpdateControl::rescheduleAfter); + return statusUpdateControl; + } + } + + return updateControl; + } + + private static boolean isStableBlueGreenJobStatus( + FlinkBlueGreenDeploymentState currentState, + FlinkBlueGreenDeploymentStatus deploymentStatus) { + // Suspension is represented as ACTIVE_* with parent jobStatus SUSPENDED, not as a separate + // BlueGreen state. Transitional or failing states must continue waiting for finalization. + if (currentState != ACTIVE_BLUE && currentState != ACTIVE_GREEN) { + return false; + } + + var jobStatus = deploymentStatus.getJobStatus(); + var jobState = jobStatus == null ? null : jobStatus.getState(); + return jobState == JobStatus.RUNNING || jobState == JobStatus.SUSPENDED; + } + public static void logAndThrow(String message) { throw new RuntimeException(message); } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java index 81bbe0436c..1dca42e078 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java @@ -859,6 +859,18 @@ private static void resetTransitionMarkers(FlinkBlueGreenDeploymentStatus status status.setSavepointTriggerId(null); } + public static boolean isGenerationObserved(BlueGreenContext context) { + return Objects.equals( + context.getDeploymentStatus().getObservedGeneration(), + context.getBgDeployment().getMetadata().getGeneration()); + } + + public static boolean isGenerationStable(BlueGreenContext context) { + return Objects.equals( + context.getDeploymentStatus().getLastStableGeneration(), + context.getBgDeployment().getMetadata().getGeneration()); + } + public static UpdateControl patchStatusUpdateControl( BlueGreenContext context, FlinkBlueGreenDeploymentState deploymentState, @@ -884,6 +896,13 @@ public static UpdateControl patchStatusUpdateControl( deploymentStatus.setError(null); } + if (jobState == JobStatus.RUNNING || jobState == JobStatus.SUSPENDED) { + deploymentStatus.setLastStableGeneration( + flinkBlueGreenDeployment.getMetadata().getGeneration()); + } + + deploymentStatus.setObservedGeneration( + flinkBlueGreenDeployment.getMetadata().getGeneration()); deploymentStatus.setLastReconciledTimestamp(java.time.Instant.now().toString()); flinkBlueGreenDeployment.setStatus(deploymentStatus); return UpdateControl.patchStatus(flinkBlueGreenDeployment); diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java index a1a64aa488..70bc34282c 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java @@ -52,6 +52,7 @@ import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import org.jetbrains.annotations.NotNull; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -128,6 +129,118 @@ public void verifyBasicDeployment(FlinkVersion flinkVersion, String initialSavep executeBasicDeployment(flinkVersion, blueGreenDeployment, true, initialSavepointPath); } + @Test + public void verifyObservedGenerationUpdatedWhenNewSpecObserved() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.STATELESS); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + Long previousGeneration = rs.deployment.getMetadata().getGeneration(); + assertNotNull(previousGeneration); + assertEquals(previousGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + + simulateSpecChange(rs.deployment, UUID.randomUUID().toString()); + Long newGeneration = previousGeneration + 1; + rs.deployment.getMetadata().setGeneration(newGeneration); + kubernetesClient.resource(rs.deployment).createOrReplace(); + + rs = reconcile(rs.deployment); + + assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState()); + assertEquals(JobStatus.RECONCILING, rs.reconciledStatus.getJobStatus().getState()); + } + + @Test + public void verifyGenerationStatusSyncedForExistingStableStatus() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.STATELESS); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + rs.deployment.getStatus().setObservedGeneration(null); + rs.deployment.getStatus().setLastStableGeneration(null); + + rs = reconcile(rs.deployment); + + assertTrue(rs.updateControl.isPatchStatus()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getObservedGeneration()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getLastStableGeneration()); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState()); + } + + @Test + public void verifyStableGenerationNotUpdatedWhenTransitionFailsBeforeStart() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.STATELESS); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + Long previousGeneration = rs.deployment.getMetadata().getGeneration(); + simulateSpecChange(rs.deployment, UUID.randomUUID().toString()); + Long newGeneration = previousGeneration + 1; + rs.deployment.getMetadata().setGeneration(newGeneration); + kubernetesClient.resource(rs.deployment).createOrReplace(); + simulateJobFailure(getFlinkDeployments().get(0)); + + rs = reconcile(rs.deployment); + + assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + assertEquals(JobStatus.FAILING, rs.reconciledStatus.getJobStatus().getState()); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + } + + @Test + public void verifyStableGenerationNotUpdatedDuringSavepointHandoff() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.SAVEPOINT); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + Long previousGeneration = rs.deployment.getMetadata().getGeneration(); + simulateSpecChange(rs.deployment, UUID.randomUUID().toString()); + Long newGeneration = previousGeneration + 1; + rs.deployment.getMetadata().setGeneration(newGeneration); + kubernetesClient.resource(rs.deployment).createOrReplace(); + + rs = handleSavepoint(rs); + + assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState()); + } + @ParameterizedTest @MethodSource("flinkVersionsAndSavepointPathsAndUpgradeModes") public void verifyBasicTransition( @@ -1181,6 +1294,12 @@ public void verifyInitialSuspendedIsBlockedThenDeploysOnRunning(FlinkVersion fli JobStatus.SUSPENDED, rs.reconciledStatus.getJobStatus().getState(), "Job status should be SUSPENDED when initial deployment is blocked"); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getObservedGeneration()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getLastStableGeneration()); // Flip to RUNNING and reconcile again bg = rs.deployment; @@ -1507,6 +1626,12 @@ private void assertFinalized( rs.reconciledStatus.getLastReconciledSpec()); assertEquals(expectedBGDeploymentState, rs.reconciledStatus.getBlueGreenState()); assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getObservedGeneration()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getLastStableGeneration()); assertEquals(0, instantStrToMillis(rs.reconciledStatus.getDeploymentReadyTimestamp())); assertEquals(0, instantStrToMillis(rs.reconciledStatus.getAbortTimestamp())); @@ -1796,6 +1921,7 @@ private static FlinkBlueGreenDeployment buildSessionCluster( .withCreationTimestamp(Instant.now().toString()) .withUid(UUID.randomUUID().toString()) .withResourceVersion("1") + .withGeneration(1L) .build()); var bgDeploymentSpec = getTestFlinkDeploymentSpec(version); diff --git a/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml index 0f8ef636fb..1ea315a672 100644 --- a/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml +++ b/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml @@ -10938,6 +10938,10 @@ spec: type: "string" lastReconciledTimestamp: type: "string" + lastStableGeneration: + type: "integer" + observedGeneration: + type: "integer" savepointTriggerId: type: "string" type: "object"