Create new PRRL using global checkpoint

DaveCTurner · DaveCTurner · commit 5fb8bda58b7e · 2019-07-29T11:06:08.000+01:00
By cloning the primary's lease we need not worry about creating a lease that
retains history which has already been discarded.
diff --git a/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java b/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java
@@ -290,6 +290,35 @@ public RetentionLease addRetentionLease(
         return retentionLease;
     }
 
+    /**
+     * Atomically clones an existing retention lease to a new ID.
+     *
+     * @param sourceLeaseId the identifier of the source retention lease
+     * @param targetLeaseId the identifier of the retention lease to create
+     * @param listener      the callback when the retention lease is successfully added and synced to replicas
+     * @return the new retention lease
+     * @throws RetentionLeaseNotFoundException      if the specified source retention lease does not exist
+     * @throws RetentionLeaseAlreadyExistsException if the specified target retention lease already exists
+     */
+    RetentionLease cloneRetentionLease(String sourceLeaseId, String targetLeaseId, ActionListener<ReplicationResponse> listener) {
+        Objects.requireNonNull(listener);
+        final RetentionLease retentionLease;
+        final RetentionLeases currentRetentionLeases;
+        synchronized (this) {
+            if (getRetentionLeases().contains(sourceLeaseId) == false) {
+                throw new RetentionLeaseNotFoundException(sourceLeaseId);
+            }
+            final RetentionLease sourceLease = getRetentionLeases().get(sourceLeaseId);
+            retentionLease = innerAddRetentionLease(targetLeaseId, sourceLease.retainingSequenceNumber(), sourceLease.source());
+            currentRetentionLeases = retentionLeases;
+        }
+
+        // Syncing here may not be strictly necessary, because this new lease isn't retaining any extra history that wasn't previously
+        // retained by the source lease; however we prefer to sync anyway since we expect to do so whenever creating a new lease.
+        onSyncRetentionLeases.accept(currentRetentionLeases, listener);
+        return retentionLease;
+    }
+
     /**
      * Adds a new retention lease, but does not synchronise it with the rest of the replication group.
      *
@@ -446,6 +475,12 @@ public void addPeerRecoveryRetentionLease(String nodeId, long globalCheckpoint,
         addRetentionLease(getPeerRecoveryRetentionLeaseId(nodeId), globalCheckpoint + 1, PEER_RECOVERY_RETENTION_LEASE_SOURCE, listener);
     }
 
+    public RetentionLease cloneLocalPeerRecoveryRetentionLease(String nodeId, ActionListener<ReplicationResponse> listener) {
+        return cloneRetentionLease(
+            getPeerRecoveryRetentionLeaseId(getPeerRecoveryRetentionLeaseId(routingTable.primaryShard())),
+            getPeerRecoveryRetentionLeaseId(nodeId), listener);
+    }
+
     public void removePeerRecoveryRetentionLease(String nodeId, ActionListener<ReplicationResponse> listener) {
         removeRetentionLease(getPeerRecoveryRetentionLeaseId(nodeId), listener);
     }
diff --git a/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java b/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java
@@ -40,6 +40,7 @@
 import org.elasticsearch.Assertions;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
@@ -2591,9 +2592,16 @@ public boolean isRelocatedPrimary() {
 
     public void addPeerRecoveryRetentionLease(String nodeId, long globalCheckpoint, ActionListener<ReplicationResponse> listener) {
         assert assertPrimaryMode();
+        // only needed for BWC reasons involving rolling upgrades from versions that do not support PRRLs:
+        assert indexSettings.getIndexVersionCreated().before(Version.V_7_4_0);
         replicationTracker.addPeerRecoveryRetentionLease(nodeId, globalCheckpoint, listener);
     }
 
+    public RetentionLease cloneLocalPeerRecoveryRetentionLease(String nodeId, ActionListener<ReplicationResponse> listener) {
+        assert assertPrimaryMode();
+        return replicationTracker.cloneLocalPeerRecoveryRetentionLease(nodeId, listener);
+    }
+
     public void removePeerRecoveryRetentionLease(String nodeId, ActionListener<ReplicationResponse> listener) {
         assert assertPrimaryMode();
         replicationTracker.removePeerRecoveryRetentionLease(nodeId, listener);
diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java b/server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java
@@ -31,6 +31,7 @@
 import org.apache.lucene.util.ArrayUtil;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.StepListener;
 import org.elasticsearch.action.support.ThreadedActionListener;
@@ -177,6 +178,11 @@ && isTargetSameHistory()
                 && shard.hasCompleteHistoryOperations("peer-recovery", request.startingSeqNo())
                 && (useRetentionLeases == false
                 || (retentionLeaseRef.get() != null && retentionLeaseRef.get().retainingSequenceNumber() <= request.startingSeqNo()));
+            // NB check hasCompleteHistoryOperations when computing isSequenceNumberBasedRecovery, even if there is a retention lease,
+            // because when doing a rolling upgrade from earlier than 7.4 we may create some leases that are initially unsatisfied. It's
+            // possible there are other cases where we cannot satisfy all leases, because that's not a property we currently expect to hold.
+            // Also it's pretty cheap when soft deletes are enabled, and it'd be a disaster if we tried a sequence-number-based recovery
+            // without having a complete history.
 
             if (isSequenceNumberBasedRecovery && useRetentionLeases) {
                 // all the history we need is retained by an existing retention lease, so we do not need a separate retention lock
@@ -271,19 +277,32 @@ && isTargetSameHistory()
                 if (useRetentionLeases && isSequenceNumberBasedRecovery == false) {
                     // We can in general use retention leases for peer recovery, but there is no lease for the target node right now.
                     runUnderPrimaryPermit(() -> {
-                            // Start the lease off retaining all the history needed from the local checkpoint of the safe commit that we've
-                            // just established on the replica. This primary is certainly retaining such history, but other replicas might
-                            // not be. No big deal if this recovery succeeds, but if this primary fails then the new primary might have to
-                            // repeat phase 1 to recover this replica.
-                            final long localCheckpointOfSafeCommit = startingSeqNo - 1;
-                            logger.trace("creating new retention lease at [{}]", localCheckpointOfSafeCommit);
-                            shard.addPeerRecoveryRetentionLease(request.targetNode().getId(), localCheckpointOfSafeCommit,
-                                new ThreadedActionListener<>(logger, shard.getThreadPool(),
-                                    ThreadPool.Names.GENERIC, establishRetentionLeaseStep, false));
+                            // Clone the peer recovery retention lease belonging to the source shard. We are retaining history between the
+                            // the local checkpoint of the safe commit we're creating and this lease's retained seqno with the retention
+                            // lock, and by cloning an existing lease we (approximately) know that all our peers are also retaining history
+                            // as requested by the cloned lease. If the recovery now fails before copying enough history over then a
+                            // subsequent attempt will find this lease, determine it is not enough, and fall back to a file-based recovery.
+                            //
+                            // (approximately) because we do not guarantee to be able to satisfy every lease on every peer.
+                            logger.trace("cloning primary's retention lease");
+                            try {
+                                final RetentionLease clonedLease = shard.cloneLocalPeerRecoveryRetentionLease(request.targetNode().getId(),
+                                    new ThreadedActionListener<>(logger, shard.getThreadPool(),
+                                        ThreadPool.Names.GENERIC, establishRetentionLeaseStep, false));
+                                logger.trace("cloned primary's retention lease as [{}]", clonedLease);
+                            } catch (RetentionLeaseNotFoundException e) {
+                                // it's possible that the primary has no retention lease yet if we are doing a rolling upgrade from a
+                                // version before 7.4, and in that case we just create a lease using the local checkpoint of the safe commit
+                                // which we're using for recovery as a conservative estimate for the global checkpoint.
+                                assert shard.indexSettings().getIndexVersionCreated().before(Version.V_7_4_0);
+                                final long estimatedGlobalCheckpoint = startingSeqNo - 1;
+                                shard.addPeerRecoveryRetentionLease(request.targetNode().getId(),
+                                    estimatedGlobalCheckpoint, new ThreadedActionListener<>(logger, shard.getThreadPool(),
+                                        ThreadPool.Names.GENERIC, establishRetentionLeaseStep, false));
+                                logger.trace("created retention lease with estimated checkpoint of [{}]", estimatedGlobalCheckpoint);
+                            }
                     }, shardId + " establishing retention lease for [" + request.targetAllocationId() + "]",
                         shard, cancellableThreads, logger);
-                    // all the history we need is now retained by a retention lease so we can close the retention lock
-                    retentionLock.close();
                 } else {
                     establishRetentionLeaseStep.onResponse(null);
                 }
@@ -313,7 +332,7 @@ && isTargetSameHistory()
                 final Translog.Snapshot phase2Snapshot = shard.getHistoryOperations("peer-recovery", startingSeqNo);
                 resources.add(phase2Snapshot);
 
-                if (useRetentionLeases == false) {
+                if (useRetentionLeases == false || isSequenceNumberBasedRecovery == false) {
                     // we can release the retention lock here because the snapshot itself will retain the required operations.
                     retentionLock.close();
                 }
diff --git a/server/src/test/java/org/elasticsearch/index/seqno/ReplicationTrackerRetentionLeaseTests.java b/server/src/test/java/org/elasticsearch/index/seqno/ReplicationTrackerRetentionLeaseTests.java
@@ -20,6 +20,8 @@
 package org.elasticsearch.index.seqno;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.cluster.routing.AllocationId;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
@@ -247,6 +249,103 @@ public void testRemoveRetentionLease() {
         }
     }
 
+    public void testCloneRetentionLease() {
+        final AllocationId allocationId = AllocationId.newInitializing();
+        final AtomicReference<ReplicationTracker> replicationTrackerRef = new AtomicReference<>();
+        final AtomicLong timeReference = new AtomicLong();
+        final AtomicBoolean synced = new AtomicBoolean();
+        final ReplicationTracker replicationTracker = new ReplicationTracker(
+            new ShardId("test", "_na", 0),
+            allocationId.getId(),
+            IndexSettingsModule.newIndexSettings("test", Settings.EMPTY),
+            randomLongBetween(1, Long.MAX_VALUE),
+            UNASSIGNED_SEQ_NO,
+            value -> {},
+            timeReference::get,
+            (leases, listener) -> {
+                assertFalse(Thread.holdsLock(replicationTrackerRef.get()));
+                assertTrue(synced.compareAndSet(false, true));
+                listener.onResponse(new ReplicationResponse());
+            });
+        replicationTrackerRef.set(replicationTracker);
+        replicationTracker.updateFromMaster(
+            randomNonNegativeLong(),
+            Collections.singleton(allocationId.getId()),
+            routingTable(Collections.emptySet(), allocationId));
+        replicationTracker.activatePrimaryMode(SequenceNumbers.NO_OPS_PERFORMED);
+
+        final long addTime = randomLongBetween(timeReference.get(), Long.MAX_VALUE);
+        timeReference.set(addTime);
+        final long minimumRetainingSequenceNumber = randomLongBetween(SequenceNumbers.NO_OPS_PERFORMED, Long.MAX_VALUE);
+        final PlainActionFuture<ReplicationResponse> addFuture = new PlainActionFuture<>();
+        replicationTracker.addRetentionLease("source", minimumRetainingSequenceNumber, "test-source", addFuture);
+        addFuture.actionGet();
+        assertTrue(synced.get());
+        synced.set(false);
+
+        final long cloneTime = randomLongBetween(timeReference.get(), Long.MAX_VALUE);
+        timeReference.set(cloneTime);
+        final PlainActionFuture<ReplicationResponse> cloneFuture = new PlainActionFuture<>();
+        final RetentionLease clonedLease = replicationTracker.cloneRetentionLease("source", "target", cloneFuture);
+        cloneFuture.actionGet();
+        assertTrue(synced.get());
+        synced.set(false);
+
+        assertThat(clonedLease.id(), equalTo("target"));
+        assertThat(clonedLease.retainingSequenceNumber(), equalTo(minimumRetainingSequenceNumber));
+        assertThat(clonedLease.timestamp(), equalTo(cloneTime));
+        assertThat(clonedLease.source(), equalTo("test-source"));
+
+        assertThat(replicationTracker.getRetentionLeases().get("target"), equalTo(clonedLease));
+    }
+
+    public void testCloneNonexistentRetentionLease() {
+        final AllocationId allocationId = AllocationId.newInitializing();
+        final ReplicationTracker replicationTracker = new ReplicationTracker(
+            new ShardId("test", "_na", 0),
+            allocationId.getId(),
+            IndexSettingsModule.newIndexSettings("test", Settings.EMPTY),
+            randomLongBetween(1, Long.MAX_VALUE),
+            UNASSIGNED_SEQ_NO,
+            value -> {},
+            () -> 0L,
+            (leases, listener) -> { });
+        replicationTracker.updateFromMaster(
+            randomNonNegativeLong(),
+            Collections.singleton(allocationId.getId()),
+            routingTable(Collections.emptySet(), allocationId));
+        replicationTracker.activatePrimaryMode(SequenceNumbers.NO_OPS_PERFORMED);
+
+        assertThat(expectThrows(RetentionLeaseNotFoundException.class,
+            () -> replicationTracker.cloneRetentionLease("nonexistent-lease-id", "target", ActionListener.wrap(() -> {}))).getMessage(),
+            equalTo("retention lease with ID [nonexistent-lease-id] not found"));
+    }
+
+    public void testCloneDuplicateRetentionLease() {
+        final AllocationId allocationId = AllocationId.newInitializing();
+        final ReplicationTracker replicationTracker = new ReplicationTracker(
+            new ShardId("test", "_na", 0),
+            allocationId.getId(),
+            IndexSettingsModule.newIndexSettings("test", Settings.EMPTY),
+            randomLongBetween(1, Long.MAX_VALUE),
+            UNASSIGNED_SEQ_NO,
+            value -> {},
+            () -> 0L,
+            (leases, listener) -> { });
+        replicationTracker.updateFromMaster(
+            randomNonNegativeLong(),
+            Collections.singleton(allocationId.getId()),
+            routingTable(Collections.emptySet(), allocationId));
+        replicationTracker.activatePrimaryMode(SequenceNumbers.NO_OPS_PERFORMED);
+
+        replicationTracker.addRetentionLease("source", randomLongBetween(0L, Long.MAX_VALUE), "test-source", ActionListener.wrap(() -> {}));
+        replicationTracker.addRetentionLease("exists", randomLongBetween(0L, Long.MAX_VALUE), "test-source", ActionListener.wrap(() -> {}));
+
+        assertThat(expectThrows(RetentionLeaseAlreadyExistsException.class,
+            () -> replicationTracker.cloneRetentionLease("source", "exists", ActionListener.wrap(() -> {}))).getMessage(),
+            equalTo("retention lease with ID [exists] already exists"));
+    }
+
     public void testRemoveNotFound() {
         final AllocationId allocationId = AllocationId.newInitializing();
         long primaryTerm = randomLongBetween(1, Long.MAX_VALUE);
