[FLINK-37709][metrics] add file descriptor jvm metrics

Author: HuangZhenQiu

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java

@@ -58,6 +58,9 @@ private MetricNames() {}
     public static final String MEMORY_COMMITTED = "Committed";
     public static final String MEMORY_MAX = "Max";
 
+    public static final String FILE_DESCRIPTOR_MAX = "Max";
+    public static final String FILE_DESCRIPTOR_OPEN = "Open";
+
     public static final String IS_BACK_PRESSURED = "isBackPressured";
 
     public static final String CHECKPOINT_ALIGNMENT_TIME = "checkpointAlignmentTime";

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java

@@ -139,6 +139,7 @@ public static void instantiateStatusMetrics(MetricGroup metricGroup) {
         instantiateMemoryMetrics(jvm.addGroup(METRIC_GROUP_MEMORY));
         instantiateThreadMetrics(jvm.addGroup("Threads"));
         instantiateCPUMetrics(jvm.addGroup("CPU"));
+        instantiateFileDescriptorMetrics(jvm.addGroup("FileDescriptor"));
     }
 
     public static void instantiateFlinkMemoryMetricGroup(
@@ -338,6 +339,30 @@ static void instantiateMetaspaceMemoryMetrics(final MetricGroup parentMetricGrou
         }
     }
 
+    static void instantiateFileDescriptorMetrics(MetricGroup metrics) {
+        try {
+            final com.sun.management.OperatingSystemMXBean mxBean =
+                    (com.sun.management.OperatingSystemMXBean)
+                            ManagementFactory.getOperatingSystemMXBean();
+
+            if (mxBean instanceof com.sun.management.UnixOperatingSystemMXBean) {
+                com.sun.management.UnixOperatingSystemMXBean unixMXBean =
+                        (com.sun.management.UnixOperatingSystemMXBean) mxBean;
+                metrics.<Long, Gauge<Long>>gauge("Max", unixMXBean::getMaxFileDescriptorCount);
+                metrics.<Long, Gauge<Long>>gauge("Open", unixMXBean::getOpenFileDescriptorCount);
+
+            } else {
+                throw new UnsupportedOperationException(
+                        "Can't find com.sun.management.UnixOperatingSystemMXBean in JVM.");
+            }
+        } catch (Exception e) {
+            LOG.warn(
+                    "Cannot access com.sun.management.UnixOperatingSystemMXBean.getOpenFileDescriptorCount()"
+                            + " - FileDescriptor metrics will not be available.",
+                    e);
+        }
+    }
+
     private static void instantiateMemoryUsageMetrics(
             final MetricGroup metricGroup, final Supplier<MemoryUsage> memoryUsageSupplier) {
         metricGroup.<Long, Gauge<Long>>gauge(

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java

@@ -177,6 +177,16 @@ void testHeapMetricsCompleteness() {
         assertThat(heapMetrics.get(MetricNames.MEMORY_MAX)).isNotNull();
     }
 
+    @Test
+    void testFileDescriptorMetricsCompleteness() {
+        final InterceptingOperatorMetricGroup heapMetrics = new InterceptingOperatorMetricGroup();
+
+        MetricUtils.instantiateFileDescriptorMetrics(heapMetrics);
+
+        assertThat(heapMetrics.get(MetricNames.FILE_DESCRIPTOR_MAX)).isNotNull();
+        assertThat(heapMetrics.get(MetricNames.FILE_DESCRIPTOR_OPEN)).isNotNull();
+    }
+
     /**
      * Tests that heap/non-heap metrics do not rely on a static MemoryUsage instance.
      *