[1.3.6-prepare][Improvement][Docker/K8s] Improve feedback (#5195)
* [1.3.6-prepare][Config] Revise configuration comments * [1.3.6-prepare][Improvement][K8s] Optimize long host to short host * [1.3.6-prepare][Improvement][Docker] Remove docker restriction in worker group * [1.3.6-prepare][UI] Improve long host display in k8s * [1.3.6-prepare][Server] Optimize log display * [1.3.6-prepare][K8s] Fix the problem that DATA_BASEDIR_PATH, RESOURCE_STORAGE_TYPE, and RESOURCE_UPLOAD_PATH are not set and cannot be deployed * [1.3.6-prepare][K8s] Fix the wrong memory and always Runtime.getRuntime().availableProcessors()=1 in docker/k8s
This commit is contained in:
Shiwen Cheng
GitHub
parent
be169b6390
commit
405d62f4b9
|
|
@ -30,7 +30,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>master execute thread number</description>
|
||||
<description>master execute thread number to limit process instances</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -39,7 +39,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>master execute task number in parallel</description>
|
||||
<description>master execute task number in parallel per process instance</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -48,7 +48,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>master dispatch task number</description>
|
||||
<description>master dispatch task number per batch</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -63,7 +63,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>master heartbeat interval</description>
|
||||
<description>master heartbeat interval, the unit is second</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -81,7 +81,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>master commit task interval</description>
|
||||
<description>master commit task interval, the unit is millisecond</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -90,13 +90,13 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>only less than cpu avg load, master server can work. default value -1: the number of cpu cores * 2</description>
|
||||
<description>master max cpuload avg, only higher than the system cpu load average, master server can schedule. default value -1: the number of cpu cores * 2</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
<name>master.reserved.memory</name>
|
||||
<value>0.3</value>
|
||||
<description>only larger than reserved memory, master server can work. default value 0.3, the unit is G</description>
|
||||
<description>master reserved memory, only lower than system available memory, master server can schedule. default value 0.3, the unit is G</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
</configuration>
|
||||
|
|
@ -30,7 +30,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>worker execute thread number</description>
|
||||
<description>worker execute thread number to limit task instances</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -39,7 +39,7 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>worker heartbeat interval</description>
|
||||
<description>worker heartbeat interval, the unit is second</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
@ -48,13 +48,13 @@
|
|||
<value-attributes>
|
||||
<type>int</type>
|
||||
</value-attributes>
|
||||
<description>only less than cpu avg load, worker server can work. default value -1: the number of cpu cores * 2</description>
|
||||
<description>worker max cpuload avg, only higher than the system cpu load average, worker server can be dispatched tasks. default value -1: the number of cpu cores * 2</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
<name>worker.reserved.memory</name>
|
||||
<value>0.3</value>
|
||||
<description>only larger than reserved memory, worker server can work. default value 0.3, the unit is G</description>
|
||||
<description>worker reserved memory, only lower than system available memory, worker server can be dispatched tasks. default value 0.3, the unit is G</description>
|
||||
<on-ambari-upgrade add="true"/>
|
||||
</property>
|
||||
<property>
|
||||
|
|
|
|||
|
|
@ -18,29 +18,29 @@
|
|||
# master listen port
|
||||
#master.listen.port=5678
|
||||
|
||||
# master execute thread number
|
||||
# master execute thread number to limit process instances
|
||||
master.exec.threads=${MASTER_EXEC_THREADS}
|
||||
|
||||
# master execute task number in parallel
|
||||
# master execute task number in parallel per process instance
|
||||
master.exec.task.num=${MASTER_EXEC_TASK_NUM}
|
||||
|
||||
# master dispatch task number
|
||||
# master dispatch task number per batch
|
||||
master.dispatch.task.num=${MASTER_DISPATCH_TASK_NUM}
|
||||
|
||||
# master host selector to select a suitable worker, default value: LowerWeight. Optional values include Random, RoundRobin, LowerWeight
|
||||
master.host.selector=${MASTER_HOST_SELECTOR}
|
||||
|
||||
# master heartbeat interval
|
||||
# master heartbeat interval, the unit is second
|
||||
master.heartbeat.interval=${MASTER_HEARTBEAT_INTERVAL}
|
||||
|
||||
# master commit task retry times
|
||||
master.task.commit.retryTimes=${MASTER_TASK_COMMIT_RETRYTIMES}
|
||||
|
||||
# master commit task interval
|
||||
# master commit task interval, the unit is millisecond
|
||||
master.task.commit.interval=${MASTER_TASK_COMMIT_INTERVAL}
|
||||
|
||||
# only less than cpu avg load, master server can work. default value -1: the number of cpu cores * 2
|
||||
# master max cpuload avg, only higher than the system cpu load average, master server can schedule. default value -1: the number of cpu cores * 2
|
||||
master.max.cpuload.avg=${MASTER_MAX_CPULOAD_AVG}
|
||||
|
||||
# only larger than reserved memory, master server can work. default value 0.3, the unit is G
|
||||
# master reserved memory, only lower than system available memory, master server can schedule. default value 0.3, the unit is G
|
||||
master.reserved.memory=${MASTER_RESERVED_MEMORY}
|
||||
|
|
|
|||
|
|
@ -18,16 +18,16 @@
|
|||
# worker listener port
|
||||
#worker.listen.port=1234
|
||||
|
||||
# worker execute thread number
|
||||
# worker execute thread number to limit task instances
|
||||
worker.exec.threads=${WORKER_EXEC_THREADS}
|
||||
|
||||
# worker heartbeat interval
|
||||
# worker heartbeat interval, the unit is second
|
||||
worker.heartbeat.interval=${WORKER_HEARTBEAT_INTERVAL}
|
||||
|
||||
# only less than cpu avg load, worker server can work. default value -1: the number of cpu cores * 2
|
||||
# worker max cpuload avg, only higher than the system cpu load average, worker server can be dispatched tasks. default value -1: the number of cpu cores * 2
|
||||
worker.max.cpuload.avg=${WORKER_MAX_CPULOAD_AVG}
|
||||
|
||||
# only larger than reserved memory, worker server can work. default value 0.3, the unit is G
|
||||
# worker reserved memory, only lower than system available memory, worker server can be dispatched tasks. default value 0.3, the unit is G
|
||||
worker.reserved.memory=${WORKER_RESERVED_MEMORY}
|
||||
|
||||
# default worker groups separated by comma, like 'worker.groups=default,test'
|
||||
|
|
|
|||
|
|
@ -184,7 +184,7 @@ Create a zookeeper environment variables.
|
|||
Create a common fs_s3a environment variables.
|
||||
*/}}
|
||||
{{- define "dolphinscheduler.fs_s3a.env_vars" -}}
|
||||
{{- if eq .Values.common.configmap.RESOURCE_STORAGE_TYPE "S3" -}}
|
||||
{{- if eq (default "HDFS" .Values.common.configmap.RESOURCE_STORAGE_TYPE) "S3" -}}
|
||||
- name: FS_S3A_SECRET_KEY
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
|
|
@ -230,7 +230,7 @@ Create a fsFileResourcePersistence volumeMount.
|
|||
*/}}
|
||||
{{- define "dolphinscheduler.fsFileResource.volumeMount" -}}
|
||||
{{- if .Values.common.fsFileResourcePersistence.enabled -}}
|
||||
- mountPath: {{ .Values.common.configmap.RESOURCE_UPLOAD_PATH | quote }}
|
||||
- mountPath: {{ default "/dolphinscheduler" .Values.common.configmap.RESOURCE_UPLOAD_PATH | quote }}
|
||||
name: {{ include "dolphinscheduler.fullname" . }}-fs-file
|
||||
{{- end -}}
|
||||
{{- end -}}
|
||||
|
|
@ -14,7 +14,7 @@
|
|||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
#
|
||||
{{- if eq .Values.common.configmap.RESOURCE_STORAGE_TYPE "S3" }}
|
||||
{{- if eq (default "HDFS" .Values.common.configmap.RESOURCE_STORAGE_TYPE) "S3" }}
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
|
|
|
|||
|
|
@ -101,7 +101,7 @@ spec:
|
|||
failureThreshold: {{ .Values.worker.readinessProbe.failureThreshold }}
|
||||
{{- end }}
|
||||
volumeMounts:
|
||||
- mountPath: {{ .Values.common.configmap.DATA_BASEDIR_PATH | quote }}
|
||||
- mountPath: {{ default "/tmp/dolphinscheduler" .Values.common.configmap.DATA_BASEDIR_PATH | quote }}
|
||||
name: {{ include "dolphinscheduler.fullname" . }}-worker-data
|
||||
- mountPath: "/opt/dolphinscheduler/logs"
|
||||
name: {{ include "dolphinscheduler.fullname" . }}-worker-logs
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@
|
|||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.dolphinscheduler.api;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
|
|
@ -22,19 +23,15 @@ import org.springframework.boot.web.servlet.ServletComponentScan;
|
|||
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
|
||||
import org.springframework.context.annotation.ComponentScan;
|
||||
import org.springframework.context.annotation.FilterType;
|
||||
import springfox.documentation.swagger2.annotations.EnableSwagger2;
|
||||
|
||||
@SpringBootApplication
|
||||
@ServletComponentScan
|
||||
@ComponentScan(basePackages = {"org.apache.dolphinscheduler"},
|
||||
excludeFilters = @ComponentScan.Filter(type = FilterType.REGEX,
|
||||
pattern = "org.apache.dolphinscheduler.server.*"))
|
||||
|
||||
@ComponentScan(value = "org.apache.dolphinscheduler",
|
||||
excludeFilters = @ComponentScan.Filter(type = FilterType.REGEX, pattern = "org.apache.dolphinscheduler.server.*"))
|
||||
public class ApiApplicationServer extends SpringBootServletInitializer {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(ApiApplicationServer.class, args);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -78,10 +78,6 @@ public class WorkerGroupService extends BaseService {
|
|||
if (checkAdmin(loginUser, result)) {
|
||||
return result;
|
||||
}
|
||||
if (Constants.DOCKER_MODE && !Constants.KUBERNETES_MODE) {
|
||||
putMsg(result, Status.CREATE_WORKER_GROUP_FORBIDDEN_IN_DOCKER);
|
||||
return result;
|
||||
}
|
||||
if (StringUtils.isEmpty(name)) {
|
||||
putMsg(result, Status.NAME_NULL);
|
||||
return result;
|
||||
|
|
@ -297,10 +293,6 @@ public class WorkerGroupService extends BaseService {
|
|||
if (checkAdmin(loginUser, result)) {
|
||||
return result;
|
||||
}
|
||||
if (Constants.DOCKER_MODE && !Constants.KUBERNETES_MODE) {
|
||||
putMsg(result, Status.DELETE_WORKER_GROUP_FORBIDDEN_IN_DOCKER);
|
||||
return result;
|
||||
}
|
||||
WorkerGroup workerGroup = workerGroupMapper.selectById(id);
|
||||
if (workerGroup == null) {
|
||||
putMsg(result, Status.DELETE_WORKER_GROUP_NOT_EXIST);
|
||||
|
|
|
|||
|
|
@ -367,22 +367,11 @@ public final class Constants {
|
|||
*/
|
||||
public static final int DEFAULT_MASTER_CPU_LOAD = Runtime.getRuntime().availableProcessors() * 2;
|
||||
|
||||
/**
|
||||
* master reserved memory
|
||||
*/
|
||||
public static final double DEFAULT_MASTER_RESERVED_MEMORY = OSUtils.totalMemorySize() / 10;
|
||||
|
||||
/**
|
||||
* worker cpu load
|
||||
*/
|
||||
public static final int DEFAULT_WORKER_CPU_LOAD = Runtime.getRuntime().availableProcessors() * 2;
|
||||
|
||||
/**
|
||||
* worker reserved memory
|
||||
*/
|
||||
public static final double DEFAULT_WORKER_RESERVED_MEMORY = OSUtils.totalMemorySize() / 10;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* default log cache rows num,output when reach the number
|
||||
|
|
|
|||
|
|
@ -86,6 +86,9 @@ public class NetUtils {
|
|||
if (STS_PATTERN.matcher(host).find()) {
|
||||
return String.format("%s.%s", host, host.replaceFirst("\\d+$", "headless"));
|
||||
}
|
||||
} else if (canonicalHost.contains(".")) {
|
||||
String[] items = canonicalHost.split("\\.");
|
||||
return String.format("%s.%s", items[0], items[1]);
|
||||
}
|
||||
return canonicalHost;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,11 +16,8 @@
|
|||
*/
|
||||
package org.apache.dolphinscheduler.common.utils;
|
||||
|
||||
import org.apache.dolphinscheduler.common.Constants;
|
||||
import org.apache.dolphinscheduler.common.shell.ShellExecutor;
|
||||
|
||||
import org.apache.commons.configuration.Configuration;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.IOException;
|
||||
|
|
@ -56,23 +53,19 @@ public class OSUtils {
|
|||
|
||||
public static final ThreadLocal<Logger> taskLoggerThreadLocal = new ThreadLocal<>();
|
||||
|
||||
private static final Pattern STS_PATTERN = Pattern.compile("-\\d+$"); // StatefulSet pattern
|
||||
private static final Pattern IP_PATTERN = Pattern.compile("\\d{1,3}(\\.\\d{1,3}){3,5}$");
|
||||
|
||||
private static final SystemInfo SI = new SystemInfo();
|
||||
public static final String TWO_DECIMAL = "0.00";
|
||||
|
||||
/**
|
||||
* return -1 when the function can not get hardware env info
|
||||
* e.g {@link OSUtils#loadAverage()} {@link OSUtils#cpuUsage()}
|
||||
*/
|
||||
public static final double NEGATIVE_ONE = -1;
|
||||
/**
|
||||
* return -1 when the function can not get hardware env info
|
||||
* e.g {@link OSUtils#loadAverage()} {@link OSUtils#cpuUsage()}
|
||||
*/
|
||||
public static final double NEGATIVE_ONE = -1;
|
||||
|
||||
private static HardwareAbstractionLayer hal = SI.getHardware();
|
||||
|
||||
private OSUtils() {}
|
||||
|
||||
|
||||
/**
|
||||
* get memory usage
|
||||
* Keep 2 decimal
|
||||
|
|
@ -80,14 +73,13 @@ public class OSUtils {
|
|||
*/
|
||||
public static double memoryUsage() {
|
||||
GlobalMemory memory = hal.getMemory();
|
||||
double memoryUsage = (memory.getTotal() - memory.getAvailable() - memory.getSwapUsed()) * 0.1 / memory.getTotal() * 10;
|
||||
double memoryUsage = (memory.getTotal() - memory.getAvailable()) * 1.0 / memory.getTotal();
|
||||
|
||||
DecimalFormat df = new DecimalFormat(TWO_DECIMAL);
|
||||
df.setRoundingMode(RoundingMode.HALF_UP);
|
||||
return Double.parseDouble(df.format(memoryUsage));
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* get available physical memory size
|
||||
*
|
||||
|
|
@ -96,12 +88,11 @@ public class OSUtils {
|
|||
*/
|
||||
public static double availablePhysicalMemorySize() {
|
||||
GlobalMemory memory = hal.getMemory();
|
||||
double availablePhysicalMemorySize = (memory.getAvailable() + memory.getSwapUsed()) /1024.0/1024/1024;
|
||||
double availablePhysicalMemorySize = memory.getAvailable() / 1024.0 / 1024 / 1024;
|
||||
|
||||
DecimalFormat df = new DecimalFormat(TWO_DECIMAL);
|
||||
df.setRoundingMode(RoundingMode.HALF_UP);
|
||||
return Double.parseDouble(df.format(availablePhysicalMemorySize));
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -112,14 +103,13 @@ public class OSUtils {
|
|||
*/
|
||||
public static double totalMemorySize() {
|
||||
GlobalMemory memory = hal.getMemory();
|
||||
double availablePhysicalMemorySize = memory.getTotal() /1024.0/1024/1024;
|
||||
double totalPhysicalMemorySize = memory.getTotal() / 1024.0 / 1024 / 1024;
|
||||
|
||||
DecimalFormat df = new DecimalFormat(TWO_DECIMAL);
|
||||
df.setRoundingMode(RoundingMode.HALF_UP);
|
||||
return Double.parseDouble(df.format(availablePhysicalMemorySize));
|
||||
return Double.parseDouble(df.format(totalPhysicalMemorySize));
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* load average
|
||||
*
|
||||
|
|
@ -421,7 +411,6 @@ public class OSUtils {
|
|||
return getOSName().startsWith("Mac");
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* whether is windows
|
||||
* @return true if windows
|
||||
|
|
@ -440,42 +429,22 @@ public class OSUtils {
|
|||
|
||||
/**
|
||||
* check memory and cpu usage
|
||||
* @param systemCpuLoad systemCpuLoad
|
||||
* @param systemReservedMemory systemReservedMemory
|
||||
* @param maxCpuloadAvg maxCpuloadAvg
|
||||
* @param reservedMemory reservedMemory
|
||||
* @return check memory and cpu usage
|
||||
*/
|
||||
public static Boolean checkResource(double systemCpuLoad, double systemReservedMemory){
|
||||
public static Boolean checkResource(double maxCpuloadAvg, double reservedMemory) {
|
||||
// system load average
|
||||
double loadAverage = OSUtils.loadAverage();
|
||||
// system available physical memory
|
||||
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
|
||||
|
||||
if(loadAverage > systemCpuLoad || availablePhysicalMemorySize < systemReservedMemory){
|
||||
logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}", availablePhysicalMemorySize , loadAverage);
|
||||
if (loadAverage > maxCpuloadAvg || availablePhysicalMemorySize < reservedMemory) {
|
||||
logger.warn("current cpu load average {} is too high or available memory {}G is too low, under max.cpuload.avg={} and reserved.memory={}G",
|
||||
loadAverage, availablePhysicalMemorySize, maxCpuloadAvg, reservedMemory);
|
||||
return false;
|
||||
}else{
|
||||
} else {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* check memory and cpu usage
|
||||
* @param conf conf
|
||||
* @param isMaster is master
|
||||
* @return check memory and cpu usage
|
||||
*/
|
||||
public static Boolean checkResource(Configuration conf, Boolean isMaster){
|
||||
double systemCpuLoad;
|
||||
double systemReservedMemory;
|
||||
|
||||
if(Boolean.TRUE.equals(isMaster)){
|
||||
systemCpuLoad = conf.getDouble(Constants.MASTER_MAX_CPULOAD_AVG, Constants.DEFAULT_MASTER_CPU_LOAD);
|
||||
systemReservedMemory = conf.getDouble(Constants.MASTER_RESERVED_MEMORY, Constants.DEFAULT_MASTER_RESERVED_MEMORY);
|
||||
}else{
|
||||
systemCpuLoad = conf.getDouble(Constants.WORKER_MAX_CPULOAD_AVG, Constants.DEFAULT_WORKER_CPU_LOAD);
|
||||
systemReservedMemory = conf.getDouble(Constants.WORKER_RESERVED_MEMORY, Constants.DEFAULT_WORKER_RESERVED_MEMORY);
|
||||
}
|
||||
return checkResource(systemCpuLoad,systemReservedMemory);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ public class NetUtilsTest {
|
|||
when(address.getHostAddress()).thenReturn("172.17.0.15");
|
||||
assertEquals("172.17.0.15", NetUtils.getHost(address));
|
||||
CommonTest.setFinalStatic(Constants.class.getDeclaredField("KUBERNETES_MODE"), true);
|
||||
assertEquals("dolphinscheduler-worker-0.dolphinscheduler-worker-headless.default.svc.cluster.local", NetUtils.getHost(address));
|
||||
assertEquals("dolphinscheduler-worker-0.dolphinscheduler-worker-headless", NetUtils.getHost(address));
|
||||
address = mock(InetAddress.class);
|
||||
when(address.getCanonicalHostName()).thenReturn("dolphinscheduler-worker-0");
|
||||
when(address.getHostName()).thenReturn("dolphinscheduler-worker-0");
|
||||
|
|
|
|||
|
|
@ -16,11 +16,6 @@
|
|||
*/
|
||||
package org.apache.dolphinscheduler.common.utils;
|
||||
|
||||
import org.apache.dolphinscheduler.common.Constants;
|
||||
|
||||
import org.apache.commons.configuration.Configuration;
|
||||
import org.apache.commons.configuration.PropertiesConfiguration;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
|
||||
|
|
@ -95,29 +90,6 @@ public class OSUtilsTest {
|
|||
Assert.assertTrue(resource);
|
||||
resource = OSUtils.checkResource(0,Double.MAX_VALUE);
|
||||
Assert.assertFalse(resource);
|
||||
|
||||
Configuration configuration = new PropertiesConfiguration();
|
||||
|
||||
configuration.setProperty(Constants.MASTER_MAX_CPULOAD_AVG,100);
|
||||
configuration.setProperty(Constants.MASTER_RESERVED_MEMORY,0);
|
||||
resource = OSUtils.checkResource(configuration,true);
|
||||
Assert.assertTrue(resource);
|
||||
|
||||
configuration.setProperty(Constants.MASTER_MAX_CPULOAD_AVG,0);
|
||||
configuration.setProperty(Constants.MASTER_RESERVED_MEMORY,Double.MAX_VALUE);
|
||||
resource = OSUtils.checkResource(configuration,true);
|
||||
Assert.assertFalse(resource);
|
||||
|
||||
configuration.setProperty(Constants.WORKER_MAX_CPULOAD_AVG,100);
|
||||
configuration.setProperty(Constants.WORKER_RESERVED_MEMORY,0);
|
||||
resource = OSUtils.checkResource(configuration,false);
|
||||
Assert.assertTrue(resource);
|
||||
|
||||
configuration.setProperty(Constants.WORKER_MAX_CPULOAD_AVG,0);
|
||||
configuration.setProperty(Constants.WORKER_RESERVED_MEMORY,Double.MAX_VALUE);
|
||||
resource = OSUtils.checkResource(configuration,false);
|
||||
Assert.assertFalse(resource);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -151,7 +151,7 @@ public class TaskPriorityQueueConsumer extends Thread{
|
|||
result = dispatcher.dispatch(executionContext);
|
||||
}
|
||||
} catch (ExecuteException e) {
|
||||
logger.error("dispatch error",e);
|
||||
logger.error("dispatch error: {}",e.getMessage());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,12 +26,13 @@ import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteExce
|
|||
import org.apache.dolphinscheduler.server.master.dispatch.executor.ExecutorManager;
|
||||
import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager;
|
||||
import org.apache.dolphinscheduler.server.master.dispatch.host.HostManager;
|
||||
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* executor dispatcher
|
||||
*/
|
||||
|
|
@ -84,8 +85,8 @@ public class ExecutorDispatcher implements InitializingBean {
|
|||
|
||||
Host host = hostManager.select(context);
|
||||
if (StringUtils.isEmpty(host.getAddress())) {
|
||||
throw new ExecuteException(String.format("fail to execute : %s due to no suitable worker , " +
|
||||
"current task need to %s worker group execute",
|
||||
throw new ExecuteException(String.format("fail to execute : %s due to no suitable worker, "
|
||||
+ "current task needs worker group %s to execute",
|
||||
context.getCommand(),context.getWorkerGroup()));
|
||||
}
|
||||
context.setHost(host);
|
||||
|
|
|
|||
|
|
@ -156,8 +156,8 @@ public class LowerWeightHostManager extends CommonHostManager {
|
|||
String[] parts = heartbeat.split(Constants.COMMA);
|
||||
int status = Integer.parseInt(parts[8]);
|
||||
if (status == Constants.ABNORMAL_NODE_STATUS) {
|
||||
logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}",
|
||||
Double.parseDouble(parts[3]), Double.parseDouble(parts[2]));
|
||||
logger.warn("worker {} current cpu load average {} is too high or available memory {}G is too low",
|
||||
addr, Double.parseDouble(parts[2]), Double.parseDouble(parts[3]));
|
||||
return null;
|
||||
}
|
||||
double cpu = Double.parseDouble(parts[0]);
|
||||
|
|
|
|||
|
|
@ -73,14 +73,12 @@ public class HeartBeatTask implements Runnable {
|
|||
}
|
||||
}
|
||||
|
||||
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
|
||||
double loadAverage = OSUtils.loadAverage();
|
||||
|
||||
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
|
||||
int status = Constants.NORAML_NODE_STATUS;
|
||||
|
||||
if (availablePhysicalMemorySize < reservedMemory
|
||||
|| loadAverage > maxCpuloadAvg) {
|
||||
logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}", availablePhysicalMemorySize, loadAverage);
|
||||
if (loadAverage > maxCpuloadAvg || availablePhysicalMemorySize < reservedMemory) {
|
||||
logger.warn("current cpu load average {} is too high or available memory {}G is too low, under max.cpuload.avg={} and reserved.memory={}G",
|
||||
loadAverage, availablePhysicalMemorySize, maxCpuloadAvg, reservedMemory);
|
||||
status = Constants.ABNORMAL_NODE_STATUS;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -18,29 +18,29 @@
|
|||
# master listen port
|
||||
#master.listen.port=5678
|
||||
|
||||
# master execute thread number
|
||||
# master execute thread number to limit process instances
|
||||
#master.exec.threads=100
|
||||
|
||||
# master execute task number in parallel
|
||||
# master execute task number in parallel per process instance
|
||||
#master.exec.task.num=20
|
||||
|
||||
# master dispatch task number
|
||||
# master dispatch task number per batch
|
||||
#master.dispatch.task.num=3
|
||||
|
||||
# master host selector to select a suitable worker, default value: LowerWeight. Optional values include Random, RoundRobin, LowerWeight
|
||||
#master.host.selector=LowerWeight
|
||||
|
||||
# master heartbeat interval
|
||||
# master heartbeat interval, the unit is second
|
||||
#master.heartbeat.interval=10
|
||||
|
||||
# master commit task retry times
|
||||
#master.task.commit.retryTimes=5
|
||||
|
||||
# master commit task interval
|
||||
# master commit task interval, the unit is millisecond
|
||||
#master.task.commit.interval=1000
|
||||
|
||||
# only less than cpu avg load, master server can work. default value -1: the number of cpu cores * 2
|
||||
# master max cpuload avg, only higher than the system cpu load average, master server can schedule. default value -1: the number of cpu cores * 2
|
||||
#master.max.cpuload.avg=-1
|
||||
|
||||
# only larger than reserved memory, master server can work. default value 0.3, the unit is G
|
||||
# master reserved memory, only lower than system available memory, master server can schedule. default value 0.3, the unit is G
|
||||
#master.reserved.memory=0.3
|
||||
|
|
|
|||
|
|
@ -18,16 +18,16 @@
|
|||
# worker listener port
|
||||
#worker.listen.port=1234
|
||||
|
||||
# worker execute thread number
|
||||
# worker execute thread number to limit task instances
|
||||
#worker.exec.threads=100
|
||||
|
||||
# worker heartbeat interval
|
||||
# worker heartbeat interval, the unit is second
|
||||
#worker.heartbeat.interval=10
|
||||
|
||||
# only less than cpu avg load, worker server can work. default value -1: the number of cpu cores * 2
|
||||
# worker max cpuload avg, only higher than the system cpu load average, worker server can be dispatched tasks. default value -1: the number of cpu cores * 2
|
||||
#worker.max.cpuload.avg=-1
|
||||
|
||||
# only larger than reserved memory, worker server can work. default value 0.3, the unit is G
|
||||
# worker reserved memory, only lower than system available memory, worker server can be dispatched tasks. default value 0.3, the unit is G
|
||||
#worker.reserved.memory=0.3
|
||||
|
||||
# default worker groups separated by comma, like 'worker.groups=default,test'
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@
|
|||
<th scope="col" style="min-width: 135px">
|
||||
<span>{{$t('Executor')}}</span>
|
||||
</th>
|
||||
<th scope="col" style="max-width: 600px">
|
||||
<th scope="col" style="min-width: 490px">
|
||||
<span>{{$t('host')}}</span>
|
||||
</th>
|
||||
<th scope="col" style="min-width: 230px">
|
||||
|
|
@ -93,7 +93,7 @@
|
|||
<span v-if="item.executorName">{{item.executorName}}</span>
|
||||
<span v-else>-</span>
|
||||
</td>
|
||||
<td style="max-width: 600px"><span style="padding-right: 5px">{{item.host || '-'}}</span></td>
|
||||
<td style="min-width: 490px"><span style="padding-right: 10px">{{item.host || '-'}}</span></td>
|
||||
<td style="z-index: inherit;">
|
||||
<div v-show="item.disabled">
|
||||
<x-button type="info"
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@
|
|||
<th scope="col" style="min-width: 130px">
|
||||
<span>{{$t('End Time')}}</span>
|
||||
</th>
|
||||
<th scope="col" style="max-width: 600px">
|
||||
<th scope="col" style="min-width: 250px">
|
||||
<span>{{$t('host')}}</span>
|
||||
</th>
|
||||
<th scope="col" style="min-width: 70px">
|
||||
|
|
@ -87,7 +87,7 @@
|
|||
<span v-if="item.endTime">{{item.endTime | formatDate}}</span>
|
||||
<span v-else>-</span>
|
||||
</td>
|
||||
<td style="max-width: 600px"><span style="padding-right: 5px">{{item.host || '-'}}</span></td>
|
||||
<td style="min-width: 250px"><span style="padding-right: 10px">{{item.host || '-'}}</span></td>
|
||||
<td><span>{{item.duration || '-'}}</span></td>
|
||||
<td><span>{{item.retryTimes}}</span></td>
|
||||
<td>
|
||||
|
|
|
|||
|
|
@ -19,7 +19,8 @@
|
|||
ref="popup"
|
||||
:ok-text="item ? $t('Edit') : $t('Submit')"
|
||||
:nameText="item ? $t('Edit worker group') : $t('Create worker group')"
|
||||
@ok="_ok">
|
||||
@ok="_ok"
|
||||
style="width: 700px;">
|
||||
<template slot="content">
|
||||
<div class="create-worker-model">
|
||||
<m-list-box-f>
|
||||
|
|
|
|||
2
pom.xml
2
pom.xml
|
|
@ -91,7 +91,7 @@
|
|||
<postgresql.version>42.1.4</postgresql.version>
|
||||
<hive.jdbc.version>2.1.0</hive.jdbc.version>
|
||||
<commons.io.version>2.4</commons.io.version>
|
||||
<oshi.core.version>3.5.0</oshi.core.version>
|
||||
<oshi.core.version>3.9.1</oshi.core.version>
|
||||
<clickhouse.jdbc.version>0.1.52</clickhouse.jdbc.version>
|
||||
<mssql.jdbc.version>6.1.0.jre8</mssql.jdbc.version>
|
||||
<jsp-2.1.version>6.1.14</jsp-2.1.version>
|
||||
|
|
|
|||
|
|
@ -100,7 +100,7 @@ case $startStop in
|
|||
(start)
|
||||
if [ "$DOCKER" = "true" ]; then
|
||||
echo start $command in docker
|
||||
export DOLPHINSCHEDULER_OPTS="$DOLPHINSCHEDULER_OPTS -XX:+UseContainerSupport -XX:+UnlockExperimentalVMOptions -XX:+UseCGroupMemoryLimitForHeap"
|
||||
export DOLPHINSCHEDULER_OPTS="$DOLPHINSCHEDULER_OPTS -XX:-UseContainerSupport"
|
||||
exec_command="$LOG_FILE $DOLPHINSCHEDULER_OPTS -classpath $DOLPHINSCHEDULER_CONF_DIR:$DOLPHINSCHEDULER_LIB_JARS $CLASS"
|
||||
$JAVA_HOME/bin/java $exec_command
|
||||
else
|
||||
|
|
|
|||
|
|
@ -165,7 +165,7 @@ mybatis-spring-2.0.2.jar
|
|||
netty-3.6.2.Final.jar
|
||||
netty-all-4.1.53.Final.jar
|
||||
opencsv-2.3.jar
|
||||
oshi-core-3.5.0.jar
|
||||
oshi-core-3.9.1.jar
|
||||
paranamer-2.3.jar
|
||||
parquet-hadoop-bundle-1.8.1.jar
|
||||
poi-3.17.jar
|
||||
|
|
@ -211,7 +211,6 @@ swagger-models-1.5.24.jar
|
|||
skywalking-agent.jar
|
||||
spring-commons-8.4.0.jar
|
||||
tephra-api-0.6.0.jar
|
||||
threetenbp-1.3.6.jar
|
||||
transaction-api-1.1.jar
|
||||
validation-api-2.0.1.Final.jar
|
||||
xercesImpl-2.9.1.jar
|
||||
|
|
|
|||
Loading…
Reference in New Issue