[GitHub] [flink] dmvk commented on a change in pull request #18143: [FLINK-23976][metrics] Add additional job status metrics

[GitBox <gi...@apache.org>]

dmvk commented on a change in pull request #18143:
URL: https://github.com/apache/flink/pull/18143#discussion_r783101544



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
##########
@@ -599,20 +608,117 @@ public ExecutionGraph getExecutionGraph() {
     @Override
     public final void startScheduling() {
         mainThreadExecutor.assertRunningInMainThread();
-        registerJobMetrics(jobManagerJobMetricGroup, executionGraph, this::getNumberOfRestarts);
+        registerJobMetrics(
+                jobManagerJobMetricGroup,
+                executionGraph,
+                this::getNumberOfRestarts,
+                ignored -> {},
+                executionGraph.getStatusTimestamp(JobStatus.INITIALIZING),
+                jobStatusMetricsSettings);
         operatorCoordinatorHandler.startAllOperatorCoordinators();
         startSchedulingInternal();
     }
 
     public static void registerJobMetrics(

Review comment:
       Would it make sense to get rid of the `jobStatusListenerConsumer`? I've found it bit confusing at first.
   ```suggestion
       public static JobStatusListener registerJobStatusMetrics(
   ```

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
##########
@@ -599,20 +608,117 @@ public ExecutionGraph getExecutionGraph() {
     @Override
     public final void startScheduling() {
         mainThreadExecutor.assertRunningInMainThread();
-        registerJobMetrics(jobManagerJobMetricGroup, executionGraph, this::getNumberOfRestarts);
+        registerJobMetrics(
+                jobManagerJobMetricGroup,
+                executionGraph,
+                this::getNumberOfRestarts,
+                ignored -> {},

Review comment:
       How does this work for the default scheduler if we don't listen on status changes?

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
##########
@@ -496,8 +495,7 @@ public void testNumRestartsMetric() throws Exception {
     public void testStatusMetrics() throws Exception {
         final CompletableFuture<UpTimeGauge> upTimeMetricFuture = new CompletableFuture<>();
         final CompletableFuture<DownTimeGauge> downTimeMetricFuture = new CompletableFuture<>();
-        final CompletableFuture<RestartTimeGauge> restartTimeMetricFuture =
-                new CompletableFuture<>();
+        new CompletableFuture<>();

Review comment:
       typo?

##########
File path: flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java
##########
@@ -215,5 +218,89 @@
                                     + "faster updating metrics. Increase this value if the metric fetcher causes too much load. Setting this value to 0 "
                                     + "disables the metric fetching completely.");
 
+    /** Controls which job status metrics will be exposed. */
+    public static final ConfigOption<List<JobStatusMetrics>> JOB_STATUS_METRICS =
+            key("metrics.job.status.enable")

Review comment:
       this is just to reduce cardinality or is there any other reason?




-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org