+
+
+
+ {t('monitor.worker.thread_pool_usage')}
+
+ }
+ >
+
{item && (
)}
@@ -182,6 +269,38 @@ const worker = defineComponent({
data={zkDirectoryRef}
onConfirmModal={onConfirmModal}
/>
+
+
+ {diskDetailListRef.length === 0 &&
+ t('monitor.worker.data_disk_usage_empty')}
+ {diskDetailListRef.map((d: any) =>
+ h(
+ NSpace,
+ { justify: 'space-between' },
+ {
+ default: () => [
+ h('span', d.diskPath || '-'),
+ h(
+ NTag,
+ { type: 'info', size: 'small' },
+ {
+ default: () =>
+ `${((d.usedPercentage || 0) * 100).toFixed(2)}%`
+ }
+ )
+ ]
+ }
+ )
+ )}
+
+
)
}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java
index 47d0f9c7dd45..6ad6318f9a54 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java
@@ -20,10 +20,14 @@
import org.apache.dolphinscheduler.common.enums.ServerStatus;
import org.apache.dolphinscheduler.common.lifecycle.ServerLifeCycleManager;
import org.apache.dolphinscheduler.common.model.BaseHeartBeatTask;
+import org.apache.dolphinscheduler.common.model.DiskUsageDetail;
import org.apache.dolphinscheduler.common.model.WorkerHeartBeat;
+import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.NetUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
+import org.apache.dolphinscheduler.meter.metrics.DiskUsagePercentageThresholdsRule;
+import org.apache.dolphinscheduler.meter.metrics.DiskUsageUtils;
import org.apache.dolphinscheduler.meter.metrics.MetricsProvider;
import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
import org.apache.dolphinscheduler.registry.api.RegistryClient;
@@ -33,6 +37,11 @@
import org.apache.dolphinscheduler.server.worker.metrics.WorkerServerMetrics;
import org.apache.dolphinscheduler.task.executor.container.ITaskExecutorContainer;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
@@ -67,6 +76,8 @@ public WorkerHeartBeatTask(@NonNull WorkerConfig workerConfig,
public WorkerHeartBeat getHeartBeat() {
SystemMetrics systemMetrics = metricsProvider.getSystemMetrics();
+ final List diskUsageDetails = buildDiskUsageDetails();
+
return WorkerHeartBeat.builder()
.startupTime(ServerLifeCycleManager.getServerStartupTime())
.reportTime(System.currentTimeMillis())
@@ -79,6 +90,7 @@ public WorkerHeartBeat getHeartBeat() {
.jvmNonHeapMax(systemMetrics.getJvmNonHeapMax())
.memoryUsage(systemMetrics.getSystemMemoryUsedPercentage())
.diskUsage(systemMetrics.getDiskUsedPercentage())
+ .diskUsageDetails(diskUsageDetails)
.processId(processId)
.workerHostWeight(workerConfig.getHostWeight())
.threadPoolUsage(taskExecutorContainer.slotUsage())
@@ -90,6 +102,37 @@ public WorkerHeartBeat getHeartBeat() {
.build();
}
+ private List buildDiskUsageDetails() {
+ final Set monitoredPaths = new HashSet<>();
+ // Always monitor data basedir path.
+ monitoredPaths.add(FileUtils.DATA_BASEDIR);
+
+ final List rules =
+ workerConfig.getServerLoadProtection().getMaxDiskUsagePercentageThresholdsRules();
+ if (rules != null) {
+ for (DiskUsagePercentageThresholdsRule rule : rules) {
+ if (rule == null) {
+ continue;
+ }
+ if (rule.getDiskPath() == null || rule.getDiskPath().trim().isEmpty()) {
+ continue;
+ }
+ monitoredPaths.add(rule.getDiskPath().trim());
+ }
+ }
+
+ final List details = new ArrayList<>();
+ for (String path : monitoredPaths) {
+ DiskUsageUtils.getDiskUsage(path).ifPresent(diskUsage -> details.add(DiskUsageDetail.builder()
+ .diskPath(diskUsage.getDiskPath())
+ .usedBytes(diskUsage.getUsedBytes())
+ .totalBytes(diskUsage.getTotalBytes())
+ .usedPercentage(diskUsage.getUsedPercentage())
+ .build()));
+ }
+ return details;
+ }
+
@Override
public void writeHeartBeat(final WorkerHeartBeat workerHeartBeat) {
final String failoverNodePath = RegistryUtils.getFailoveredNodePath(workerHeartBeat);
diff --git a/dolphinscheduler-worker/src/main/resources/application.yaml b/dolphinscheduler-worker/src/main/resources/application.yaml
index ef1793fb7e7a..247bb7eeb52f 100644
--- a/dolphinscheduler-worker/src/main/resources/application.yaml
+++ b/dolphinscheduler-worker/src/main/resources/application.yaml
@@ -58,7 +58,16 @@ worker:
# Worker max System memory usage , when the master's system memory usage is smaller then this value, master server can execute workflow.
max-system-memory-usage-percentage-thresholds: 0.8
# Worker max disk usage , when the worker's disk usage is smaller then this value, worker server can be dispatched tasks.
+ # Deprecated: Use max-disk-usage-percentage-thresholds-rules instead
max-disk-usage-percentage-thresholds: 0.8
+ # Worker disk usage threshold rules for monitoring multiple paths
+ # Example:
+ # max-disk-usage-percentage-thresholds-rules:
+ # - disk-path: /data1
+ # usage-percentage-thresholds: 0.9
+ # - disk-path: /data2
+ # usage-percentage-thresholds: 0.8
+ max-disk-usage-percentage-thresholds-rules: []
task-execute-threads-full-policy: REJECT
physical-task-config:
# The number of threads in the Physical task engine that used to execute tasks