Fix searchable snapshot stats fwd/bwd seeking counters (#52760)

This commit fixes the backward/forward seeking counters in 
searchable snapshot statistics. These counters use the current 
file pointer position and the new seeking position to compute the 
seeking stats, but at the time the stat is computed the current 
file pointer position is already updated to the new position, 
making the delta always equals to 0.

Author: tlrx

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

@@ -86,10 +86,16 @@ public Map<String, IndexInputStats> getStats() {
     }
 
     // pkg private for tests
-    @Nullable IndexInputStats getStats(String name) {
+    @Nullable
+    IndexInputStats getStats(String name) {
         return stats.get(name);
     }
 
+    // pkg private so tests can override
+    IndexInputStats createIndexInputStats(final long fileLength) {
+        return new IndexInputStats(fileLength);
+    }
+
     public void close() throws IOException {
         super.close();
         // Ideally we could let the cache evict/remove cached files by itself after the
@@ -101,7 +107,7 @@ public void close() throws IOException {
     public IndexInput openInput(final String name, final IOContext context) throws IOException {
         ensureOpen();
         final long fileLength = fileLength(name);
-        return new CacheBufferedIndexInput(name, fileLength, context, stats.computeIfAbsent(name, n -> new IndexInputStats(fileLength)));
+        return new CacheBufferedIndexInput(name, fileLength, context, stats.computeIfAbsent(name, n -> createIndexInputStats(fileLength)));
     }
 
     private class CacheFileReference implements CacheFile.EvictionListener {
@@ -184,6 +190,8 @@ public class CacheBufferedIndexInput extends BufferedIndexInput {
 
         // last read position is kept around in order to detect (non)contiguous reads for stats
         private long lastReadPosition;
+        // last seek position is kept around in order to detect forward/backward seeks for stats
+        private long lastSeekPosition;
 
         CacheBufferedIndexInput(String fileName, long fileLength, IOContext ioContext, IndexInputStats stats) {
             this(new CacheFileReference(fileName, fileLength), ioContext, stats,
@@ -201,6 +209,8 @@ private CacheBufferedIndexInput(CacheFileReference cacheFileReference, IOContext
             this.end = offset + length;
             this.closed = new AtomicBoolean(false);
             this.isClone = isClone;
+            this.lastReadPosition = this.offset;
+            this.lastSeekPosition = this.offset;
         }
 
         @Override
@@ -260,6 +270,7 @@ protected void readInternal(final byte[] buffer, final int offset, final int len
             assert totalBytesRead == length : "partial read operation, read [" + totalBytesRead + "] bytes of [" + length + "]";
             stats.incrementBytesRead(lastReadPosition, position, totalBytesRead);
             lastReadPosition = position + totalBytesRead;
+            lastSeekPosition = lastReadPosition;
         }
 
         int readCacheFile(FileChannel fc, long end, long position, byte[] buffer, int offset, long length) throws IOException {
@@ -302,7 +313,9 @@ 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);
+            final long position = pos + this.offset;
+            stats.incrementSeeks(lastSeekPosition, position);
+            lastSeekPosition = position;
         }
 
         @Override

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

@@ -156,7 +156,7 @@ public TimedCounter getCachedBytesWritten() {
     }
 
     @SuppressForbidden(reason = "Handles Long.MIN_VALUE before using Math.abs()")
-    private boolean isLargeSeek(long delta) {
+    boolean isLargeSeek(long delta) {
         return delta != Long.MIN_VALUE && Math.abs(delta) > seekingThreshold;
     }
 

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/xpack/searchablesnapshots/cache/CacheBufferedIndexInputStatsTests.java

@@ -294,6 +294,132 @@ public void testReadBytesNonContiguously() throws Exception {
         });
     }
 
+    public void testForwardSeeks() throws Exception {
+        // use default cache service settings
+        final CacheService cacheService = new CacheService(Settings.EMPTY);
+
+        executeTestCase(cacheService, (fileName, fileContent, cacheDirectory) -> {
+            final IOContext ioContext = newIOContext(random());
+            try (IndexInput indexInput = cacheDirectory.openInput(fileName, ioContext)) {
+                IndexInput input = indexInput;
+                if (randomBoolean()) {
+                    final long sliceOffset = randomLongBetween(0L, input.length() - 1L);
+                    final long sliceLength = randomLongBetween(1L, input.length() - sliceOffset);
+                    input = input.slice("slice", sliceOffset, sliceLength);
+                }
+                if (randomBoolean()) {
+                    input = input.clone();
+                }
+
+                final IndexInputStats inputStats = cacheDirectory.getStats(fileName);
+                final IndexInputStats.Counter forwardSmallSeeksCounter = inputStats.getForwardSmallSeeks();
+                assertCounter(forwardSmallSeeksCounter, 0L, 0L, 0L, 0L);
+
+                long totalSmallSeeks = 0L;
+                long countSmallSeeks = 0L;
+                long minSmallSeeks = Long.MAX_VALUE;
+                long maxSmallSeeks = Long.MIN_VALUE;
+
+                final IndexInputStats.Counter forwardLargeSeeksCounter = inputStats.getForwardLargeSeeks();
+                assertCounter(forwardLargeSeeksCounter, 0L, 0L, 0L, 0L);
+
+                long totalLargeSeeks = 0L;
+                long countLargeSeeks = 0L;
+                long minLargeSeeks = Long.MAX_VALUE;
+                long maxLargeSeeks = Long.MIN_VALUE;
+
+                while (input.getFilePointer() < input.length()) {
+                    long moveForward = randomLongBetween(1L, input.length() - input.getFilePointer());
+                    input.seek(input.getFilePointer() + moveForward);
+
+                    if (inputStats.isLargeSeek(moveForward)) {
+                        minLargeSeeks = (moveForward < minLargeSeeks) ? moveForward : minLargeSeeks;
+                        maxLargeSeeks = (moveForward > maxLargeSeeks) ? moveForward : maxLargeSeeks;
+                        totalLargeSeeks += moveForward;
+                        countLargeSeeks += 1;
+
+                        assertCounter(forwardLargeSeeksCounter, totalLargeSeeks, countLargeSeeks, minLargeSeeks, maxLargeSeeks);
+
+                    } else {
+                        minSmallSeeks = (moveForward < minSmallSeeks) ? moveForward : minSmallSeeks;
+                        maxSmallSeeks = (moveForward > maxSmallSeeks) ? moveForward : maxSmallSeeks;
+                        totalSmallSeeks += moveForward;
+                        countSmallSeeks += 1;
+
+                        assertCounter(forwardSmallSeeksCounter, totalSmallSeeks, countSmallSeeks, minSmallSeeks, maxSmallSeeks);
+                    }
+                }
+            } catch (IOException e) {
+                throw new AssertionError(e);
+            }
+        });
+    }
+
+    public void testBackwardSeeks() throws Exception {
+        // use default cache service settings
+        final CacheService cacheService = new CacheService(Settings.EMPTY);
+
+        executeTestCase(cacheService, (fileName, fileContent, cacheDirectory) -> {
+            final IOContext ioContext = newIOContext(random());
+            try (IndexInput indexInput = cacheDirectory.openInput(fileName, ioContext)) {
+                IndexInput input = indexInput;
+                if (randomBoolean()) {
+                    final long sliceOffset = randomLongBetween(0L, input.length() - 1L);
+                    final long sliceLength = randomLongBetween(1L, input.length() - sliceOffset);
+                    input = input.slice("slice", sliceOffset, sliceLength);
+                }
+                if (randomBoolean()) {
+                    input = input.clone();
+                }
+
+                final IndexInputStats inputStats = cacheDirectory.getStats(fileName);
+                final IndexInputStats.Counter backwardSmallSeeks = inputStats.getBackwardSmallSeeks();
+                assertCounter(backwardSmallSeeks, 0L, 0L, 0L, 0L);
+
+                long totalSmallSeeks = 0L;
+                long countSmallSeeks = 0L;
+                long minSmallSeeks = Long.MAX_VALUE;
+                long maxSmallSeeks = Long.MIN_VALUE;
+
+                final IndexInputStats.Counter backwardLargeSeeks = inputStats.getBackwardLargeSeeks();
+                assertCounter(backwardLargeSeeks, 0L, 0L, 0L, 0L);
+
+                long totalLargeSeeks = 0L;
+                long countLargeSeeks = 0L;
+                long minLargeSeeks = Long.MAX_VALUE;
+                long maxLargeSeeks = Long.MIN_VALUE;
+
+                input.seek(input.length());
+                assertThat(input.getFilePointer(), equalTo(input.length()));
+
+                do {
+                    long moveBackward = -1L * randomLongBetween(1L, input.getFilePointer());
+                    input.seek(input.getFilePointer() + moveBackward);
+
+                    if (inputStats.isLargeSeek(moveBackward)) {
+                        minLargeSeeks = (moveBackward < minLargeSeeks) ? moveBackward : minLargeSeeks;
+                        maxLargeSeeks = (moveBackward > maxLargeSeeks) ? moveBackward : maxLargeSeeks;
+                        totalLargeSeeks += moveBackward;
+                        countLargeSeeks += 1;
+
+                        assertCounter(backwardLargeSeeks, totalLargeSeeks, countLargeSeeks, minLargeSeeks, maxLargeSeeks);
+
+                    } else {
+                        minSmallSeeks = (moveBackward < minSmallSeeks) ? moveBackward : minSmallSeeks;
+                        maxSmallSeeks = (moveBackward > maxSmallSeeks) ? moveBackward : maxSmallSeeks;
+                        totalSmallSeeks += moveBackward;
+                        countSmallSeeks += 1;
+
+                        assertCounter(backwardSmallSeeks, totalSmallSeeks, countSmallSeeks, minSmallSeeks, maxSmallSeeks);
+                    }
+
+                } while (input.getFilePointer() > 0L);
+            } catch (IOException e) {
+                throw new AssertionError(e);
+            }
+        });
+    }
+
     private static void executeTestCase(CacheService cacheService, TriConsumer<String, byte[], CacheDirectory> test) throws Exception {
         final byte[] fileContent = randomUnicodeOfLength(randomIntBetween(10, MAX_FILE_LENGTH)).getBytes(StandardCharsets.UTF_8);
         final String fileName = randomAlphaOfLength(10);
@@ -302,10 +428,21 @@ private static void executeTestCase(CacheService cacheService, TriConsumer<Strin
         final ShardId shardId = new ShardId("_name", "_uuid", 0);
         final AtomicLong fakeClock = new AtomicLong();
 
+        final Long seekingThreshold = randomBoolean() ? randomLongBetween(1L, fileContent.length) : null;
+
         try (CacheService ignored = cacheService;
              Directory directory = newDirectory();
-             CacheDirectory cacheDirectory = new CacheDirectory(directory, cacheService, createTempDir(), snapshotId, indexId, shardId,
-                 () -> fakeClock.addAndGet(FAKE_CLOCK_ADVANCE_NANOS))
+             CacheDirectory cacheDirectory =
+                 new CacheDirectory(directory, cacheService, createTempDir(), snapshotId, indexId, shardId,
+                     () -> fakeClock.addAndGet(FAKE_CLOCK_ADVANCE_NANOS)) {
+                     @Override
+                     IndexInputStats createIndexInputStats(long fileLength) {
+                         if (seekingThreshold == null) {
+                             return super.createIndexInputStats(fileLength);
+                         }
+                         return new IndexInputStats(fileLength, seekingThreshold);
+                     }
+                 }
         ) {
             cacheService.start();
             assertThat(cacheDirectory.getStats(fileName), nullValue());