Enable a long translog retention policy by default

core/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -118,7 +118,7 @@ public final class IndexSettings {
      * the chance of ops based recoveries.
      **/
     public static final Setting<TimeValue> INDEX_TRANSLOG_RETENTION_AGE_SETTING =
-        Setting.timeSetting("index.translog.retention.age", TimeValue.timeValueMillis(-1), TimeValue.timeValueMillis(-1), Property.Dynamic,
+        Setting.timeSetting("index.translog.retention.age", TimeValue.timeValueHours(12), TimeValue.timeValueMillis(-1), Property.Dynamic,
             Property.IndexScope);
 
     /**
@@ -127,7 +127,7 @@ public final class IndexSettings {
      * the chance of ops based recoveries.
      **/
     public static final Setting<ByteSizeValue> INDEX_TRANSLOG_RETENTION_SIZE_SETTING =
-        Setting.byteSizeSetting("index.translog.retention.size", new ByteSizeValue(-1, ByteSizeUnit.MB), Property.Dynamic,
+        Setting.byteSizeSetting("index.translog.retention.size", new ByteSizeValue(512, ByteSizeUnit.MB), Property.Dynamic,
             Property.IndexScope);
 
     /**

core/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -803,6 +803,12 @@ public final boolean refreshNeeded() {
      */
     public abstract CommitId flush() throws EngineException;
 
+    /**
+     * Rolls the translog generation and cleans unneeded.
+     */
+    public abstract void rollTranslogGeneration() throws EngineException;
+
+
     /**
      * Force merges to 1 segment
      */

core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -1215,7 +1215,7 @@ final boolean tryRenewSyncCommit() {
             ensureOpen();
             ensureCanFlush();
             String syncId = lastCommittedSegmentInfos.getUserData().get(SYNC_COMMIT_ID);
-            if (syncId != null && translog.totalOperations() == 0 && indexWriter.hasUncommittedChanges()) {
+            if (syncId != null && translog.uncommittedOperations() == 0 && indexWriter.hasUncommittedChanges()) {
                 logger.trace("start renewing sync commit [{}]", syncId);
                 commitIndexWriter(indexWriter, translog, syncId);
                 logger.debug("successfully sync committed. sync id [{}].", syncId);
@@ -1317,6 +1317,25 @@ public CommitId flush(boolean force, boolean waitIfOngoing) throws EngineExcepti
         return new CommitId(newCommitId);
     }
 
+    @Override
+    public void rollTranslogGeneration() throws EngineException {
+        try (ReleasableLock ignored = readLock.acquire()) {
+            ensureOpen();
+            translog.rollGeneration();
+            translog.trimUnreferencedReaders();
+        } catch (AlreadyClosedException e) {
+            failOnTragicEvent(e);
+            throw e;
+        } catch (Exception e) {
+            try {
+                failEngine("translog trimming failed", e);
+            } catch (Exception inner) {
+                e.addSuppressed(inner);
+            }
+            throw new EngineException(shardId, "failed to roll translog", e);
+        }
+    }
+
     private void pruneDeletedTombstones() {
         long timeMSec = engineConfig.getThreadPool().relativeTimeInMillis();
 

core/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -921,13 +921,11 @@ public Engine.CommitId flush(FlushRequest request) {
     }
 
     /**
-     * Rolls the tranlog generation.
-     *
-     * @throws IOException if any file operations on the translog throw an I/O exception
+     * Rolls the tranlog generation and cleans unneeded.
      */
-    private void rollTranslogGeneration() throws IOException {
+    private void rollTranslogGeneration() {
         final Engine engine = getEngine();
-        engine.getTranslog().rollGeneration();
+        engine.rollTranslogGeneration();
     }
 
     public void forceMerge(ForceMergeRequest forceMerge) throws IOException {
@@ -2142,7 +2140,7 @@ public void onFailure(final Exception e) {
                         }
 
                         @Override
-                        protected void doRun() throws IOException {
+                        protected void doRun() throws Exception {
                             rollTranslogGeneration();
                         }
 

core/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java

@@ -80,7 +80,8 @@ void setChunkSize(ByteSizeValue chunkSize) { // only settable for tests
 
     public void resync(IndexShard indexShard, ActionListener<ResyncTask> listener) throws IOException {
         try (Translog.View view = indexShard.acquireTranslogView()) {
-            Translog.Snapshot snapshot = view.snapshot();
+            final long startingSeqNo = indexShard.getGlobalCheckpoint() + 1;
+            Translog.Snapshot snapshot = view.snapshot(startingSeqNo);
             ShardId shardId = indexShard.shardId();
 
             // Wrap translog snapshot to make it synchronized as it is accessed by different threads through SnapshotSender.
@@ -104,7 +105,7 @@ public synchronized Translog.Operation next() throws IOException {
             };
 
             resync(shardId, indexShard.routingEntry().allocationId().getId(), wrappedSnapshot,
-                indexShard.getGlobalCheckpoint() + 1, listener);
+                startingSeqNo, listener);
         }
     }
 

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

@@ -367,17 +367,31 @@ long getMinFileGeneration() {
 
 
     /**
-     * Returns the number of operations in the transaction files that aren't committed to lucene..
+     * Returns the number of operations in the translog files that aren't committed to lucene.
      */
-    public int totalOperations() {
+    public int uncommittedOperations() {
         return totalOperations(deletionPolicy.getMinTranslogGenerationForRecovery());
     }
 
     /**
      * Returns the size in bytes of the translog files that aren't committed to lucene.
      */
+    public long uncommittedSizeInBytes() {
+        return sizeInBytesByMinGen(deletionPolicy.getMinTranslogGenerationForRecovery());
+    }
+
+    /**
+     * Returns the number of operations in the translog files
+     */
+    public int totalOperations() {
+        return totalOperations(-1);
+    }
+
+    /**
+     * Returns the size in bytes of the v files
+     */
     public long sizeInBytes() {
-        return sizeInBytes(deletionPolicy.getMinTranslogGenerationForRecovery());
+        return sizeInBytesByMinGen(-1);
     }
 
     /**
@@ -394,9 +408,19 @@ private int totalOperations(long minGeneration) {
     }
 
     /**
-     * Returns the size in bytes of the translog files that aren't committed to lucene.
+     * Returns the number of operations in the transaction files that aren't committed to lucene..
+     */
+    private int totalOperationsInGensAboveSeqNo(long minSeqNo) {
+        try (ReleasableLock ignored = readLock.acquire()) {
+            ensureOpen();
+            return readersAboveMinSeqNo(minSeqNo).mapToInt(BaseTranslogReader::totalOperations).sum();
+        }
+    }
+
+    /**
+     * Returns the size in bytes of the translog files above the given generation
      */
-    private long sizeInBytes(long minGeneration) {
+    private long sizeInBytesByMinGen(long minGeneration) {
         try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
             return Stream.concat(readers.stream(), Stream.of(current))
@@ -406,6 +430,16 @@ private long sizeInBytes(long minGeneration) {
         }
     }
 
+    /**
+     * Returns the size in bytes of the translog files with ops above the given seqNo
+     */
+    private long sizeOfGensAboveSeqNoInBytes(long minSeqNo) {
+        try (ReleasableLock ignored = readLock.acquire()) {
+            ensureOpen();
+            return readersAboveMinSeqNo(minSeqNo).mapToLong(BaseTranslogReader::sizeInBytes).sum();
+        }
+    }
+
     /**
      * Creates a new translog for the specified generation.
      *
@@ -493,7 +527,7 @@ public Location add(final Operation operation) throws IOException {
      * @return {@code true} if the translog should be flushed
      */
     public boolean shouldFlush() {
-        final long size = this.sizeInBytes();
+        final long size = this.uncommittedSizeInBytes();
         return size > this.indexSettings.getFlushThresholdSize().getBytes();
     }
 
@@ -560,14 +594,34 @@ public Snapshot newSnapshot(long minGeneration) {
         }
     }
 
+    private Stream<? extends BaseTranslogReader> readersAboveMinSeqNo(long minSeqNo) {
+        assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread() :
+        "callers of readersAboveMinSeqNo must hold a lock: readLock ["
+            + readLock.isHeldByCurrentThread() + "], writeLock [" + readLock.isHeldByCurrentThread() + "]";
+        return Stream.concat(readers.stream(), Stream.of(current))
+            .filter(reader -> {
+                final long maxSeqNo = reader.getCheckpoint().maxSeqNo;
+                return maxSeqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO || maxSeqNo >= minSeqNo;
+            });
+    }
+
+    private Snapshot createSnapshotFromMinSeqNo(long minSeqNo) {
+        try (ReleasableLock ignored = readLock.acquire()) {
+            ensureOpen();
+            Snapshot[] snapshots = readersAboveMinSeqNo(minSeqNo).map(BaseTranslogReader::newSnapshot).toArray(Snapshot[]::new);
+            return new MultiSnapshot(snapshots);
+        }
+    }
+
     /**
      * Returns a view into the current translog that is guaranteed to retain all current operations
      * while receiving future ones as well
      */
     public Translog.View newView() {
         try (ReleasableLock lock = readLock.acquire()) {
             ensureOpen();
-            final long viewGen = deletionPolicy.acquireTranslogGenForView();
+            final long viewGen = getMinFileGeneration();
+            deletionPolicy.acquireTranslogGenForView(viewGen);
             try {
                 return new View(viewGen);
             } catch (Exception e) {
@@ -674,7 +728,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());
+            return new TranslogStats(totalOperations(), sizeInBytes(), uncommittedOperations(), uncommittedSizeInBytes());
         }
     }
 
@@ -698,35 +752,36 @@ public TranslogDeletionPolicy getDeletionPolicy() {
     public class View implements Closeable {
 
         AtomicBoolean closed = new AtomicBoolean();
-        final long minGeneration;
-
-        View(long minGeneration) {
-            this.minGeneration = minGeneration;
-        }
+        final long viewGenToRelease;
 
-        /** this smallest translog generation in this view */
-        public long minTranslogGeneration() {
-            return minGeneration;
+        View(long viewGenToRelease) {
+            this.viewGenToRelease = viewGenToRelease;
         }
 
         /**
-         * The total number of operations in the view.
+         * The total number of operations in the view files which contain an operation with a sequence number
+         * above the given min sequence numbers. This will be the number of operations in snapshot taken
+         * by calling {@link #snapshot(long)} with the same parameter.
          */
-        public int totalOperations() {
-            return Translog.this.totalOperations(minGeneration);
+        public int estimateTotalOperations(long minSequenceNumber) {
+            return Translog.this.totalOperationsInGensAboveSeqNo(minSequenceNumber);
         }
 
         /**
-         * Returns the size in bytes of the files behind the view.
+         * The total size of the view files which contain an operation with a sequence number
+         * above the given min sequence numbers. These are the files that would need to be read by snapshot
+         * acquired {@link #snapshot(long)} with the same parameter.
          */
-        public long sizeInBytes() {
-            return Translog.this.sizeInBytes(minGeneration);
+        public long estimateSizeInBytes(long minSequenceNumber) {
+            return Translog.this.sizeOfGensAboveSeqNoInBytes(minSequenceNumber);
         }
 
-        /** create a snapshot from this view */
-        public Snapshot snapshot() {
+        /**
+         * create a snapshot from this view, containing all
+         * operations from the given sequence number and up (with potentially some more) */
+        public Snapshot snapshot(long minSequenceNumber) {
             ensureOpen();
-            return Translog.this.newSnapshot(minGeneration);
+            return Translog.this.createSnapshotFromMinSeqNo(minSequenceNumber);
         }
 
         void ensureOpen() {
@@ -738,8 +793,8 @@ void ensureOpen() {
         @Override
         public void close() throws IOException {
             if (closed.getAndSet(true) == false) {
-                logger.trace("closing view starting at translog [{}]", minGeneration);
-                deletionPolicy.releaseTranslogGenView(minGeneration);
+                logger.trace("closing view starting at translog [{}]", viewGenToRelease);
+                deletionPolicy.releaseTranslogGenView(viewGenToRelease);
                 trimUnreferencedReaders();
                 closeFilesIfNoPendingViews();
             }
@@ -1663,4 +1718,12 @@ public String getTranslogUUID() {
         return translogUUID;
     }
 
+
+    TranslogWriter getCurrent() {
+        return current;
+    }
+
+    List<TranslogReader> getReaders() {
+        return readers;
+    }
 }

core/src/main/java/org/elasticsearch/index/translog/TranslogDeletionPolicy.java

@@ -69,9 +69,8 @@ public synchronized void setRetentionAgeInMillis(long ageInMillis) {
      * acquires the basis generation for a new view. Any translog generation above, and including, the returned generation
      * will not be deleted until a corresponding call to {@link #releaseTranslogGenView(long)} is called.
      */
-    synchronized long acquireTranslogGenForView() {
-        translogRefCounts.computeIfAbsent(minTranslogGenerationForRecovery, l -> Counter.newCounter(false)).addAndGet(1);
-        return minTranslogGenerationForRecovery;
+    synchronized void acquireTranslogGenForView(final long genForView) {
+        translogRefCounts.computeIfAbsent(genForView, l -> Counter.newCounter(false)).addAndGet(1);
     }
 
     /** returns the number of generations that were acquired for views */
@@ -80,7 +79,7 @@ synchronized int pendingViewsCount() {
     }
 
     /**
-     * releases a generation that was acquired by {@link #acquireTranslogGenForView()}
+     * releases a generation that was acquired by {@link #acquireTranslogGenForView(long)}
      */
     synchronized void releaseTranslogGenView(long translogGen) {
         Counter current = translogRefCounts.get(translogGen);
@@ -154,4 +153,9 @@ private long getMinTranslogGenRequiredByViews() {
     public synchronized long getMinTranslogGenerationForRecovery() {
         return minTranslogGenerationForRecovery;
     }
+
+    synchronized long getViewCount(long viewGen) {
+        final Counter counter = translogRefCounts.get(viewGen);
+        return counter == null ? 0 : counter.get();
+    }
 }

core/src/main/java/org/elasticsearch/index/translog/TranslogSnapshot.java

@@ -99,7 +99,7 @@ public String toString() {
         return "TranslogSnapshot{" +
                 "readOperations=" + readOperations +
                 ", position=" + position +
-                ", totalOperations=" + totalOperations +
+                ", estimateTotalOperations=" + totalOperations +
                 ", length=" + length +
                 ", reusableBuffer=" + reusableBuffer +
                 '}';