Add support to clone existing pinned timestamp

Sachin Kale · Sachin Kale · commit bbda09f1c2cf · 2024-08-31T18:28:15.000+05:30
Signed-off-by: Sachin Kale &lt;kalsac@amazon.com&gt;
diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStorePinnedTimestampsIT.java
@@ -8,6 +8,7 @@
 
 package org.opensearch.remotestore;
 
+import org.opensearch.action.LatchedActionListener;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
@@ -17,6 +18,7 @@
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
 public class RemoteStorePinnedTimestampsIT extends RemoteStoreBaseIntegTestCase {
@@ -75,10 +77,25 @@ public void testTimestampPinUnpin() throws Exception {
 
         remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
 
-        // This should be a no-op as pinning entity is different
-        remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "no-snapshot", noOpActionListener);
         // Unpinning already pinned entity
         remoteStorePinnedTimestampService.unpinTimestamp(timestamp2, "ss3", noOpActionListener);
+
+        // This should fail as timestamp is not pinned by pinning entity
+        CountDownLatch latch = new CountDownLatch(1);
+        remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "no-snapshot", new LatchedActionListener<>(new ActionListener<Void>() {
+            @Override
+            public void onResponse(Void unused) {
+                // onResponse should not get called.
+                fail();
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                assertTrue(e instanceof IllegalArgumentException);
+            }
+        }, latch));
+        latch.await();
+
         // Adding different entity to already pinned timestamp
         remoteStorePinnedTimestampService.pinTimestamp(timestamp3, "ss5", noOpActionListener);
 
@@ -93,4 +110,74 @@ public void testTimestampPinUnpin() throws Exception {
 
         remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
     }
+
+    public void testPinnedTimestampClone() throws Exception {
+        prepareCluster(1, 1, INDEX_NAME, 0, 2);
+        ensureGreen(INDEX_NAME);
+
+        RemoteStorePinnedTimestampService remoteStorePinnedTimestampService = internalCluster().getInstance(
+            RemoteStorePinnedTimestampService.class,
+            primaryNodeName(INDEX_NAME)
+        );
+
+        long timestamp1 = System.currentTimeMillis() + 30000L;
+        long timestamp2 = System.currentTimeMillis() + 60000L;
+        long timestamp3 = System.currentTimeMillis() + 900000L;
+        remoteStorePinnedTimestampService.pinTimestamp(timestamp1, "ss2", noOpActionListener);
+        remoteStorePinnedTimestampService.pinTimestamp(timestamp2, "ss3", noOpActionListener);
+        remoteStorePinnedTimestampService.pinTimestamp(timestamp3, "ss4", noOpActionListener);
+
+        // Clone timestamp1
+        remoteStorePinnedTimestampService.cloneTimestamp(timestamp1, "ss2", "ss2-2", noOpActionListener);
+
+        // With clone, set of pinned timestamp will not change
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1));
+        assertBusy(
+            () -> assertEquals(Set.of(timestamp1, timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2())
+        );
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
+
+        // Clone timestamp1 but provide invalid existing entity
+        CountDownLatch latch = new CountDownLatch(1);
+        remoteStorePinnedTimestampService.cloneTimestamp(
+            timestamp1,
+            "ss3",
+            "ss2-3",
+            new LatchedActionListener<>(new ActionListener<Void>() {
+                @Override
+                public void onResponse(Void unused) {
+                    // onResponse should not get called.
+                    fail();
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    assertTrue(e instanceof IllegalArgumentException);
+                }
+            }, latch)
+        );
+        latch.await();
+
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1));
+        assertBusy(
+            () -> assertEquals(Set.of(timestamp1, timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2())
+        );
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
+
+        // Now we have timestamp1 pinned by 2 entities, unpin 1, this should not change set of pinned timestamps
+        remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "ss2", noOpActionListener);
+
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1));
+        assertBusy(
+            () -> assertEquals(Set.of(timestamp1, timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2())
+        );
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
+
+        // Now unpin second entity as well, set of pinned timestamp should be reduced by 1
+        remoteStorePinnedTimestampService.unpinTimestamp(timestamp1, "ss2-2", noOpActionListener);
+
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueSeconds(1));
+        assertBusy(() -> assertEquals(Set.of(timestamp2, timestamp3), RemoteStorePinnedTimestampService.getPinnedTimestamps().v2()));
+        remoteStorePinnedTimestampService.rescheduleAsyncUpdatePinnedTimestampTask(TimeValue.timeValueMinutes(3));
+    }
 }
diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java b/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java
@@ -542,8 +542,11 @@ public static List<String> filterOutMetadataFilesBasedOnAge(
         if (RemoteStoreSettings.isPinnedTimestampsEnabled() == false) {
             return new ArrayList<>(metadataFiles);
         }
-        long maximumAllowedTimestamp = lastSuccessfulFetchOfPinnedTimestamps - RemoteStoreSettings.getPinnedTimestampsLookbackInterval()
-            .getMillis();
+        // We allow now() - loopback interval to be pinned. Also, the actual pinning can take at most loopback interval
+        // This means the pinned timestamp can be available for read after at most (2 * loopback interval)
+        long maximumAllowedTimestamp = lastSuccessfulFetchOfPinnedTimestamps - (2 * RemoteStoreSettings
+            .getPinnedTimestampsLookbackInterval()
+            .getMillis());
         List<String> metadataFilesWithMinAge = new ArrayList<>();
         for (String metadataFileName : metadataFiles) {
             long metadataTimestamp = getTimestampFunction.apply(metadataFileName);
diff --git a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java
@@ -108,13 +108,61 @@ public void pinTimestamp(long timestamp, String pinningEntity, ActionListener<Vo
                 "Timestamp to be pinned is less than current timestamp - value of cluster.remote_store.pinned_timestamps.lookback_interval"
             );
         }
+        long startTime = System.nanoTime();
         try {
             logger.debug("Pinning timestamp = {} against entity = {}", timestamp, pinningEntity);
             blobContainer.writeBlob(getBlobName(timestamp, pinningEntity), new ByteArrayInputStream(new byte[0]), 0, true);
+            long elapsedTime = System.nanoTime() - startTime;
+            if (elapsedTime > RemoteStoreSettings.getPinnedTimestampsLookbackInterval().nanos()) {
+                String errorMessage = String.format(
+                    "Timestamp pinning took %s nanoseconds which is more than limit of %s nanoseconds, failing the operation",
+                    elapsedTime,
+                    RemoteStoreSettings.getPinnedTimestampsLookbackInterval().nanos()
+                );
+                unpinTimestamp(timestamp, pinningEntity, ActionListener.wrap(() -> listener.onFailure(new RuntimeException(errorMessage))));
+            } else {
+                listener.onResponse(null);
+            }
+        } catch (IOException e) {
+            listener.onFailure(e);
+        }
+    }
+
+    /**
+     * Clones a timestamp by creating a new pinning entity for an existing timestamp.
+     *
+     * This method attempts to create a new pinning entity for a given timestamp that is already
+     * associated with an existing pinning entity. If the timestamp exists for the existing entity,
+     * a new blob is created for the new pinning entity. If the timestamp doesn't exist for the
+     * existing entity, the operation fails with an IllegalArgumentException.
+     *
+     * @param timestamp The timestamp to be cloned.
+     * @param existingPinningEntity The name of the existing entity that has pinned the timestamp.
+     * @param newPinningEntity The name of the new entity to pin the timestamp to.
+     * @param listener An ActionListener that will be notified of the operation's success or failure.
+     *                 On success, onResponse will be called with null. On failure, onFailure will
+     *                 be called with the appropriate exception.
+     */
+    public void cloneTimestamp(long timestamp, String existingPinningEntity, String newPinningEntity, ActionListener<Void> listener) {
+        try {
+            logger.debug(
+                "cloning timestamp = {} with existing pinningEntity = {} with new pinningEntity = {}",
+                timestamp,
+                existingPinningEntity,
+                newPinningEntity
+            );
+            String blobName = getBlobName(timestamp, existingPinningEntity);
+            if (blobContainer.blobExists(blobName)) {
+                logger.debug("Pinning timestamp = {} against entity = {}", timestamp, newPinningEntity);
+                blobContainer.writeBlob(getBlobName(timestamp, newPinningEntity), new ByteArrayInputStream(new byte[0]), 0, true);
+                listener.onResponse(null);
+            } else {
+                String errorMessage = String.format("Timestamp: %s is not pinned by existing entity: %s", timestamp, existingPinningEntity);
+                listener.onFailure(new IllegalArgumentException(errorMessage));
+            }
         } catch (IOException e) {
             listener.onFailure(e);
         }
-        listener.onResponse(null);
     }
 
     private String getBlobName(long timestamp, String pinningEntity) {
@@ -147,13 +195,14 @@ public void unpinTimestamp(long timestamp, String pinningEntity, ActionListener<
             String blobName = getBlobName(timestamp, pinningEntity);
             if (blobContainer.blobExists(blobName)) {
                 blobContainer.deleteBlobsIgnoringIfNotExists(List.of(blobName));
+                listener.onResponse(null);
             } else {
-                logger.warn("Timestamp: {} is not pinned by entity: {}", timestamp, pinningEntity);
+                String errorMessage = String.format("Timestamp: %s is not pinned by entity: %s", timestamp, pinningEntity);
+                listener.onFailure(new IllegalArgumentException(errorMessage));
             }
         } catch (IOException e) {
             listener.onFailure(e);
         }
-        listener.onResponse(null);
     }
 
     @Override
