[fix] [broker] incorrect CPU usage collected by pulsar

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));
     }
 

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> 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());
+        }
+    }
 }