Add translog files age to Translog Stats (#28613)

rest-api-spec/src/main/resources/rest-api-spec/test/indices.stats/20_translog.yml

@@ -33,6 +33,7 @@ setup:
 # non empty generation with one op may be smaller or larger than that.
 #  - gt: { indices.test.primaries.translog.uncommitted_size_in_bytes: $creation_size }
   - match: { indices.test.primaries.translog.uncommitted_operations: 1 }
+  - gte: { indices.test.primaries.translog.earliest_last_modified_age: 0 }
 
   - do:
       indices.flush:
@@ -46,6 +47,7 @@ setup:
   ## creation translog size has some overhead due to an initial empty generation that will be trimmed later
   - lt: { indices.test.primaries.translog.uncommitted_size_in_bytes: $creation_size }
   - match: { indices.test.primaries.translog.uncommitted_operations: 0 }
+  - gte: { indices.test.primaries.translog.earliest_last_modified_age: 0 }
 
   - do:
       indices.put_settings:
@@ -67,3 +69,4 @@ setup:
   - match: { indices.test.primaries.translog.operations: 0 }
   - lte: { indices.test.primaries.translog.uncommitted_size_in_bytes: $creation_size }
   - match: { indices.test.primaries.translog.uncommitted_operations: 0 }
+  - gte: { indices.test.primaries.translog.earliest_last_modified_age: 0 }

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -385,6 +385,26 @@ public long sizeInBytes() {
         return sizeInBytesByMinGen(-1);
     }
 
+    long earliestLastModifiedAge() {
+        try (ReleasableLock ignored = readLock.acquire()) {
+            ensureOpen();
+            return findEarliestLastModifiedAge(System.currentTimeMillis(), readers, current);
+        } catch (IOException e) {
+            throw new TranslogException(shardId, "Unable to get the earliest last modified time for the transaction log");
+        }
+    }
+
+    /**
+     * Returns the age of the oldest entry in the translog files in seconds
+     */
+    static long findEarliestLastModifiedAge(long currentTime, Iterable<TranslogReader> readers, TranslogWriter writer) throws IOException {
+        long earliestTime = currentTime;
+        for (BaseTranslogReader r : readers) {
+            earliestTime = Math.min(r.getLastModifiedTime(), earliestTime);
+        }
+        return Math.max(0, currentTime - Math.min(earliestTime, writer.getLastModifiedTime()));
+    }
+
     /**
      * Returns the number of operations in the transaction files that aren't committed to lucene..
      */
@@ -738,7 +758,7 @@ private void closeOnTragicEvent(Exception ex) {
     public TranslogStats stats() {
         // acquire lock to make the two numbers roughly consistent (no file change half way)
         try (ReleasableLock lock = readLock.acquire()) {
-            return new TranslogStats(totalOperations(), sizeInBytes(), uncommittedOperations(), uncommittedSizeInBytes());
+            return new TranslogStats(totalOperations(), sizeInBytes(), uncommittedOperations(), uncommittedSizeInBytes(), earliestLastModifiedAge());
         }
     }
 

server/src/main/java/org/elasticsearch/index/translog/TranslogStats.java

@@ -34,11 +34,13 @@ public class TranslogStats implements Streamable, ToXContentFragment {
     private int numberOfOperations;
     private long uncommittedSizeInBytes;
     private int  uncommittedOperations;
+    private long earliestLastModifiedAge;
 
     public TranslogStats() {
     }
 
-    public TranslogStats(int numberOfOperations, long translogSizeInBytes, int uncommittedOperations, long uncommittedSizeInBytes) {
+    public TranslogStats(int numberOfOperations, long translogSizeInBytes, int uncommittedOperations, long uncommittedSizeInBytes,
+                         long earliestLastModifiedAge) {
         if (numberOfOperations < 0) {
             throw new IllegalArgumentException("numberOfOperations must be >= 0");
         }
@@ -51,10 +53,14 @@ public TranslogStats(int numberOfOperations, long translogSizeInBytes, int uncom
         if (uncommittedSizeInBytes < 0) {
             throw new IllegalArgumentException("uncommittedSizeInBytes must be >= 0");
         }
+        if (earliestLastModifiedAge < 0) {
+            throw new IllegalArgumentException("earliestLastModifiedAge must be >= 0");
+        }
         this.numberOfOperations = numberOfOperations;
         this.translogSizeInBytes = translogSizeInBytes;
         this.uncommittedSizeInBytes = uncommittedSizeInBytes;
         this.uncommittedOperations = uncommittedOperations;
+        this.earliestLastModifiedAge = earliestLastModifiedAge;
     }
 
     public void add(TranslogStats translogStats) {
@@ -66,6 +72,8 @@ public void add(TranslogStats translogStats) {
         this.translogSizeInBytes += translogStats.translogSizeInBytes;
         this.uncommittedOperations += translogStats.uncommittedOperations;
         this.uncommittedSizeInBytes += translogStats.uncommittedSizeInBytes;
+        this.earliestLastModifiedAge =
+            Math.min(this.earliestLastModifiedAge, translogStats.earliestLastModifiedAge);
     }
 
     public long getTranslogSizeInBytes() {
@@ -86,13 +94,16 @@ public int getUncommittedOperations() {
         return uncommittedOperations;
     }
 
+    public long getEarliestLastModifiedAge() { return earliestLastModifiedAge; }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject("translog");
         builder.field("operations", numberOfOperations);
         builder.byteSizeField("size_in_bytes", "size", translogSizeInBytes);
         builder.field("uncommitted_operations", uncommittedOperations);
         builder.byteSizeField("uncommitted_size_in_bytes", "uncommitted_size", uncommittedSizeInBytes);
+        builder.field("earliest_last_modified_age", earliestLastModifiedAge);
         builder.endObject();
         return builder;
     }
@@ -113,6 +124,9 @@ public void readFrom(StreamInput in) throws IOException {
             uncommittedOperations = numberOfOperations;
             uncommittedSizeInBytes = translogSizeInBytes;
         }
+        if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            earliestLastModifiedAge = in.readVLong();
+        }
     }
 
     @Override
@@ -123,5 +137,8 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeVInt(uncommittedOperations);
             out.writeVLong(uncommittedSizeInBytes);
         }
+        if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            out.writeVLong(earliestLastModifiedAge);
+        }
     }
 }

