[jindo] Log Jindo stream and memory statistics

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,18 +137,28 @@ 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
     public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
         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,22 +390,33 @@ 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;
         }
     }
 
     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();
+            }
         }
     }
 

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<Duration> 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<String, String> CASE_SENSITIVE_KEYS =
             new HashMap<String, String>() {
                 {
@@ -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<JindoHadoopSystem, String> createFileSystem(
                 };
 
         if (allowCache) {
-            return CACHE.computeIfAbsent(
-                    new CacheKey(options, scheme, authority), key -> supplier.get());
+            Pair<JindoHadoopSystem, String> 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) {