diff --git a/docs/content.zh/docs/ops/metrics.md b/docs/content.zh/docs/ops/metrics.md
index dd9649a4a96a4..a4fc89215ddb5 100644
--- a/docs/content.zh/docs/ops/metrics.md
+++ b/docs/content.zh/docs/ops/metrics.md
@@ -632,6 +632,35 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
+### File Descriptors
+
+
+
+ | Scope |
+ Infix |
+ Metrics |
+ Description |
+ Type |
+
+
+
+
+ | Job-/TaskManager |
+ Status.FileDescriptor.Max |
+ Count |
+ The max number of file descriptors. |
+ Gauge |
+
+
+ | Job-/TaskManager |
+ Status.FileDescriptor.Open |
+ Count |
+ The total open of file descriptors. |
+ Gauge |
+
+
+
+
### Threads
diff --git a/docs/content/docs/ops/metrics.md b/docs/content/docs/ops/metrics.md
index 73d0c1ac96808..9b5003755da19 100644
--- a/docs/content/docs/ops/metrics.md
+++ b/docs/content/docs/ops/metrics.md
@@ -624,6 +624,35 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
+### File Descriptors
+
+
+
+ | Scope |
+ Infix |
+ Metrics |
+ Description |
+ Type |
+
+
+
+
+ | Job-/TaskManager |
+ Status.FileDescriptor.Max |
+ Count |
+ The max number of file descriptors. |
+ Gauge |
+
+
+ | Job-/TaskManager |
+ Status.FileDescriptor.Open |
+ Count |
+ The total open of file descriptors. |
+ Gauge |
+
+
+
+
### Threads
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java
index 22f59ee4e9bbe..4301808e58759 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java
+++ b/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";
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java
index 6b0c818fad646..915548be41a51 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java
+++ b/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.>gauge("Max", unixMXBean::getMaxFileDescriptorCount);
+ metrics.>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 memoryUsageSupplier) {
metricGroup.>gauge(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java
index b8f791c74181f..2195e8c4b265d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java
+++ b/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.
*