Add stats for time spent fetching data while searching snapshots

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

@@ -9,6 +9,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.repositories.IndexId;
@@ -134,15 +135,15 @@ public static class CacheIndexInputStats implements Writeable, ToXContentObject
         private final Counter contiguousReads;
         private final Counter nonContiguousReads;
         private final Counter cachedBytesRead;
-        private final Counter cachedBytesWritten;
-        private final Counter directBytesRead;
+        private final TimedCounter cachedBytesWritten;
+        private final TimedCounter directBytesRead;
 
         public CacheIndexInputStats(String fileName, long fileLength, long openCount, long innerCount, long closeCount,
                                     Counter forwardSmallSeeks, Counter backwardSmallSeeks,
                                     Counter forwardLargeSeeks, Counter backwardLargeSeeks,
                                     Counter contiguousReads, Counter nonContiguousReads,
-                                    Counter cachedBytesRead, Counter cachedBytesWritten,
-                                    Counter directBytesRead) {
+                                    Counter cachedBytesRead, TimedCounter cachedBytesWritten,
+                                    TimedCounter directBytesRead) {
             this.fileName = fileName;
             this.fileLength = fileLength;
             this.openCount = openCount;
@@ -172,8 +173,8 @@ public CacheIndexInputStats(String fileName, long fileLength, long openCount, lo
             this.contiguousReads = new Counter(in);
             this.nonContiguousReads = new Counter(in);
             this.cachedBytesRead = new Counter(in);
-            this.cachedBytesWritten = new Counter(in);
-            this.directBytesRead = new Counter(in);
+            this.cachedBytesWritten = new TimedCounter(in);
+            this.directBytesRead = new TimedCounter(in);
         }
 
         @Override
@@ -243,11 +244,11 @@ public Counter getCachedBytesRead() {
             return cachedBytesRead;
         }
 
-        public Counter getCachedBytesWritten() {
+        public TimedCounter getCachedBytesWritten() {
             return cachedBytesWritten;
         }
 
-        public Counter getDirectBytesRead() {
+        public TimedCounter getDirectBytesRead() {
             return directBytesRead;
         }
 
@@ -347,18 +348,22 @@ public void writeTo(final StreamOutput out) throws IOException {
         }
 
         @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        public final XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
             {
                 builder.field("count", count);
                 builder.field("sum", total);
                 builder.field("min", min);
                 builder.field("max", max);
+                innerToXContent(builder, params);
             }
             builder.endObject();
             return builder;
         }
 
+        void innerToXContent(XContentBuilder builder, Params params) throws IOException {
+        }
+
         public long getCount() {
             return count;
         }
@@ -395,4 +400,54 @@ public int hashCode() {
             return Objects.hash(count, total, min, max);
         }
     }
+
+    public static class TimedCounter extends Counter {
+
+        private final long totalNanoseconds;
+
+        public TimedCounter(long count, long total, long min, long max, long totalNanoseconds) {
+            super(count, total, min, max);
+            this.totalNanoseconds = totalNanoseconds;
+        }
+
+        TimedCounter(StreamInput in) throws IOException {
+            super(in);
+            totalNanoseconds = in.readZLong();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeZLong(totalNanoseconds);
+        }
+
+        @Override
+        void innerToXContent(XContentBuilder builder, Params params) throws IOException {
+            if (builder.humanReadable()) {
+                builder.field("time", TimeValue.timeValueNanos(totalNanoseconds).toString());
+            }
+            builder.field("time_in_nanos", totalNanoseconds);
+        }
+
+        @Override
+        public boolean equals(Object other) {
+            if (this == other) {
+                return true;
+            }
+            if (other == null || getClass() != other.getClass()) {
+                return false;
+            }
+            if (super.equals(other) == false) {
+                return false;
+            }
+            TimedCounter that = (TimedCounter) other;
+            return totalNanoseconds == that.totalNanoseconds;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(super.hashCode(), totalNanoseconds);
+        }
+    }
+
 }

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

@@ -14,6 +14,7 @@
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats.CacheIndexInputStats;
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats.Counter;
+import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats.TimedCounter;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -45,11 +46,15 @@ private CacheIndexInputStats randomCacheIndexInputStats() {
             randomCounter(), randomCounter(),
             randomCounter(), randomCounter(),
             randomCounter(), randomCounter(),
-            randomCounter(), randomCounter(),
-            randomCounter());
+            randomCounter(), randomTimedCounter(),
+            randomTimedCounter());
     }
 
     private Counter randomCounter() {
         return new Counter(randomLong(), randomLong(), randomLong(), randomLong());
     }
+
+    private TimedCounter randomTimedCounter() {
+        return new TimedCounter(randomLong(), randomLong(), randomLong(), randomLong(), randomLong());
+    }
 }

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

