Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[improve][broker] Support cgroup v2 by using jdk.internal.platform.Metrics in Pulsar Loadbalancer #16832

Merged
merged 9 commits into from
Apr 28, 2023
Merged

[improve][broker] Support cgroup v2 by using jdk.internal.platform.Metrics in Pulsar Loadbalancer #16832

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
1346100
Support cgroupV2 for by using `jdk.internal.platform.Metrics` in Puls…
coderzc Jul 28, 2022
10b4d9b
exports `jdk.internal.platform`
coderzc Jul 28, 2022
38bc31c
Use reflection to get metrics
coderzc Jul 28, 2022
c58105a
fall back get metrics by path
coderzc Aug 2, 2022
70b721d
Make sure really uses the jdk.internal.platform.Metrics
coderzc Aug 8, 2022
39062b8
Use `@VisibleForTesting` instead of reflection.
coderzc Aug 15, 2022
09297ed
improve somme comment
coderzc Aug 15, 2022
514cde7
Merge branch 'master' into cgroup_metrics
coderzc Apr 27, 2023
3ccdaba
fix code style
coderzc Apr 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bin/pulsar
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,8 @@ if [[ -z "$IS_JAVA_8" ]]; then
# netty.DnsResolverUtil
# JvmDefaultGCMetricsLogger
OPTS="$OPTS --add-opens java.base/sun.net=ALL-UNNAMED --add-opens java.management/sun.management=ALL-UNNAMED"
# LinuxInfoUtils
OPTS="$OPTS --add-opens java.base/jdk.internal.platform=ALL-UNNAMED"
fi

OPTS="-cp $PULSAR_CLASSPATH $OPTS"
Expand Down
1 change: 1 addition & 0 deletions buildtools/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@
<test.additional.args>
--add-opens java.base/jdk.internal.loader=ALL-UNNAMED
--add-opens java.base/java.lang=ALL-UNNAMED <!--Mockito-->
--add-opens java.base/jdk.internal.platform=ALL-UNNAMED <!--LinuxInfoUtils-->
</test.additional.args>
</properties>

Expand Down
1 change: 1 addition & 0 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ flexible messaging model and an intuitive client API.</description>
--add-opens java.base/java.io=ALL-UNNAMED <!--Bookkeeper NativeIO -->
--add-opens java.base/sun.net=ALL-UNNAMED <!--netty.DnsResolverUtil-->
--add-opens java.management/sun.management=ALL-UNNAMED <!--JvmDefaultGCMetricsLogger-->
--add-opens java.base/jdk.internal.platform=ALL-UNNAMED <!--LinuxInfoUtils-->
</test.additional.args>
<testReuseFork>true</testReuseFork>
<testForkCount>4</testForkCount>
Expand Down
66 changes: 57 additions & 9 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@
*/
package org.apache.pulsar.broker.loadbalance;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Charsets;
import java.io.IOException;
import java.lang.reflect.Method;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
Expand All @@ -44,13 +46,38 @@ 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";

// proc states
private static final String PROC_STAT_PATH = "/proc/stat";
private static final String NIC_PATH = "/sys/class/net/";
// NIC type
private static final int ARPHRD_ETHER = 1;
private static final String NIC_SPEED_TEMPLATE = "/sys/class/net/%s/speed";

private static Object /*jdk.internal.platform.Metrics*/ metrics;
private static Method getMetricsProviderMethod;
private static Method getCpuQuotaMethod;
private static Method getCpuPeriodMethod;
private static Method getCpuUsageMethod;

static {
try {
metrics = Class.forName("jdk.internal.platform.Container").getMethod("metrics")
.invoke(null);
if (metrics != null) {
getMetricsProviderMethod = metrics.getClass().getMethod("getProvider");
getMetricsProviderMethod.setAccessible(true);
getCpuQuotaMethod = metrics.getClass().getMethod("getCpuQuota");
getCpuQuotaMethod.setAccessible(true);
getCpuPeriodMethod = metrics.getClass().getMethod("getCpuPeriod");
getCpuPeriodMethod.setAccessible(true);
getCpuUsageMethod = metrics.getClass().getMethod("getCpuUsage");
getCpuUsageMethod.setAccessible(true);
}
} catch (Throwable e) {
log.warn("Failed to get runtime metrics", e);
}
}

