This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 321e21d5270 KAFKA-20223 Extended I/O Metrics Collection (#22454)
321e21d5270 is described below
commit 321e21d52708333a13bffb3a8cff253ef1158065
Author: Sahil Devgon <[email protected]>
AuthorDate: Tue Jun 30 22:02:42 2026 +0530
KAFKA-20223 Extended I/O Metrics Collection (#22454)
Extend the existing I/O metrics collection with the 5 missing metrics
(rchar, wchar, syscr, syscw, and cancelled_write_bytes) that are
essential for diagnosing performance issues and optimising Kafka
deployments.
Details in the KIP below
https: //cwiki.apache.org/confluence/x/co48G
Reviewers: Ken Huang <[email protected]>, PoAn Yang
<[email protected]>, Chia-Ping Tsai <[email protected]>
---
.../main/scala/kafka/server/ControllerServer.scala | 3 +-
core/src/main/scala/kafka/server/KafkaBroker.scala | 3 +-
.../scala/unit/kafka/metrics/MetricsTest.scala | 12 +-
.../server/metrics/LinuxIoMetricsCollector.java | 179 +++++++++++++++++----
.../metrics/LinuxIoMetricsCollectorTest.java | 59 ++++++-
5 files changed, 211 insertions(+), 45 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala
b/core/src/main/scala/kafka/server/ControllerServer.scala
index d2a24fe0bed..56f036f090f 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/core/src/main/scala/kafka/server/ControllerServer.scala
@@ -143,8 +143,7 @@ class ControllerServer(
linuxIoMetricsCollector = new LinuxIoMetricsCollector("/proc", time)
if (linuxIoMetricsCollector.usable()) {
- metricsGroup.newGauge("linux-disk-read-bytes", () =>
linuxIoMetricsCollector.readBytes())
- metricsGroup.newGauge("linux-disk-write-bytes", () =>
linuxIoMetricsCollector.writeBytes())
+ linuxIoMetricsCollector.registerMetrics(metricsGroup)
}
authorizerPlugin = config.createNewAuthorizer(metrics,
ProcessRole.ControllerRole.toString)
diff --git a/core/src/main/scala/kafka/server/KafkaBroker.scala
b/core/src/main/scala/kafka/server/KafkaBroker.scala
index 7b13bd93d3d..55bf1323029 100644
--- a/core/src/main/scala/kafka/server/KafkaBroker.scala
+++ b/core/src/main/scala/kafka/server/KafkaBroker.scala
@@ -118,7 +118,6 @@ trait KafkaBroker extends Logging {
private val linuxIoMetricsCollector = new LinuxIoMetricsCollector("/proc",
Time.SYSTEM)
if (linuxIoMetricsCollector.usable()) {
- metricsGroup.newGauge("linux-disk-read-bytes", () =>
linuxIoMetricsCollector.readBytes())
- metricsGroup.newGauge("linux-disk-write-bytes", () =>
linuxIoMetricsCollector.writeBytes())
+ linuxIoMetricsCollector.registerMetrics(metricsGroup)
}
}
diff --git a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
index 64b506e0065..df85330f431 100644
--- a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
+++ b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
@@ -108,12 +108,20 @@ class MetricsTest extends KafkaServerTestHarness with
Logging {
@Test
def testLinuxIoMetrics(): Unit = {
- // Check if linux-disk-{read,write}-bytes metrics either do or do not
exist depending on whether we are or are not
+ // Check if the linux-disk-* I/O metrics either do or do not exist
depending on whether we are or are not
// able to collect those metrics on the platform where this test is
running.
val usable = new LinuxIoMetricsCollector("/proc", Time.SYSTEM).usable()
val expectedCount = if (usable) 1 else 0
val metrics = KafkaYammerMetrics.defaultRegistry.allMetrics
- Set("linux-disk-read-bytes", "linux-disk-write-bytes").foreach(name =>
+ Set(
+ "linux-disk-read-bytes",
+ "linux-disk-write-bytes",
+ "linux-disk-rchar",
+ "linux-disk-wchar",
+ "linux-disk-syscr",
+ "linux-disk-syscw",
+ "linux-disk-cancelled-write-bytes"
+ ).foreach(name =>
assertEquals(metrics.keySet.asScala.count(_.getMBeanName ==
s"$requiredKafkaServerPrefix=$name"), expectedCount))
}
diff --git
a/server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java
b/server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java
index e34a4d5c426..6df4b41aece 100644
---
a/server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java
+++
b/server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java
@@ -26,6 +26,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* Retrieves Linux /proc/self/io metrics.
@@ -35,13 +36,24 @@ public class LinuxIoMetricsCollector {
private static final Logger LOG =
LoggerFactory.getLogger(LinuxIoMetricsCollector.class);
private static final String READ_BYTES_PREFIX = "read_bytes: ";
private static final String WRITE_BYTES_PREFIX = "write_bytes: ";
+ private static final String RCHAR_PREFIX = "rchar: ";
+ private static final String WCHAR_PREFIX = "wchar: ";
+ private static final String SYSCR_PREFIX = "syscr: ";
+ private static final String SYSCW_PREFIX = "syscw: ";
+ private static final String CANCELLED_WRITE_BYTES_PREFIX =
"cancelled_write_bytes: ";
private final Time time;
private final Path path;
+ private final AtomicBoolean isUpdating = new AtomicBoolean(false);
- private long lastUpdateMs = -1L;
- private long cachedReadBytes = 0L;
- private long cachedWriteBytes = 0L;
+ private volatile long lastUpdateMs = -1L;
+ private volatile long cachedReadBytes = 0L;
+ private volatile long cachedWriteBytes = 0L;
+ private volatile long cachedRchar = 0L;
+ private volatile long cachedWchar = 0L;
+ private volatile long cachedSyscr = 0L;
+ private volatile long cachedSyscw = 0L;
+ private volatile long cachedCancelledWriteBytes = 0L;
public LinuxIoMetricsCollector(String procRoot, Time time) {
this.time = time;
@@ -49,22 +61,94 @@ public class LinuxIoMetricsCollector {
}
public long readBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
- }
- return cachedReadBytes;
- }
+ refreshIfStale();
+ return cachedReadBytes;
}
public long writeBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
+ refreshIfStale();
+ return cachedWriteBytes;
+ }
+
+ /**
+ * Returns the total number of characters read (includes cached reads).
+ * This value represents all read operations, including those satisfied by
the page cache.
+ */
+ public long rchar() {
+ refreshIfStale();
+ return cachedRchar;
+ }
+
+ /**
+ * Returns the total number of characters written (includes cached writes).
+ * This value represents all write operations, including those that may
not have reached disk.
+ */
+ public long wchar() {
+ refreshIfStale();
+ return cachedWchar;
+ }
+
+ /**
+ * Returns the number of read system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscr() {
+ refreshIfStale();
+ return cachedSyscr;
+ }
+
+ /**
+ * Returns the number of write system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscw() {
+ refreshIfStale();
+ return cachedSyscw;
+ }
+
+ /**
+ * Returns the number of bytes that were cancelled before being written.
+ * This can occur when a write is truncated or cancelled.
+ */
+ public long cancelledWriteBytes() {
+ refreshIfStale();
+ return cachedCancelledWriteBytes;
+ }
+
+ /**
+ * Registers all 7 Linux I/O metrics with the given metrics group. Should
be called only
+ * after {@link #usable()} returns true.
+ */
+ public void registerMetrics(KafkaMetricsGroup metricsGroup) {
+ metricsGroup.newGauge("linux-disk-read-bytes", this::readBytes);
+ metricsGroup.newGauge("linux-disk-write-bytes", this::writeBytes);
+ metricsGroup.newGauge("linux-disk-rchar", this::rchar);
+ metricsGroup.newGauge("linux-disk-wchar", this::wchar);
+ metricsGroup.newGauge("linux-disk-syscr", this::syscr);
+ metricsGroup.newGauge("linux-disk-syscw", this::syscw);
+ metricsGroup.newGauge("linux-disk-cancelled-write-bytes",
this::cancelledWriteBytes);
+ }
+
+ /**
+ * Refreshes all cached values from /proc/self/io if more than one
millisecond has elapsed
+ * since the last refresh. Uses an {@link AtomicBoolean} CAS to coordinate
so that only
+ * one thread performs the update at a time; other concurrent callers fall
through and
+ * read whatever is currently cached. This is safe because {@link
#updateValues} never
+ * publishes intermediate {@code -1L} state to the cached fields.
+ */
+ private void refreshIfStale() {
+ long currentMs = time.milliseconds();
+ if (currentMs != lastUpdateMs) {
+ if (isUpdating.compareAndSet(false, true)) {
+ try {
+ currentMs = time.milliseconds();
+ if (currentMs != lastUpdateMs) {
+ updateValues(currentMs);
+ }
+ } finally {
+ isUpdating.set(false);
+ }
}
- return cachedWriteBytes;
}
}
@@ -79,27 +163,60 @@ public class LinuxIoMetricsCollector {
* read_bytes: 0
* write_bytes: 0
* cancelled_write_bytes: 0
+ *
+ * <p>Parses into local variables first, then publishes to the volatile
cached fields only
+ * after parsing succeeds. This prevents lock-free readers from observing
the transient
+ * -1L state during an update. {@code lastUpdateMs} is written last on the
success path so
+ * any thread observing the new timestamp via a volatile read also
observes all updated
+ * cached fields.
+ *
+ * <p>On parse/IO failure, the cached fields are left untouched (readers
retain the
+ * last-known-good values) but {@code lastUpdateMs} is still updated to
{@code now} so that
+ * subsequent getters within the same ms-window do not re-trigger the
failed read and
+ * re-log the warning. This bounds log output to at most one warning per
ms-window
+ * (i.e. once per JMX scrape) when the file is persistently unreadable.
*/
private boolean updateValues(long now) {
- synchronized (this) {
- try {
- cachedReadBytes = -1L;
- cachedWriteBytes = -1L;
- List<String> lines = Files.readAllLines(path,
StandardCharsets.UTF_8);
- for (String line : lines) {
- if (line.startsWith(READ_BYTES_PREFIX)) {
- cachedReadBytes =
Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
- } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
- cachedWriteBytes =
Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
- }
+ long readBytes = -1L;
+ long writeBytes = -1L;
+ long rchar = -1L;
+ long wchar = -1L;
+ long syscr = -1L;
+ long syscw = -1L;
+ long cancelledWriteBytes = -1L;
+ try {
+ List<String> lines = Files.readAllLines(path,
StandardCharsets.UTF_8);
+ for (String line : lines) {
+ if (line.startsWith(READ_BYTES_PREFIX)) {
+ readBytes =
Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
+ } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
+ writeBytes =
Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
+ } else if (line.startsWith(RCHAR_PREFIX)) {
+ rchar =
Long.parseLong(line.substring(RCHAR_PREFIX.length()));
+ } else if (line.startsWith(WCHAR_PREFIX)) {
+ wchar =
Long.parseLong(line.substring(WCHAR_PREFIX.length()));
+ } else if (line.startsWith(SYSCR_PREFIX)) {
+ syscr =
Long.parseLong(line.substring(SYSCR_PREFIX.length()));
+ } else if (line.startsWith(SYSCW_PREFIX)) {
+ syscw =
Long.parseLong(line.substring(SYSCW_PREFIX.length()));
+ } else if (line.startsWith(CANCELLED_WRITE_BYTES_PREFIX)) {
+ cancelledWriteBytes =
Long.parseLong(line.substring(CANCELLED_WRITE_BYTES_PREFIX.length()));
}
- lastUpdateMs = now;
- return true;
- } catch (Throwable t) {
- LOG.warn("Unable to update IO metrics", t);
- return false;
}
+ } catch (Throwable t) {
+ LOG.warn("Unable to update IO metrics", t);
+ lastUpdateMs = now;
+ return false;
}
+ cachedReadBytes = readBytes;
+ cachedWriteBytes = writeBytes;
+ cachedRchar = rchar;
+ cachedWchar = wchar;
+ cachedSyscr = syscr;
+ cachedSyscw = syscw;
+ cachedCancelledWriteBytes = cancelledWriteBytes;
+ lastUpdateMs = now;
+ return true;
}
public boolean usable() {
diff --git
a/server/src/test/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollectorTest.java
b/server/src/test/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollectorTest.java
index 8c96c5c5c10..3a13a143278 100644
---
a/server/src/test/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollectorTest.java
+++
b/server/src/test/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollectorTest.java
@@ -39,23 +39,65 @@ public class LinuxIoMetricsCollectorTest {
public void testReadProcFile() throws IOException {
TestDirectory testDirectory = new TestDirectory();
Time time = new MockTime(0L, 100L, 1000L);
- testDirectory.writeProcFile(123L, 456L);
+ testDirectory.writeProcFile(123L, 456L, 1000L, 2000L, 10L, 20L, 5L);
LinuxIoMetricsCollector collector = new
LinuxIoMetricsCollector(testDirectory.baseDir.getAbsolutePath(), time);
// Test that we can read the values we wrote.
assertTrue(collector.usable());
assertEquals(123L, collector.readBytes());
assertEquals(456L, collector.writeBytes());
- testDirectory.writeProcFile(124L, 457L);
+ assertEquals(1000L, collector.rchar());
+ assertEquals(2000L, collector.wchar());
+ assertEquals(10L, collector.syscr());
+ assertEquals(20L, collector.syscw());
+ assertEquals(5L, collector.cancelledWriteBytes());
+ testDirectory.writeProcFile(124L, 457L, 1001L, 2001L, 11L, 21L, 6L);
// The previous values should still be cached.
assertEquals(123L, collector.readBytes());
assertEquals(456L, collector.writeBytes());
+ assertEquals(1000L, collector.rchar());
+ assertEquals(2000L, collector.wchar());
+ assertEquals(10L, collector.syscr());
+ assertEquals(20L, collector.syscw());
+ assertEquals(5L, collector.cancelledWriteBytes());
// Update the time, and the values should be re-read.
time.sleep(1);
assertEquals(124L, collector.readBytes());
assertEquals(457L, collector.writeBytes());
+ assertEquals(1001L, collector.rchar());
+ assertEquals(2001L, collector.wchar());
+ assertEquals(11L, collector.syscr());
+ assertEquals(21L, collector.syscw());
+ assertEquals(6L, collector.cancelledWriteBytes());
+ }
+
+ @Test
+ public void testAllMetricsWithRealWorldValues() throws IOException {
+ TestDirectory testDirectory = new TestDirectory();
+ Time time = new MockTime(0L, 100L, 1000L);
+ // Simulate real-world values where rchar/wchar are much larger than
read_bytes/write_bytes
+ // (due to page cache hits)
+ testDirectory.writeProcFile(
+ 1048576L, // read_bytes: 1 MB actually read from disk
+ 2097152L, // write_bytes: 2 MB actually written to disk
+ 10485760L, // rchar: 10 MB total reads (9 MB from cache)
+ 20971520L, // wchar: 20 MB total writes (18 MB cached)
+ 150L, // syscr: 150 read syscalls
+ 300L, // syscw: 300 write syscalls
+ 524288L // cancelled_write_bytes: 512 KB cancelled
+ );
+ LinuxIoMetricsCollector collector = new
LinuxIoMetricsCollector(testDirectory.baseDir.getAbsolutePath(), time);
+
+ assertTrue(collector.usable());
+ assertEquals(1048576L, collector.readBytes());
+ assertEquals(2097152L, collector.writeBytes());
+ assertEquals(10485760L, collector.rchar());
+ assertEquals(20971520L, collector.wchar());
+ assertEquals(150L, collector.syscr());
+ assertEquals(300L, collector.syscw());
+ assertEquals(524288L, collector.cancelledWriteBytes());
}
@Test
@@ -78,14 +120,15 @@ public class LinuxIoMetricsCollectorTest {
selfDir =
Files.createDirectories(baseDir.toPath().resolve("self"));
}
- void writeProcFile(long readBytes, long writeBytes) throws IOException
{
- String bld = "rchar: 0\n" +
- "wchar: 0\n" +
- "syschr: 0\n" +
- "syscw: 0\n" +
+ void writeProcFile(long readBytes, long writeBytes, long rchar, long
wchar,
+ long syscr, long syscw, long cancelledWriteBytes)
throws IOException {
+ String bld = "rchar: " + rchar + "\n" +
+ "wchar: " + wchar + "\n" +
+ "syscr: " + syscr + "\n" +
+ "syscw: " + syscw + "\n" +
"read_bytes: " + readBytes + "\n" +
"write_bytes: " + writeBytes + "\n" +
- "cancelled_write_bytes: 0\n";
+ "cancelled_write_bytes: " + cancelledWriteBytes +
"\n";
Files.writeString(selfDir.resolve("io"), bld);
}
}
