Add optimized / direct read stats for non-cached files

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/searchablesnapshots/SearchableSnapshotShardStats.java

@@ -118,14 +118,12 @@ public int hashCode() {
         return Objects.hash(shardRouting, snapshotId, indexId, inputStats);
     }
 
-
     public static class CacheIndexInputStats implements Writeable, ToXContentObject {
 
         private final String fileName;
         private final long fileLength;
 
         private final long openCount;
-        private final long innerCount;
         private final long closeCount;
 
         private final Counter forwardSmallSeeks;
@@ -137,17 +135,17 @@ public static class CacheIndexInputStats implements Writeable, ToXContentObject
         private final Counter cachedBytesRead;
         private final TimedCounter cachedBytesWritten;
         private final TimedCounter directBytesRead;
+        private final TimedCounter optimizedBytesRead;
 
-        public CacheIndexInputStats(String fileName, long fileLength, long openCount, long innerCount, long closeCount,
+        public CacheIndexInputStats(String fileName, long fileLength, long openCount, long closeCount,
                                     Counter forwardSmallSeeks, Counter backwardSmallSeeks,
                                     Counter forwardLargeSeeks, Counter backwardLargeSeeks,
                                     Counter contiguousReads, Counter nonContiguousReads,
                                     Counter cachedBytesRead, TimedCounter cachedBytesWritten,
-                                    TimedCounter directBytesRead) {
+                                    TimedCounter directBytesRead, TimedCounter optimizedBytesRead) {
             this.fileName = fileName;
             this.fileLength = fileLength;
             this.openCount = openCount;
-            this.innerCount = innerCount;
             this.closeCount = closeCount;
             this.forwardSmallSeeks = forwardSmallSeeks;
             this.backwardSmallSeeks = backwardSmallSeeks;
@@ -158,13 +156,13 @@ public CacheIndexInputStats(String fileName, long fileLength, long openCount, lo
             this.cachedBytesRead = cachedBytesRead;
             this.cachedBytesWritten = cachedBytesWritten;
             this.directBytesRead = directBytesRead;
+            this.optimizedBytesRead = optimizedBytesRead;
         }
 
         CacheIndexInputStats(final StreamInput in) throws IOException {
             this.fileName = in.readString();
             this.fileLength = in.readVLong();
             this.openCount = in.readVLong();
-            this.innerCount = in.readVLong();
             this.closeCount = in.readVLong();
             this.forwardSmallSeeks = new Counter(in);
             this.backwardSmallSeeks = new Counter(in);
@@ -175,14 +173,14 @@ public CacheIndexInputStats(String fileName, long fileLength, long openCount, lo
             this.cachedBytesRead = new Counter(in);
             this.cachedBytesWritten = new TimedCounter(in);
             this.directBytesRead = new TimedCounter(in);
+            this.optimizedBytesRead = new TimedCounter(in);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(fileName);
             out.writeVLong(fileLength);
             out.writeVLong(openCount);
-            out.writeVLong(innerCount);
             out.writeVLong(closeCount);
 
             forwardSmallSeeks.writeTo(out);
@@ -194,6 +192,7 @@ public void writeTo(StreamOutput out) throws IOException {
             cachedBytesRead.writeTo(out);
             cachedBytesWritten.writeTo(out);
             directBytesRead.writeTo(out);
+            optimizedBytesRead.writeTo(out);
         }
 
         public String getFileName() {
@@ -208,10 +207,6 @@ public long getOpenCount() {
             return openCount;
         }
 
-        public long getInnerCount() {
-            return innerCount;
-        }
-
         public long getCloseCount() {
             return closeCount;
         }
@@ -252,20 +247,24 @@ public TimedCounter getDirectBytesRead() {
             return directBytesRead;
         }
 
+        public TimedCounter getOptimizedBytesRead() {
+            return optimizedBytesRead;
+        }
+
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
             {
                 builder.field("name", getFileName());
                 builder.field("length", getFileLength());
                 builder.field("open_count", getOpenCount());
-                builder.field("inner_count", getInnerCount());
                 builder.field("close_count", getCloseCount());
                 builder.field("contiguous_bytes_read", getContiguousReads());
                 builder.field("non_contiguous_bytes_read", getNonContiguousReads());
                 builder.field("cached_bytes_read", getCachedBytesRead());
                 builder.field("cached_bytes_written", getCachedBytesWritten());
                 builder.field("direct_bytes_read", getDirectBytesRead());
+                builder.field("optimized_bytes_read", getOptimizedBytesRead());
                 {
                     builder.startObject("forward_seeks");
                     builder.field("small", getForwardSmallSeeks());
@@ -293,7 +292,6 @@ public boolean equals(Object other) {
             CacheIndexInputStats stats = (CacheIndexInputStats) other;
             return fileLength == stats.fileLength
                 && openCount == stats.openCount
-                && innerCount == stats.innerCount
                 && closeCount == stats.closeCount
                 && Objects.equals(fileName, stats.fileName)
                 && Objects.equals(forwardSmallSeeks, stats.forwardSmallSeeks)
@@ -304,17 +302,18 @@ public boolean equals(Object other) {
                 && Objects.equals(nonContiguousReads, stats.nonContiguousReads)
                 && Objects.equals(cachedBytesRead, stats.cachedBytesRead)
                 && Objects.equals(cachedBytesWritten, stats.cachedBytesWritten)
-                && Objects.equals(directBytesRead, stats.directBytesRead);
+                && Objects.equals(directBytesRead, stats.directBytesRead)
+                && Objects.equals(optimizedBytesRead, stats.optimizedBytesRead);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(fileName, fileLength, openCount, innerCount, closeCount,
+            return Objects.hash(fileName, fileLength, openCount, closeCount,
                 forwardSmallSeeks, backwardSmallSeeks,
                 forwardLargeSeeks, backwardLargeSeeks,
                 contiguousReads, nonContiguousReads,
                 cachedBytesRead, cachedBytesWritten,
-                directBytesRead);
+                directBytesRead, optimizedBytesRead);
         }
     }
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/searchablesnapshots/SearchableSnapshotShardStatsTests.java

@@ -42,12 +42,12 @@ protected SearchableSnapshotShardStats createTestInstance() {
 
     private CacheIndexInputStats randomCacheIndexInputStats() {
         return new CacheIndexInputStats(randomAlphaOfLength(10), randomNonNegativeLong(),
-            randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(),
+            randomNonNegativeLong(), randomNonNegativeLong(),
             randomCounter(), randomCounter(),
             randomCounter(), randomCounter(),
             randomCounter(), randomCounter(),
             randomCounter(), randomTimedCounter(),
-            randomTimedCounter());
+            randomTimedCounter(), randomTimedCounter());
     }
 
     private Counter randomCounter() {

x-pack/plugin/searchable-snapshots/qa/rest/src/test/resources/rest-api-spec/test/stats.yml

@@ -147,7 +147,6 @@ teardown:
   - is_true: indices.docs.shards.0.0.files.0.name
   - gt:      { indices.docs.shards.0.0.files.0.length: 0 }
   - gt:      { indices.docs.shards.0.0.files.0.open_count: 0 }
-  - gte:     { indices.docs.shards.0.0.files.0.inner_count: 0 }
   - gt:      { indices.docs.shards.0.0.files.0.close_count: 0 }
 
   - gte:     { indices.docs.shards.0.0.files.0.contiguous_bytes_read.count: 0 }
@@ -179,6 +178,13 @@ teardown:
   - gte:     { indices.docs.shards.0.0.files.0.direct_bytes_read.time_in_nanos: 0 }
   - is_false:  indices.docs.shards.0.0.files.0.direct_bytes_read.time
 
+  - gte:     { indices.docs.shards.0.0.files.0.optimized_bytes_read.count: 0 }
+  - gte:     { indices.docs.shards.0.0.files.0.optimized_bytes_read.sum: 0 }
+  - gte:     { indices.docs.shards.0.0.files.0.optimized_bytes_read.min: 0 }
+  - gte:     { indices.docs.shards.0.0.files.0.optimized_bytes_read.max: 0 }
+  - gte:     { indices.docs.shards.0.0.files.0.optimized_bytes_read.time_in_nanos: 0 }
+  - is_false:  indices.docs.shards.0.0.files.0.optimized_bytes_read.time
+
   - gte:     { indices.docs.shards.0.0.files.0.forward_seeks.small.count: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.forward_seeks.small.sum: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.forward_seeks.small.min: 0 }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/BaseSearchableSnapshotIndexInput.java

@@ -16,33 +16,88 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public abstract class BaseSearchableSnapshotIndexInput extends BufferedIndexInput {
 
     protected final BlobContainer blobContainer;
     protected final FileInfo fileInfo;
     protected final IOContext context;
+    protected final IndexInputStats stats;
+    protected final long offset;
+    protected final long length;
 
-    public BaseSearchableSnapshotIndexInput(String resourceDesc, BlobContainer blobContainer, FileInfo fileInfo, IOContext context) {
+    // the following are only mutable so they can be adjusted after cloning/slicing
+    protected volatile boolean isClone;
+    private AtomicBoolean closed;
+
+    public BaseSearchableSnapshotIndexInput(
+        String resourceDesc,
+        BlobContainer blobContainer,
+        FileInfo fileInfo,
+        IOContext context,
+        IndexInputStats stats,
+        long offset,
+        long length
+    ) {
         super(resourceDesc, context);
         this.blobContainer = Objects.requireNonNull(blobContainer);
         this.fileInfo = Objects.requireNonNull(fileInfo);
         this.context = Objects.requireNonNull(context);
         assert fileInfo.metadata().hashEqualsContents() == false
             : "this method should only be used with blobs that are NOT stored in metadata's hash field (fileInfo: " + fileInfo + ')';
+        this.stats = Objects.requireNonNull(stats);
+        this.offset = offset;
+        this.length = length;
+        this.closed = new AtomicBoolean(false);
+        this.isClone = false;
     }
 
     public BaseSearchableSnapshotIndexInput(
         String resourceDesc,
         BlobContainer blobContainer,
         FileInfo fileInfo,
         IOContext context,
+        IndexInputStats stats,
+        long offset,
+        long length,
         int bufferSize
     ) {
-        this(resourceDesc, blobContainer, fileInfo, context);
+        this(resourceDesc, blobContainer, fileInfo, context, stats, offset, length);
         setBufferSize(bufferSize);
     }
 
+    @Override
+    public final long length() {
+        return length;
+    }
+
+    @Override
+    public BaseSearchableSnapshotIndexInput clone() {
+        final BaseSearchableSnapshotIndexInput clone = (BaseSearchableSnapshotIndexInput) super.clone();
+        clone.closed = new AtomicBoolean(false);
+        clone.isClone = true;
+        return clone;
+    }
+
+    protected void ensureOpen() throws IOException {
+        if (closed.get()) {
+            throw new IOException(toString() + " is closed");
+        }
+    }
+
+    @Override
+    public final void close() throws IOException {
+        if (closed.compareAndSet(false, true)) {
+            if (isClone == false) {
+                stats.incrementCloseCount();
+            }
+            innerClose();
+        }
+    }
+
+    public abstract void innerClose() throws IOException;
+
     protected InputStream openInputStream(final long position, final long length) throws IOException {
         assert assertCurrentThreadMayAccessBlobStore();
         final long startPart = getPartNumberForPosition(position);

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/IndexInputStats.java

@@ -13,6 +13,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.LongConsumer;
+import java.util.function.LongSupplier;
 
 /**
  * {@link IndexInputStats} records stats for a given {@link CachedBlobContainerIndexInput}.
@@ -24,9 +25,9 @@ public class IndexInputStats {
 
     private final long fileLength;
     private final long seekingThreshold;
+    private final LongSupplier currentTimeNanos;
 
     private final LongAdder opened = new LongAdder();
-    private final LongAdder inner = new LongAdder();
     private final LongAdder closed = new LongAdder();
 
     private final Counter forwardSmallSeeks = new Counter();
@@ -39,25 +40,30 @@ public class IndexInputStats {
     private final Counter nonContiguousReads = new Counter();
 
     private final TimedCounter directBytesRead = new TimedCounter();
+    private final TimedCounter optimizedBytesRead = new TimedCounter();
 
     private final Counter cachedBytesRead = new Counter();
     private final TimedCounter cachedBytesWritten = new TimedCounter();
 
-    public IndexInputStats(long fileLength) {
-        this(fileLength, SEEKING_THRESHOLD.getBytes());
+    public IndexInputStats(long fileLength, LongSupplier currentTimeNanos) {
+        this(fileLength, SEEKING_THRESHOLD.getBytes(), currentTimeNanos);
     }
 
-    public IndexInputStats(long fileLength, long seekingThreshold) {
+    public IndexInputStats(long fileLength, long seekingThreshold, LongSupplier currentTimeNanos) {
         this.fileLength = fileLength;
         this.seekingThreshold = seekingThreshold;
+        this.currentTimeNanos = currentTimeNanos;
     }
 
-    public void incrementOpenCount() {
-        opened.increment();
+    /**
+     * @return the current time in nanoseconds that should be used to measure statistics.
+     */
+    public long currentTimeNanos() {
+        return currentTimeNanos.getAsLong();
     }
 
-    public void incrementInnerOpenCount() {
-        inner.increment();
+    public void incrementOpenCount() {
+        opened.increment();
     }
 
     public void incrementCloseCount() {
@@ -76,6 +82,10 @@ public void addDirectBytesRead(int bytesRead, long nanoseconds) {
         directBytesRead.add(bytesRead, nanoseconds);
     }
 
+    public void addOptimizedBytesRead(int bytesRead, long nanoseconds) {
+        optimizedBytesRead.add(bytesRead, nanoseconds);
+    }
+
     public void incrementBytesRead(long previousPosition, long currentPosition, int bytesRead) {
         LongConsumer incBytesRead = (previousPosition == currentPosition) ? contiguousReads::add : nonContiguousReads::add;
         incBytesRead.accept(bytesRead);
@@ -110,10 +120,6 @@ public LongAdder getOpened() {
         return opened;
     }
 
-    public LongAdder getInnerOpened() {
-        return inner;
-    }
-
     public LongAdder getClosed() {
         return closed;
     }
@@ -146,6 +152,10 @@ public TimedCounter getDirectBytesRead() {
         return directBytesRead;
     }
 
+    public TimedCounter getOptimizedBytesRead() {
+        return optimizedBytesRead;
+    }
+
     public Counter getCachedBytesRead() {
         return cachedBytesRead;
     }