/**
* Determine whether the OS is the linux kernel.
Expand All @@ -65,9 +92,14 @@ public static boolean isLinux() {
*/
public static boolean isCGroupEnabled() {
try {
return Files.exists(Paths.get(CGROUPS_CPU_USAGE_PATH));
if (metrics == null) {
return Files.exists(Paths.get(CGROUPS_CPU_USAGE_PATH));
}
String provider = (String) getMetricsProviderMethod.invoke(metrics);
log.info("[LinuxInfo] The system metrics provider is: {}", provider);
return provider.contains("cgroup");
} catch (Exception e) {
log.warn("[LinuxInfo] Failed to check cgroup CPU usage file: {}", e.getMessage());
log.warn("[LinuxInfo] Failed to check cgroup CPU: {}", e.getMessage());
return false;
}
}
Expand All @@ -80,13 +112,21 @@ public static boolean isCGroupEnabled() {
public static double getTotalCpuLimit(boolean isCGroupsEnabled) {
if (isCGroupsEnabled) {
try {
long quota = readLongFromFile(Paths.get(CGROUPS_CPU_LIMIT_QUOTA_PATH));
long period = readLongFromFile(Paths.get(CGROUPS_CPU_LIMIT_PERIOD_PATH));
long quota;
long period;
if (metrics != null && getCpuQuotaMethod != null && getCpuPeriodMethod != null) {
quota = (long) getCpuQuotaMethod.invoke(metrics);
period = (long) getCpuPeriodMethod.invoke(metrics);
} else {
quota = readLongFromFile(Paths.get(CGROUPS_CPU_LIMIT_QUOTA_PATH));
period = readLongFromFile(Paths.get(CGROUPS_CPU_LIMIT_PERIOD_PATH));
}

if (quota > 0) {
return 100.0 * quota / period;
}
} catch (IOException e) {
log.warn("[LinuxInfo] Failed to read CPU quotas from cgroups", e);
} catch (Exception e) {
log.warn("[LinuxInfo] Failed to read CPU quotas from cgroup", e);
// Fallback to availableProcessors
}
}
Expand All @@ -98,11 +138,14 @@ public static double getTotalCpuLimit(boolean isCGroupsEnabled) {
* Get CGroup cpu usage.
* @return Cpu usage
*/
public static double getCpuUsageForCGroup() {
public static long getCpuUsageForCGroup() {
try {
if (metrics != null && getCpuUsageMethod != null) {
return (long) getCpuUsageMethod.invoke(metrics);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For backward compatibility, don't we need to multiply the limit in calculateBrokerHostUsage?

public void calculateBrokerHostUsage() {
... 
       double totalCpuLimit = getTotalCpuLimit(isCGroupsEnabled);
        if (isCGroupsEnabled && metrics != null && getCpuUsageMethod != null) {
            // cgroup cpuUsage is already scaled, [0.0, 1.0]
            usage.setCpu(new ResourceUsage(cpuUsage * totalCpuLimit, totalCpuLimit));
        } else {
            usage.setCpu(new ResourceUsage(cpuUsage, totalCpuLimit));
        }

Copy link
Member Author

@coderzc coderzc Apr 28, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The Metrics.getCpuUsage will return the aggregate time, so I think we don't need to multiply the limit.

    /**
     * Returns the aggregate time, in nanoseconds, consumed by all
     * tasks in the Isolation Group.
     *
     * @return Time in nanoseconds, -1 if unknown or
     *         -2 if the metric is not supported.
     *
     */
    public long getCpuUsage();

Copy link
Contributor

@heesung-sn heesung-sn Apr 28, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see. Thanks for the check.

Ref:
https://github.com/openjdk/jdk17/blob/master/src/java.base/share/classes/jdk/internal/platform/Metrics.java

https://github.com/openjdk/jdk17/tree/master/src/java.base/linux/classes/jdk/internal/platform

}
return readLongFromFile(Paths.get(CGROUPS_CPU_USAGE_PATH));
} catch (IOException e) {
log.error("[LinuxInfo] Failed to read CPU usage from {}", CGROUPS_CPU_USAGE_PATH, e);
} catch (Exception e) {
log.error("[LinuxInfo] Failed to read CPU usage from cgroup", e);
return -1;
}
}
Expand Down Expand Up @@ -242,6 +285,11 @@ private static double readDoubleFromFile(Path path) throws IOException {
return Double.parseDouble(readTrimStringFromFile(path));
}

@VisibleForTesting
public static Object getMetrics() {
return metrics;
}

@AllArgsConstructor
public enum NICUsageType {
// transport
Expand Down
2 changes: 1 addition & 1 deletion ...ker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ private double getTotalCpuUsage(double elapsedTimeSeconds) {
}

private double getTotalCpuUsageForCGroup(double elapsedTimeSeconds) {
double usage = getCpuUsageForCGroup();
double usage = (double) getCpuUsageForCGroup();
double currentUsage = usage - lastCpuUsage;
lastCpuUsage = usage;
return 100 * currentUsage / elapsedTimeSeconds / TimeUnit.SECONDS.toNanos(1);
Expand Down
26 changes: 26 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/SimpleBrokerStartTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Optional;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -29,6 +31,7 @@
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
import org.testng.Assert;
import org.testng.annotations.Test;

@Slf4j
Expand Down Expand Up @@ -97,4 +100,27 @@ public void testNoNICSpeed() throws Exception {
}


@Test
public void testCGroupMetrics() {
if (!LinuxInfoUtils.isLinux()) {
return;
}

boolean existsCGroup = Files.exists(Paths.get("/sys/fs/cgroup"));
boolean cGroupEnabled = LinuxInfoUtils.isCGroupEnabled();
Assert.assertEquals(cGroupEnabled, existsCGroup);

double totalCpuLimit = LinuxInfoUtils.getTotalCpuLimit(cGroupEnabled);
log.info("totalCpuLimit: {}", totalCpuLimit);
Assert.assertTrue(totalCpuLimit > 0.0);

if (cGroupEnabled) {
Assert.assertNotNull(LinuxInfoUtils.getMetrics());

long cpuUsageForCGroup = LinuxInfoUtils.getCpuUsageForCGroup();
log.info("cpuUsageForCGroup: {}", cpuUsageForCGroup);
Assert.assertTrue(cpuUsageForCGroup > 0);
}
}

}