diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java index 61f34ef4901ba..b56ef0a417c20 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java @@ -47,6 +47,7 @@ public class LinuxInfoUtils { private static final String CGROUPS_CPU_USAGE_PATH = "/sys/fs/cgroup/cpu/cpuacct.usage"; private static final String CGROUPS_CPU_LIMIT_QUOTA_PATH = "/sys/fs/cgroup/cpu/cpu.cfs_quota_us"; private static final String CGROUPS_CPU_LIMIT_PERIOD_PATH = "/sys/fs/cgroup/cpu/cpu.cfs_period_us"; + private static final String CGROUPS_CPU_CPUSET_CPUS = "/sys/fs/cgroup/cpuset/cpuset.cpus"; // proc states private static final String PROC_STAT_PATH = "/proc/stat"; @@ -105,6 +106,26 @@ public static boolean isCGroupEnabled() { } } + /** + * Get total cpu count. + * @return Total cpu count + */ + public static int getTotalCpuCount() throws IOException { + int totalCpuCount = 0; + String[] ranges = readTrimStringFromFile(Paths.get(CGROUPS_CPU_CPUSET_CPUS)).split(","); + for (String range : ranges) { + if (!range.contains("-")) { + totalCpuCount++; + } else { + int dashIndex = range.indexOf('-'); + int left = Integer.parseInt(range.substring(0, dashIndex)); + int right = Integer.parseInt(range.substring(dashIndex + 1)); + totalCpuCount += right - left + 1; + } + } + return totalCpuCount; + } + /** * Get total cpu limit. * @param isCGroupsEnabled Whether CGroup is enabled @@ -126,6 +147,10 @@ public static double getTotalCpuLimit(boolean isCGroupsEnabled) { if (quota > 0) { return 100.0 * quota / period; } + int totalCpuCount = getTotalCpuCount(); + if (totalCpuCount > 0) { + return 100 * totalCpuCount; + } } catch (Exception e) { log.warn("[LinuxInfo] Failed to read CPU quotas from cgroup", e); // Fallback to availableProcessors @@ -300,11 +325,11 @@ private static Path getReplacedNICPath(String template, String nic) { return Paths.get(String.format(template, nic)); } - private static String readTrimStringFromFile(Path path) throws IOException { + public static String readTrimStringFromFile(Path path) throws IOException { return new String(Files.readAllBytes(path), StandardCharsets.UTF_8).trim(); } - private static long readLongFromFile(Path path) throws IOException { + public static long readLongFromFile(Path path) throws IOException { return Long.parseLong(readTrimStringFromFile(path)); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtilsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtilsTest.java index ac21b30bdde51..25d51b7ec9025 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtilsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtilsTest.java @@ -21,6 +21,8 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mockStatic; import static org.testng.Assert.assertEquals; + +import java.io.IOException; import java.nio.file.Files; import java.util.stream.Stream; import lombok.extern.slf4j.Slf4j; @@ -47,4 +49,26 @@ public void testGetCpuUsageForEntireHost(){ assertEquals(LinuxInfoUtils.getCpuUsageForEntireHost(), resourceUsage); } } + + /** + * simulate reading contents in /sys/fs/cgroup/cpuset/cpuset.cpus to get the number of Cpus + * and return the limit of cpu. + */ + @Test + public void testGetTotalCpuCountAndLimit() throws IOException { + try (MockedStatic linuxInfoUtils = mockStatic(LinuxInfoUtils.class)) { + linuxInfoUtils.when(() -> LinuxInfoUtils.readTrimStringFromFile(any())).thenReturn("0-2,16,20-30"); + linuxInfoUtils.when(() -> LinuxInfoUtils.getTotalCpuCount()).thenCallRealMethod(); + assertEquals(LinuxInfoUtils.getTotalCpuCount(), 15); + + // set quota to -1. + linuxInfoUtils.when(() -> LinuxInfoUtils.readLongFromFile(any())).thenReturn(-1L); + linuxInfoUtils.when(() -> LinuxInfoUtils.getTotalCpuLimit(true)).thenCallRealMethod(); + assertEquals(LinuxInfoUtils.getTotalCpuLimit(true), 1500); + + // invalid CGroup CPU settings test, fallback to JVM reported CPU quota. + linuxInfoUtils.when(() -> LinuxInfoUtils.readTrimStringFromFile(any())).thenReturn("0-a,1-30"); + assertEquals(LinuxInfoUtils.getTotalCpuLimit(true), 100 * Runtime.getRuntime().availableProcessors()); + } + } }