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..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 @@ -37,13 +37,17 @@ 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; 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; @@ -133,9 +137,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException { JindoHadoopSystem fs = pair.getKey(); String sysType = pair.getValue(); FSDataInputStream fsInput = fs.open(hadoopPath); + 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 @@ -143,8 +149,16 @@ 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); + FSDataOutputStream fsOutput = + getFileSystem(hadoopPath, shouldCache).create(hadoopPath, overwrite); return new HadoopPositionOutputStream( - getFileSystem(hadoopPath, shouldCache).create(hadoopPath, overwrite)); + 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 @@ -290,8 +304,14 @@ private static class HadoopSeekableInputStream extends SeekableInputStream { protected final FSDataInputStream in; - private HadoopSeekableInputStream(FSDataInputStream in) { + @Nullable protected final JindoIOStats.StreamCounter stats; + + private final AtomicBoolean closed = new AtomicBoolean(); + + private HadoopSeekableInputStream( + FSDataInputStream in, @Nullable JindoIOStats.StreamCounter stats) { this.in = in; + this.stats = stats; } @Override @@ -316,17 +336,29 @@ public long getPos() throws IOException { @Override public int read() throws IOException { - return in.read(); + int b = in.read(); + if (b >= 0 && stats != null) { + stats.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); + if (stats != null) { + stats.bytes(n); + } + return n; } @Override 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.compareAndSet(false, true)) { + stats.closed(); + } } /** @@ -358,13 +390,18 @@ 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 { - return in.read(position, bytes, off, len); + int n = in.read(position, bytes, off, len); + if (stats != null) { + stats.bytes(n); + } + return n; } } @@ -372,8 +409,14 @@ private static class HadoopPositionOutputStream extends PositionOutputStream { private final FSDataOutputStream out; - private HadoopPositionOutputStream(FSDataOutputStream out) { + @Nullable private final JindoIOStats.StreamCounter stats; + + private final AtomicBoolean closed = new AtomicBoolean(); + + private HadoopPositionOutputStream( + FSDataOutputStream out, @Nullable JindoIOStats.StreamCounter stats) { this.out = out; + this.stats = stats; } @Override @@ -384,16 +427,25 @@ public long getPos() throws IOException { @Override public void write(int b) throws IOException { out.write(b); + if (stats != null) { + stats.bytes(1); + } } @Override public void write(byte[] b) throws IOException { out.write(b); + if (stats != null) { + stats.bytes(b.length); + } } @Override public void write(byte[] b, int off, int len) throws IOException { out.write(b, off, len); + if (stats != null) { + stats.bytes(len); + } } @Override @@ -404,6 +456,10 @@ public void flush() throws IOException { @Override 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.compareAndSet(false, 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 71468672af9f..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 @@ -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; @@ -40,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; @@ -73,6 +77,18 @@ 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 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() { { @@ -92,6 +108,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 +126,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()) { @@ -206,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 @@ -293,13 +314,24 @@ 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(); } } + @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 new file mode 100644 index 000000000000..ee6dcddf67db --- /dev/null +++ b/paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java @@ -0,0 +1,671 @@ +/* + * 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.annotation.VisibleForTesting; +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.ArrayList; +import java.util.Collections; +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; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * 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 { + + private static final Logger LOG = LoggerFactory.getLogger(JindoIOStats.class); + + static final StreamCounter INPUT = new StreamCounter(); + static final StreamCounter OUTPUT = new StreamCounter(); + + 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 shortest positive interval is used. + */ + @Nullable + static StreamCounter track(StreamCounter counter, @Nullable Duration interval) { + if (interval == null || interval.isZero() || interval.isNegative()) { + return null; + } + 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() + || System.currentTimeMillis() < startRetryMillis + || !RUNNING.compareAndSet(false, true)) { + return; + } + 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. */ + 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 active = new AtomicLong(); + private final AtomicLong peak = new AtomicLong(); + private final LongAdder bytes = new LongAdder(); + + void opened() { + opened.incrementAndGet(); + peak.accumulateAndGet(active.incrementAndGet(), Math::max); + } + + void closed() { + closed.incrementAndGet(); + active.decrementAndGet(); + } + + void bytes(long n) { + if (n > 0) { + bytes.add(n); + } + } + + long open() { + return active.get(); + } + + long openedTotal() { + return opened.get(); + } + + long closedTotal() { + return closed.get(); + } + + long bytesTotal() { + return bytes.sum(); + } + + /** + * 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 until the streams go idle. */ + 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 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 long lastEvents; + + 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 lastReport = System.currentTimeMillis(); + while (true) { + try { + Thread.sleep(Math.min(SAMPLE_MILLIS, intervalMillis.get())); + } catch (InterruptedException e) { + running.set(false); + return; + } + try { + String alarm = checkAlarm(); + if (alarm != null) { + LOG.warn(alarm); + } + long now = System.currentTimeMillis(); + // 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()) { + return; + } + } + } catch (Throwable t) { + LOG.debug("Failed to collect Jindo IO stats.", t); + } + } + } + + /** Returns a WARN message when the working set of a cgroup level enters its next alarm. */ + @Nullable + String checkAlarm() { + 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 (alarmed == null) { + return null; + } + return String.format( + "Container memory working set is at %d%% of the limit. %s", + Math.round(alarmedRatio * 100), snapshot(alarmed)); + } + + /** True while streams are open or when any stream opened or closed since last call. */ + boolean hasActivity() { + 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() { + 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, cgroup); + return sb.toString(); + } + + private static void appendStreams( + StringBuilder sb, String name, StreamCounter counter, String bytesName) { + long open = counter.open(); + sb.append(' ') + .append(name) + .append(" open=") + .append(open) + .append(" peak=") + .append(counter.takePeak(open)) + .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 static void appendCgroup(StringBuilder sb, @Nullable Cgroup cgroup) { + 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; + } + + /** One memory cgroup (v1 or v2) level 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; + } + + /** + * 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 Collections.emptyList(); + } + 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); + } + } + 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 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) { + 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(usageFile(v2)); + } + + /** 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/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 new file mode 100644 index 000000000000..444746a524b0 --- /dev/null +++ b/paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/JindoIOStatsTest.java @@ -0,0 +1,711 @@ +/* + * 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.MultiPartUploadStore; +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 javax.annotation.Nullable; + +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; +import java.time.Duration; +import java.util.ArrayList; +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.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; +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 + 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(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 + 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, INTERVAL); + 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 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 inOpen = JindoIOStats.INPUT.open(); + 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(); + + // 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(); + 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 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( + new AtomicLong(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(); + 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 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( + new AtomicLong(60_000), + Collections.emptyList(), + java.nio.file.Paths.get("/nonexistent"), + new AtomicBoolean()); + 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"); + + 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); + assertThat(cgroup.workingSet()).isEqualTo(1900000000L); + assertThat(cgroup.peak()).isEqualTo(2100000000L); + assertThat(cgroup.limitHits()).isEqualTo(12); + 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( + new AtomicLong(60_000), levels, tempDir, new AtomicBoolean()); + + 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 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"); + 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"); + + 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); + 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)).isEmpty(); + } + + @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( + new AtomicLong(60_000), + JindoIOStats.Cgroup.detect(procCgroup, root), + tempDir, + new AtomicBoolean()); + + 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 long events() { + return JindoIOStats.INPUT.openedTotal() + + JindoIOStats.INPUT.closedTotal() + + JindoIOStats.OUTPUT.openedTotal() + + 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()) { + 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, @Nullable Duration ioStatsInterval) { + this.fs = fs; + this.ioStatsInterval = ioStatsInterval; + } + + @Nullable + @Override + protected Duration ioStatsInterval() { + return ioStatsInterval; + } + + @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); + } + } +}