Add cache directory low-level instrumentation (#51637)

This commit is a first step to add instrumentation to the CacheDirectory 
added in #50693. It adds a new mutable IndexInputStats object that allows 
to track various information about how CacheBufferedIndexInput interacts 
with the underlying cache file to satisfy the read operations. It keep tracks 
of small/large forward/backward seekings as well as the total number of 
bytes read from the IndexInput and the total number of bytes read/written 
from/to the CacheFile.

Note that the stats do not reflect the exact usage that Lucene does of the 
IndexInputs opened through a CacheDirectory: IndexInputStats is not aware 
of the read operations that are directly served at a higher level by the 
internal BufferedIndexInput's buffer. Instead it tracks what really hit the disk.

Author: tlrx

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/cache/CacheDirectory.java

@@ -5,6 +5,9 @@
  */
 package org.elasticsearch.xpack.searchablesnapshots.cache;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.Directory;
@@ -14,6 +17,7 @@
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.io.Channels;
+import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.repositories.IndexId;
@@ -25,6 +29,7 @@
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -34,8 +39,10 @@
  */
 public class CacheDirectory extends FilterDirectory {
 
+    private static final Logger logger = LogManager.getLogger(CacheDirectory.class);
     private static final int COPY_BUFFER_SIZE = 8192;
 
+    private final Map<String, IndexInputStats> stats;
     private final CacheService cacheService;
     private final SnapshotId snapshotId;
     private final IndexId indexId;
@@ -45,6 +52,7 @@ public class CacheDirectory extends FilterDirectory {
     public CacheDirectory(Directory in, CacheService cacheService, Path cacheDir, SnapshotId snapshotId, IndexId indexId, ShardId shardId)
         throws IOException {
         super(in);
+        this.stats = ConcurrentCollections.newConcurrentMapWithAggressiveConcurrency();
         this.cacheService = Objects.requireNonNull(cacheService);
         this.cacheDir = Files.createDirectories(cacheDir);
         this.snapshotId = Objects.requireNonNull(snapshotId);
@@ -56,6 +64,11 @@ private CacheKey createCacheKey(String fileName) {
         return new CacheKey(snapshotId, indexId, shardId, fileName);
     }
 
+    // pkg private for tests
+    @Nullable IndexInputStats getStats(String name) {
+        return stats.get(name);
+    }
+
     public void close() throws IOException {
         super.close();
         // Ideally we could let the cache evict/remove cached files by itself after the
@@ -66,7 +79,8 @@ public void close() throws IOException {
     @Override
     public IndexInput openInput(final String name, final IOContext context) throws IOException {
         ensureOpen();
-        return new CacheBufferedIndexInput(name, fileLength(name), context);
+        final long fileLength = fileLength(name);
+        return new CacheBufferedIndexInput(name, fileLength, context, stats.computeIfAbsent(name, n -> new IndexInputStats(fileLength)));
     }
 
     private class CacheFileReference implements CacheFile.EvictionListener {
@@ -141,22 +155,28 @@ public class CacheBufferedIndexInput extends BufferedIndexInput {
         private final long offset;
         private final long end;
         private final CacheFileReference cacheFileReference;
+        private final IndexInputStats stats;
 
         // the following are only mutable so they can be adjusted after cloning
         private AtomicBoolean closed;
         private boolean isClone;
 
-        CacheBufferedIndexInput(String fileName, long fileLength, IOContext ioContext) {
-            this(new CacheFileReference(fileName, fileLength), ioContext,
+        // last read position is kept around in order to detect (non)contiguous reads for stats
+        private long lastReadPosition;
+
+        CacheBufferedIndexInput(String fileName, long fileLength, IOContext ioContext, IndexInputStats stats) {
+            this(new CacheFileReference(fileName, fileLength), ioContext, stats,
                 "CachedBufferedIndexInput(" + fileName + ")", 0L, fileLength, false);
+            stats.incrementOpenCount();
         }
 
-        private CacheBufferedIndexInput(CacheFileReference cacheFileReference, IOContext ioContext, String desc, long offset, long length,
-                                        boolean isClone) {
+        private CacheBufferedIndexInput(CacheFileReference cacheFileReference, IOContext ioContext, IndexInputStats stats,
+                                        String desc, long offset, long length, boolean isClone) {
             super(desc, ioContext);
             this.ioContext = ioContext;
             this.offset = offset;
             this.cacheFileReference = cacheFileReference;
+            this.stats = stats;
             this.end = offset + length;
             this.closed = new AtomicBoolean(false);
             this.isClone = isClone;
@@ -171,6 +191,7 @@ public long length() {
         public void close() {
             if (closed.compareAndSet(false, true)) {
                 if (isClone == false) {
+                    stats.incrementCloseCount();
                     cacheFileReference.releaseOnClose();
                 }
             }
@@ -180,20 +201,21 @@ public void close() {
         protected void readInternal(final byte[] buffer, final int offset, final int length) throws IOException {
             final long position = getFilePointer() + this.offset;
 
-            int bytesRead = 0;
-            while (bytesRead < length) {
-                final long pos = position + bytesRead;
-                final int off = offset + bytesRead;
-                final int len = length - bytesRead;
+            int totalBytesRead = 0;
+            while (totalBytesRead < length) {
+                final long pos = position + totalBytesRead;
+                final int off = offset + totalBytesRead;
+                final int len = length - totalBytesRead;
 
+                int bytesRead = 0;
                 try {
                     final CacheFile cacheFile = cacheFileReference.get();
                     if (cacheFile == null) {
                         throw new AlreadyClosedException("Failed to acquire a non-evicted cache file");
                     }
 
                     try (ReleasableLock ignored = cacheFile.fileLock()) {
-                        bytesRead += cacheFile.fetchRange(pos,
+                        bytesRead = cacheFile.fetchRange(pos,
                             (start, end) -> readCacheFile(cacheFile.getChannel(), end, pos, buffer, off, len),
                             (start, end) -> writeCacheFile(cacheFile.getChannel(), start, end))
                             .get();
@@ -202,33 +224,42 @@ protected void readInternal(final byte[] buffer, final int offset, final int len
                     if (e instanceof AlreadyClosedException || (e.getCause() != null && e.getCause() instanceof AlreadyClosedException)) {
                         try {
                             // cache file was evicted during the range fetching, read bytes directly from source
-                            bytesRead += readDirectly(pos, pos + len, buffer, off);
+                            bytesRead = readDirectly(pos, pos + len, buffer, off);
                             continue;
                         } catch (Exception inner) {
                             e.addSuppressed(inner);
                         }
                     }
                     throw new IOException("Fail to read data from cache", e);
 
+                } finally {
+                    totalBytesRead += bytesRead;
                 }
             }
-            assert bytesRead == length : "partial read operation, read [" + bytesRead + "] bytes of [" + length + "]";
+            assert totalBytesRead == length : "partial read operation, read [" + totalBytesRead + "] bytes of [" + length + "]";
+            stats.incrementBytesRead(lastReadPosition, position, totalBytesRead);
+            lastReadPosition = position + totalBytesRead;
         }
 
         int readCacheFile(FileChannel fc, long end, long position, byte[] buffer, int offset, long length) throws IOException {
             assert assertFileChannelOpen(fc);
-            return Channels.readFromFileChannel(fc, position, buffer, offset, Math.toIntExact(Math.min(length, end - position)));
+            int bytesRead = Channels.readFromFileChannel(fc, position, buffer, offset, Math.toIntExact(Math.min(length, end - position)));
+            stats.addCachedBytesRead(bytesRead);
+            return bytesRead;
         }
 
         @SuppressForbidden(reason = "Use positional writes on purpose")
         void writeCacheFile(FileChannel fc, long start, long end) throws IOException {
             assert assertFileChannelOpen(fc);
             final byte[] copyBuffer = new byte[Math.toIntExact(Math.min(COPY_BUFFER_SIZE, end - start))];
+            logger.trace(() -> new ParameterizedMessage("writing range [{}-{}] to cache file [{}]", start, end, cacheFileReference));
+
+            int bytesCopied = 0;
             try (IndexInput input = in.openInput(cacheFileReference.getFileName(), ioContext)) {
+                stats.incrementInnerOpenCount();
                 if (start > 0) {
                     input.seek(start);
                 }
-                int bytesCopied = 0;
                 long remaining = end - start;
                 while (remaining > 0) {
                     final int size = (remaining < copyBuffer.length) ? Math.toIntExact(remaining) : copyBuffer.length;
@@ -237,6 +268,7 @@ void writeCacheFile(FileChannel fc, long start, long end) throws IOException {
                     bytesCopied += size;
                     remaining -= size;
                 }
+                stats.addCachedBytesWritten(bytesCopied);
             }
         }
 
@@ -247,6 +279,7 @@ protected void seekInternal(long pos) throws IOException {
             } else if (pos < 0L) {
                 throw new IOException("Seeking to negative position [" + pos + "] for " + toString());
             }
+            stats.incrementSeeks(getFilePointer(), pos);
         }
 
         @Override
@@ -263,7 +296,7 @@ public IndexInput slice(String sliceDescription, long offset, long length) {
                 throw new IllegalArgumentException("slice() " + sliceDescription + " out of bounds: offset=" + offset
                     + ",length=" + length + ",fileLength=" + this.length() + ": " + this);
             }
-            return new CacheBufferedIndexInput(cacheFileReference, ioContext,
+            return new CacheBufferedIndexInput(cacheFileReference, ioContext, stats,
                 getFullSliceDescription(sliceDescription), this.offset + offset, length, true);
         }
 
@@ -280,9 +313,12 @@ public String toString() {
 
         private int readDirectly(long start, long end, byte[] buffer, int offset) throws IOException {
             final byte[] copyBuffer = new byte[Math.toIntExact(Math.min(COPY_BUFFER_SIZE, end - start))];
+            logger.trace(() ->
+                new ParameterizedMessage("direct reading of range [{}-{}] for cache file [{}]", start, end, cacheFileReference));
 
             int bytesCopied = 0;
             try (IndexInput input = in.openInput(cacheFileReference.getFileName(), ioContext)) {
+                stats.incrementInnerOpenCount();
                 if (start > 0) {
                     input.seek(start);
                 }
@@ -294,6 +330,7 @@ private int readDirectly(long start, long end, byte[] buffer, int offset) throws
                     bytesCopied += len;
                     remaining -= len;
                 }
+                stats.addDirectBytesRead(bytesCopied);
             }
             return bytesCopied;
         }