natas_hw
/
dolphinscheduler0701
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add log for NetUtils (#14578)

This commit is contained in:
Wenjun Ruan 2023-07-18 21:17:17 +08:00 committed by GitHub
parent c0d776381c
commit 399d0d0ce9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 74 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/config/AlertConfig.java
View File

@ -19,6 +19,8 @@ package org.apache.dolphinscheduler.alert.config;
import org.apache.dolphinscheduler.common.utils.NetUtils;
import org.apache.commons.lang3.StringUtils;
import java.time.Duration;
import lombok.Data;
@ -56,7 +58,10 @@ public final class AlertConfig implements Validator {
errors.rejectValue("heartbeat-interval", null, "should be a valid duration");
}
alertConfig.setAlertServerAddress(NetUtils.getAddr(port));
if (StringUtils.isEmpty(alertServerAddress)) {
alertConfig.setAlertServerAddress(NetUtils.getAddr(alertConfig.getPort()));
}
printConfig();
}

101
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/NetUtils.java
View File

@ -121,9 +121,9 @@ public class NetUtils {
InetAddress localAddress = null;
try {
NetworkInterface networkInterface = findNetworkInterface();
if (networkInterface != null) {
Enumeration<InetAddress> addresses = networkInterface.getInetAddresses();
Optional<NetworkInterface> networkInterface = findNetworkInterface();
if (networkInterface.isPresent()) {
Enumeration<InetAddress> addresses = networkInterface.get().getInetAddresses();
while (addresses.hasMoreElements()) {
Optional<InetAddress> addressOp = toValidAddress(addresses.nextElement());
if (addressOp.isPresent()) {
@ -137,8 +137,8 @@ public class NetUtils {
}
}
}
}
}
localAddress = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
log.warn("InetAddress get LocalHost exception", e);
@ -154,12 +154,15 @@ public class NetUtils {
if (address instanceof Inet6Address) {
Inet6Address v6Address = (Inet6Address) address;
if (isPreferIPV6Address()) {
return Optional.ofNullable(normalizeV6Address(v6Address));
InetAddress inetAddress = normalizeV6Address(v6Address);
log.debug("The host prefer ipv6 address, will use ipv6 address: {} directly", inetAddress);
return Optional.ofNullable(inetAddress);
}
}
if (isValidV4Address(address)) {
return Optional.of(address);
}
log.warn("The address of the host is invalid, address: {}", address);
return Optional.empty();
}
@ -202,7 +205,7 @@ public class NetUtils {
*
* @return If no {@link NetworkInterface} is available , return <code>null</code>
*/
private static NetworkInterface findNetworkInterface() {
private static Optional<NetworkInterface> findNetworkInterface() {
List<NetworkInterface> validNetworkInterfaces = emptyList();
@ -211,19 +214,19 @@ public class NetUtils {
} catch (SocketException e) {
log.warn("ValidNetworkInterfaces exception", e);
}
if (CollectionUtils.isEmpty(validNetworkInterfaces)) {
log.warn("ValidNetworkInterfaces is empty");
return Optional.empty();
}
NetworkInterface result = null;
// Try to specify config NetWork Interface
for (NetworkInterface networkInterface : validNetworkInterfaces) {
if (isSpecifyNetworkInterface(networkInterface)) {
result = networkInterface;
break;
}
Optional<NetworkInterface> specifyNetworkInterface =
validNetworkInterfaces.stream().filter(NetUtils::isSpecifyNetworkInterface).findFirst();
if (specifyNetworkInterface.isPresent()) {
log.info("Use specified NetworkInterface: {}", specifyNetworkInterface.get());
return specifyNetworkInterface;
}
if (null != result) {
return result;
}
return findAddress(validNetworkInterfaces);
}
@ -239,8 +242,10 @@ public class NetUtils {
NetworkInterface networkInterface = interfaces.nextElement();
// ignore
if (ignoreNetworkInterface(networkInterface)) {
log.debug("Info NetworkInterface: {}", networkInterface);
continue;
}
log.info("Found valid NetworkInterface: {}", networkInterface);
validNetworkInterfaces.add(networkInterface);
}
return validNetworkInterfaces;
@ -265,34 +270,40 @@ public class NetUtils {
return Objects.equals(networkInterface.getDisplayName(), preferredNetworkInterface);
}
private static NetworkInterface findAddress(List<NetworkInterface> validNetworkInterfaces) {
private static Optional<NetworkInterface> findAddress(List<NetworkInterface> validNetworkInterfaces) {
if (CollectionUtils.isEmpty(validNetworkInterfaces)) {
return null;
return Optional.empty();
}
String networkPriority = PropertyUtils.getString(Constants.DOLPHIN_SCHEDULER_NETWORK_PRIORITY_STRATEGY,
NETWORK_PRIORITY_DEFAULT);
if (NETWORK_PRIORITY_DEFAULT.equalsIgnoreCase(networkPriority)) {
return findAddressByDefaultPolicy(validNetworkInterfaces);
} else if (NETWORK_PRIORITY_INNER.equalsIgnoreCase(networkPriority)) {
return findInnerAddress(validNetworkInterfaces);
} else if (NETWORK_PRIORITY_OUTER.equalsIgnoreCase(networkPriority)) {
return findOuterAddress(validNetworkInterfaces);
} else {
log.error("There is no matching network card acquisition policy!");
return null;
switch (networkPriority) {
case NETWORK_PRIORITY_DEFAULT:
log.debug("Use default NetworkInterface acquisition policy");
return findAddressByDefaultPolicy(validNetworkInterfaces);
case NETWORK_PRIORITY_INNER:
log.debug("Use inner NetworkInterface acquisition policy");
return findInnerAddress(validNetworkInterfaces);
case NETWORK_PRIORITY_OUTER:
log.debug("Use outer NetworkInterface acquisition policy");
return findOuterAddress(validNetworkInterfaces);
default:
log.error("There is no matching network card acquisition policy!");
return Optional.empty();
}
}
private static NetworkInterface findAddressByDefaultPolicy(List<NetworkInterface> validNetworkInterfaces) {
NetworkInterface networkInterface;
networkInterface = findInnerAddress(validNetworkInterfaces);
if (networkInterface == null) {
networkInterface = findOuterAddress(validNetworkInterfaces);
if (networkInterface == null) {
networkInterface = validNetworkInterfaces.get(0);
}
private static Optional<NetworkInterface> findAddressByDefaultPolicy(List<NetworkInterface> validNetworkInterfaces) {
Optional<NetworkInterface> innerAddress = findInnerAddress(validNetworkInterfaces);
if (innerAddress.isPresent()) {
log.debug("Found inner NetworkInterface: {}", innerAddress.get());
return innerAddress;
}
return networkInterface;
Optional<NetworkInterface> outerAddress = findOuterAddress(validNetworkInterfaces);
if (outerAddress.isPresent()) {
log.debug("Found outer NetworkInterface: {}", outerAddress.get());
return outerAddress;
}
return Optional.empty();
}
/**
@ -300,35 +311,39 @@ public class NetUtils {
*
* @return If no {@link NetworkInterface} is available , return <code>null</code>
*/
private static NetworkInterface findInnerAddress(List<NetworkInterface> validNetworkInterfaces) {
private static Optional<NetworkInterface> findInnerAddress(List<NetworkInterface> validNetworkInterfaces) {
if (CollectionUtils.isEmpty(validNetworkInterfaces)) {
return Optional.empty();
}
NetworkInterface networkInterface = null;
for (NetworkInterface ni : validNetworkInterfaces) {
Enumeration<InetAddress> address = ni.getInetAddresses();
while (address.hasMoreElements()) {
InetAddress ip = address.nextElement();
if (ip.isSiteLocalAddress()
&& !ip.isLoopbackAddress()) {
networkInterface = ni;
return Optional.of(ni);
}
}
}
return networkInterface;
return Optional.empty();
}
private static NetworkInterface findOuterAddress(List<NetworkInterface> validNetworkInterfaces) {
NetworkInterface networkInterface = null;
private static Optional<NetworkInterface> findOuterAddress(List<NetworkInterface> validNetworkInterfaces) {
if (CollectionUtils.isEmpty(validNetworkInterfaces)) {
return Optional.empty();
}
for (NetworkInterface ni : validNetworkInterfaces) {
Enumeration<InetAddress> address = ni.getInetAddresses();
while (address.hasMoreElements()) {
InetAddress ip = address.nextElement();
if (!ip.isSiteLocalAddress()
&& !ip.isLoopbackAddress()) {
networkInterface = ni;
return Optional.of(ni);
}
}
}
return networkInterface;
return Optional.empty();
}
}

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

@ -26,6 +26,8 @@ import org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostSelect
import org.apache.dolphinscheduler.server.master.processor.queue.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import org.apache.commons.lang3.StringUtils;
import java.time.Duration;
import lombok.Data;
@ -157,8 +159,10 @@ public class MasterConfig implements Validator {
if (masterConfig.getWorkerGroupRefreshInterval().getSeconds() < 10) {
errors.rejectValue("worker-group-refresh-interval", null, "should >= 10s");
}
if (StringUtils.isEmpty(masterConfig.getMasterAddress())) {
masterConfig.setMasterAddress(NetUtils.getAddr(masterConfig.getListenPort()));
}
masterConfig.setMasterAddress(NetUtils.getAddr(masterConfig.getListenPort()));
masterConfig.setMasterRegistryPath(
RegistryNodeType.MASTER.getRegistryPath() + "/" + masterConfig.getMasterAddress());
printConfig();

6
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
View File

@ -24,6 +24,8 @@ import org.apache.dolphinscheduler.registry.api.ConnectStrategyProperties;
import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.apache.commons.lang3.StringUtils;
import java.time.Duration;
import lombok.Data;
@ -80,7 +82,9 @@ public class WorkerConfig implements Validator {
if (workerConfig.getMaxCpuLoadAvg() <= 0) {
workerConfig.setMaxCpuLoadAvg(Runtime.getRuntime().availableProcessors() * 2);
}
workerConfig.setWorkerAddress(NetUtils.getAddr(workerConfig.getListenPort()));
if (StringUtils.isEmpty(workerConfig.getWorkerAddress())) {
workerConfig.setWorkerAddress(NetUtils.getAddr(workerConfig.getListenPort()));
}
workerConfig.setWorkerRegistryPath(REGISTRY_DOLPHINSCHEDULER_WORKERS + "/" + workerConfig.getWorkerAddress());
printConfig();