diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java index fc6c4116e0cbe..a3be6f86d8eed 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java @@ -30,6 +30,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Locale; import java.util.Optional; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -100,7 +101,7 @@ public SystemResourceUsage getBrokerHostUsage() { @Override public void calculateBrokerHostUsage() { - List nics = getNics(); + List nics = getUsablePhysicalNICs(); double totalNicLimit = getTotalNicLimitKbps(nics); double totalNicUsageTx = getTotalNicUsageTxKb(nics); double totalNicUsageRx = getTotalNicUsageRxKb(nics); @@ -212,9 +213,11 @@ private ResourceUsage getMemUsage() { return new ResourceUsage(total - free, total); } - private List getNics() { + private List getUsablePhysicalNICs() { try (Stream stream = Files.list(Paths.get("/sys/class/net/"))) { - return stream.filter(this::isPhysicalNic).map(path -> path.getFileName().toString()) + return stream.filter(this::isPhysicalNic) + .filter(this::isUsable) + .map(path -> path.getFileName().toString()) .collect(Collectors.toList()); } catch (IOException e) { log.error("Failed to find NICs", e); @@ -223,7 +226,7 @@ private List getNics() { } public int getNicCount() { - return getNics().size(); + return getUsablePhysicalNICs().size(); } private boolean isPhysicalNic(Path path) { @@ -241,6 +244,25 @@ private boolean isPhysicalNic(Path path) { } } + private boolean isUsable(Path nicPath) { + try { + String operstate = readTrimStringFromFile(nicPath.resolve("operstate")); + Operstate operState = Operstate.valueOf(operstate.toUpperCase(Locale.ROOT)); + switch (operState) { + case UP: + case UNKNOWN: + case DORMANT: + return true; + default: + return false; + } + } catch (Exception e) { + log.warn("[LinuxInfo] Failed to read {} NIC operstate, the detail is: {}", nicPath, e.getMessage()); + // Read operstate got error. + return false; + } + } + private Path getNicSpeedPath(String nic) { return Paths.get(String.format("/sys/class/net/%s/speed", nic)); } @@ -294,4 +316,32 @@ private double getTotalNicUsageTxKb(List nics) { private static long readLongFromFile(String path) throws IOException { return Long.parseLong(new String(Files.readAllBytes(Paths.get(path)), Charsets.UTF_8).trim()); } + + private static String readTrimStringFromFile(Path path) throws IOException { + return new String(Files.readAllBytes(path), StandardCharsets.UTF_8).trim(); + } + + + /** + * TLV IFLA_OPERSTATE + * contains RFC2863 state of the interface in numeric representation: + * See ... + */ + enum Operstate { + // Interface is in unknown state, neither driver nor userspace has set + // operational state. Interface must be considered for user data as + // setting operational state has not been implemented in every driver. + UNKNOWN, + // Interface is unable to transfer data on L1, f.e. ethernet is not + // plugged or interface is ADMIN down. + DOWN, + // Interfaces stacked on an interface that is IF_OPER_DOWN show this + // state (f.e. VLAN). + LOWERLAYERDOWN, + // Interface is L1 up, but waiting for an external event, f.e. for a + // protocol to establish. (802.1X) + DORMANT, + // Interface is operational up and can be used. + UP + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadReportNetworkLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadReportNetworkLimitTest.java index 2d9dc1abb0efc..15dadd89dd12f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadReportNetworkLimitTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadReportNetworkLimitTest.java @@ -29,7 +29,7 @@ @Test(groups = "broker") public class LoadReportNetworkLimitTest extends MockedPulsarServiceBaseTest { - int nicCount; + int usableNicCount; @BeforeClass @Override @@ -39,7 +39,7 @@ public void setup() throws Exception { super.internalSetup(); if (SystemUtils.IS_OS_LINUX) { - nicCount = new LinuxBrokerHostUsageImpl(pulsar).getNicCount(); + usableNicCount = new LinuxBrokerHostUsageImpl(pulsar).getNicCount(); } } @@ -56,8 +56,8 @@ public void checkLoadReportNicSpeed() throws Exception { LoadManagerReport report = admin.brokerStats().getLoadReport(); if (SystemUtils.IS_OS_LINUX) { - assertEquals(report.getBandwidthIn().limit, nicCount * 5.4 * 1000 * 1000); - assertEquals(report.getBandwidthOut().limit, nicCount * 5.4 * 1000 * 1000); + assertEquals(report.getBandwidthIn().limit, usableNicCount * 5.4 * 1000 * 1000); + assertEquals(report.getBandwidthOut().limit, usableNicCount * 5.4 * 1000 * 1000); } else { // On non-Linux system we don't report the network usage assertEquals(report.getBandwidthIn().limit, -1.0);