[FLINK-29251] Send CREATED status and Cancel event via FlinkResourceListener

morhidi · tweise · commit 84b08d00f675 · 2022-09-14T07:29:09.000-05:00
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
@@ -84,7 +84,14 @@ public FlinkDeploymentController(
 
     @Override
     public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
-        LOG.info("Deleting FlinkDeployment");
+        String msg = "Cleaning up " + FlinkDeployment.class.getSimpleName();
+        LOG.info(msg);
+        eventRecorder.triggerEvent(
+                flinkApp,
+                EventRecorder.Type.Normal,
+                EventRecorder.Reason.Cleanup,
+                EventRecorder.Component.Operator,
+                msg);
         statusRecorder.updateStatusFromCache(flinkApp);
         try {
             observerFactory.getOrCreate(flinkApp).observe(flinkApp, context);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
@@ -107,7 +107,14 @@ public UpdateControl<FlinkSessionJob> reconcile(
 
     @Override
     public DeleteControl cleanup(FlinkSessionJob sessionJob, Context context) {
-        LOG.info("Deleting FlinkSessionJob");
+        String msg = "Cleaning up " + FlinkSessionJob.class.getSimpleName();
+        LOG.info(msg);
+        eventRecorder.triggerEvent(
+                sessionJob,
+                EventRecorder.Type.Normal,
+                EventRecorder.Reason.Cleanup,
+                EventRecorder.Component.Operator,
+                msg);
         statusRecorder.removeCachedStatus(sessionJob);
         return reconciler.cleanup(sessionJob, context);
     }
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
@@ -147,7 +147,7 @@ private void updateJobStatus(
             eventRecorder.triggerEvent(
                     resource,
                     EventRecorder.Type.Normal,
-                    EventRecorder.Reason.StatusChanged,
+                    EventRecorder.Reason.JobStatusChanged,
                     EventRecorder.Component.Job,
                     message);
         }
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -187,7 +187,7 @@ public DeleteControl cleanupInternal(FlinkDeployment deployment, Context<?> cont
             if (eventRecorder.triggerEvent(
                     deployment,
                     EventRecorder.Type.Warning,
-                    EventRecorder.Reason.Cleanup,
+                    EventRecorder.Reason.CleanupFailed,
                     EventRecorder.Component.Operator,
                     error)) {
                 LOG.warn(error);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
@@ -121,9 +121,10 @@ public enum Reason {
         SpecChanged,
         Rollback,
         Submit,
-        StatusChanged,
+        JobStatusChanged,
         SavepointError,
         Cleanup,
+        CleanupFailed,
         Missing,
         ValidationError
     }
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
@@ -27,6 +27,7 @@
 import org.apache.flink.kubernetes.operator.listener.AuditUtils;
 import org.apache.flink.kubernetes.operator.listener.FlinkResourceListener;
 import org.apache.flink.kubernetes.operator.metrics.MetricManager;
+import org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -140,6 +141,9 @@ public void updateStatusFromCache(CR resource) {
         } else {
             // Initialize cache with current status copy
             statusCache.put(key, objectMapper.convertValue(resource.getStatus(), ObjectNode.class));
+            if (ResourceLifecycleState.CREATED.equals(resource.getStatus().getLifecycleState())) {
+                statusUpdateListener.accept(resource, resource.getStatus());
+            }
         }
         metricManager.onUpdate(resource);
     }
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -117,7 +117,7 @@ public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) throws Exception
         assertEquals(
                 org.apache.flink.api.common.JobStatus.RECONCILING.name(),
                 appCluster.getStatus().getJobStatus().getState());
-        assertEquals(2, testController.getInternalStatusUpdateCount());
+        assertEquals(3, testController.getInternalStatusUpdateCount());
         assertFalse(updateControl.isUpdateStatus());
         assertEquals(
                 Optional.of(
@@ -140,7 +140,7 @@ public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) throws Exception
         assertEquals(
                 org.apache.flink.api.common.JobStatus.RECONCILING.name(),
                 appCluster.getStatus().getJobStatus().getState());
-        assertEquals(3, testController.getInternalStatusUpdateCount());
+        assertEquals(4, testController.getInternalStatusUpdateCount());
         assertFalse(updateControl.isUpdateStatus());
         assertEquals(
                 Optional.of(
@@ -154,7 +154,7 @@ public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) throws Exception
         assertEquals(
                 org.apache.flink.api.common.JobStatus.RUNNING.name(),
                 appCluster.getStatus().getJobStatus().getState());
-        assertEquals(4, testController.getInternalStatusUpdateCount());
+        assertEquals(5, testController.getInternalStatusUpdateCount());
         assertFalse(updateControl.isUpdateStatus());
         assertEquals(
                 Optional.of(
@@ -169,7 +169,7 @@ public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) throws Exception
         assertEquals(
                 org.apache.flink.api.common.JobStatus.RUNNING.name(),
                 appCluster.getStatus().getJobStatus().getState());
-        assertEquals(4, testController.getInternalStatusUpdateCount());
+        assertEquals(5, testController.getInternalStatusUpdateCount());
         assertFalse(updateControl.isUpdateStatus());
         assertEquals(
                 Optional.of(
@@ -195,7 +195,7 @@ public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) throws Exception
         assertEquals(
                 org.apache.flink.api.common.JobStatus.RUNNING.name(),
                 appCluster.getStatus().getJobStatus().getState());
-        assertEquals(5, testController.getInternalStatusUpdateCount());
+        assertEquals(6, testController.getInternalStatusUpdateCount());
         assertFalse(updateControl.isUpdateStatus());
 
         reconciliationStatus = appCluster.getStatus().getReconciliationStatus();
@@ -482,7 +482,7 @@ public void verifyStatelessUpgrade(FlinkVersion flinkVersion) throws Exception {
 
         assertEquals(1, testController.events().size());
         assertEquals(
-                EventRecorder.Reason.StatusChanged,
+                EventRecorder.Reason.JobStatusChanged,
                 EventRecorder.Reason.valueOf(testController.events().poll().getReason()));
 
         // Upgrade job
@@ -530,7 +530,7 @@ public void verifyStatelessUpgrade(FlinkVersion flinkVersion) throws Exception {
         testController.reconcile(appCluster, context);
         assertEquals(1, testController.events().size());
         assertEquals(
-                EventRecorder.Reason.StatusChanged,
+                EventRecorder.Reason.JobStatusChanged,
                 EventRecorder.Reason.valueOf(testController.events().poll().getReason()));
 
         // Suspend job
@@ -592,7 +592,7 @@ public void verifyStatelessUpgrade(FlinkVersion flinkVersion) throws Exception {
                         .collect(Collectors.toList());
         assertEquals(1, statusEvents.size());
         assertEquals(
-                EventRecorder.Reason.StatusChanged,
+                EventRecorder.Reason.JobStatusChanged,
                 EventRecorder.Reason.valueOf(statusEvents.get(0).getReason()));
 
         assertEquals(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
@@ -58,36 +58,40 @@ public void testListeners() {
         var eventRecorder = EventRecorder.create(kubernetesClient, listeners);
 
         var deployment = TestUtils.buildApplicationCluster();
-        statusRecorder.updateStatusFromCache(deployment);
 
-        statusRecorder.patchAndCacheStatus(deployment);
         assertTrue(listener1.updates.isEmpty());
         assertTrue(listener2.updates.isEmpty());
         assertTrue(listener1.events.isEmpty());
         assertTrue(listener2.events.isEmpty());
 
-        deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
-
-        statusRecorder.patchAndCacheStatus(deployment);
+        statusRecorder.updateStatusFromCache(deployment);
+        assertEquals(1, listener1.updates.size());
+        statusRecorder.updateStatusFromCache(deployment);
         assertEquals(1, listener1.updates.size());
         assertEquals(deployment, listener1.updates.get(0).getFlinkResource());
 
-        assertEquals(1, listener2.updates.size());
-        assertEquals(deployment, listener2.updates.get(0).getFlinkResource());
-        assertEquals(
-                listener1.updates.get(0).getTimestamp(), listener2.updates.get(0).getTimestamp());
+        deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
+        statusRecorder.patchAndCacheStatus(deployment);
+        assertEquals(2, listener1.updates.size());
+        assertEquals(deployment, listener1.updates.get(1).getFlinkResource());
 
         deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
         statusRecorder.patchAndCacheStatus(deployment);
+        assertEquals(3, listener1.updates.size());
+        assertEquals(deployment, listener1.updates.get(2).getFlinkResource());
 
-        assertEquals(2, listener1.updates.size());
-        assertEquals(deployment, listener1.updates.get(0).getFlinkResource());
-        assertEquals(2, listener2.updates.size());
-        assertEquals(deployment, listener2.updates.get(0).getFlinkResource());
+        for (int i = 0; i < listener1.updates.size(); i++) {
+            assertEquals(
+                    listener1.updates.get(i).getTimestamp(),
+                    listener2.updates.get(i).getTimestamp());
+            assertEquals(
+                    listener1.updates.get(i).getFlinkResource(),
+                    listener2.updates.get(i).getFlinkResource());
+        }
 
         var updateContext =
                 (FlinkResourceListener.StatusUpdateContext<FlinkDeployment, FlinkDeploymentStatus>)
-                        listener1.updates.get(1);
+                        listener1.updates.get(2);
         assertEquals(
                 JobManagerDeploymentStatus.ERROR,
                 updateContext.getPreviousStatus().getJobManagerDeploymentStatus());

Original file line number	Diff line number	Diff line change
`@@ -147,7 +147,7 @@ private void updateJobStatus(`
`147`	`147`	`eventRecorder.triggerEvent(`
`148`	`148`	`resource,`
`149`	`149`	`EventRecorder.Type.Normal,`
`150`		`- EventRecorder.Reason.StatusChanged,`
	`150`	`+ EventRecorder.Reason.JobStatusChanged,`
`151`	`151`	`EventRecorder.Component.Job,`
`152`	`152`	`message);`
`153`	`153`	`}`