[FLINK-27609] Tracking flink-version and flink-revision in FlinkDeplo…

…ymentStatus
apache · May 16, 2022 · b403fcd · b403fcd
1 parent be54f2b
commit b403fcd
Show file tree

Hide file tree

Showing 7 changed files with 85 additions and 0 deletions.
diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md
@@ -185,6 +185,7 @@ This page serves as a full reference for FlinkDeployment custom resource definit
 | ----------| ---- | ---- |
 | jobStatus | org.apache.flink.kubernetes.operator.crd.status.JobStatus | Last observed status of the Flink job on Application/Session cluster. |
 | error | java.lang.String | Error information about the FlinkDeployment/FlinkSessionJob. |
+| clusterInfo | java.util.Map<java.lang.String,java.lang.String> | Config information from running clusters. |
 | jobManagerDeploymentStatus | org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus | Last observed status of the JobManager deployment. |
 | reconciliationStatus | org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentReconciliationStatus | Status of the last reconcile operation. |
 

diff --git a/.../src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java b/.../src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java
@@ -27,6 +27,9 @@
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /** Last observed status of the Flink deployment. */
 @Experimental
 @Data
@@ -37,6 +40,9 @@
 @SuperBuilder
 public class FlinkDeploymentStatus extends CommonStatus<FlinkDeploymentSpec> {
 
+    /** Config information from running clusters. */
+    private Map<String, String> clusterInfo = new HashMap<>();
+
     /** Last observed status of the JobManager deployment. */
     private JobManagerDeploymentStatus jobManagerDeploymentStatus =
             JobManagerDeploymentStatus.MISSING;

diff --git a/.../org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java b/.../org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -48,6 +48,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 /** The base observer. */
@@ -85,16 +86,29 @@ public void observe(FlinkDeployment flinkApp, Context context) {
         if (!isJmDeploymentReady(flinkApp)) {
             observeJmDeployment(flinkApp, context, observeConfig);
         }
+
         if (isJmDeploymentReady(flinkApp)) {
+            observeClusterInfo(flinkApp, observeConfig);
             if (observeFlinkCluster(flinkApp, context, observeConfig)) {
                 if (reconciliationStatus.getState() != ReconciliationState.ROLLED_BACK) {
                     reconciliationStatus.markReconciledSpecAsStable();
                 }
             }
         }
+
         clearErrorsIfDeploymentIsHealthy(flinkApp);
     }
 
+    private void observeClusterInfo(FlinkDeployment flinkApp, Configuration configuration) {
+        try {
+            Map<String, String> clusterInfo = flinkService.getClusterinfo(configuration);
+            flinkApp.getStatus().setClusterInfo(clusterInfo);
+            logger.debug("ClusterInfo: {}", clusterInfo);
+        } catch (Exception e) {
+            logger.warn("Exception while fetching cluster info", e);
+        }
+    }
+
     protected void observeJmDeployment(
             FlinkDeployment flinkApp, Context context, Configuration effectiveConfig) {
         FlinkDeploymentStatus deploymentStatus = flinkApp.getStatus();

diff --git a/...tes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/...tes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -56,6 +56,8 @@
 import org.apache.flink.runtime.rest.RestClient;
 import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
 import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.DashboardConfigurationHeaders;
 import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
 import org.apache.flink.runtime.rest.messages.TriggerId;
@@ -102,6 +104,8 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
@@ -643,6 +647,35 @@ public SavepointFetchResult fetchSavepointInfo(
         }
     }
 
+    public Map<String, String> getClusterinfo(Configuration conf) throws Exception {
+        Map<String, String> runtimeVersion = new HashMap<>();
+
+        try (RestClusterClient<String> clusterClient =
+                (RestClusterClient<String>) getClusterClient(conf)) {
+
+            DashboardConfiguration dashboardConfiguration =
+                    clusterClient
+                            .sendRequest(
+                                    DashboardConfigurationHeaders.getInstance(),
+                                    EmptyMessageParameters.getInstance(),
+                                    EmptyRequestBody.getInstance())
+                            .get(
+                                    configManager
+                                            .getOperatorConfiguration()
+                                            .getFlinkClientTimeout()
+                                            .toSeconds(),
+                                    TimeUnit.SECONDS);
+
+            runtimeVersion.put(
+                    DashboardConfiguration.FIELD_NAME_FLINK_VERSION,
+                    dashboardConfiguration.getFlinkVersion());
+            runtimeVersion.put(
+                    DashboardConfiguration.FIELD_NAME_FLINK_REVISION,
+                    dashboardConfiguration.getFlinkRevision());
+        }
+        return runtimeVersion;
+    }
+
     public PodList getJmPodList(FlinkDeployment deployment, Configuration conf) {
         final String namespace = conf.getString(KubernetesConfigOptions.NAMESPACE);
         final String clusterId;

diff --git a/...etes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/...etes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -40,6 +40,7 @@
 import org.apache.flink.runtime.client.JobStatusMessage;
 import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
 
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.PodList;
@@ -64,6 +65,12 @@
 /** Flink service mock for tests. */
 public class TestingFlinkService extends FlinkService {
 
+    public static final Map<String, String> CLUSTER_INFO =
+            Map.of(
+                    DashboardConfiguration.FIELD_NAME_FLINK_VERSION, "15.0.0",
+                    DashboardConfiguration.FIELD_NAME_FLINK_REVISION,
+                            "1234567 @ 1970-01-01T00:00:00+00:00");
+
     private int savepointCounter = 0;
     private int triggerCounter = 0;
 
@@ -326,4 +333,9 @@ public SubmittedJobInfo(
             this.effectiveConfig = effectiveConfig;
         }
     }
+
+    @Override
+    public Map<String, String> getClusterinfo(Configuration conf) throws Exception {
+        return CLUSTER_INFO;
+    }
 }
diff --git a/...ava/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java b/...ava/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -270,4 +270,19 @@ public void observeListJobsError() {
                         });
         assertEquals(podFailedMessage, exception.getMessage());
     }
+
+    @Test
+    public void observeClusterInfo() {
+        TestingFlinkService flinkService = new TestingFlinkService();
+        ApplicationObserver observer =
+                new ApplicationObserver(
+                        null, flinkService, configManager, new TestingStatusHelper<>());
+        FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+        bringToReadyStatus(deployment);
+        observer.observe(deployment, readyContext);
+        assertEquals(
+                JobManagerDeploymentStatus.READY,
+                deployment.getStatus().getJobManagerDeploymentStatus());
+        assertEquals(TestingFlinkService.CLUSTER_INFO, deployment.getStatus().getClusterInfo());
+    }
 }
diff --git a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
@@ -9084,6 +9084,10 @@ spec:
             type: object
           status:
             properties:
+              clusterInfo:
+                additionalProperties:
+                  type: string
+                type: object
               jobManagerDeploymentStatus:
                 enum:
                 - READY