From 7c5dfe6babfd38e43ba0ae9f2fd88aa6d829b653 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Mon, 26 Oct 2026 13:42:49 +0800 Subject: [PATCH 1/4] [jindo] Log JVM-wide Jindo stream and memory statistics Compaction TaskManagers reading through paimon-jindo are OOMKilled by the container while the Java heap is far from full: JindoSDK keeps a readahead cache of tens of MiB per open read stream outside the JVM, and glibc keeps most of it resident after the streams close. Nothing in the logs tells how many streams were open or how much native memory Jindo held when the container died. JindoIOStats keeps JVM-wide counters of the streams opened through JindoFileIO (open, peak since the previous line, opened, closed, bytes). When jindo.io-stats.interval is set, a daemon thread logs one INFO line per interval while streams are open or have opened/closed since the previous line: Jindo IO stats: input open=16 peak=16 opened=16 closed=0 read=384MiB; output open=0 ...; jindo prefetchCache=346MiB/4423MiB readStreams=16 writeStreams=0 memInsufficient=0; process rss=1003MiB anon=939MiB file=63MiB shmem=0MiB threads=42 heap=69MiB/494MiB/494MiB direct=0MiB metaspace=13MiB jemalloc=false; cgroup usage=943MiB/1100MiB workingSet=943MiB peak=943MiB limitHits=0 oomKills=0 The jindo section comes from the native metrics dump (STATIC_ONLY JSON, about 0.4 ms), read only through cached file systems, which are never closed. The thread also samples the memory cgroup (v1 or v2) every second and logs the same line at WARN when the working set (usage minus inactive file) first crosses 80%, 90% and 95% of the limit, resetting below 70%, so the last line before an OOM kill shows what was open. The reporter is off by default (interval 0), to be turned on, for example with 1 min, for jobs that keep getting OOMKilled. A sample costs about 50 us and a full line about 0.2 ms plus the dump. Verified on JDK 11 against OSS with jindosdk 6.10.2: 16 streams in a 960 MiB memcg logged "working set is at 90% of the limit" with prefetchCache=308MiB right before the kernel OOM-killed the JVM; after all streams closed, prefetchCache dropped to 0 MiB while anon stayed at 934 MiB, which is glibc retention rather than Jindo. --- .../paimon/jindo/HadoopCompliantFileIO.java | 48 +- .../org/apache/paimon/jindo/JindoFileIO.java | 24 +- .../org/apache/paimon/jindo/JindoIOStats.java | 546 ++++++++++++++++++ .../apache/paimon/jindo/JindoIOStatsTest.java | 293 ++++++++++ 4 files changed, 902 insertions(+), 9 deletions(-) create mode 100644 paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java create mode 100644 paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java index 64fc3e07c708..b9822a4bfff9 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java @@ -133,6 +133,7 @@ public SeekableInputStream newInputStream(Path path) throws IOException { JindoHadoopSystem fs = pair.getKey(); String sysType = pair.getValue(); FSDataInputStream fsInput = fs.open(hadoopPath); + JindoIOStats.INPUT.opened(); return "jobj".equalsIgnoreCase(sysType) ? new VectoredReadableInputStream(fsInput) : new HadoopSeekableInputStream(fsInput); @@ -143,8 +144,10 @@ public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws org.apache.hadoop.fs.Path hadoopPath = path(path); boolean shouldCache = writeCacheEnabled && shouldCache(path); LOG.debug("OutputStream should cache {} for path {}", shouldCache, path); - return new HadoopPositionOutputStream( - getFileSystem(hadoopPath, shouldCache).create(hadoopPath, overwrite)); + FSDataOutputStream fsOutput = + getFileSystem(hadoopPath, shouldCache).create(hadoopPath, overwrite); + JindoIOStats.OUTPUT.opened(); + return new HadoopPositionOutputStream(fsOutput); } @Override @@ -290,6 +293,8 @@ private static class HadoopSeekableInputStream extends SeekableInputStream { protected final FSDataInputStream in; + private boolean closed; + private HadoopSeekableInputStream(FSDataInputStream in) { this.in = in; } @@ -316,17 +321,31 @@ public long getPos() throws IOException { @Override public int read() throws IOException { - return in.read(); + int b = in.read(); + if (b >= 0) { + JindoIOStats.INPUT.bytes(1); + } + return b; } @Override public int read(byte[] b, int off, int len) throws IOException { - return in.read(b, off, len); + int n = in.read(b, off, len); + JindoIOStats.INPUT.bytes(n); + return n; } @Override public void close() throws IOException { - in.close(); + if (closed) { + return; + } + closed = true; + try { + in.close(); + } finally { + JindoIOStats.INPUT.closed(); + } } /** @@ -364,7 +383,9 @@ private VectoredReadableInputStream(FSDataInputStream in) { @Override public int pread(long position, byte[] bytes, int off, int len) throws IOException { - return in.read(position, bytes, off, len); + int n = in.read(position, bytes, off, len); + JindoIOStats.INPUT.bytes(n); + return n; } } @@ -372,6 +393,8 @@ private static class HadoopPositionOutputStream extends PositionOutputStream { private final FSDataOutputStream out; + private boolean closed; + private HadoopPositionOutputStream(FSDataOutputStream out) { this.out = out; } @@ -384,16 +407,19 @@ public long getPos() throws IOException { @Override public void write(int b) throws IOException { out.write(b); + JindoIOStats.OUTPUT.bytes(1); } @Override public void write(byte[] b) throws IOException { out.write(b); + JindoIOStats.OUTPUT.bytes(b.length); } @Override public void write(byte[] b, int off, int len) throws IOException { out.write(b, off, len); + JindoIOStats.OUTPUT.bytes(len); } @Override @@ -403,7 +429,15 @@ public void flush() throws IOException { @Override public void close() throws IOException { - out.close(); + if (closed) { + return; + } + closed = true; + try { + out.close(); + } finally { + JindoIOStats.OUTPUT.closed(); + } } } diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java index 71468672af9f..beb2c8674f6b 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java @@ -24,6 +24,8 @@ import org.apache.paimon.fs.HadoopOptionsProvider; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.ConfigOptions; import org.apache.paimon.options.Options; import org.apache.paimon.plugin.PluginLoader; import org.apache.paimon.utils.IOUtils; @@ -73,6 +75,14 @@ public class JindoFileIO extends HadoopCompliantFileIO implements HadoopOptionsP private static final String OSS_SECURITY_TOKEN = "fs.oss.securityToken"; private static final String OSS_SHOW_DIR_TIMESTAMP = "fs.oss.show-dir-timestamp"; + private static final ConfigOption IO_STATS_INTERVAL = + ConfigOptions.key("jindo.io-stats.interval") + .durationType() + .defaultValue(Duration.ZERO) + .withDescription( + "Interval of logging JVM-wide Jindo stream and memory statistics, " + + "e.g. 1 min. 0 (the default) disables it."); + private static final Map CASE_SENSITIVE_KEYS = new HashMap() { { @@ -92,6 +102,7 @@ public class JindoFileIO extends HadoopCompliantFileIO implements HadoopOptionsP private Options hadoopOptions; private Options hadoopOptionsWithCache; private boolean allowCache = true; + private Duration ioStatsInterval = IO_STATS_INTERVAL.defaultValue(); private transient BlobPresigner blobPresigner; public JindoFileIO() {} @@ -109,6 +120,7 @@ public boolean isObjectStore() { public void configure(CatalogContext context) { super.configure(context); allowCache = context.options().get(FILE_IO_ALLOW_CACHE); + ioStatsInterval = context.options().get(IO_STATS_INTERVAL); hadoopOptions = new Options(); // read all configuration with prefix 'CONFIG_PREFIXES' for (String key : context.options().keySet()) { @@ -258,6 +270,10 @@ protected Pair createFileSystem( org.apache.hadoop.fs.Path path, boolean enableCache) { final String scheme = path.toUri().getScheme(); final String authority = path.toUri().getAuthority(); + // FileIOs deserialized from older versions have no interval + if (ioStatsInterval != null) { + JindoIOStats.start(ioStatsInterval); + } Options options = enableCache ? hadoopOptionsWithCache : hadoopOptions; Supplier> supplier = () -> { @@ -293,8 +309,12 @@ protected Pair createFileSystem( }; if (allowCache) { - return CACHE.computeIfAbsent( - new CacheKey(options, scheme, authority), key -> supplier.get()); + Pair pair = + CACHE.computeIfAbsent( + new CacheKey(options, scheme, authority), key -> supplier.get()); + // cached file systems are never closed, so the reporter may call into them + JindoIOStats.registerNativeMetrics(pair.getKey(), path); + return pair; } else { return supplier.get(); } diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java new file mode 100644 index 000000000000..6f71faaf8556 --- /dev/null +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java @@ -0,0 +1,546 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.jindo; + +import org.apache.paimon.utils.JsonSerdeUtil; +import org.apache.paimon.utils.Pair; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode; + +import com.aliyun.jindodata.common.JindoConstant; +import com.aliyun.jindodata.common.JindoHadoopSystem; +import com.aliyun.jindodata.store.JindoMetricsStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.lang.management.BufferPoolMXBean; +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryPoolMXBean; +import java.lang.management.MemoryUsage; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * JVM-wide counters of Jindo streams, logged periodically with Jindo native, process and container + * memory; a WARN is logged when the container working set nears its limit. + */ +public final class JindoIOStats { + + private static final Logger LOG = LoggerFactory.getLogger(JindoIOStats.class); + + static final StreamCounter INPUT = new StreamCounter(); + static final StreamCounter OUTPUT = new StreamCounter(); + + private static final AtomicBoolean STARTED = new AtomicBoolean(false); + private static volatile Pair nativeSource; + + private JindoIOStats() {} + + /** Starts the reporter thread once per JVM; a zero or negative interval disables it. */ + static void start(Duration interval) { + if (interval.isZero() || interval.isNegative() || !STARTED.compareAndSet(false, true)) { + return; + } + Reporter reporter = + new Reporter( + interval.toMillis(), + Cgroup.detect(Paths.get("/proc/self/cgroup"), Paths.get("/sys/fs/cgroup")), + Paths.get("/proc/self")); + Thread thread = new Thread(reporter::run, "paimon-jindo-io-stats"); + thread.setDaemon(true); + thread.start(); + } + + /** Registers a file system that is never closed, used to dump the Jindo native metrics. */ + static void registerNativeMetrics(JindoHadoopSystem fs, org.apache.hadoop.fs.Path path) { + if (nativeSource == null) { + nativeSource = Pair.of(fs, path); + } + } + + /** Open, close and byte counters of one stream direction. */ + static final class StreamCounter { + + private final AtomicLong opened = new AtomicLong(); + private final AtomicLong closed = new AtomicLong(); + private final AtomicLong peak = new AtomicLong(); + private final LongAdder bytes = new LongAdder(); + + void opened() { + long open = opened.incrementAndGet() - closed.get(); + peak.accumulateAndGet(open, Math::max); + } + + void closed() { + closed.incrementAndGet(); + } + + void bytes(long n) { + if (n > 0) { + bytes.add(n); + } + } + + long open() { + return opened.get() - closed.get(); + } + + long openedTotal() { + return opened.get(); + } + + long closedTotal() { + return closed.get(); + } + + long bytesTotal() { + return bytes.sum(); + } + + /** Returns the peak since the previous log line and restarts it from the current count. */ + long takePeak() { + return peak.getAndSet(open()); + } + } + + /** Samples memory every second and logs the statistics. */ + static final class Reporter { + + private static final long SAMPLE_MILLIS = 1000; + private static final double[] ALARM_RATIOS = {0.8, 0.9, 0.95}; + private static final double ALARM_RESET_RATIO = 0.7; + + private final long intervalMillis; + @Nullable private final Cgroup cgroup; + private final Path procSelf; + + @Nullable private JindoMetricsStore nativeStore; + private boolean nativeDisabled; + @Nullable private String jemalloc; + private int alarmLevel; + private long lastEvents; + + Reporter(long intervalMillis, @Nullable Cgroup cgroup, Path procSelf) { + this.intervalMillis = intervalMillis; + this.cgroup = cgroup; + this.procSelf = procSelf; + } + + void run() { + long nextReport = System.currentTimeMillis() + intervalMillis; + while (true) { + try { + Thread.sleep(SAMPLE_MILLIS); + } catch (InterruptedException e) { + return; + } + try { + String alarm = checkAlarm(); + if (alarm != null) { + LOG.warn(alarm); + } + long now = System.currentTimeMillis(); + if (now >= nextReport) { + nextReport = now + intervalMillis; + if (hasActivity()) { + LOG.info(snapshot()); + } + } + } catch (Throwable t) { + LOG.debug("Failed to collect Jindo IO stats.", t); + } + } + } + + /** Returns a WARN message when the working set enters a higher alarm level. */ + @Nullable + String checkAlarm() { + if (cgroup == null) { + return null; + } + long limit = cgroup.limit(); + long workingSet = cgroup.workingSet(); + if (limit <= 0 || workingSet < 0) { + return null; + } + double ratio = (double) workingSet / limit; + if (ratio < ALARM_RESET_RATIO) { + alarmLevel = 0; + } + int level = 0; + while (level < ALARM_RATIOS.length && ratio >= ALARM_RATIOS[level]) { + level++; + } + if (level <= alarmLevel) { + return null; + } + alarmLevel = level; + return String.format( + "Container memory working set is at %d%% of the limit. %s", + Math.round(ratio * 100), snapshot()); + } + + /** True while streams are open or when any stream opened or closed since last call. */ + boolean hasActivity() { + long events = + INPUT.openedTotal() + + INPUT.closedTotal() + + OUTPUT.openedTotal() + + OUTPUT.closedTotal(); + boolean active = events != lastEvents || INPUT.open() > 0 || OUTPUT.open() > 0; + lastEvents = events; + return active; + } + + String snapshot() { + StringBuilder sb = new StringBuilder("Jindo IO stats:"); + appendStreams(sb, "input", INPUT, "read"); + appendStreams(sb, "output", OUTPUT, "written"); + appendNative(sb); + appendProcess(sb); + appendCgroup(sb); + return sb.toString(); + } + + private static void appendStreams( + StringBuilder sb, String name, StreamCounter counter, String bytesName) { + sb.append(' ') + .append(name) + .append(" open=") + .append(counter.open()) + .append(" peak=") + .append(counter.takePeak()) + .append(" opened=") + .append(counter.openedTotal()) + .append(" closed=") + .append(counter.closedTotal()) + .append(' ') + .append(bytesName) + .append('=') + .append(mib(counter.bytesTotal())) + .append(';'); + } + + private void appendNative(StringBuilder sb) { + Map metrics = nativeMetrics(); + if (metrics == null) { + sb.append(" jindo unavailable;"); + return; + } + long memInsufficient = 0; + for (Map.Entry e : metrics.entrySet()) { + if (e.getKey().startsWith("jindosdk_prefetch_fallback_mem_insufficient")) { + memInsufficient += e.getValue(); + } + } + sb.append(" jindo prefetchCache=") + .append(mib(get(metrics, "jindosdk_prefetch_cache_size", 0))) + .append('/') + .append(mib(get(metrics, "jindosdk_prefetch_cache_capacity", -1))) + .append(" readStreams=") + .append( + get(metrics, "jindosdk_read_stream_open_count", 0) + - get(metrics, "jindosdk_read_stream_close_count", 0)) + .append(" writeStreams=") + .append( + get(metrics, "jindosdk_write_stream_open_count", 0) + - get(metrics, "jindosdk_write_stream_close_count", 0)) + .append(" memInsufficient=") + .append(memInsufficient) + .append(';'); + } + + @Nullable + private Map nativeMetrics() { + Pair source = nativeSource; + if (nativeDisabled || source == null) { + return null; + } + try { + if (nativeStore == null) { + nativeStore = source.getLeft().getMetricsStore(source.getRight()); + if (nativeStore == null) { + nativeDisabled = true; + return null; + } + } + return parseNativeMetrics( + nativeStore.dumpMetrics( + JindoConstant.MetricType.STATIC_ONLY, + JindoConstant.MetricFormat.JSON)); + } catch (Throwable t) { + nativeDisabled = true; + LOG.debug("Jindo native metrics are unavailable.", t); + return null; + } + } + + private void appendProcess(StringBuilder sb) { + Map status = readStatus(procSelf.resolve("status")); + MemoryUsage heap = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage(); + sb.append(" process rss=") + .append(mib(status.getOrDefault("VmRSS", -1L))) + .append(" anon=") + .append(mib(status.getOrDefault("RssAnon", -1L))) + .append(" file=") + .append(mib(status.getOrDefault("RssFile", -1L))) + .append(" shmem=") + .append(mib(status.getOrDefault("RssShmem", -1L))) + .append(" threads=") + .append(status.getOrDefault("Threads", -1L)) + .append(" heap=") + .append(mib(heap.getUsed())) + .append('/') + .append(mib(heap.getCommitted())) + .append('/') + .append(mib(heap.getMax())) + .append(" direct=") + .append(mib(directUsed())) + .append(" metaspace=") + .append(mib(metaspaceUsed())) + .append(" jemalloc=") + .append(jemalloc()) + .append(';'); + } + + private void appendCgroup(StringBuilder sb) { + if (cgroup == null) { + sb.append(" cgroup unavailable"); + return; + } + sb.append(" cgroup usage=") + .append(mib(cgroup.usage())) + .append('/') + .append(mib(cgroup.limit())) + .append(" workingSet=") + .append(mib(cgroup.workingSet())) + .append(" peak=") + .append(mib(cgroup.peak())) + .append(" limitHits=") + .append(cgroup.limitHits()) + .append(" oomKills=") + .append(cgroup.oomKills()); + } + + private String jemalloc() { + if (jemalloc == null) { + try { + jemalloc = "false"; + for (String line : Files.readAllLines(procSelf.resolve("maps"))) { + if (line.contains("/libjemalloc")) { + jemalloc = "true"; + break; + } + } + } catch (IOException e) { + jemalloc = "unknown"; + } + } + return jemalloc; + } + + private static long directUsed() { + for (BufferPoolMXBean pool : + ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class)) { + if ("direct".equals(pool.getName())) { + return pool.getMemoryUsed(); + } + } + return -1; + } + + private static long metaspaceUsed() { + for (MemoryPoolMXBean pool : ManagementFactory.getMemoryPoolMXBeans()) { + if ("Metaspace".equals(pool.getName())) { + return pool.getUsage().getUsed(); + } + } + return -1; + } + + /** Reads /proc/self/status; sizes in bytes, counts as is. */ + private static Map readStatus(Path file) { + Map values = new HashMap<>(); + try { + for (String line : Files.readAllLines(file)) { + int colon = line.indexOf(':'); + if (colon < 0) { + continue; + } + String[] parts = line.substring(colon + 1).trim().split("\\s+"); + try { + long value = Long.parseLong(parts[0]); + boolean kb = parts.length > 1 && "kB".equals(parts[1]); + values.put(line.substring(0, colon), kb ? value * 1024 : value); + } catch (NumberFormatException ignored) { + } + } + } catch (IOException ignored) { + } + return values; + } + + private static long get(Map metrics, String name, long defaultValue) { + return metrics.getOrDefault(name, defaultValue); + } + + private static String mib(long bytes) { + return bytes < 0 ? "?" : (bytes >> 20) + "MiB"; + } + } + + /** Sums the values of each metric in a Jindo JSON metrics dump. */ + static Map parseNativeMetrics(String json) throws IOException { + Map values = new HashMap<>(); + for (JsonNode metric : JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(json)) { + long sum = 0; + for (JsonNode m : metric.path("metrics")) { + sum += m.path("value").asLong(); + } + values.merge(metric.path("name").asText(), sum, Long::sum); + } + return values; + } + + /** Memory cgroup (v1 or v2) of this process; unreadable values are -1. */ + static final class Cgroup { + + private static final Pattern V1_MEMORY = + Pattern.compile("^\\d+:(?:[^:]*,)?memory(?:,[^:]*)?:(.*)$"); + + private final Path dir; + private final boolean v2; + + private Cgroup(Path dir, boolean v2) { + this.dir = dir; + this.v2 = v2; + } + + /** Uses the process's own cgroup when visible, else the mount root (container view). */ + @Nullable + static Cgroup detect(Path procSelfCgroup, Path root) { + List lines; + try { + lines = Files.readAllLines(procSelfCgroup); + } catch (IOException e) { + return null; + } + if (Files.exists(root.resolve("cgroup.controllers"))) { + for (String line : lines) { + if (line.startsWith("0::")) { + Path own = resolve(root, line.substring(3)); + if (Files.exists(own.resolve("memory.current"))) { + return new Cgroup(own, true); + } + } + } + return Files.exists(root.resolve("memory.current")) ? new Cgroup(root, true) : null; + } + Path memoryRoot = root.resolve("memory"); + for (String line : lines) { + Matcher matcher = V1_MEMORY.matcher(line); + if (matcher.matches()) { + Path own = resolve(memoryRoot, matcher.group(1)); + if (Files.exists(own.resolve("memory.usage_in_bytes"))) { + return new Cgroup(own, false); + } + } + } + return Files.exists(memoryRoot.resolve("memory.usage_in_bytes")) + ? new Cgroup(memoryRoot, false) + : null; + } + + private static Path resolve(Path root, String relative) { + String trimmed = relative.startsWith("/") ? relative.substring(1) : relative; + return trimmed.isEmpty() ? root : root.resolve(trimmed); + } + + long limit() { + if (v2) { + return readLong("memory.max"); + } + long limit = readLong("memory.limit_in_bytes"); + long hierarchical = readKey("memory.stat", "hierarchical_memory_limit"); + if (hierarchical > 0 && (limit < 0 || hierarchical < limit)) { + limit = hierarchical; + } + return limit >= Long.MAX_VALUE / 2 ? -1 : limit; + } + + long usage() { + return readLong(v2 ? "memory.current" : "memory.usage_in_bytes"); + } + + /** Usage minus inactive file pages, the part the kernel cannot reclaim cheaply. */ + long workingSet() { + long usage = usage(); + long inactiveFile = + readKey("memory.stat", v2 ? "inactive_file" : "total_inactive_file"); + return usage < 0 ? -1 : Math.max(0, usage - Math.max(0, inactiveFile)); + } + + long peak() { + return readLong(v2 ? "memory.peak" : "memory.max_usage_in_bytes"); + } + + long limitHits() { + return v2 ? readKey("memory.events", "max") : readLong("memory.failcnt"); + } + + long oomKills() { + return readKey(v2 ? "memory.events" : "memory.oom_control", "oom_kill"); + } + + private long readLong(String file) { + try { + String value = new String(Files.readAllBytes(dir.resolve(file))).trim(); + return "max".equals(value) ? -1 : Long.parseLong(value); + } catch (IOException | NumberFormatException e) { + return -1; + } + } + + private long readKey(String file, String key) { + try { + for (String line : Files.readAllLines(dir.resolve(file))) { + int space = line.indexOf(' '); + if (space > 0 && line.substring(0, space).equals(key)) { + return Long.parseLong(line.substring(space + 1).trim()); + } + } + } catch (IOException | NumberFormatException ignored) { + } + return -1; + } + } +} diff --git a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java new file mode 100644 index 000000000000..c574f3b20a27 --- /dev/null +++ b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java @@ -0,0 +1,293 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.jindo; + +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.utils.Pair; + +import com.aliyun.jindodata.common.JindoHadoopSystem; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.PositionedReadable; +import org.apache.hadoop.fs.Seekable; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests for {@link JindoIOStats}. */ +public class JindoIOStatsTest { + + @TempDir java.nio.file.Path tempDir; + + @Test + public void testStreamCounterPeak() { + JindoIOStats.StreamCounter counter = new JindoIOStats.StreamCounter(); + counter.opened(); + counter.opened(); + counter.opened(); + counter.closed(); + counter.bytes(10); + counter.bytes(-1); + + assertThat(counter.open()).isEqualTo(2); + assertThat(counter.bytesTotal()).isEqualTo(10); + assertThat(counter.takePeak()).isEqualTo(3); + assertThat(counter.takePeak()).isEqualTo(2); + } + + @Test + public void testFileIOStreamsAreCounted() throws Exception { + JindoHadoopSystem fs = mock(JindoHadoopSystem.class); + when(fs.open(any(org.apache.hadoop.fs.Path.class))) + .thenAnswer(i -> new FSDataInputStream(new SeekableBytes(new byte[100]))); + when(fs.create(any(org.apache.hadoop.fs.Path.class), anyBoolean())) + .thenAnswer(i -> new FSDataOutputStream(new ByteArrayOutputStream(), null)); + StubFileIO fileIO = new StubFileIO(fs); + Path path = new Path("oss://bucket/db/t/data.parquet"); + long inOpened = JindoIOStats.INPUT.openedTotal(); + long inClosed = JindoIOStats.INPUT.closedTotal(); + long read = JindoIOStats.INPUT.bytesTotal(); + long outOpened = JindoIOStats.OUTPUT.openedTotal(); + long outClosed = JindoIOStats.OUTPUT.closedTotal(); + long written = JindoIOStats.OUTPUT.bytesTotal(); + + SeekableInputStream in = fileIO.newInputStream(path); + assertThat(in.read(new byte[64], 0, 64)).isEqualTo(64); + in.read(); + in.close(); + in.close(); + PositionOutputStream out = fileIO.newOutputStream(path, true); + out.write(new byte[30]); + out.write(new byte[30], 5, 20); + out.write(1); + out.close(); + out.close(); + + assertThat(JindoIOStats.INPUT.openedTotal() - inOpened).isEqualTo(1); + assertThat(JindoIOStats.INPUT.closedTotal() - inClosed).isEqualTo(1); + assertThat(JindoIOStats.INPUT.bytesTotal() - read).isEqualTo(65); + assertThat(JindoIOStats.OUTPUT.openedTotal() - outOpened).isEqualTo(1); + assertThat(JindoIOStats.OUTPUT.closedTotal() - outClosed).isEqualTo(1); + assertThat(JindoIOStats.OUTPUT.bytesTotal() - written).isEqualTo(51); + } + + @Test + public void testActivityIncludesCloses() { + JindoIOStats.Reporter reporter = + new JindoIOStats.Reporter(60_000, null, java.nio.file.Paths.get("/nonexistent")); + reporter.hasActivity(); + assertThat(reporter.hasActivity()).isFalse(); + + JindoIOStats.INPUT.opened(); + assertThat(reporter.hasActivity()).isTrue(); + assertThat(reporter.hasActivity()).isTrue(); + JindoIOStats.INPUT.closed(); + assertThat(reporter.hasActivity()).isTrue(); + assertThat(reporter.hasActivity()).isFalse(); + } + + @Test + public void testParseNativeMetrics() throws IOException { + String json = + "[{\"name\":\"jindosdk_prefetch_cache_size\",\"help\":\"prefetch cache size\"," + + "\"type\":\"gauge\",\"metrics\":[{\"labels\":{},\"value\":157286400}]}," + + "{\"name\":\"jindosdk_read_stream_open_count\",\"help\":\"\"," + + "\"type\":\"counter\",\"metrics\":[{\"labels\":{},\"value\":8}]}," + + "{\"name\":\"jindosdk_getObject_latency\",\"help\":\"\"," + + "\"type\":\"summary\",\"quantiles\":[0.5],\"metrics\":" + + "[{\"labels\":{},\"quantiles\":[1.0],\"count\":3,\"sum\":3.0}]}]"; + + Map metrics = JindoIOStats.parseNativeMetrics(json); + + assertThat(metrics) + .containsEntry("jindosdk_prefetch_cache_size", 157286400L) + .containsEntry("jindosdk_read_stream_open_count", 8L) + .containsEntry("jindosdk_getObject_latency", 0L); + } + + @Test + public void testCgroupV2() throws IOException { + java.nio.file.Path root = tempDir.resolve("cgroup"); + java.nio.file.Path own = root.resolve("kubepods/pod1"); + Files.createDirectories(own); + write(root.resolve("cgroup.controllers"), "memory"); + write(own.resolve("memory.current"), "2000000000"); + write(own.resolve("memory.max"), "2442133504"); + write(own.resolve("memory.peak"), "2100000000"); + write(own.resolve("memory.stat"), "anon 1700000000\ninactive_file 100000000\n"); + write(own.resolve("memory.events"), "low 0\nhigh 0\nmax 12\noom 1\noom_kill 1\n"); + java.nio.file.Path procCgroup = + write(tempDir.resolve("cgroup-file"), "0::/kubepods/pod1\n"); + + JindoIOStats.Cgroup cgroup = JindoIOStats.Cgroup.detect(procCgroup, root); + + assertThat(cgroup).isNotNull(); + assertThat(cgroup.limit()).isEqualTo(2442133504L); + assertThat(cgroup.usage()).isEqualTo(2000000000L); + assertThat(cgroup.workingSet()).isEqualTo(1900000000L); + assertThat(cgroup.peak()).isEqualTo(2100000000L); + assertThat(cgroup.limitHits()).isEqualTo(12); + assertThat(cgroup.oomKills()).isEqualTo(1); + } + + @Test + public void testCgroupV1FallsBackToMountRoot() throws IOException { + java.nio.file.Path root = tempDir.resolve("cgroup"); + java.nio.file.Path memory = root.resolve("memory"); + Files.createDirectories(memory); + write(memory.resolve("memory.usage_in_bytes"), "1500000000"); + write(memory.resolve("memory.limit_in_bytes"), "9223372036854771712"); + write(memory.resolve("memory.max_usage_in_bytes"), "1600000000"); + write(memory.resolve("memory.failcnt"), "3"); + write( + memory.resolve("memory.stat"), + "hierarchical_memory_limit 2442133504\ntotal_inactive_file 500000000\n"); + write( + memory.resolve("memory.oom_control"), + "oom_kill_disable 0\nunder_oom 0\noom_kill 0\n"); + java.nio.file.Path procCgroup = + write(tempDir.resolve("cgroup-file"), "4:cpu,cpuacct:/a\n9:memory:/kubepods/x\n"); + + JindoIOStats.Cgroup cgroup = JindoIOStats.Cgroup.detect(procCgroup, root); + + assertThat(cgroup).isNotNull(); + assertThat(cgroup.limit()).isEqualTo(2442133504L); + assertThat(cgroup.workingSet()).isEqualTo(1000000000L); + assertThat(cgroup.peak()).isEqualTo(1600000000L); + assertThat(cgroup.limitHits()).isEqualTo(3); + assertThat(cgroup.oomKills()).isEqualTo(0); + } + + @Test + public void testNoCgroup() { + assertThat(JindoIOStats.Cgroup.detect(tempDir.resolve("missing"), tempDir)).isNull(); + } + + @Test + public void testAlarmLevels() throws IOException { + java.nio.file.Path root = tempDir.resolve("cgroup"); + Files.createDirectories(root); + write(root.resolve("cgroup.controllers"), "memory"); + write(root.resolve("memory.max"), "1000"); + write(root.resolve("memory.stat"), "inactive_file 0\n"); + write(root.resolve("memory.current"), "500"); + java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/\n"); + JindoIOStats.Reporter reporter = + new JindoIOStats.Reporter( + 60_000, JindoIOStats.Cgroup.detect(procCgroup, root), tempDir); + + assertThat(reporter.checkAlarm()).isNull(); + write(root.resolve("memory.current"), "810"); + assertThat(reporter.checkAlarm()) + .startsWith("Container memory working set is at 81% of the limit.") + .contains("Jindo IO stats: input open=") + .contains("cgroup usage=0MiB/0MiB"); + write(root.resolve("memory.current"), "850"); + assertThat(reporter.checkAlarm()).isNull(); + write(root.resolve("memory.current"), "960"); + assertThat(reporter.checkAlarm()).contains("at 96% of the limit"); + write(root.resolve("memory.current"), "600"); + assertThat(reporter.checkAlarm()).isNull(); + write(root.resolve("memory.current"), "800"); + assertThat(reporter.checkAlarm()).contains("at 80% of the limit"); + } + + private static java.nio.file.Path write(java.nio.file.Path file, String content) + throws IOException { + return Files.write(file, content.getBytes(StandardCharsets.UTF_8)); + } + + private static class StubFileIO extends HadoopCompliantFileIO { + + private final JindoHadoopSystem fs; + + private StubFileIO(JindoHadoopSystem fs) { + this.fs = fs; + } + + @Override + public void configure(CatalogContext context) {} + + @Override + public boolean isObjectStore() { + return true; + } + + @Override + protected Pair createFileSystem( + org.apache.hadoop.fs.Path path, boolean enableCache) { + return Pair.of(fs, "oss"); + } + } + + private static class SeekableBytes extends ByteArrayInputStream + implements Seekable, PositionedReadable { + + private SeekableBytes(byte[] bytes) { + super(bytes); + } + + @Override + public void seek(long pos) { + this.pos = (int) pos; + } + + @Override + public long getPos() { + return pos; + } + + @Override + public boolean seekToNewSource(long targetPos) { + return false; + } + + @Override + public int read(long position, byte[] buffer, int offset, int length) { + int n = Math.min(length, count - (int) position); + System.arraycopy(buf, (int) position, buffer, offset, n); + return n; + } + + @Override + public void readFully(long position, byte[] buffer, int offset, int length) { + read(position, buffer, offset, length); + } + + @Override + public void readFully(long position, byte[] buffer) { + read(position, buffer, 0, buffer.length); + } + } +} From 622f101e0d967126e28110855a041990458fb63a Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Mon, 26 Oct 2026 20:39:39 +0800 Subject: [PATCH 2/4] [jindo] Address review of the Jindo IO statistics - Always delegate close() again and count a stream closed only after the delegate close succeeds, so a cleanup retry after a failed close still reaches Jindo (e.g. BTreeIndexWriter calls closeQuietly after close fails). - Paimon never closes FileIOs, so the reporter cannot be tied to their lifecycle. It now starts on the first counted stream open, stops after an interval without stream activity, and is restarted by the next open. Its context class loader is the one that loaded paimon-jindo, so it no longer pins the loader of whichever job opened the first stream. - Keep a single active counter and reset the peak with a CAS loop, so a log line never shows a negative open count or a peak below the open count. - Cgroup limits are hierarchical: walk from the process's cgroup to the mount root and report and alarm on the level whose working set is closest to its limit, reading limit, usage and events from that same level. - An interval of 0 now also skips counting. Each stream captures its counter when it is opened, so its open and close stay balanced. --- .../paimon/jindo/HadoopCompliantFileIO.java | 87 ++++--- .../org/apache/paimon/jindo/JindoFileIO.java | 18 +- .../org/apache/paimon/jindo/JindoIOStats.java | 168 ++++++++++---- .../apache/paimon/jindo/JindoIOStatsTest.java | 212 +++++++++++++++++- 4 files changed, 394 insertions(+), 91 deletions(-) diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java index b9822a4bfff9..2514e0adcdef 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java @@ -37,9 +37,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.IOException; import java.io.UncheckedIOException; import java.net.URI; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -133,10 +136,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException { JindoHadoopSystem fs = pair.getKey(); String sysType = pair.getValue(); FSDataInputStream fsInput = fs.open(hadoopPath); - JindoIOStats.INPUT.opened(); + JindoIOStats.StreamCounter stats = + JindoIOStats.track(JindoIOStats.INPUT, ioStatsInterval()); return "jobj".equalsIgnoreCase(sysType) - ? new VectoredReadableInputStream(fsInput) - : new HadoopSeekableInputStream(fsInput); + ? new VectoredReadableInputStream(fsInput, stats) + : new HadoopSeekableInputStream(fsInput, stats); } @Override @@ -146,8 +150,14 @@ public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws LOG.debug("OutputStream should cache {} for path {}", shouldCache, path); FSDataOutputStream fsOutput = getFileSystem(hadoopPath, shouldCache).create(hadoopPath, overwrite); - JindoIOStats.OUTPUT.opened(); - return new HadoopPositionOutputStream(fsOutput); + return new HadoopPositionOutputStream( + fsOutput, JindoIOStats.track(JindoIOStats.OUTPUT, ioStatsInterval())); + } + + /** Interval of logging {@link JindoIOStats}; null or zero leaves the streams uncounted. */ + @Nullable + protected Duration ioStatsInterval() { + return null; } @Override @@ -293,10 +303,14 @@ private static class HadoopSeekableInputStream extends SeekableInputStream { protected final FSDataInputStream in; + @Nullable protected final JindoIOStats.StreamCounter stats; + private boolean closed; - private HadoopSeekableInputStream(FSDataInputStream in) { + private HadoopSeekableInputStream( + FSDataInputStream in, @Nullable JindoIOStats.StreamCounter stats) { this.in = in; + this.stats = stats; } @Override @@ -322,8 +336,8 @@ public long getPos() throws IOException { @Override public int read() throws IOException { int b = in.read(); - if (b >= 0) { - JindoIOStats.INPUT.bytes(1); + if (b >= 0 && stats != null) { + stats.bytes(1); } return b; } @@ -331,20 +345,19 @@ public int read() throws IOException { @Override public int read(byte[] b, int off, int len) throws IOException { int n = in.read(b, off, len); - JindoIOStats.INPUT.bytes(n); + if (stats != null) { + stats.bytes(n); + } return n; } @Override public void close() throws IOException { - if (closed) { - return; - } - closed = true; - try { - in.close(); - } finally { - JindoIOStats.INPUT.closed(); + in.close(); + // a close that throws is not counted, so that a retried close still reaches Jindo + if (stats != null && !closed) { + closed = true; + stats.closed(); } } @@ -377,14 +390,17 @@ public void skipFully(long bytes) throws IOException { private static class VectoredReadableInputStream extends HadoopSeekableInputStream implements VectoredReadable { - private VectoredReadableInputStream(FSDataInputStream in) { - super(in); + private VectoredReadableInputStream( + FSDataInputStream in, @Nullable JindoIOStats.StreamCounter stats) { + super(in, stats); } @Override public int pread(long position, byte[] bytes, int off, int len) throws IOException { int n = in.read(position, bytes, off, len); - JindoIOStats.INPUT.bytes(n); + if (stats != null) { + stats.bytes(n); + } return n; } } @@ -393,10 +409,14 @@ private static class HadoopPositionOutputStream extends PositionOutputStream { private final FSDataOutputStream out; + @Nullable private final JindoIOStats.StreamCounter stats; + private boolean closed; - private HadoopPositionOutputStream(FSDataOutputStream out) { + private HadoopPositionOutputStream( + FSDataOutputStream out, @Nullable JindoIOStats.StreamCounter stats) { this.out = out; + this.stats = stats; } @Override @@ -407,19 +427,25 @@ public long getPos() throws IOException { @Override public void write(int b) throws IOException { out.write(b); - JindoIOStats.OUTPUT.bytes(1); + if (stats != null) { + stats.bytes(1); + } } @Override public void write(byte[] b) throws IOException { out.write(b); - JindoIOStats.OUTPUT.bytes(b.length); + if (stats != null) { + stats.bytes(b.length); + } } @Override public void write(byte[] b, int off, int len) throws IOException { out.write(b, off, len); - JindoIOStats.OUTPUT.bytes(len); + if (stats != null) { + stats.bytes(len); + } } @Override @@ -429,14 +455,11 @@ public void flush() throws IOException { @Override public void close() throws IOException { - if (closed) { - return; - } - closed = true; - try { - out.close(); - } finally { - JindoIOStats.OUTPUT.closed(); + out.close(); + // a close that throws is not counted, so that a retried close still reaches Jindo + if (stats != null && !closed) { + closed = true; + stats.closed(); } } } diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java index beb2c8674f6b..a02f19d7780f 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java @@ -42,6 +42,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.IOException; import java.io.UncheckedIOException; import java.net.URI; @@ -80,8 +82,9 @@ public class JindoFileIO extends HadoopCompliantFileIO implements HadoopOptionsP .durationType() .defaultValue(Duration.ZERO) .withDescription( - "Interval of logging JVM-wide Jindo stream and memory statistics, " - + "e.g. 1 min. 0 (the default) disables it."); + "Interval of logging JVM-wide Jindo stream and memory statistics " + + "while Jindo streams are in use, e.g. 1 min. 0 (the default) " + + "disables it, and streams of this FileIO are not counted."); private static final Map CASE_SENSITIVE_KEYS = new HashMap() { @@ -270,10 +273,6 @@ protected Pair createFileSystem( org.apache.hadoop.fs.Path path, boolean enableCache) { final String scheme = path.toUri().getScheme(); final String authority = path.toUri().getAuthority(); - // FileIOs deserialized from older versions have no interval - if (ioStatsInterval != null) { - JindoIOStats.start(ioStatsInterval); - } Options options = enableCache ? hadoopOptionsWithCache : hadoopOptions; Supplier> supplier = () -> { @@ -320,6 +319,13 @@ protected Pair createFileSystem( } } + @Nullable + @Override + protected Duration ioStatsInterval() { + // FileIOs deserialized from older versions have no interval + return ioStatsInterval; + } + @Override public synchronized void close() { if (blobPresigner != null) { diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java index 6f71faaf8556..8e292047383c 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java @@ -40,6 +40,8 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -50,8 +52,9 @@ import java.util.regex.Pattern; /** - * JVM-wide counters of Jindo streams, logged periodically with Jindo native, process and container - * memory; a WARN is logged when the container working set nears its limit. + * JVM-wide counters of the Jindo streams of FileIOs that enable them, logged periodically with + * Jindo native, process and container memory; a WARN is logged when the container working set nears + * its limit. */ public final class JindoIOStats { @@ -60,26 +63,49 @@ public final class JindoIOStats { static final StreamCounter INPUT = new StreamCounter(); static final StreamCounter OUTPUT = new StreamCounter(); - private static final AtomicBoolean STARTED = new AtomicBoolean(false); + private static final AtomicLong INTERVAL_MILLIS = new AtomicLong(); + private static final AtomicBoolean RUNNING = new AtomicBoolean(false); private static volatile Pair nativeSource; private JindoIOStats() {} - /** Starts the reporter thread once per JVM; a zero or negative interval disables it. */ - static void start(Duration interval) { - if (interval.isZero() || interval.isNegative() || !STARTED.compareAndSet(false, true)) { + /** + * Counts a stream just opened by a FileIO with this interval and returns its counter, or null + * when the interval is null, zero or negative. The first positive interval in the JVM is used. + */ + @Nullable + static StreamCounter track(StreamCounter counter, @Nullable Duration interval) { + if (interval == null || interval.isZero() || interval.isNegative()) { + return null; + } + INTERVAL_MILLIS.compareAndSet(0, Math.max(1, interval.toMillis())); + counter.opened(); + startReporter(); + return counter; + } + + private static void startReporter() { + if (RUNNING.get() || !RUNNING.compareAndSet(false, true)) { return; } Reporter reporter = new Reporter( - interval.toMillis(), + INTERVAL_MILLIS.get(), Cgroup.detect(Paths.get("/proc/self/cgroup"), Paths.get("/sys/fs/cgroup")), Paths.get("/proc/self")); Thread thread = new Thread(reporter::run, "paimon-jindo-io-stats"); thread.setDaemon(true); + // the inherited loader may belong to the job that opened the stream, keep it collectable + thread.setContextClassLoader(JindoIOStats.class.getClassLoader()); thread.start(); } + /** Returns true when the reporter stopped; streams opened meanwhile keep it running. */ + private static boolean stopReporter() { + RUNNING.set(false); + return (INPUT.open() == 0 && OUTPUT.open() == 0) || !RUNNING.compareAndSet(false, true); + } + /** Registers a file system that is never closed, used to dump the Jindo native metrics. */ static void registerNativeMetrics(JindoHadoopSystem fs, org.apache.hadoop.fs.Path path) { if (nativeSource == null) { @@ -92,16 +118,18 @@ static final class StreamCounter { private final AtomicLong opened = new AtomicLong(); private final AtomicLong closed = new AtomicLong(); + private final AtomicLong active = new AtomicLong(); private final AtomicLong peak = new AtomicLong(); private final LongAdder bytes = new LongAdder(); void opened() { - long open = opened.incrementAndGet() - closed.get(); - peak.accumulateAndGet(open, Math::max); + opened.incrementAndGet(); + peak.accumulateAndGet(active.incrementAndGet(), Math::max); } void closed() { closed.incrementAndGet(); + active.decrementAndGet(); } void bytes(long n) { @@ -111,7 +139,7 @@ void bytes(long n) { } long open() { - return opened.get() - closed.get(); + return active.get(); } long openedTotal() { @@ -126,13 +154,21 @@ long bytesTotal() { return bytes.sum(); } - /** Returns the peak since the previous log line and restarts it from the current count. */ - long takePeak() { - return peak.getAndSet(open()); + /** + * Returns the peak since the previous call, at least {@code open}, and restarts it from the + * current count without losing a concurrent update. + */ + long takePeak(long open) { + while (true) { + long previous = peak.get(); + if (peak.compareAndSet(previous, active.get())) { + return Math.max(previous, open); + } + } } } - /** Samples memory every second and logs the statistics. */ + /** Samples memory every second and logs the statistics until the streams go idle. */ static final class Reporter { private static final long SAMPLE_MILLIS = 1000; @@ -140,7 +176,7 @@ static final class Reporter { private static final double ALARM_RESET_RATIO = 0.7; private final long intervalMillis; - @Nullable private final Cgroup cgroup; + private final List cgroups; private final Path procSelf; @Nullable private JindoMetricsStore nativeStore; @@ -149,18 +185,21 @@ static final class Reporter { private int alarmLevel; private long lastEvents; - Reporter(long intervalMillis, @Nullable Cgroup cgroup, Path procSelf) { + Reporter(long intervalMillis, List cgroups, Path procSelf) { this.intervalMillis = intervalMillis; - this.cgroup = cgroup; + this.cgroups = cgroups; this.procSelf = procSelf; } + /** Runs until an interval passes without stream activity; the next open starts another. */ void run() { + long sampleMillis = Math.min(SAMPLE_MILLIS, intervalMillis); long nextReport = System.currentTimeMillis() + intervalMillis; while (true) { try { - Thread.sleep(SAMPLE_MILLIS); + Thread.sleep(sampleMillis); } catch (InterruptedException e) { + RUNNING.set(false); return; } try { @@ -173,6 +212,8 @@ void run() { nextReport = now + intervalMillis; if (hasActivity()) { LOG.info(snapshot()); + } else if (stopReporter()) { + return; } } } catch (Throwable t) { @@ -184,6 +225,7 @@ void run() { /** Returns a WARN message when the working set enters a higher alarm level. */ @Nullable String checkAlarm() { + Cgroup cgroup = Cgroup.closestToLimit(cgroups); if (cgroup == null) { return null; } @@ -233,12 +275,13 @@ String snapshot() { private static void appendStreams( StringBuilder sb, String name, StreamCounter counter, String bytesName) { + long open = counter.open(); sb.append(' ') .append(name) .append(" open=") - .append(counter.open()) + .append(open) .append(" peak=") - .append(counter.takePeak()) + .append(counter.takePeak(open)) .append(" opened=") .append(counter.openedTotal()) .append(" closed=") @@ -333,6 +376,7 @@ private void appendProcess(StringBuilder sb) { } private void appendCgroup(StringBuilder sb) { + Cgroup cgroup = Cgroup.closestToLimit(cgroups); if (cgroup == null) { sb.append(" cgroup unavailable"); return; @@ -431,7 +475,7 @@ static Map parseNativeMetrics(String json) throws IOException { return values; } - /** Memory cgroup (v1 or v2) of this process; unreadable values are -1. */ + /** One memory cgroup (v1 or v2) level of this process; unreadable values are -1. */ static final class Cgroup { private static final Pattern V1_MEMORY = @@ -445,39 +489,73 @@ private Cgroup(Path dir, boolean v2) { this.v2 = v2; } - /** Uses the process's own cgroup when visible, else the mount root (container view). */ - @Nullable - static Cgroup detect(Path procSelfCgroup, Path root) { + /** + * Returns the levels from the process's own cgroup up to the mount root when it is visible, + * else the mount root alone (container view); empty when there is no memory cgroup. + */ + static List detect(Path procSelfCgroup, Path root) { List lines; try { lines = Files.readAllLines(procSelfCgroup); } catch (IOException e) { - return null; + return Collections.emptyList(); } - if (Files.exists(root.resolve("cgroup.controllers"))) { - for (String line : lines) { - if (line.startsWith("0::")) { - Path own = resolve(root, line.substring(3)); - if (Files.exists(own.resolve("memory.current"))) { - return new Cgroup(own, true); - } + boolean v2 = Files.exists(root.resolve("cgroup.controllers")); + Path mount = (v2 ? root : root.resolve("memory")).normalize(); + for (String line : lines) { + String relative = null; + if (v2 && line.startsWith("0::")) { + relative = line.substring(3); + } else if (!v2) { + Matcher matcher = V1_MEMORY.matcher(line); + if (matcher.matches()) { + relative = matcher.group(1); } } - return Files.exists(root.resolve("memory.current")) ? new Cgroup(root, true) : null; - } - Path memoryRoot = root.resolve("memory"); - for (String line : lines) { - Matcher matcher = V1_MEMORY.matcher(line); - if (matcher.matches()) { - Path own = resolve(memoryRoot, matcher.group(1)); - if (Files.exists(own.resolve("memory.usage_in_bytes"))) { - return new Cgroup(own, false); + if (relative != null) { + Path own = resolve(mount, relative).normalize(); + if (own.startsWith(mount) && Files.exists(own.resolve(usageFile(v2)))) { + return levels(mount, own, v2); } } } - return Files.exists(memoryRoot.resolve("memory.usage_in_bytes")) - ? new Cgroup(memoryRoot, false) - : null; + return levels(mount, mount, v2); + } + + /** + * Limits are hierarchical: the level whose working set is closest to its limit, or the + * leaf. + */ + @Nullable + static Cgroup closestToLimit(List levels) { + Cgroup closest = null; + double closestRatio = -1; + for (Cgroup level : levels) { + long limit = level.limit(); + long workingSet = level.workingSet(); + if (limit > 0 && workingSet >= 0 && (double) workingSet / limit > closestRatio) { + closest = level; + closestRatio = (double) workingSet / limit; + } + } + if (closest == null && !levels.isEmpty()) { + return levels.get(0); + } + return closest; + } + + private static List levels(Path mount, Path leaf, boolean v2) { + List levels = new ArrayList<>(); + for (Path dir = leaf; dir != null && dir.startsWith(mount); dir = dir.getParent()) { + if (Files.exists(dir.resolve(usageFile(v2)))) { + levels.add(new Cgroup(dir, v2)); + } + } + return levels; + } + + private static String usageFile(boolean v2) { + return v2 ? "memory.current" : "memory.usage_in_bytes"; } private static Path resolve(Path root, String relative) { @@ -498,7 +576,7 @@ long limit() { } long usage() { - return readLong(v2 ? "memory.current" : "memory.usage_in_bytes"); + return readLong(usageFile(v2)); } /** Usage minus inactive file pages, the part the kernel cannot reclaim cheaply. */ diff --git a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java index c574f3b20a27..5b8581f553ff 100644 --- a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java +++ b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java @@ -32,22 +32,36 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import javax.annotation.Nullable; + import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** Tests for {@link JindoIOStats}. */ public class JindoIOStatsTest { + // the first positive interval wins in the JVM, so every test uses the same one + private static final Duration INTERVAL = Duration.ofMillis(10); + @TempDir java.nio.file.Path tempDir; @Test @@ -62,8 +76,41 @@ public void testStreamCounterPeak() { assertThat(counter.open()).isEqualTo(2); assertThat(counter.bytesTotal()).isEqualTo(10); - assertThat(counter.takePeak()).isEqualTo(3); - assertThat(counter.takePeak()).isEqualTo(2); + assertThat(counter.takePeak(counter.open())).isEqualTo(3); + assertThat(counter.takePeak(counter.open())).isEqualTo(2); + } + + @Test + public void testConcurrentCountsStayConsistent() throws InterruptedException { + JindoIOStats.StreamCounter counter = new JindoIOStats.StreamCounter(); + AtomicBoolean stop = new AtomicBoolean(); + List workers = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + Thread worker = + new Thread( + () -> { + while (!stop.get()) { + counter.opened(); + counter.closed(); + } + }); + worker.start(); + workers.add(worker); + } + try { + for (int i = 0; i < 100_000; i++) { + long open = counter.open(); + long peak = counter.takePeak(open); + assertThat(open).isBetween(0L, 4L); + assertThat(peak).isBetween(open, 4L); + } + } finally { + stop.set(true); + for (Thread worker : workers) { + worker.join(); + } + } + assertThat(counter.open()).isZero(); } @Test @@ -73,7 +120,7 @@ public void testFileIOStreamsAreCounted() throws Exception { .thenAnswer(i -> new FSDataInputStream(new SeekableBytes(new byte[100]))); when(fs.create(any(org.apache.hadoop.fs.Path.class), anyBoolean())) .thenAnswer(i -> new FSDataOutputStream(new ByteArrayOutputStream(), null)); - StubFileIO fileIO = new StubFileIO(fs); + StubFileIO fileIO = new StubFileIO(fs, INTERVAL); Path path = new Path("oss://bucket/db/t/data.parquet"); long inOpened = JindoIOStats.INPUT.openedTotal(); long inClosed = JindoIOStats.INPUT.closedTotal(); @@ -102,10 +149,98 @@ public void testFileIOStreamsAreCounted() throws Exception { assertThat(JindoIOStats.OUTPUT.bytesTotal() - written).isEqualTo(51); } + @Test + public void testDisabledStreamsAreNotCounted() throws Exception { + JindoHadoopSystem fs = mock(JindoHadoopSystem.class); + when(fs.open(any(org.apache.hadoop.fs.Path.class))) + .thenAnswer(i -> new FSDataInputStream(new SeekableBytes(new byte[100]))); + when(fs.create(any(org.apache.hadoop.fs.Path.class), anyBoolean())) + .thenAnswer(i -> new FSDataOutputStream(new ByteArrayOutputStream(), null)); + Path path = new Path("oss://bucket/db/t/data.parquet"); + long events = events(); + long bytes = JindoIOStats.INPUT.bytesTotal() + JindoIOStats.OUTPUT.bytesTotal(); + + for (Duration interval : new Duration[] {Duration.ZERO, null}) { + StubFileIO fileIO = new StubFileIO(fs, interval); + try (SeekableInputStream in = fileIO.newInputStream(path)) { + assertThat(in.read(new byte[64], 0, 64)).isEqualTo(64); + } + try (PositionOutputStream out = fileIO.newOutputStream(path, true)) { + out.write(new byte[30]); + out.write(1); + } + } + + assertThat(events()).isEqualTo(events); + assertThat(JindoIOStats.INPUT.bytesTotal() + JindoIOStats.OUTPUT.bytesTotal()) + .isEqualTo(bytes); + } + + @Test + public void testFailedCloseIsRetried() throws Exception { + FSDataInputStream fsInput = mock(FSDataInputStream.class); + doThrow(new IOException("close failed")).doNothing().when(fsInput).close(); + FSDataOutputStream fsOutput = mock(FSDataOutputStream.class); + doThrow(new IOException("close failed")).doNothing().when(fsOutput).close(); + JindoHadoopSystem fs = mock(JindoHadoopSystem.class); + when(fs.open(any(org.apache.hadoop.fs.Path.class))).thenReturn(fsInput); + when(fs.create(any(org.apache.hadoop.fs.Path.class), anyBoolean())).thenReturn(fsOutput); + StubFileIO fileIO = new StubFileIO(fs, INTERVAL); + Path path = new Path("oss://bucket/db/t/data.parquet"); + long inClosed = JindoIOStats.INPUT.closedTotal(); + long outClosed = JindoIOStats.OUTPUT.closedTotal(); + + SeekableInputStream in = fileIO.newInputStream(path); + PositionOutputStream out = fileIO.newOutputStream(path, true); + assertThatThrownBy(in::close).hasMessage("close failed"); + assertThatThrownBy(out::close).hasMessage("close failed"); + assertThat(JindoIOStats.INPUT.closedTotal() - inClosed).isZero(); + assertThat(JindoIOStats.OUTPUT.closedTotal() - outClosed).isZero(); + in.close(); + in.close(); + out.close(); + out.close(); + + verify(fsInput, times(3)).close(); + verify(fsOutput, times(3)).close(); + assertThat(JindoIOStats.INPUT.closedTotal() - inClosed).isEqualTo(1); + assertThat(JindoIOStats.OUTPUT.closedTotal() - outClosed).isEqualTo(1); + } + + @Test + public void testReporterStopsWhenIdle() throws Exception { + Thread previous = reporterThread(); + if (previous != null) { + previous.join(10_000); + } + JindoHadoopSystem fs = mock(JindoHadoopSystem.class); + when(fs.open(any(org.apache.hadoop.fs.Path.class))) + .thenAnswer(i -> new FSDataInputStream(new SeekableBytes(new byte[10]))); + StubFileIO fileIO = new StubFileIO(fs, INTERVAL); + Path path = new Path("oss://bucket/db/t/data.parquet"); + + for (int round = 0; round < 2; round++) { + SeekableInputStream in = fileIO.newInputStream(path); + Thread reporter = reporterThread(); + assertThat(reporter).isNotNull(); + assertThat(reporter.isDaemon()).isTrue(); + assertThat(reporter.getContextClassLoader()) + .isSameAs(JindoIOStats.class.getClassLoader()); + // an open stream keeps the reporter running across intervals + Thread.sleep(INTERVAL.toMillis() * 5); + assertThat(reporter.isAlive()).isTrue(); + + in.close(); + reporter.join(10_000); + assertThat(reporter.isAlive()).isFalse(); + } + } + @Test public void testActivityIncludesCloses() { JindoIOStats.Reporter reporter = - new JindoIOStats.Reporter(60_000, null, java.nio.file.Paths.get("/nonexistent")); + new JindoIOStats.Reporter( + 60_000, Collections.emptyList(), java.nio.file.Paths.get("/nonexistent")); reporter.hasActivity(); assertThat(reporter.hasActivity()).isFalse(); @@ -150,8 +285,10 @@ public void testCgroupV2() throws IOException { java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/kubepods/pod1\n"); - JindoIOStats.Cgroup cgroup = JindoIOStats.Cgroup.detect(procCgroup, root); + List levels = JindoIOStats.Cgroup.detect(procCgroup, root); + assertThat(levels).hasSize(1); + JindoIOStats.Cgroup cgroup = JindoIOStats.Cgroup.closestToLimit(levels); assertThat(cgroup).isNotNull(); assertThat(cgroup.limit()).isEqualTo(2442133504L); assertThat(cgroup.usage()).isEqualTo(2000000000L); @@ -161,6 +298,38 @@ public void testCgroupV2() throws IOException { assertThat(cgroup.oomKills()).isEqualTo(1); } + @Test + public void testCgroupV2AncestorLimit() throws IOException { + java.nio.file.Path root = tempDir.resolve("cgroup"); + java.nio.file.Path pod = root.resolve("pod"); + java.nio.file.Path child = pod.resolve("tm"); + Files.createDirectories(child); + write(root.resolve("cgroup.controllers"), "memory"); + write(pod.resolve("memory.max"), "1000"); + write(pod.resolve("memory.current"), "900"); + write(pod.resolve("memory.stat"), "inactive_file 0\n"); + write(pod.resolve("memory.events"), "max 7\noom_kill 2\n"); + write(child.resolve("memory.max"), "max"); + write(child.resolve("memory.current"), "100"); + write(child.resolve("memory.stat"), "inactive_file 0\n"); + write(child.resolve("memory.events"), "max 0\noom_kill 0\n"); + java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/pod/tm\n"); + + List levels = JindoIOStats.Cgroup.detect(procCgroup, root); + JindoIOStats.Reporter reporter = new JindoIOStats.Reporter(60_000, levels, tempDir); + + assertThat(levels).hasSize(2); + assertThat(reporter.checkAlarm()) + .startsWith("Container memory working set is at 90% of the limit.") + .contains("limitHits=7 oomKills=2"); + + // a limited child still loses to an ancestor that is closer to its limit + write(child.resolve("memory.max"), "500"); + assertThat(JindoIOStats.Cgroup.closestToLimit(levels).limit()).isEqualTo(1000); + write(child.resolve("memory.current"), "480"); + assertThat(JindoIOStats.Cgroup.closestToLimit(levels).limit()).isEqualTo(500); + } + @Test public void testCgroupV1FallsBackToMountRoot() throws IOException { java.nio.file.Path root = tempDir.resolve("cgroup"); @@ -179,8 +348,10 @@ public void testCgroupV1FallsBackToMountRoot() throws IOException { java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "4:cpu,cpuacct:/a\n9:memory:/kubepods/x\n"); - JindoIOStats.Cgroup cgroup = JindoIOStats.Cgroup.detect(procCgroup, root); + List levels = JindoIOStats.Cgroup.detect(procCgroup, root); + assertThat(levels).hasSize(1); + JindoIOStats.Cgroup cgroup = JindoIOStats.Cgroup.closestToLimit(levels); assertThat(cgroup).isNotNull(); assertThat(cgroup.limit()).isEqualTo(2442133504L); assertThat(cgroup.workingSet()).isEqualTo(1000000000L); @@ -191,7 +362,7 @@ public void testCgroupV1FallsBackToMountRoot() throws IOException { @Test public void testNoCgroup() { - assertThat(JindoIOStats.Cgroup.detect(tempDir.resolve("missing"), tempDir)).isNull(); + assertThat(JindoIOStats.Cgroup.detect(tempDir.resolve("missing"), tempDir)).isEmpty(); } @Test @@ -228,12 +399,37 @@ private static java.nio.file.Path write(java.nio.file.Path file, String content) return Files.write(file, content.getBytes(StandardCharsets.UTF_8)); } + private static long events() { + return JindoIOStats.INPUT.openedTotal() + + JindoIOStats.INPUT.closedTotal() + + JindoIOStats.OUTPUT.openedTotal() + + JindoIOStats.OUTPUT.closedTotal(); + } + + @Nullable + private static Thread reporterThread() { + for (Thread thread : Thread.getAllStackTraces().keySet()) { + if ("paimon-jindo-io-stats".equals(thread.getName()) && thread.isAlive()) { + return thread; + } + } + return null; + } + private static class StubFileIO extends HadoopCompliantFileIO { private final JindoHadoopSystem fs; + @Nullable private final Duration ioStatsInterval; - private StubFileIO(JindoHadoopSystem fs) { + private StubFileIO(JindoHadoopSystem fs, @Nullable Duration ioStatsInterval) { this.fs = fs; + this.ioStatsInterval = ioStatsInterval; + } + + @Nullable + @Override + protected Duration ioStatsInterval() { + return ioStatsInterval; } @Override From bc6f7a4db5ca30b6853e80e65bad02df772c34d5 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 27 Oct 2026 00:16:07 +0800 Subject: [PATCH 3/4] [jindo] Fix races in Jindo IO stream accounting and reporter shutdown - Guard the one-time close accounting with an AtomicBoolean. Two concurrent successful closes of one stream could both see it open and decrement the active count twice, leaving open at -1 and keeping the reporter running. - A stream opened between the reporter's idle check and its shutdown saw it running and started no replacement, so its events were never reported. After clearing the running flag, stop() now also checks for events since the idle check and keeps running when there are any, logging them at the next interval. --- .../paimon/jindo/HadoopCompliantFileIO.java | 11 +- .../org/apache/paimon/jindo/JindoIOStats.java | 40 ++++--- .../apache/paimon/jindo/JindoIOStatsTest.java | 102 +++++++++++++++++- 3 files changed, 129 insertions(+), 24 deletions(-) diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java index 2514e0adcdef..56f55315e1ef 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java @@ -47,6 +47,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.paimon.rest.RESTCatalogOptions.IO_CACHE_ENABLED; import static org.apache.paimon.rest.RESTCatalogOptions.IO_CACHE_POLICY; @@ -305,7 +306,7 @@ private static class HadoopSeekableInputStream extends SeekableInputStream { @Nullable protected final JindoIOStats.StreamCounter stats; - private boolean closed; + private final AtomicBoolean closed = new AtomicBoolean(); private HadoopSeekableInputStream( FSDataInputStream in, @Nullable JindoIOStats.StreamCounter stats) { @@ -355,8 +356,7 @@ public int read(byte[] b, int off, int len) throws IOException { public void close() throws IOException { in.close(); // a close that throws is not counted, so that a retried close still reaches Jindo - if (stats != null && !closed) { - closed = true; + if (stats != null && closed.compareAndSet(false, true)) { stats.closed(); } } @@ -411,7 +411,7 @@ private static class HadoopPositionOutputStream extends PositionOutputStream { @Nullable private final JindoIOStats.StreamCounter stats; - private boolean closed; + private final AtomicBoolean closed = new AtomicBoolean(); private HadoopPositionOutputStream( FSDataOutputStream out, @Nullable JindoIOStats.StreamCounter stats) { @@ -457,8 +457,7 @@ public void flush() throws IOException { public void close() throws IOException { out.close(); // a close that throws is not counted, so that a retried close still reaches Jindo - if (stats != null && !closed) { - closed = true; + if (stats != null && closed.compareAndSet(false, true)) { stats.closed(); } } diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java index 8e292047383c..fc1c310c59fb 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java @@ -92,7 +92,8 @@ private static void startReporter() { new Reporter( INTERVAL_MILLIS.get(), Cgroup.detect(Paths.get("/proc/self/cgroup"), Paths.get("/sys/fs/cgroup")), - Paths.get("/proc/self")); + Paths.get("/proc/self"), + RUNNING); Thread thread = new Thread(reporter::run, "paimon-jindo-io-stats"); thread.setDaemon(true); // the inherited loader may belong to the job that opened the stream, keep it collectable @@ -100,12 +101,6 @@ private static void startReporter() { thread.start(); } - /** Returns true when the reporter stopped; streams opened meanwhile keep it running. */ - private static boolean stopReporter() { - RUNNING.set(false); - return (INPUT.open() == 0 && OUTPUT.open() == 0) || !RUNNING.compareAndSet(false, true); - } - /** Registers a file system that is never closed, used to dump the Jindo native metrics. */ static void registerNativeMetrics(JindoHadoopSystem fs, org.apache.hadoop.fs.Path path) { if (nativeSource == null) { @@ -178,6 +173,7 @@ static final class Reporter { private final long intervalMillis; private final List cgroups; private final Path procSelf; + private final AtomicBoolean running; @Nullable private JindoMetricsStore nativeStore; private boolean nativeDisabled; @@ -185,10 +181,11 @@ static final class Reporter { private int alarmLevel; private long lastEvents; - Reporter(long intervalMillis, List cgroups, Path procSelf) { + Reporter(long intervalMillis, List cgroups, Path procSelf, AtomicBoolean running) { this.intervalMillis = intervalMillis; this.cgroups = cgroups; this.procSelf = procSelf; + this.running = running; } /** Runs until an interval passes without stream activity; the next open starts another. */ @@ -199,7 +196,7 @@ void run() { try { Thread.sleep(sampleMillis); } catch (InterruptedException e) { - RUNNING.set(false); + running.set(false); return; } try { @@ -212,7 +209,7 @@ void run() { nextReport = now + intervalMillis; if (hasActivity()) { LOG.info(snapshot()); - } else if (stopReporter()) { + } else if (stop()) { return; } } @@ -253,16 +250,29 @@ String checkAlarm() { /** True while streams are open or when any stream opened or closed since last call. */ boolean hasActivity() { - long events = - INPUT.openedTotal() - + INPUT.closedTotal() - + OUTPUT.openedTotal() - + OUTPUT.closedTotal(); + long events = events(); boolean active = events != lastEvents || INPUT.open() > 0 || OUTPUT.open() > 0; lastEvents = events; return active; } + /** + * Called after an idle {@link #hasActivity()}; returns true when stopped. Opens since then + * saw it running and started no other reporter, so their events keep it running. + */ + boolean stop() { + running.set(false); + boolean missed = events() != lastEvents || INPUT.open() > 0 || OUTPUT.open() > 0; + return !missed || !running.compareAndSet(false, true); + } + + private static long events() { + return INPUT.openedTotal() + + INPUT.closedTotal() + + OUTPUT.openedTotal() + + OUTPUT.closedTotal(); + } + String snapshot() { StringBuilder sb = new StringBuilder("Jindo IO stats:"); appendStreams(sb, "input", INPUT, "read"); diff --git a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java index 5b8581f553ff..1d4649967250 100644 --- a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java +++ b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java @@ -36,6 +36,7 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.Closeable; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -44,7 +45,12 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -207,6 +213,58 @@ public void testFailedCloseIsRetried() throws Exception { assertThat(JindoIOStats.OUTPUT.closedTotal() - outClosed).isEqualTo(1); } + @Test + public void testConcurrentCloseIsCountedOnce() throws Exception { + JindoHadoopSystem fs = mock(JindoHadoopSystem.class); + when(fs.open(any(org.apache.hadoop.fs.Path.class))) + .thenAnswer(i -> new FSDataInputStream(new SeekableBytes(new byte[1]))); + when(fs.create(any(org.apache.hadoop.fs.Path.class), anyBoolean())) + .thenAnswer(i -> new FSDataOutputStream(new ByteArrayOutputStream(), null)); + StubFileIO fileIO = new StubFileIO(fs, INTERVAL); + Path path = new Path("oss://bucket/db/t/data.parquet"); + long inOpen = JindoIOStats.INPUT.open(); + long inClosed = JindoIOStats.INPUT.closedTotal(); + long outOpen = JindoIOStats.OUTPUT.open(); + long outClosed = JindoIOStats.OUTPUT.closedTotal(); + int rounds = 1000; + List streams = new ArrayList<>(); + for (int i = 0; i < rounds; i++) { + streams.add(fileIO.newInputStream(path)); + streams.add(fileIO.newOutputStream(path, true)); + } + + closeConcurrently(streams); + + assertThat(JindoIOStats.INPUT.closedTotal() - inClosed).isEqualTo(rounds); + assertThat(JindoIOStats.INPUT.open()).isEqualTo(inOpen); + assertThat(JindoIOStats.OUTPUT.closedTotal() - outClosed).isEqualTo(rounds); + assertThat(JindoIOStats.OUTPUT.open()).isEqualTo(outOpen); + } + + @Test + public void testStopKeepsEventsSinceIdleCheck() { + AtomicBoolean running = new AtomicBoolean(true); + JindoIOStats.Reporter reporter = + new JindoIOStats.Reporter( + 60_000, + Collections.emptyList(), + java.nio.file.Paths.get("/nonexistent"), + running); + reporter.hasActivity(); + assertThat(reporter.hasActivity()).isFalse(); + + // the stream sees the reporter running, so it starts no other one + JindoIOStats.INPUT.opened(); + JindoIOStats.INPUT.closed(); + assertThat(reporter.stop()).isFalse(); + assertThat(running).isTrue(); + assertThat(reporter.hasActivity()).isTrue(); + + assertThat(reporter.hasActivity()).isFalse(); + assertThat(reporter.stop()).isTrue(); + assertThat(running).isFalse(); + } + @Test public void testReporterStopsWhenIdle() throws Exception { Thread previous = reporterThread(); @@ -240,7 +298,10 @@ public void testReporterStopsWhenIdle() throws Exception { public void testActivityIncludesCloses() { JindoIOStats.Reporter reporter = new JindoIOStats.Reporter( - 60_000, Collections.emptyList(), java.nio.file.Paths.get("/nonexistent")); + 60_000, + Collections.emptyList(), + java.nio.file.Paths.get("/nonexistent"), + new AtomicBoolean()); reporter.hasActivity(); assertThat(reporter.hasActivity()).isFalse(); @@ -316,7 +377,8 @@ public void testCgroupV2AncestorLimit() throws IOException { java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/pod/tm\n"); List levels = JindoIOStats.Cgroup.detect(procCgroup, root); - JindoIOStats.Reporter reporter = new JindoIOStats.Reporter(60_000, levels, tempDir); + JindoIOStats.Reporter reporter = + new JindoIOStats.Reporter(60_000, levels, tempDir, new AtomicBoolean()); assertThat(levels).hasSize(2); assertThat(reporter.checkAlarm()) @@ -376,7 +438,10 @@ public void testAlarmLevels() throws IOException { java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/\n"); JindoIOStats.Reporter reporter = new JindoIOStats.Reporter( - 60_000, JindoIOStats.Cgroup.detect(procCgroup, root), tempDir); + 60_000, + JindoIOStats.Cgroup.detect(procCgroup, root), + tempDir, + new AtomicBoolean()); assertThat(reporter.checkAlarm()).isNull(); write(root.resolve("memory.current"), "810"); @@ -406,6 +471,37 @@ private static long events() { + JindoIOStats.OUTPUT.closedTotal(); } + // two threads meet before each stream and close it together, spinning so neither one parks + private static void closeConcurrently(List streams) throws Exception { + AtomicInteger arrived = new AtomicInteger(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + List> closers = new ArrayList<>(); + for (int t = 0; t < 2; t++) { + closers.add( + executor.submit( + () -> { + for (int i = 0; i < streams.size(); i++) { + arrived.incrementAndGet(); + while (arrived.get() < 2 * (i + 1)) { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + Thread.yield(); + } + streams.get(i).close(); + } + return null; + })); + } + for (Future closer : closers) { + closer.get(60, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + } + @Nullable private static Thread reporterThread() { for (Thread thread : Thread.getAllStackTraces().keySet()) { From 9af315ffbe6c58a39c0a51a36e6ffa19761769e2 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 27 Oct 2026 00:52:40 +0800 Subject: [PATCH 4/4] [jindo] Keep Jindo IO stats best effort and cover multipart uploads - Starting the reporter must not fail I/O: the stream is already open when it is counted, so a failure to create the thread (e.g. no native thread near the memory limit) is now logged, the running flag is cleared, and a later stream retries after one interval. - Use the shortest interval set by any FileIO instead of the first one, and read it on every sample so that a running reporter picks it up. - Keep one alarm level per cgroup level. With a single level, a parent's 95% WARN muted a child that later became the level closest to its limit. - Count Jindo multipart (two-phase) output streams, which bypassed the output counters. Such a stream holds no native stream and cannot retry closeForCommit(), so even a failed attempt counts it closed. - Say that the counters cover the class loader of paimon-jindo rather than the whole JVM, which differs when jobs bundle their own copy. --- .../org/apache/paimon/jindo/JindoFileIO.java | 14 +- .../org/apache/paimon/jindo/JindoIOStats.java | 135 +++++++++++------- .../jindo/JindoTwoPhaseOutputStream.java | 41 +++++- .../apache/paimon/jindo/JindoIOStatsTest.java | 134 ++++++++++++++++- 4 files changed, 266 insertions(+), 58 deletions(-) diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java index a02f19d7780f..5afb528a6234 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java @@ -82,9 +82,12 @@ public class JindoFileIO extends HadoopCompliantFileIO implements HadoopOptionsP .durationType() .defaultValue(Duration.ZERO) .withDescription( - "Interval of logging JVM-wide Jindo stream and memory statistics " - + "while Jindo streams are in use, e.g. 1 min. 0 (the default) " - + "disables it, and streams of this FileIO are not counted."); + "Interval of logging Jindo stream and memory statistics while Jindo " + + "streams are in use, e.g. 1 min. Stream counts cover the " + + "class loader of paimon-jindo, which is the whole JVM unless " + + "jobs bundle their own copy. The shortest interval set in " + + "it is used. 0 (the default) disables it, and streams of " + + "this FileIO are not counted."); private static final Map CASE_SENSITIVE_KEYS = new HashMap() { @@ -221,7 +224,10 @@ public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite return super.newTwoPhaseOutputStream(path, overwrite); } return new JindoTwoPhaseOutputStream( - new JindoMultiPartUpload(mpuStore, fs.getWorkingDirectory()), hadoopPath, path); + new JindoMultiPartUpload(mpuStore, fs.getWorkingDirectory()), + hadoopPath, + path, + ioStatsInterval()); } @Override diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java index fc1c310c59fb..ee6dcddf67db 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java @@ -18,6 +18,7 @@ package org.apache.paimon.jindo; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.utils.JsonSerdeUtil; import org.apache.paimon.utils.Pair; @@ -45,6 +46,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; @@ -52,9 +54,8 @@ import java.util.regex.Pattern; /** - * JVM-wide counters of the Jindo streams of FileIOs that enable them, logged periodically with - * Jindo native, process and container memory; a WARN is logged when the container working set nears - * its limit. + * Counters of the Jindo streams in the class loader of paimon-jindo, which is the whole JVM unless + * jobs bundle their own copy, logged with native, process and container memory and a WARN near OOM. */ public final class JindoIOStats { @@ -63,42 +64,62 @@ public final class JindoIOStats { static final StreamCounter INPUT = new StreamCounter(); static final StreamCounter OUTPUT = new StreamCounter(); - private static final AtomicLong INTERVAL_MILLIS = new AtomicLong(); + static final AtomicLong INTERVAL_MILLIS = new AtomicLong(); private static final AtomicBoolean RUNNING = new AtomicBoolean(false); private static volatile Pair nativeSource; + private static volatile long startRetryMillis; + + @VisibleForTesting + static volatile ThreadFactory reporterThreads = r -> new Thread(r, "paimon-jindo-io-stats"); private JindoIOStats() {} /** * Counts a stream just opened by a FileIO with this interval and returns its counter, or null - * when the interval is null, zero or negative. The first positive interval in the JVM is used. + * when the interval is null, zero or negative. The shortest positive interval is used. */ @Nullable static StreamCounter track(StreamCounter counter, @Nullable Duration interval) { if (interval == null || interval.isZero() || interval.isNegative()) { return null; } - INTERVAL_MILLIS.compareAndSet(0, Math.max(1, interval.toMillis())); + long millis = Math.max(1, interval.toMillis()); + long current = INTERVAL_MILLIS.get(); + if (current == 0 || millis < current) { + INTERVAL_MILLIS.accumulateAndGet(millis, (c, m) -> c == 0 ? m : Math.min(c, m)); + } counter.opened(); startReporter(); return counter; } private static void startReporter() { - if (RUNNING.get() || !RUNNING.compareAndSet(false, true)) { + if (RUNNING.get() + || System.currentTimeMillis() < startRetryMillis + || !RUNNING.compareAndSet(false, true)) { return; } - Reporter reporter = - new Reporter( - INTERVAL_MILLIS.get(), - Cgroup.detect(Paths.get("/proc/self/cgroup"), Paths.get("/sys/fs/cgroup")), - Paths.get("/proc/self"), - RUNNING); - Thread thread = new Thread(reporter::run, "paimon-jindo-io-stats"); - thread.setDaemon(true); - // the inherited loader may belong to the job that opened the stream, keep it collectable - thread.setContextClassLoader(JindoIOStats.class.getClassLoader()); - thread.start(); + try { + Reporter reporter = + new Reporter( + INTERVAL_MILLIS, + Cgroup.detect( + Paths.get("/proc/self/cgroup"), Paths.get("/sys/fs/cgroup")), + Paths.get("/proc/self"), + RUNNING); + Thread thread = reporterThreads.newThread(reporter::run); + thread.setDaemon(true); + // the inherited loader may be a job's, which must stay collectable + thread.setContextClassLoader(JindoIOStats.class.getClassLoader()); + thread.start(); + } catch (Throwable t) { + // e.g. no native thread near the memory limit; the stream must still reach its caller + startRetryMillis = System.currentTimeMillis() + INTERVAL_MILLIS.get(); + RUNNING.set(false); + LOG.warn( + "Failed to start the Jindo IO stats reporter, retrying with a later stream.", + t); + } } /** Registers a file system that is never closed, used to dump the Jindo native metrics. */ @@ -170,31 +191,35 @@ static final class Reporter { private static final double[] ALARM_RATIOS = {0.8, 0.9, 0.95}; private static final double ALARM_RESET_RATIO = 0.7; - private final long intervalMillis; + private final AtomicLong intervalMillis; private final List cgroups; private final Path procSelf; private final AtomicBoolean running; + private final int[] alarmLevels; @Nullable private JindoMetricsStore nativeStore; private boolean nativeDisabled; @Nullable private String jemalloc; - private int alarmLevel; private long lastEvents; - Reporter(long intervalMillis, List cgroups, Path procSelf, AtomicBoolean running) { + Reporter( + AtomicLong intervalMillis, + List cgroups, + Path procSelf, + AtomicBoolean running) { this.intervalMillis = intervalMillis; this.cgroups = cgroups; this.procSelf = procSelf; this.running = running; + this.alarmLevels = new int[cgroups.size()]; } /** Runs until an interval passes without stream activity; the next open starts another. */ void run() { - long sampleMillis = Math.min(SAMPLE_MILLIS, intervalMillis); - long nextReport = System.currentTimeMillis() + intervalMillis; + long lastReport = System.currentTimeMillis(); while (true) { try { - Thread.sleep(sampleMillis); + Thread.sleep(Math.min(SAMPLE_MILLIS, intervalMillis.get())); } catch (InterruptedException e) { running.set(false); return; @@ -205,8 +230,9 @@ void run() { LOG.warn(alarm); } long now = System.currentTimeMillis(); - if (now >= nextReport) { - nextReport = now + intervalMillis; + // read every time, so that a shorter interval set by a later FileIO applies + if (now - lastReport >= intervalMillis.get()) { + lastReport = now; if (hasActivity()) { LOG.info(snapshot()); } else if (stop()) { @@ -219,33 +245,41 @@ void run() { } } - /** Returns a WARN message when the working set enters a higher alarm level. */ + /** Returns a WARN message when the working set of a cgroup level enters its next alarm. */ @Nullable String checkAlarm() { - Cgroup cgroup = Cgroup.closestToLimit(cgroups); - if (cgroup == null) { - return null; - } - long limit = cgroup.limit(); - long workingSet = cgroup.workingSet(); - if (limit <= 0 || workingSet < 0) { - return null; - } - double ratio = (double) workingSet / limit; - if (ratio < ALARM_RESET_RATIO) { - alarmLevel = 0; - } - int level = 0; - while (level < ALARM_RATIOS.length && ratio >= ALARM_RATIOS[level]) { - level++; + Cgroup alarmed = null; + double alarmedRatio = 0; + // each level keeps its own alarm, so a parent's WARN does not mute a child's + for (int i = 0; i < cgroups.size(); i++) { + Cgroup cgroup = cgroups.get(i); + long limit = cgroup.limit(); + long workingSet = cgroup.workingSet(); + if (limit <= 0 || workingSet < 0) { + continue; + } + double ratio = (double) workingSet / limit; + if (ratio < ALARM_RESET_RATIO) { + alarmLevels[i] = 0; + } + int level = 0; + while (level < ALARM_RATIOS.length && ratio >= ALARM_RATIOS[level]) { + level++; + } + if (level > alarmLevels[i]) { + alarmLevels[i] = level; + if (ratio > alarmedRatio) { + alarmed = cgroup; + alarmedRatio = ratio; + } + } } - if (level <= alarmLevel) { + if (alarmed == null) { return null; } - alarmLevel = level; return String.format( "Container memory working set is at %d%% of the limit. %s", - Math.round(ratio * 100), snapshot()); + Math.round(alarmedRatio * 100), snapshot(alarmed)); } /** True while streams are open or when any stream opened or closed since last call. */ @@ -274,12 +308,16 @@ private static long events() { } String snapshot() { + return snapshot(Cgroup.closestToLimit(cgroups)); + } + + private String snapshot(@Nullable Cgroup cgroup) { StringBuilder sb = new StringBuilder("Jindo IO stats:"); appendStreams(sb, "input", INPUT, "read"); appendStreams(sb, "output", OUTPUT, "written"); appendNative(sb); appendProcess(sb); - appendCgroup(sb); + appendCgroup(sb, cgroup); return sb.toString(); } @@ -385,8 +423,7 @@ private void appendProcess(StringBuilder sb) { .append(';'); } - private void appendCgroup(StringBuilder sb) { - Cgroup cgroup = Cgroup.closestToLimit(cgroups); + private static void appendCgroup(StringBuilder sb, @Nullable Cgroup cgroup) { if (cgroup == null) { sb.append(" cgroup unavailable"); return; diff --git a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoTwoPhaseOutputStream.java b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoTwoPhaseOutputStream.java index 3bf5cde03cef..26338eda8604 100644 --- a/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoTwoPhaseOutputStream.java +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoTwoPhaseOutputStream.java @@ -22,18 +22,57 @@ import org.apache.paimon.fs.MultiPartUploadTwoPhaseOutputStream; import org.apache.paimon.fs.Path; +import javax.annotation.Nullable; + import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; /** Jindo implementation of TwoPhaseOutputStream using multipart upload. */ public class JindoTwoPhaseOutputStream extends MultiPartUploadTwoPhaseOutputStream { + @Nullable private final JindoIOStats.StreamCounter stats; + + private final AtomicBoolean closed = new AtomicBoolean(); + public JindoTwoPhaseOutputStream( MultiPartUploadStore multiPartUploadStore, org.apache.hadoop.fs.Path hadoopPath, - Path targetPath) + Path targetPath, + @Nullable Duration ioStatsInterval) throws IOException { super(multiPartUploadStore, hadoopPath, targetPath); + // counted once the upload has started, so a failed start leaves nothing open + this.stats = JindoIOStats.track(JindoIOStats.OUTPUT, ioStatsInterval); + } + + @Override + public void write(int b) throws IOException { + super.write(b); + if (stats != null) { + stats.bytes(1); + } + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + super.write(b, off, len); + if (stats != null) { + stats.bytes(len); + } + } + + @Override + public Committer closeForCommit() throws IOException { + try { + return super.closeForCommit(); + } finally { + // even a failed attempt ends the stream: it holds no native stream and cannot retry + if (stats != null && closed.compareAndSet(false, true)) { + stats.closed(); + } + } } @Override diff --git a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java index 1d4649967250..444746a524b0 100644 --- a/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java +++ b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java @@ -19,6 +19,7 @@ package org.apache.paimon.jindo; import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.fs.MultiPartUploadStore; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.SeekableInputStream; @@ -48,9 +49,11 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -193,6 +196,7 @@ public void testFailedCloseIsRetried() throws Exception { when(fs.create(any(org.apache.hadoop.fs.Path.class), anyBoolean())).thenReturn(fsOutput); StubFileIO fileIO = new StubFileIO(fs, INTERVAL); Path path = new Path("oss://bucket/db/t/data.parquet"); + long inOpen = JindoIOStats.INPUT.open(); long inClosed = JindoIOStats.INPUT.closedTotal(); long outClosed = JindoIOStats.OUTPUT.closedTotal(); @@ -202,6 +206,17 @@ public void testFailedCloseIsRetried() throws Exception { assertThatThrownBy(out::close).hasMessage("close failed"); assertThat(JindoIOStats.INPUT.closedTotal() - inClosed).isZero(); assertThat(JindoIOStats.OUTPUT.closedTotal() - outClosed).isZero(); + + // until a close succeeds the stream may hold native memory, so it stays open and reported + assertThat(JindoIOStats.INPUT.open() - inOpen).isEqualTo(1); + JindoIOStats.Reporter reporter = + new JindoIOStats.Reporter( + new AtomicLong(60_000), + Collections.emptyList(), + java.nio.file.Paths.get("/nonexistent"), + new AtomicBoolean()); + reporter.hasActivity(); + assertThat(reporter.hasActivity()).isTrue(); in.close(); in.close(); out.close(); @@ -246,7 +261,7 @@ public void testStopKeepsEventsSinceIdleCheck() { AtomicBoolean running = new AtomicBoolean(true); JindoIOStats.Reporter reporter = new JindoIOStats.Reporter( - 60_000, + new AtomicLong(60_000), Collections.emptyList(), java.nio.file.Paths.get("/nonexistent"), running); @@ -294,11 +309,88 @@ public void testReporterStopsWhenIdle() throws Exception { } } + @Test + public void testReporterStartFailureKeepsStreams() throws Exception { + Thread previous = reporterThread(); + if (previous != null) { + previous.join(10_000); + } + JindoHadoopSystem fs = mock(JindoHadoopSystem.class); + when(fs.open(any(org.apache.hadoop.fs.Path.class))) + .thenAnswer(i -> new FSDataInputStream(new SeekableBytes(new byte[10]))); + StubFileIO fileIO = new StubFileIO(fs, INTERVAL); + Path path = new Path("oss://bucket/db/t/data.parquet"); + long inOpen = JindoIOStats.INPUT.open(); + ThreadFactory threads = JindoIOStats.reporterThreads; + try { + JindoIOStats.reporterThreads = + r -> { + throw new OutOfMemoryError("unable to create native thread"); + }; + try (SeekableInputStream in = fileIO.newInputStream(path)) { + assertThat(in.read(new byte[10], 0, 10)).isEqualTo(10); + assertThat(JindoIOStats.INPUT.open() - inOpen).isEqualTo(1); + assertThat(reporterThread()).isNull(); + } + } finally { + JindoIOStats.reporterThreads = threads; + } + + // a later stream starts the reporter once the retry delay has passed + Thread.sleep(INTERVAL.toMillis() * 3); + try (SeekableInputStream in = fileIO.newInputStream(path)) { + assertThat(reporterThread()).isNotNull(); + } + assertThat(JindoIOStats.INPUT.open()).isEqualTo(inOpen); + } + + @Test + public void testShortestIntervalWins() { + JindoIOStats.StreamCounter counter = new JindoIOStats.StreamCounter(); + JindoIOStats.track(counter, INTERVAL); + long interval = JindoIOStats.INTERVAL_MILLIS.get(); + assertThat(interval).isLessThanOrEqualTo(INTERVAL.toMillis()); + + JindoIOStats.track(counter, Duration.ofHours(1)); + assertThat(JindoIOStats.INTERVAL_MILLIS.get()).isEqualTo(interval); + JindoIOStats.track(counter, Duration.ofMillis(interval - 1)); + assertThat(JindoIOStats.INTERVAL_MILLIS.get()).isEqualTo(interval - 1); + } + + @Test + public void testTwoPhaseOutputStreamIsCounted() throws Exception { + @SuppressWarnings("unchecked") + MultiPartUploadStore store = + mock(MultiPartUploadStore.class); + when(store.startMultiPartUpload(any())).thenReturn("upload"); + org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path("oss://bucket/t/f"); + Path path = new Path("oss://bucket/t/f"); + long opened = JindoIOStats.OUTPUT.openedTotal(); + long closed = JindoIOStats.OUTPUT.closedTotal(); + long written = JindoIOStats.OUTPUT.bytesTotal(); + + JindoTwoPhaseOutputStream out = + new JindoTwoPhaseOutputStream(store, hadoopPath, path, INTERVAL); + out.write(new byte[30]); + out.write(new byte[30], 5, 20); + out.write(1); + out.closeForCommit(); + out.close(); + new JindoTwoPhaseOutputStream(store, hadoopPath, path, null).close(); + when(store.startMultiPartUpload(any())).thenThrow(new IOException("start failed")); + assertThatThrownBy(() -> new JindoTwoPhaseOutputStream(store, hadoopPath, path, INTERVAL)) + .hasMessage("start failed"); + + assertThat(JindoIOStats.OUTPUT.openedTotal() - opened).isEqualTo(1); + assertThat(JindoIOStats.OUTPUT.closedTotal() - closed).isEqualTo(1); + assertThat(JindoIOStats.OUTPUT.bytesTotal() - written).isEqualTo(51); + } + @Test public void testActivityIncludesCloses() { JindoIOStats.Reporter reporter = new JindoIOStats.Reporter( - 60_000, + new AtomicLong(60_000), Collections.emptyList(), java.nio.file.Paths.get("/nonexistent"), new AtomicBoolean()); @@ -378,7 +470,8 @@ public void testCgroupV2AncestorLimit() throws IOException { List levels = JindoIOStats.Cgroup.detect(procCgroup, root); JindoIOStats.Reporter reporter = - new JindoIOStats.Reporter(60_000, levels, tempDir, new AtomicBoolean()); + new JindoIOStats.Reporter( + new AtomicLong(60_000), levels, tempDir, new AtomicBoolean()); assertThat(levels).hasSize(2); assertThat(reporter.checkAlarm()) @@ -392,6 +485,39 @@ public void testCgroupV2AncestorLimit() throws IOException { assertThat(JindoIOStats.Cgroup.closestToLimit(levels).limit()).isEqualTo(500); } + @Test + public void testAlarmsPerCgroupLevel() throws IOException { + java.nio.file.Path root = tempDir.resolve("cgroup"); + java.nio.file.Path pod = root.resolve("pod"); + java.nio.file.Path child = pod.resolve("tm"); + Files.createDirectories(child); + write(root.resolve("cgroup.controllers"), "memory"); + write(pod.resolve("memory.max"), "1000"); + write(pod.resolve("memory.current"), "960"); + write(pod.resolve("memory.stat"), "inactive_file 0\n"); + write(pod.resolve("memory.events"), "max 7\noom_kill 0\n"); + write(child.resolve("memory.max"), "500"); + write(child.resolve("memory.current"), "100"); + write(child.resolve("memory.stat"), "inactive_file 0\n"); + write(child.resolve("memory.events"), "max 3\noom_kill 0\n"); + java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/pod/tm\n"); + JindoIOStats.Reporter reporter = + new JindoIOStats.Reporter( + new AtomicLong(60_000), + JindoIOStats.Cgroup.detect(procCgroup, root), + tempDir, + new AtomicBoolean()); + + assertThat(reporter.checkAlarm()).contains("at 96% of the limit").contains("limitHits=7"); + // the parent eases but stays above the reset ratio while the child fills up + write(pod.resolve("memory.current"), "750"); + write(child.resolve("memory.current"), "450"); + assertThat(reporter.checkAlarm()).contains("at 90% of the limit").contains("limitHits=3"); + write(child.resolve("memory.current"), "490"); + assertThat(reporter.checkAlarm()).contains("at 98% of the limit"); + assertThat(reporter.checkAlarm()).isNull(); + } + @Test public void testCgroupV1FallsBackToMountRoot() throws IOException { java.nio.file.Path root = tempDir.resolve("cgroup"); @@ -438,7 +564,7 @@ public void testAlarmLevels() throws IOException { java.nio.file.Path procCgroup = write(tempDir.resolve("cgroup-file"), "0::/\n"); JindoIOStats.Reporter reporter = new JindoIOStats.Reporter( - 60_000, + new AtomicLong(60_000), JindoIOStats.Cgroup.detect(procCgroup, root), tempDir, new AtomicBoolean());