@@ -167,11 +167,15 @@ teardown:
   - gte:     { indices.docs.shards.0.0.files.0.cached_bytes_written.sum: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.cached_bytes_written.min: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.cached_bytes_written.max: 0 }
+  - gte:     { indices.docs.shards.0.0.files.0.cached_bytes_written.time_in_nanos: 0 }
+  - is_false:  indices.docs.shards.0.0.files.0.cached_bytes_written.time
 
   - gte:     { indices.docs.shards.0.0.files.0.direct_bytes_read.count: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.direct_bytes_read.sum: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.direct_bytes_read.min: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.direct_bytes_read.max: 0 }
+  - 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.forward_seeks.small.count: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.forward_seeks.small.sum: 0 }
@@ -190,3 +194,11 @@ teardown:
   - gte:     { indices.docs.shards.0.0.files.0.backward_seeks.large.sum: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.backward_seeks.large.min: 0 }
   - gte:     { indices.docs.shards.0.0.files.0.backward_seeks.large.max: 0 }
+
+  - do:
+      searchable_snapshots.stats:
+        index: "d*"
+        human: true
+
+  - is_true:   indices.docs.shards.0.0.files.0.cached_bytes_written.time
+  - is_true:   indices.docs.shards.0.0.files.0.direct_bytes_read.time

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

@@ -35,6 +35,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.function.Function;
+import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.index.IndexModule.INDEX_STORE_TYPE_SETTING;
@@ -74,7 +75,8 @@ public SearchableSnapshotRepository(Repository in) {
         blobStoreRepository = (BlobStoreRepository) in;
     }
 
-    private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath, CacheService cacheService) throws IOException {
+    private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath, CacheService cacheService,
+                                    LongSupplier currentTimeNanosSupplier) throws IOException {
 
         IndexId indexId = new IndexId(indexSettings.getIndex().getName(), SNAPSHOT_INDEX_ID_SETTING.get(indexSettings.getSettings()));
         BlobContainer blobContainer = blobStoreRepository.shardContainer(indexId, shardPath.getShardId().id());
@@ -86,7 +88,8 @@ private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath
         Directory directory = new SearchableSnapshotDirectory(snapshot, blobContainer);
         if (SNAPSHOT_CACHE_ENABLED_SETTING.get(indexSettings.getSettings())) {
             final Path cacheDir = shardPath.getDataPath().resolve("snapshots").resolve(snapshotId.getUUID());
-            directory = new CacheDirectory(directory, cacheService, cacheDir, snapshotId, indexId, shardPath.getShardId());
+            directory = new CacheDirectory(directory, cacheService, cacheDir, snapshotId, indexId, shardPath.getShardId(),
+                currentTimeNanosSupplier);
         }
         directory = new InMemoryNoOpCommitDirectory(directory);
 
@@ -146,7 +149,8 @@ public Repository create(RepositoryMetaData metaData, Function<String, Factory>
     }
 
     public static IndexStorePlugin.DirectoryFactory newDirectoryFactory(final Supplier<RepositoriesService> repositoriesService,
-                                                                        final Supplier<CacheService> cacheService) {
+                                                                        final Supplier<CacheService> cacheService,
+                                                                        final LongSupplier currentTimeNanosSupplier) {
         return (indexSettings, shardPath) -> {
             final RepositoriesService repositories = repositoriesService.get();
             assert repositories != null;
@@ -159,7 +163,7 @@ public static IndexStorePlugin.DirectoryFactory newDirectoryFactory(final Suppli
             final CacheService cache = cacheService.get();
             assert cache != null;
 
-            return ((SearchableSnapshotRepository) repository).makeDirectory(indexSettings, shardPath, cache);
+            return ((SearchableSnapshotRepository) repository).makeDirectory(indexSettings, shardPath, cache, currentTimeNanosSupplier);
         };
     }
 }

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

@@ -106,7 +106,7 @@ public void onRepositoriesModule(RepositoriesModule repositoriesModule) {
     @Override
     public Map<String, DirectoryFactory> getDirectoryFactories() {
         return Map.of(SearchableSnapshotRepository.SNAPSHOT_DIRECTORY_FACTORY_KEY,
-            SearchableSnapshotRepository.newDirectoryFactory(repositoriesService::get, cacheService::get));
+            SearchableSnapshotRepository.newDirectoryFactory(repositoriesService::get, cacheService::get, System::nanoTime));
     }
 
     @Override

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/TransportSearchableSnapshotsStatsAction.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats;
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats.CacheIndexInputStats;
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats.Counter;
+import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats.TimedCounter;
 import org.elasticsearch.xpack.searchablesnapshots.InMemoryNoOpCommitDirectory;
 import org.elasticsearch.xpack.searchablesnapshots.cache.CacheDirectory;
 import org.elasticsearch.xpack.searchablesnapshots.cache.IndexInputStats;
@@ -122,14 +123,18 @@ private static CacheIndexInputStats toCacheIndexInputStats(final String fileName
             toCounter(inputStats.getForwardSmallSeeks()), toCounter(inputStats.getBackwardSmallSeeks()),
             toCounter(inputStats.getForwardLargeSeeks()), toCounter(inputStats.getBackwardLargeSeeks()),
             toCounter(inputStats.getContiguousReads()), toCounter(inputStats.getNonContiguousReads()),
-            toCounter(inputStats.getCachedBytesRead()), toCounter(inputStats.getCachedBytesWritten()),
-            toCounter(inputStats.getDirectBytesRead()));
+            toCounter(inputStats.getCachedBytesRead()), toTimedCounter(inputStats.getCachedBytesWritten()),
+            toTimedCounter(inputStats.getDirectBytesRead()));
     }
 
     private static Counter toCounter(final IndexInputStats.Counter counter) {
         return new Counter(counter.count(), counter.total(), counter.min(), counter.max());
     }
 
