[Chore] Optimize the configuration in ServerLoadProtection (#17315)

Author: ruanwenjun

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java

@@ -28,7 +28,6 @@
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.validation.Errors;
@@ -56,8 +55,7 @@ public class MasterConfig implements Validator {
      */
     private Duration maxHeartbeatInterval = Duration.ofSeconds(10);
 
-    @Autowired
-    private MasterServerLoadProtection serverLoadProtection;
+    private MasterServerLoadProtectionConfig serverLoadProtection = new MasterServerLoadProtectionConfig();
 
     private Duration workerGroupRefreshInterval = Duration.ofMinutes(5);
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterServerLoadProtection.java

@@ -21,49 +21,41 @@
 import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
 import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository;
 
-import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
+import org.springframework.stereotype.Component;
+
 @Slf4j
-@Getter
+@Component
 public class MasterServerLoadProtection extends BaseServerLoadProtection {
 
-    private final int maxConcurrentWorkflowInstances;
     private final IWorkflowRepository workflowRepository;
 
+    private final MasterServerLoadProtectionConfig masterServerLoadProtectionConfig;
+
     public MasterServerLoadProtection(IWorkflowRepository workflowRepository,
-                                      int maxConcurrentWorkflowInstances,
-                                      double maxSystemCpuUsagePercentageThresholds,
-                                      double maxJvmCpuUsagePercentageThresholds,
-                                      double maxSystemMemoryUsagePercentageThresholds,
-                                      double maxDiskUsagePercentageThresholds,
-                                      boolean enabled) {
+                                      MasterConfig masterConfig) {
+        super(masterConfig.getServerLoadProtection());
+        this.masterServerLoadProtectionConfig = masterConfig.getServerLoadProtection();
         this.workflowRepository = workflowRepository;
-        this.maxConcurrentWorkflowInstances = maxConcurrentWorkflowInstances;
-        this.maxSystemCpuUsagePercentageThresholds = maxSystemCpuUsagePercentageThresholds;
-        this.maxJvmCpuUsagePercentageThresholds = maxJvmCpuUsagePercentageThresholds;
-        this.maxSystemMemoryUsagePercentageThresholds = maxSystemMemoryUsagePercentageThresholds;
-        this.maxDiskUsagePercentageThresholds = maxDiskUsagePercentageThresholds;
-        this.enabled = enabled;
     }
 
     @Override
     public boolean isOverload(SystemMetrics systemMetrics) {
-        if (!enabled) {
+        if (!masterServerLoadProtectionConfig.isEnabled()) {
             return false;
         }
 
-        // Check system metrics first
         if (super.isOverload(systemMetrics)) {
             return true;
         }
 
         // Check workflow instance count
         int currentWorkflowInstanceCount = workflowRepository.getAll().size();
-        if (currentWorkflowInstanceCount >= maxConcurrentWorkflowInstances) {
+        if (currentWorkflowInstanceCount >= masterServerLoadProtectionConfig.getMaxConcurrentWorkflowInstances()) {
             log.info(
                     "OverLoad: the workflow instance count: {} exceeds the maxConcurrentWorkflowInstances {}",
-                    currentWorkflowInstanceCount, maxConcurrentWorkflowInstances);
+                    currentWorkflowInstanceCount, masterServerLoadProtectionConfig.getMaxConcurrentWorkflowInstances());
             return true;
         }
         return false;

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterServerLoadProtectionConfig.java

@@ -17,43 +17,15 @@
 
 package org.apache.dolphinscheduler.server.master.config;
 
-import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository;
+import org.apache.dolphinscheduler.meter.metrics.BaseServerLoadProtectionConfig;
 
-import lombok.extern.slf4j.Slf4j;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
 
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class MasterServerLoadProtectionConfig extends BaseServerLoadProtectionConfig {
 
-@Slf4j
-@Configuration
-public class MasterServerLoadProtectionConfig {
+    private int maxConcurrentWorkflowInstances = Integer.MAX_VALUE;
 
-    @Bean
-    public MasterServerLoadProtection masterServerLoadProtection(
-                                                                 IWorkflowRepository workflowRepository,
-                                                                 @Value("${master.server-load-protection.max-concurrent-workflow-instances:2147483647}") int maxConcurrentWorkflowInstances,
-                                                                 @Value("${master.server-load-protection.max-system-cpu-usage-percentage-thresholds:0.7}") double maxSystemCpuUsagePercentageThresholds,
-                                                                 @Value("${master.server-load-protection.max-jvm-cpu-usage-percentage-thresholds:0.7}") double maxJvmCpuUsagePercentageThresholds,
-                                                                 @Value("${master.server-load-protection.max-system-memory-usage-percentage-thresholds:0.7}") double maxSystemMemoryUsagePercentageThresholds,
-                                                                 @Value("${master.server-load-protection.max-disk-usage-percentage-thresholds:0.7}") double maxDiskUsagePercentageThresholds,
-                                                                 @Value("${master.server-load-protection.enabled:true}") boolean enabled) {
-        MasterServerLoadProtection protection =
-                new MasterServerLoadProtection(workflowRepository,
-                        maxConcurrentWorkflowInstances,
-                        maxSystemCpuUsagePercentageThresholds,
-                        maxJvmCpuUsagePercentageThresholds,
-                        maxSystemMemoryUsagePercentageThresholds,
-                        maxDiskUsagePercentageThresholds,
-                        enabled);
-        log.info(
-                "Initialized MasterServerLoadProtection with IWorkflowRepository and maxConcurrentWorkflowInstances={}, "
-                        +
-                        "maxSystemCpuUsagePercentageThresholds={}, maxJvmCpuUsagePercentageThresholds={}, " +
-                        "maxSystemMemoryUsagePercentageThresholds={}, maxDiskUsagePercentageThresholds={}, enabled={}",
-                maxConcurrentWorkflowInstances, maxSystemCpuUsagePercentageThresholds,
-                maxJvmCpuUsagePercentageThresholds,
-                maxSystemMemoryUsagePercentageThresholds, maxDiskUsagePercentageThresholds, enabled);
-        return protection;
-    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java

@@ -68,6 +68,9 @@ public class CommandEngine extends BaseDaemonThread implements AutoCloseable {
     @Autowired
     private MasterConfig masterConfig;
 
+    @Autowired
+    private MasterServerLoadProtection masterServerLoadProtection;
+
     @Autowired
     private IWorkflowRepository workflowRepository;
 
@@ -107,12 +110,11 @@ public void close() throws Exception {
 
     @Override
     public void run() {
-        MasterServerLoadProtection serverLoadProtection = masterConfig.getServerLoadProtection();
         while (flag) {
             try {
                 // todo: if the workflow event queue is much, we need to handle the back pressure
                 SystemMetrics systemMetrics = metricsProvider.getSystemMetrics();
-                if (serverLoadProtection.isOverload(systemMetrics)) {
+                if (masterServerLoadProtection.isOverload(systemMetrics)) {
                     log.warn("The current server is overload, cannot consumes commands.");
                     MasterServerMetrics.incMasterOverload();
                     Thread.sleep(Constants.SLEEP_TIME_MILLIS);

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterHeartBeatTask.java

@@ -41,6 +41,8 @@ public class MasterHeartBeatTask extends BaseHeartBeatTask<MasterHeartBeat> {
 
     private final MasterConfig masterConfig;
 
+    private MasterServerLoadProtection masterServerLoadProtection;
+
     private final MetricsProvider metricsProvider;
 
     private final RegistryClient registryClient;
@@ -52,11 +54,13 @@ public class MasterHeartBeatTask extends BaseHeartBeatTask<MasterHeartBeat> {
     private final int processId;
 
     public MasterHeartBeatTask(@NonNull MasterConfig masterConfig,
+                               @NonNull MasterServerLoadProtection masterServerLoadProtection,
                                @NonNull MetricsProvider metricsProvider,
                                @NonNull RegistryClient registryClient,
                                @NonNull MasterCoordinator masterCoordinator) {
         super("MasterHeartBeatTask", masterConfig.getMaxHeartbeatInterval().toMillis());
         this.masterConfig = masterConfig;
+        this.masterServerLoadProtection = masterServerLoadProtection;
         this.metricsProvider = metricsProvider;
         this.registryClient = registryClient;
         this.masterCoordinator = masterCoordinator;
@@ -67,7 +71,6 @@ public MasterHeartBeatTask(@NonNull MasterConfig masterConfig,
     @Override
     public MasterHeartBeat getHeartBeat() {
         SystemMetrics systemMetrics = metricsProvider.getSystemMetrics();
-        ServerStatus serverStatus = getServerStatus(systemMetrics, masterConfig.getServerLoadProtection());
         return MasterHeartBeat.builder()
                 .startupTime(ServerLifeCycleManager.getServerStartupTime())
                 .reportTime(System.currentTimeMillis())
@@ -81,7 +84,8 @@ public MasterHeartBeat getHeartBeat() {
                 .memoryUsage(systemMetrics.getSystemMemoryUsedPercentage())
                 .diskUsage(systemMetrics.getDiskUsedPercentage())
                 .processId(processId)
-                .serverStatus(serverStatus)
+                .serverStatus(
+                        masterServerLoadProtection.isOverload(systemMetrics) ? ServerStatus.BUSY : ServerStatus.NORMAL)
                 .host(NetUtils.getHost())
                 .port(masterConfig.getListenPort())
                 .isCoordinator(masterCoordinator.isActive())
@@ -108,8 +112,4 @@ public void writeHeartBeat(final MasterHeartBeat masterHeartBeat) {
                 masterHeartBeatJson);
     }
 
-    private ServerStatus getServerStatus(final SystemMetrics systemMetrics,
-                                         final MasterServerLoadProtection masterServerLoadProtection) {
-        return masterServerLoadProtection.isOverload(systemMetrics) ? ServerStatus.BUSY : ServerStatus.NORMAL;
-    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistryClient.java

@@ -28,6 +28,7 @@
 import org.apache.dolphinscheduler.registry.api.RegistryException;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
+import org.apache.dolphinscheduler.server.master.config.MasterServerLoadProtection;
 import org.apache.dolphinscheduler.server.master.engine.MasterCoordinator;
 
 import lombok.extern.slf4j.Slf4j;
@@ -49,6 +50,9 @@ public class MasterRegistryClient implements AutoCloseable {
     @Autowired
     private MasterConfig masterConfig;
 
+    @Autowired
+    private MasterServerLoadProtection masterServerLoadProtection;
+
     @Autowired
     private MetricsProvider metricsProvider;
 
@@ -59,8 +63,8 @@ public class MasterRegistryClient implements AutoCloseable {
 
     public void start() {
         try {
-            this.masterHeartBeatTask =
-                    new MasterHeartBeatTask(masterConfig, metricsProvider, registryClient, masterCoordinator);
+            this.masterHeartBeatTask = new MasterHeartBeatTask(masterConfig, masterServerLoadProtection,
+                    metricsProvider, registryClient, masterCoordinator);
             // master registry
             registry();
             registryClient.addConnectionStateListener(new MasterConnectionStateListener(registryClient));

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/config/MasterConfigTest.java

@@ -20,49 +20,25 @@
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.mock;
 
 import org.apache.dolphinscheduler.server.master.cluster.loadbalancer.WorkerLoadBalancerConfigurationProperties;
 import org.apache.dolphinscheduler.server.master.cluster.loadbalancer.WorkerLoadBalancerType;
-import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository;
 
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
 import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.boot.test.context.TestConfiguration;
-import org.springframework.context.annotation.Bean;
-import org.springframework.test.context.TestPropertySource;
 
 @AutoConfigureMockMvc
-@SpringBootTest(classes = {
-        MasterConfig.class,
-        MasterServerLoadProtectionConfig.class,
-        MasterConfigTest.TestBeans.class
-})
-@TestPropertySource(properties = {
-        "master.server-load-protection.max-system-cpu-usage-percentage-thresholds=0.9",
-        "master.server-load-protection.max-jvm-cpu-usage-percentage-thresholds=0.9",
-        "master.server-load-protection.max-system-memory-usage-percentage-thresholds=0.9",
-        "master.server-load-protection.max-disk-usage-percentage-thresholds=0.9"
-})
+@SpringBootTest(classes = MasterConfig.class)
 public class MasterConfigTest {
 
     @Autowired
     private MasterConfig masterConfig;
 
-    @TestConfiguration
-    static class TestBeans {
-
-        @Bean
-        public IWorkflowRepository workflowRepository() {
-            return mock(IWorkflowRepository.class);
-        }
-    }
-
     @Test
     public void getServerLoadProtection() {
-        MasterServerLoadProtection serverLoadProtection = masterConfig.getServerLoadProtection();
+        MasterServerLoadProtectionConfig serverLoadProtection = masterConfig.getServerLoadProtection();
         assertTrue(serverLoadProtection.isEnabled());
         assertEquals(0.9, serverLoadProtection.getMaxSystemCpuUsagePercentageThresholds());
         assertEquals(0.9, serverLoadProtection.getMaxJvmCpuUsagePercentageThresholds());