server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -120,6 +120,8 @@
 import static org.hamcrest.Matchers.hasToString;
 import static org.hamcrest.Matchers.isIn;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.stub;
 
 @LuceneTestCase.SuppressFileSystems("ExtrasFS")
 public class TranslogTests extends ESTestCase {
@@ -366,6 +368,29 @@ protected TranslogStats stats() throws IOException {
         return stats;
     }
 
+    public void testFindEarliestLastModifiedAge() throws IOException {
+        final int numberOfReaders = scaledRandomIntBetween(1, 10);
+        long fixedTime = randomLongBetween(0, 10000000000000000L);
+        long[] periods = new long[numberOfReaders + 1];
+        long period = randomLongBetween(10000, 1000000);
+        periods[numberOfReaders] = period;
+        TranslogWriter w = mock(TranslogWriter.class);
+        stub(w.getLastModifiedTime()).toReturn(fixedTime - period);
+        assertThat(Translog.findEarliestLastModifiedAge(fixedTime, new ArrayList<>(), w), equalTo(period));
+
+        for (int i = 0; i < numberOfReaders; i++) {
+            periods[i] = randomLongBetween(10000, 1000000);
+        }
+        List<TranslogReader> readers = new ArrayList<>();
+        for (long l : periods) {
+            TranslogReader r = mock(TranslogReader.class);
+            stub(r.getLastModifiedTime()).toReturn(fixedTime - l);
+            readers.add(r);
+        }
+        assertThat(Translog.findEarliestLastModifiedAge(fixedTime, readers, w), equalTo
+            (LongStream.of(periods).max().orElse(0L)));
+    }
+
     public void testStats() throws IOException {
         // self control cleaning for test
         translog.getDeletionPolicy().setRetentionSizeInBytes(1024 * 1024);
@@ -384,6 +409,7 @@ public void testStats() throws IOException {
             assertThat(stats.getTranslogSizeInBytes(), equalTo(140L));
             assertThat(stats.getUncommittedOperations(), equalTo(1));
             assertThat(stats.getUncommittedSizeInBytes(), equalTo(140L));
+            assertThat(stats.getEarliestLastModifiedAge(), greaterThan(1L));
         }
 
         translog.add(new Translog.Delete("test", "2", 1, newUid("2")));
@@ -393,6 +419,7 @@ public void testStats() throws IOException {
             assertThat(stats.getTranslogSizeInBytes(), equalTo(189L));
             assertThat(stats.getUncommittedOperations(), equalTo(2));
             assertThat(stats.getUncommittedSizeInBytes(), equalTo(189L));
+            assertThat(stats.getEarliestLastModifiedAge(), greaterThan(1L));
         }
 
         translog.add(new Translog.Delete("test", "3", 2, newUid("3")));
@@ -402,6 +429,7 @@ public void testStats() throws IOException {
             assertThat(stats.getTranslogSizeInBytes(), equalTo(238L));
             assertThat(stats.getUncommittedOperations(), equalTo(3));
             assertThat(stats.getUncommittedSizeInBytes(), equalTo(238L));
+            assertThat(stats.getEarliestLastModifiedAge(), greaterThan(1L));
         }
 
         translog.add(new Translog.NoOp(3, 1, randomAlphaOfLength(16)));
@@ -411,6 +439,7 @@ public void testStats() throws IOException {
             assertThat(stats.getTranslogSizeInBytes(), equalTo(280L));
             assertThat(stats.getUncommittedOperations(), equalTo(4));
             assertThat(stats.getUncommittedSizeInBytes(), equalTo(280L));
+            assertThat(stats.getEarliestLastModifiedAge(), greaterThan(1L));
         }
 
         final long expectedSizeInBytes = 323L;
@@ -421,6 +450,7 @@ public void testStats() throws IOException {
             assertThat(stats.getTranslogSizeInBytes(), equalTo(expectedSizeInBytes));
             assertThat(stats.getUncommittedOperations(), equalTo(4));
             assertThat(stats.getUncommittedSizeInBytes(), equalTo(expectedSizeInBytes));
+            assertThat(stats.getEarliestLastModifiedAge(), greaterThan(1L));
         }
 
         {
@@ -438,7 +468,8 @@ public void testStats() throws IOException {
                 copy.toXContent(builder, ToXContent.EMPTY_PARAMS);
                 builder.endObject();
                 assertThat(builder.string(), equalTo("{\"translog\":{\"operations\":4,\"size_in_bytes\":" + expectedSizeInBytes
-                    + ",\"uncommitted_operations\":4,\"uncommitted_size_in_bytes\":" + expectedSizeInBytes + "}}"));
+                    + ",\"uncommitted_operations\":4,\"uncommitted_size_in_bytes\":" + expectedSizeInBytes
+                    + ",\"earliest_last_modified_age\":" + stats.getEarliestLastModifiedAge() + "}}"));
             }
         }
 