+    private static TimedCounter toTimedCounter(final IndexInputStats.TimedCounter counter) {
+        return new TimedCounter(counter.count(), counter.total(), counter.min(), counter.max(), counter.totalNanoseconds());
+    }
+
     @Nullable
     private static CacheDirectory unwrap(Directory dir) {
         while (dir != null) {

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

@@ -34,6 +34,7 @@
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.LongSupplier;
 
 /**
  * {@link CacheDirectory} uses a {@link CacheService} to cache Lucene files provided by another {@link Directory}.
@@ -49,8 +50,10 @@ public class CacheDirectory extends FilterDirectory {
     private final IndexId indexId;
     private final ShardId shardId;
     private final Path cacheDir;
+    private final LongSupplier currentTimeNanosSupplier;
 
-    public CacheDirectory(Directory in, CacheService cacheService, Path cacheDir, SnapshotId snapshotId, IndexId indexId, ShardId shardId)
+    public CacheDirectory(Directory in, CacheService cacheService, Path cacheDir, SnapshotId snapshotId, IndexId indexId, ShardId shardId,
+                          LongSupplier currentTimeNanosSupplier)
         throws IOException {
         super(in);
         this.stats = ConcurrentCollections.newConcurrentMapWithAggressiveConcurrency();
@@ -59,6 +62,7 @@ public CacheDirectory(Directory in, CacheService cacheService, Path cacheDir, Sn
         this.snapshotId = Objects.requireNonNull(snapshotId);
         this.indexId = Objects.requireNonNull(indexId);
         this.shardId = Objects.requireNonNull(shardId);
+        this.currentTimeNanosSupplier = Objects.requireNonNull(currentTimeNanosSupplier);
     }
 
     private CacheKey createCacheKey(String fileName) {
@@ -274,6 +278,7 @@ void writeCacheFile(FileChannel fc, long start, long end) throws IOException {
             int bytesCopied = 0;
             try (IndexInput input = in.openInput(cacheFileReference.getFileName(), ioContext)) {
                 stats.incrementInnerOpenCount();
+                final long startTimeNanos = currentTimeNanosSupplier.getAsLong();
                 if (start > 0) {
                     input.seek(start);
                 }
@@ -285,7 +290,8 @@ void writeCacheFile(FileChannel fc, long start, long end) throws IOException {
                     bytesCopied += size;
                     remaining -= size;
                 }
-                stats.addCachedBytesWritten(bytesCopied);
+                final long endTimeNanos = currentTimeNanosSupplier.getAsLong();
+                stats.addCachedBytesWritten(bytesCopied, endTimeNanos - startTimeNanos);
             }
         }
 
@@ -336,6 +342,7 @@ private int readDirectly(long start, long end, byte[] buffer, int offset) throws
             int bytesCopied = 0;
             try (IndexInput input = in.openInput(cacheFileReference.getFileName(), ioContext)) {
                 stats.incrementInnerOpenCount();
+                final long startTimeNanos = currentTimeNanosSupplier.getAsLong();
                 if (start > 0) {
                     input.seek(start);
                 }
@@ -347,7 +354,8 @@ private int readDirectly(long start, long end, byte[] buffer, int offset) throws
                     bytesCopied += len;
                     remaining -= len;
                 }
-                stats.addDirectBytesRead(bytesCopied);
+                final long endTimeNanos = currentTimeNanosSupplier.getAsLong();
+                stats.addDirectBytesRead(bytesCopied, endTimeNanos - startTimeNanos);
             }
             return bytesCopied;
         }