@@ -449,6 +480,7 @@ public void testStats() throws IOException {
             assertThat(stats.getTranslogSizeInBytes(), equalTo(expectedSizeInBytes));
             assertThat(stats.getUncommittedOperations(), equalTo(0));
             assertThat(stats.getUncommittedSizeInBytes(), equalTo(firstOperationPosition));
+            assertThat(stats.getEarliestLastModifiedAge(), greaterThan(1L));
         }
     }
 
@@ -478,12 +510,12 @@ public void testUncommittedOperations() throws Exception {
     }
 
     public void testTotalTests() {
-        final TranslogStats total = new TranslogStats();
+        final TranslogStats total = new TranslogStats(0, 0, 0, 0, 1);
         final int n = randomIntBetween(0, 16);
         final List<TranslogStats> statsList = new ArrayList<>(n);
         for (int i = 0; i < n; i++) {
             final TranslogStats stats = new TranslogStats(randomIntBetween(1, 4096), randomIntBetween(1, 1 << 20),
-                randomIntBetween(1, 1 << 20), randomIntBetween(1, 4096));
+                randomIntBetween(1, 1 << 20), randomIntBetween(1, 4096), randomIntBetween(1, 1 << 20));
             statsList.add(stats);
             total.add(stats);
         }
@@ -500,22 +532,30 @@ public void testTotalTests() {
         assertThat(
             total.getUncommittedSizeInBytes(),
             equalTo(statsList.stream().mapToLong(TranslogStats::getUncommittedSizeInBytes).sum()));
+        assertThat(
+            total.getEarliestLastModifiedAge(),
+            equalTo(1L));
     }
 
     public void testNegativeNumberOfOperations() {
-        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(-1, 1, 1, 1));
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(-1, 1, 1, 1, 1));
         assertThat(e, hasToString(containsString("numberOfOperations must be >= 0")));
-        e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, 1, -1, 1));
+        e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, 1, -1, 1, 1));
         assertThat(e, hasToString(containsString("uncommittedOperations must be >= 0")));
     }
 
     public void testNegativeSizeInBytes() {
-        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, -1, 1, 1));
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, -1, 1, 1, 1));
         assertThat(e, hasToString(containsString("translogSizeInBytes must be >= 0")));
-        e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, 1, 1, -1));
+        e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, 1, 1, -1, 1));
         assertThat(e, hasToString(containsString("uncommittedSizeInBytes must be >= 0")));
     }
 
+    public void testOldestEntryInSeconds() {
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new TranslogStats(1, 1, 1, 1, -1));
+        assertThat(e, hasToString(containsString("earliestLastModifiedAge must be >= 0")));
+    }
+
     public void testSnapshot() throws IOException {
         ArrayList<Translog.Operation> ops = new ArrayList<>();
         try (Translog.Snapshot snapshot = translog.newSnapshot()) {