From 11c5d8958511d66f4e8389921d4efa2de1e4ecf5 Mon Sep 17 00:00:00 2001 From: Sachin Kale Date: Sun, 25 Aug 2024 22:37:15 +0530 Subject: [PATCH 1/5] Add support to skip pinned timestamp in remote segment garbage collector Signed-off-by: Sachin Kale --- .../opensearch/remotestore/RemoteStoreIT.java | 7 +- .../store/RemoteSegmentStoreDirectory.java | 97 +++++++++++++++-- .../RemoteStorePinnedTimestampService.java | 13 ++- .../index/remote/RemoteStoreUtilsTests.java | 1 - .../BaseRemoteSegmentStoreDirectoryTests.java | 15 ++- .../RemoteSegmentStoreDirectoryTests.java | 100 ++++++++++++++++++ 6 files changed, 214 insertions(+), 19 deletions(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java index 194dce5f4a57a..589b64c3b0ab5 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java @@ -217,10 +217,7 @@ public void testStaleCommitDeletionWithInvokeFlush() throws Exception { } else { // As delete is async its possible that the file gets created before the deletion or after // deletion. - MatcherAssert.assertThat( - actualFileCount, - is(oneOf(lastNMetadataFilesToKeep - 1, lastNMetadataFilesToKeep, lastNMetadataFilesToKeep + 1)) - ); + assertTrue(actualFileCount >= lastNMetadataFilesToKeep); } }, 30, TimeUnit.SECONDS); } @@ -249,7 +246,7 @@ public void testStaleCommitDeletionWithMinSegmentFiles_3() throws Exception { Path indexPath = Path.of(segmentRepoPath + "/" + shardPath); int actualFileCount = getFileCount(indexPath); // We also allow (numberOfIterations + 1) as index creation also triggers refresh. - MatcherAssert.assertThat(actualFileCount, is(oneOf(4))); + assertTrue(actualFileCount >= 4); } public void testStaleCommitDeletionWithMinSegmentFiles_Disabled() throws Exception { diff --git a/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java b/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java index 9ff97f12015bd..10117dc1b30f3 100644 --- a/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java +++ b/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java @@ -40,6 +40,7 @@ import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadataHandler; import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint; +import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService; import org.opensearch.threadpool.ThreadPool; import java.io.FileNotFoundException; @@ -91,6 +92,8 @@ public final class RemoteSegmentStoreDirectory extends FilterDirectory implement private final RemoteStoreLockManager mdLockManager; + private final Map metadataFilePinnedTimestampMap; + private final ThreadPool threadPool; /** @@ -132,6 +135,7 @@ public RemoteSegmentStoreDirectory( this.remoteMetadataDirectory = remoteMetadataDirectory; this.mdLockManager = mdLockManager; this.threadPool = threadPool; + this.metadataFilePinnedTimestampMap = new HashMap<>(); this.logger = Loggers.getLogger(getClass(), shardId); init(); } @@ -176,6 +180,38 @@ public RemoteSegmentMetadata initializeToSpecificCommit(long primaryTerm, long c return remoteSegmentMetadata; } + /** + * Initializes the remote segment metadata to a specific timestamp. + * + * @param timestamp The timestamp to initialize the remote segment metadata to. + * @return The RemoteSegmentMetadata object corresponding to the specified timestamp, or null if no metadata file is found for that timestamp. + * @throws IOException If an I/O error occurs while reading the metadata file. + */ + public RemoteSegmentMetadata initializeToSpecificTimestamp(long timestamp) throws IOException { + List metadataFiles = remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ); + Set lockedMetadataFiles = RemoteStoreUtils.getPinnedTimestampLockedFiles( + metadataFiles, + Set.of(timestamp), + MetadataFilenameUtils::getTimestamp, + MetadataFilenameUtils::getNodeIdByPrimaryTermAndGen + ); + if (lockedMetadataFiles.isEmpty()) { + return null; + } + assert lockedMetadataFiles.size() == 1 : "Expected exactly one metadata file but got " + lockedMetadataFiles; + String metadataFile = lockedMetadataFiles.iterator().next(); + RemoteSegmentMetadata remoteSegmentMetadata = readMetadataFile(metadataFile); + if (remoteSegmentMetadata != null) { + this.segmentsUploadedToRemoteStore = new ConcurrentHashMap<>(remoteSegmentMetadata.getMetadata()); + } else { + this.segmentsUploadedToRemoteStore = new ConcurrentHashMap<>(); + } + return remoteSegmentMetadata; + } + /** * Read the latest metadata file to get the list of segments uploaded to the remote segment store. * We upload a metadata file per refresh, but it is not unique per refresh. Refresh metadata file is unique for a given commit. @@ -324,7 +360,8 @@ public static String getMetadataFilename( long translogGeneration, long uploadCounter, int metadataVersion, - String nodeId + String nodeId, + long creationTimestamp ) { return String.join( SEPARATOR, @@ -334,11 +371,30 @@ public static String getMetadataFilename( RemoteStoreUtils.invertLong(translogGeneration), RemoteStoreUtils.invertLong(uploadCounter), String.valueOf(Objects.hash(nodeId)), - RemoteStoreUtils.invertLong(System.currentTimeMillis()), + RemoteStoreUtils.invertLong(creationTimestamp), String.valueOf(metadataVersion) ); } + public static String getMetadataFilename( + long primaryTerm, + long generation, + long translogGeneration, + long uploadCounter, + int metadataVersion, + String nodeId + ) { + return getMetadataFilename( + primaryTerm, + generation, + translogGeneration, + uploadCounter, + metadataVersion, + nodeId, + System.currentTimeMillis() + ); + } + // Visible for testing static long getPrimaryTerm(String[] filenameTokens) { return RemoteStoreUtils.invertLong(filenameTokens[1]); @@ -778,6 +834,7 @@ public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException ); return; } + List sortedMetadataFileList = remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( MetadataFilenameUtils.METADATA_PREFIX, Integer.MAX_VALUE @@ -791,16 +848,44 @@ public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException return; } - List metadataFilesEligibleToDelete = new ArrayList<>( - sortedMetadataFileList.subList(lastNMetadataFilesToKeep, sortedMetadataFileList.size()) + // Check last fetch status of pinned timestamps. If stale, return. + if (RemoteStoreUtils.isPinnedTimestampStateStale()) { + logger.warn("Skipping remote segment store garbage collection as last fetch of pinned timestamp is stale"); + return; + } + + Tuple> pinnedTimestampsState = RemoteStorePinnedTimestampService.getPinnedTimestamps(); + + Set implicitLockedFiles = RemoteStoreUtils.getPinnedTimestampLockedFiles( + sortedMetadataFileList, + pinnedTimestampsState.v2(), + metadataFilePinnedTimestampMap, + MetadataFilenameUtils::getTimestamp, + MetadataFilenameUtils::getNodeIdByPrimaryTermAndGen ); - Set allLockFiles; + final Set allLockFiles = new HashSet<>(implicitLockedFiles); + try { - allLockFiles = ((RemoteStoreMetadataLockManager) mdLockManager).fetchLockedMetadataFiles(MetadataFilenameUtils.METADATA_PREFIX); + allLockFiles.addAll( + ((RemoteStoreMetadataLockManager) mdLockManager).fetchLockedMetadataFiles(MetadataFilenameUtils.METADATA_PREFIX) + ); } catch (Exception e) { logger.error("Exception while fetching segment metadata lock files, skipping deleteStaleSegments", e); return; } + + List metadataFilesEligibleToDelete = new ArrayList<>( + sortedMetadataFileList.subList(lastNMetadataFilesToKeep, sortedMetadataFileList.size()) + ); + + // Along with last N files, we need to keep files since last successful run of scheduler + long lastSuccessfulFetchOfPinnedTimestamps = pinnedTimestampsState.v1(); + metadataFilesEligibleToDelete = RemoteStoreUtils.filterOutMetadataFilesBasedOnAge( + metadataFilesEligibleToDelete, + MetadataFilenameUtils::getTimestamp, + lastSuccessfulFetchOfPinnedTimestamps + ); + List metadataFilesToBeDeleted = metadataFilesEligibleToDelete.stream() .filter(metadataFile -> allLockFiles.contains(metadataFile) == false) .collect(Collectors.toList()); diff --git a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java index c37db618c2522..e4e982bc47bb7 100644 --- a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java +++ b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java @@ -245,6 +245,14 @@ public static Tuple> getPinnedTimestamps() { return pinnedTimestampsSet; } + public RemoteStorePinnedTimestampsBlobStore pinnedTimestampsBlobStore() { + return pinnedTimestampsBlobStore; + } + + public BlobStoreTransferService blobStoreTransferService() { + return blobStoreTransferService; + } + /** * Inner class for asynchronously updating the pinned timestamp set. */ @@ -266,11 +274,12 @@ protected void runInternal() { clusterService.state().metadata().clusterUUID(), blobStoreRepository.getCompressor() ); - BlobPath path = pinnedTimestampsBlobStore.getBlobPathForUpload(remotePinnedTimestamps); - blobStoreTransferService.listAllInSortedOrder(path, remotePinnedTimestamps.getType(), 1, new ActionListener<>() { + BlobPath path = pinnedTimestampsBlobStore().getBlobPathForUpload(remotePinnedTimestamps); + blobStoreTransferService().listAllInSortedOrder(path, remotePinnedTimestamps.getType(), 1, new ActionListener<>() { @Override public void onResponse(List blobMetadata) { if (blobMetadata.isEmpty()) { + pinnedTimestampsSet = new Tuple<>(triggerTimestamp, Set.of()); return; } PinnedTimestamps pinnedTimestamps = readExistingPinnedTimestamps(blobMetadata.get(0).name(), remotePinnedTimestamps); diff --git a/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java b/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java index ceaee8337ae34..a6db37285fe6f 100644 --- a/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java +++ b/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java @@ -1081,5 +1081,4 @@ public void testIsPinnedTimestampStateStaleFeatureEnabled() { setupRemotePinnedTimestampFeature(true); assertTrue(RemoteStoreUtils.isPinnedTimestampStateStale()); } - } diff --git a/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java index ff9b62a341deb..59a77fdbc9950 100644 --- a/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java @@ -49,7 +49,8 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase { 34, 1, 1, - "node-1" + "node-1", + System.currentTimeMillis() - 300000 ); protected final String metadataFilenameDup = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( @@ -58,7 +59,8 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase { 34, 2, 1, - "node-2" + "node-2", + System.currentTimeMillis() - 400000 ); protected final String metadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( 12, @@ -66,7 +68,8 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase { 34, 1, 1, - "node-1" + "node-1", + System.currentTimeMillis() - 500000 ); protected final String metadataFilename3 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( 10, @@ -74,7 +77,8 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase { 34, 1, 1, - "node-1" + "node-1", + System.currentTimeMillis() - 600000 ); protected final String metadataFilename4 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( 10, @@ -82,7 +86,8 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase { 34, 1, 1, - "node-1" + "node-1", + System.currentTimeMillis() - 700000 ); public void setupRemoteSegmentStoreDirectory() throws IOException { diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java index 574c5bf620474..4a8173285d53c 100644 --- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java @@ -24,20 +24,29 @@ import org.apache.lucene.util.Version; import org.opensearch.common.UUIDs; import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; +import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.stream.write.WriteContext; import org.opensearch.common.io.VersionedCodecStreamWrapper; import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.lucene.store.ByteArrayIndexInput; +import org.opensearch.common.settings.Settings; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.bytes.BytesReference; import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; +import org.opensearch.gateway.remote.model.RemoteStorePinnedTimestampsBlobStore; import org.opensearch.index.remote.RemoteStoreEnums.PathHashAlgorithm; import org.opensearch.index.remote.RemoteStoreEnums.PathType; import org.opensearch.index.remote.RemoteStorePathStrategy; import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadataHandler; +import org.opensearch.index.translog.transfer.BlobStoreTransferService; +import org.opensearch.node.Node; +import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; +import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService; +import org.opensearch.repositories.RepositoriesService; +import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.test.MockLogAppender; import org.opensearch.test.junit.annotations.TestLogging; import org.opensearch.threadpool.ThreadPool; @@ -55,6 +64,7 @@ import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; import java.util.stream.Collectors; import org.mockito.Mockito; @@ -66,6 +76,7 @@ import static org.hamcrest.CoreMatchers.is; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.eq; @@ -78,9 +89,47 @@ public class RemoteSegmentStoreDirectoryTests extends BaseRemoteSegmentStoreDirectoryTests { + RemoteStorePinnedTimestampService remoteStorePinnedTimestampService; + @Before public void setup() throws IOException { setupRemoteSegmentStoreDirectory(); + + Supplier repositoriesServiceSupplier = mock(Supplier.class); + Settings settings = Settings.builder() + .put(Node.NODE_ATTRIBUTES.getKey() + RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, "remote-repo") + .build(); + RepositoriesService repositoriesService = mock(RepositoriesService.class); + when(repositoriesServiceSupplier.get()).thenReturn(repositoriesService); + BlobStoreRepository blobStoreRepository = mock(BlobStoreRepository.class); + when(repositoriesService.repository("remote-repo")).thenReturn(blobStoreRepository); + + when(threadPool.schedule(any(), any(), any())).then(invocationOnMock -> { + Runnable runnable = invocationOnMock.getArgument(0); + runnable.run(); + return null; + }).then(subsequentInvocationsOnMock -> null); + + remoteStorePinnedTimestampService = new RemoteStorePinnedTimestampService( + repositoriesServiceSupplier, + settings, + threadPool, + clusterService + ); + RemoteStorePinnedTimestampService remoteStorePinnedTimestampServiceSpy = Mockito.spy(remoteStorePinnedTimestampService); + + RemoteStorePinnedTimestampsBlobStore remoteStorePinnedTimestampsBlobStore = mock(RemoteStorePinnedTimestampsBlobStore.class); + BlobStoreTransferService blobStoreTransferService = mock(BlobStoreTransferService.class); + when(remoteStorePinnedTimestampServiceSpy.pinnedTimestampsBlobStore()).thenReturn(remoteStorePinnedTimestampsBlobStore); + when(remoteStorePinnedTimestampServiceSpy.blobStoreTransferService()).thenReturn(blobStoreTransferService); + + doAnswer(invocationOnMock -> { + ActionListener> actionListener = invocationOnMock.getArgument(3); + actionListener.onResponse(new ArrayList<>()); + return null; + }).when(blobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); + + remoteStorePinnedTimestampServiceSpy.start(); } public void testUploadedSegmentMetadataToString() { @@ -1141,6 +1190,57 @@ public void testMetadataFileNameOrder() { assertEquals(14, count); } + public void testInitializeToSpecificTimestampNoMetadataFiles() throws IOException { + when(remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder(RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, Integer.MAX_VALUE)).thenReturn(new ArrayList<>()); + assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); + } + + public void testInitializeToSpecificTimestampNoMdMatchingTimestamp() throws IOException { + String metadataPrefix = "metadata__1__2__3__4__5__"; + List metadataFiles = new ArrayList<>(); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(3000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(4000)); + + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ) + ).thenReturn(metadataFiles); + assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); + } + + public void testInitializeToSpecificTimestampMatchingMdFile() throws IOException { + String metadataPrefix = "metadata__1__2__3__4__5__"; + List metadataFiles = new ArrayList<>(); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(1000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(3000)); + + Map metadata = new HashMap<>(); + metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234::512::" + Version.LATEST.major); + metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345::1024::" + Version.LATEST.major); + + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ) + ).thenReturn(metadataFiles); + when(remoteMetadataDirectory.getBlobStream(metadataPrefix + RemoteStoreUtils.invertLong(1000))).thenReturn( + createMetadataFileBytes(metadata, indexShard.getLatestReplicationCheckpoint(), segmentInfos) + ); + + RemoteSegmentMetadata remoteSegmentMetadata = remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L); + assertNotNull(remoteSegmentMetadata); + Map uploadedSegments = remoteSegmentStoreDirectory + .getSegmentsUploadedToRemoteStore(); + assertEquals(2, uploadedSegments.size()); + assertTrue(uploadedSegments.containsKey("_0.cfe")); + assertTrue(uploadedSegments.containsKey("_0.cfs")); + } + private static class WrapperIndexOutput extends IndexOutput { public IndexOutput indexOutput; From 8406a9c89b357e229c81de0e7e9752e20770d567 Mon Sep 17 00:00:00 2001 From: Sachin Kale Date: Sun, 25 Aug 2024 22:59:43 +0530 Subject: [PATCH 2/5] Change tests to use pinned timestamp enable/disable setting Signed-off-by: Sachin Kale --- .../RemoteSegmentStoreDirectoryTests.java | 71 ++++++------------- 1 file changed, 22 insertions(+), 49 deletions(-) diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java index 4a8173285d53c..6f29d09fadfba 100644 --- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java @@ -24,29 +24,23 @@ import org.apache.lucene.util.Version; import org.opensearch.common.UUIDs; import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; -import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.stream.write.WriteContext; import org.opensearch.common.io.VersionedCodecStreamWrapper; import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.lucene.store.ByteArrayIndexInput; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.bytes.BytesReference; import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; -import org.opensearch.gateway.remote.model.RemoteStorePinnedTimestampsBlobStore; import org.opensearch.index.remote.RemoteStoreEnums.PathHashAlgorithm; import org.opensearch.index.remote.RemoteStoreEnums.PathType; import org.opensearch.index.remote.RemoteStorePathStrategy; import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadataHandler; -import org.opensearch.index.translog.transfer.BlobStoreTransferService; -import org.opensearch.node.Node; -import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; -import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService; -import org.opensearch.repositories.RepositoriesService; -import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.test.MockLogAppender; import org.opensearch.test.junit.annotations.TestLogging; import org.opensearch.threadpool.ThreadPool; @@ -64,19 +58,18 @@ import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; import java.util.stream.Collectors; import org.mockito.Mockito; import static org.opensearch.index.store.RemoteSegmentStoreDirectory.METADATA_FILES_TO_FETCH; import static org.opensearch.index.store.RemoteSegmentStoreDirectory.MetadataFilenameUtils.SEPARATOR; +import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED; import static org.opensearch.test.RemoteStoreTestUtils.createMetadataFileBytes; import static org.opensearch.test.RemoteStoreTestUtils.getDummyMetadata; import static org.hamcrest.CoreMatchers.is; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.any; -import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.eq; @@ -89,47 +82,9 @@ public class RemoteSegmentStoreDirectoryTests extends BaseRemoteSegmentStoreDirectoryTests { - RemoteStorePinnedTimestampService remoteStorePinnedTimestampService; - @Before public void setup() throws IOException { setupRemoteSegmentStoreDirectory(); - - Supplier repositoriesServiceSupplier = mock(Supplier.class); - Settings settings = Settings.builder() - .put(Node.NODE_ATTRIBUTES.getKey() + RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, "remote-repo") - .build(); - RepositoriesService repositoriesService = mock(RepositoriesService.class); - when(repositoriesServiceSupplier.get()).thenReturn(repositoriesService); - BlobStoreRepository blobStoreRepository = mock(BlobStoreRepository.class); - when(repositoriesService.repository("remote-repo")).thenReturn(blobStoreRepository); - - when(threadPool.schedule(any(), any(), any())).then(invocationOnMock -> { - Runnable runnable = invocationOnMock.getArgument(0); - runnable.run(); - return null; - }).then(subsequentInvocationsOnMock -> null); - - remoteStorePinnedTimestampService = new RemoteStorePinnedTimestampService( - repositoriesServiceSupplier, - settings, - threadPool, - clusterService - ); - RemoteStorePinnedTimestampService remoteStorePinnedTimestampServiceSpy = Mockito.spy(remoteStorePinnedTimestampService); - - RemoteStorePinnedTimestampsBlobStore remoteStorePinnedTimestampsBlobStore = mock(RemoteStorePinnedTimestampsBlobStore.class); - BlobStoreTransferService blobStoreTransferService = mock(BlobStoreTransferService.class); - when(remoteStorePinnedTimestampServiceSpy.pinnedTimestampsBlobStore()).thenReturn(remoteStorePinnedTimestampsBlobStore); - when(remoteStorePinnedTimestampServiceSpy.blobStoreTransferService()).thenReturn(blobStoreTransferService); - - doAnswer(invocationOnMock -> { - ActionListener> actionListener = invocationOnMock.getArgument(3); - actionListener.onResponse(new ArrayList<>()); - return null; - }).when(blobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); - - remoteStorePinnedTimestampServiceSpy.start(); } public void testUploadedSegmentMetadataToString() { @@ -1190,12 +1145,27 @@ public void testMetadataFileNameOrder() { assertEquals(14, count); } + private void setupRemotePinnedTimestampFeature(boolean enabled) { + RemoteStoreSettings remoteStoreSettings = new RemoteStoreSettings( + Settings.builder().put(CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), enabled).build(), + new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS) + ); + } + public void testInitializeToSpecificTimestampNoMetadataFiles() throws IOException { - when(remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder(RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, Integer.MAX_VALUE)).thenReturn(new ArrayList<>()); + setupRemotePinnedTimestampFeature(true); + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ) + ).thenReturn(new ArrayList<>()); assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); + setupRemotePinnedTimestampFeature(false); } public void testInitializeToSpecificTimestampNoMdMatchingTimestamp() throws IOException { + setupRemotePinnedTimestampFeature(true); String metadataPrefix = "metadata__1__2__3__4__5__"; List metadataFiles = new ArrayList<>(); metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); @@ -1209,9 +1179,11 @@ public void testInitializeToSpecificTimestampNoMdMatchingTimestamp() throws IOEx ) ).thenReturn(metadataFiles); assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); + setupRemotePinnedTimestampFeature(false); } public void testInitializeToSpecificTimestampMatchingMdFile() throws IOException { + setupRemotePinnedTimestampFeature(true); String metadataPrefix = "metadata__1__2__3__4__5__"; List metadataFiles = new ArrayList<>(); metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(1000)); @@ -1239,6 +1211,7 @@ public void testInitializeToSpecificTimestampMatchingMdFile() throws IOException assertEquals(2, uploadedSegments.size()); assertTrue(uploadedSegments.containsKey("_0.cfe")); assertTrue(uploadedSegments.containsKey("_0.cfs")); + setupRemotePinnedTimestampFeature(false); } private static class WrapperIndexOutput extends IndexOutput { From b2d204d8abc046341e192a960b98c600e9132543 Mon Sep 17 00:00:00 2001 From: Sachin Kale Date: Mon, 26 Aug 2024 09:16:41 +0530 Subject: [PATCH 3/5] Add separate test class for remote segment directory with pinned timestamps Signed-off-by: Sachin Kale --- .../opensearch/remotestore/RemoteStoreIT.java | 7 +- .../RemoteStorePinnedTimestampService.java | 4 +- .../BaseRemoteSegmentStoreDirectoryTests.java | 45 +-- .../RemoteSegmentStoreDirectoryTests.java | 73 ----- ...toreDirectoryWithPinnedTimestampTests.java | 292 ++++++++++++++++++ 5 files changed, 305 insertions(+), 116 deletions(-) create mode 100644 server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java index 589b64c3b0ab5..194dce5f4a57a 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java @@ -217,7 +217,10 @@ public void testStaleCommitDeletionWithInvokeFlush() throws Exception { } else { // As delete is async its possible that the file gets created before the deletion or after // deletion. - assertTrue(actualFileCount >= lastNMetadataFilesToKeep); + MatcherAssert.assertThat( + actualFileCount, + is(oneOf(lastNMetadataFilesToKeep - 1, lastNMetadataFilesToKeep, lastNMetadataFilesToKeep + 1)) + ); } }, 30, TimeUnit.SECONDS); } @@ -246,7 +249,7 @@ public void testStaleCommitDeletionWithMinSegmentFiles_3() throws Exception { Path indexPath = Path.of(segmentRepoPath + "/" + shardPath); int actualFileCount = getFileCount(indexPath); // We also allow (numberOfIterations + 1) as index creation also triggers refresh. - assertTrue(actualFileCount >= 4); + MatcherAssert.assertThat(actualFileCount, is(oneOf(4))); } public void testStaleCommitDeletionWithMinSegmentFiles_Disabled() throws Exception { diff --git a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java index e4e982bc47bb7..f7b262664d147 100644 --- a/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java +++ b/server/src/main/java/org/opensearch/node/remotestore/RemoteStorePinnedTimestampService.java @@ -219,9 +219,9 @@ private ActionListener getListenerForWriteCallResponse( private PinnedTimestamps readExistingPinnedTimestamps(String blobFilename, RemotePinnedTimestamps remotePinnedTimestamps) { remotePinnedTimestamps.setBlobFileName(blobFilename); - remotePinnedTimestamps.setFullBlobName(pinnedTimestampsBlobStore.getBlobPathForUpload(remotePinnedTimestamps)); + remotePinnedTimestamps.setFullBlobName(pinnedTimestampsBlobStore().getBlobPathForUpload(remotePinnedTimestamps)); try { - return pinnedTimestampsBlobStore.read(remotePinnedTimestamps); + return pinnedTimestampsBlobStore().read(remotePinnedTimestamps); } catch (IOException e) { throw new RuntimeException("Failed to read existing pinned timestamps", e); } diff --git a/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java index 59a77fdbc9950..2c55d26261fe0 100644 --- a/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java @@ -43,52 +43,19 @@ public class BaseRemoteSegmentStoreDirectoryTests extends IndexShardTestCase { protected SegmentInfos segmentInfos; protected ThreadPool threadPool; - protected final String metadataFilename = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( - 12, - 23, - 34, - 1, - 1, - "node-1", - System.currentTimeMillis() - 300000 - ); + protected String metadataFilename = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(12, 23, 34, 1, 1, "node-1"); - protected final String metadataFilenameDup = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + protected String metadataFilenameDup = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( 12, 23, 34, 2, 1, - "node-2", - System.currentTimeMillis() - 400000 - ); - protected final String metadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( - 12, - 13, - 34, - 1, - 1, - "node-1", - System.currentTimeMillis() - 500000 - ); - protected final String metadataFilename3 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( - 10, - 38, - 34, - 1, - 1, - "node-1", - System.currentTimeMillis() - 600000 - ); - protected final String metadataFilename4 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( - 10, - 36, - 34, - 1, - 1, - "node-1", - System.currentTimeMillis() - 700000 + "node-2" ); + protected String metadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(12, 13, 34, 1, 1, "node-1"); + protected String metadataFilename3 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(10, 38, 34, 1, 1, "node-1"); + protected String metadataFilename4 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename(10, 36, 34, 1, 1, "node-1"); public void setupRemoteSegmentStoreDirectory() throws IOException { remoteDataDirectory = mock(RemoteDirectory.class); diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java index 6f29d09fadfba..574c5bf620474 100644 --- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java @@ -28,8 +28,6 @@ import org.opensearch.common.io.VersionedCodecStreamWrapper; import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.lucene.store.ByteArrayIndexInput; -import org.opensearch.common.settings.ClusterSettings; -import org.opensearch.common.settings.Settings; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.bytes.BytesReference; import org.opensearch.core.index.Index; @@ -40,7 +38,6 @@ import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadataHandler; -import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.test.MockLogAppender; import org.opensearch.test.junit.annotations.TestLogging; import org.opensearch.threadpool.ThreadPool; @@ -64,7 +61,6 @@ import static org.opensearch.index.store.RemoteSegmentStoreDirectory.METADATA_FILES_TO_FETCH; import static org.opensearch.index.store.RemoteSegmentStoreDirectory.MetadataFilenameUtils.SEPARATOR; -import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED; import static org.opensearch.test.RemoteStoreTestUtils.createMetadataFileBytes; import static org.opensearch.test.RemoteStoreTestUtils.getDummyMetadata; import static org.hamcrest.CoreMatchers.is; @@ -1145,75 +1141,6 @@ public void testMetadataFileNameOrder() { assertEquals(14, count); } - private void setupRemotePinnedTimestampFeature(boolean enabled) { - RemoteStoreSettings remoteStoreSettings = new RemoteStoreSettings( - Settings.builder().put(CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), enabled).build(), - new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS) - ); - } - - public void testInitializeToSpecificTimestampNoMetadataFiles() throws IOException { - setupRemotePinnedTimestampFeature(true); - when( - remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( - RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, - Integer.MAX_VALUE - ) - ).thenReturn(new ArrayList<>()); - assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); - setupRemotePinnedTimestampFeature(false); - } - - public void testInitializeToSpecificTimestampNoMdMatchingTimestamp() throws IOException { - setupRemotePinnedTimestampFeature(true); - String metadataPrefix = "metadata__1__2__3__4__5__"; - List metadataFiles = new ArrayList<>(); - metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); - metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(3000)); - metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(4000)); - - when( - remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( - RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, - Integer.MAX_VALUE - ) - ).thenReturn(metadataFiles); - assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); - setupRemotePinnedTimestampFeature(false); - } - - public void testInitializeToSpecificTimestampMatchingMdFile() throws IOException { - setupRemotePinnedTimestampFeature(true); - String metadataPrefix = "metadata__1__2__3__4__5__"; - List metadataFiles = new ArrayList<>(); - metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(1000)); - metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); - metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(3000)); - - Map metadata = new HashMap<>(); - metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234::512::" + Version.LATEST.major); - metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345::1024::" + Version.LATEST.major); - - when( - remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( - RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, - Integer.MAX_VALUE - ) - ).thenReturn(metadataFiles); - when(remoteMetadataDirectory.getBlobStream(metadataPrefix + RemoteStoreUtils.invertLong(1000))).thenReturn( - createMetadataFileBytes(metadata, indexShard.getLatestReplicationCheckpoint(), segmentInfos) - ); - - RemoteSegmentMetadata remoteSegmentMetadata = remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L); - assertNotNull(remoteSegmentMetadata); - Map uploadedSegments = remoteSegmentStoreDirectory - .getSegmentsUploadedToRemoteStore(); - assertEquals(2, uploadedSegments.size()); - assertTrue(uploadedSegments.containsKey("_0.cfe")); - assertTrue(uploadedSegments.containsKey("_0.cfs")); - setupRemotePinnedTimestampFeature(false); - } - private static class WrapperIndexOutput extends IndexOutput { public IndexOutput indexOutput; diff --git a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java new file mode 100644 index 0000000000000..b4f93d706bb1e --- /dev/null +++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryWithPinnedTimestampTests.java @@ -0,0 +1,292 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.index.store; + +import org.apache.lucene.util.Version; +import org.opensearch.common.UUIDs; +import org.opensearch.common.blobstore.BlobMetadata; +import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.common.blobstore.support.PlainBlobMetadata; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.action.ActionListener; +import org.opensearch.gateway.remote.model.RemotePinnedTimestamps; +import org.opensearch.gateway.remote.model.RemoteStorePinnedTimestampsBlobStore; +import org.opensearch.index.remote.RemoteStoreUtils; +import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata; +import org.opensearch.index.translog.transfer.BlobStoreTransferService; +import org.opensearch.indices.RemoteStoreSettings; +import org.opensearch.node.Node; +import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; +import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService; +import org.opensearch.repositories.RepositoriesService; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.junit.Before; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import org.mockito.Mockito; + +import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED; +import static org.opensearch.test.RemoteStoreTestUtils.createMetadataFileBytes; +import static org.hamcrest.CoreMatchers.is; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.anyInt; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class RemoteSegmentStoreDirectoryWithPinnedTimestampTests extends RemoteSegmentStoreDirectoryTests { + + Runnable updatePinnedTimstampTask; + BlobStoreTransferService blobStoreTransferService; + RemoteStorePinnedTimestampsBlobStore remoteStorePinnedTimestampsBlobStore; + RemoteStorePinnedTimestampService remoteStorePinnedTimestampServiceSpy; + + @Before + public void setupPinnedTimestamp() throws IOException { + RemoteStoreSettings remoteStoreSettings = new RemoteStoreSettings( + Settings.builder().put(CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), true).build(), + new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS) + ); + + Supplier repositoriesServiceSupplier = mock(Supplier.class); + Settings settings = Settings.builder() + .put(Node.NODE_ATTRIBUTES.getKey() + RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, "remote-repo") + .build(); + RepositoriesService repositoriesService = mock(RepositoriesService.class); + when(repositoriesServiceSupplier.get()).thenReturn(repositoriesService); + BlobStoreRepository blobStoreRepository = mock(BlobStoreRepository.class); + when(repositoriesService.repository("remote-repo")).thenReturn(blobStoreRepository); + + when(threadPool.schedule(any(), any(), any())).then(invocationOnMock -> { + updatePinnedTimstampTask = invocationOnMock.getArgument(0); + updatePinnedTimstampTask.run(); + return null; + }).then(subsequentInvocationsOnMock -> null); + + RemoteStorePinnedTimestampService remoteStorePinnedTimestampService = new RemoteStorePinnedTimestampService( + repositoriesServiceSupplier, + settings, + threadPool, + clusterService + ); + remoteStorePinnedTimestampServiceSpy = Mockito.spy(remoteStorePinnedTimestampService); + + remoteStorePinnedTimestampsBlobStore = mock(RemoteStorePinnedTimestampsBlobStore.class); + blobStoreTransferService = mock(BlobStoreTransferService.class); + when(remoteStorePinnedTimestampServiceSpy.pinnedTimestampsBlobStore()).thenReturn(remoteStorePinnedTimestampsBlobStore); + when(remoteStorePinnedTimestampServiceSpy.blobStoreTransferService()).thenReturn(blobStoreTransferService); + + doAnswer(invocationOnMock -> { + ActionListener> actionListener = invocationOnMock.getArgument(3); + actionListener.onResponse(new ArrayList<>()); + return null; + }).when(blobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); + + remoteStorePinnedTimestampServiceSpy.start(); + + metadataWithOlderTimestamp(); + } + + private void metadataWithOlderTimestamp() { + metadataFilename = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 12, + 23, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() - 300000 + ); + metadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 12, + 13, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() - 400000 + ); + metadataFilename3 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 10, + 38, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() - 500000 + ); + metadataFilename4 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 10, + 36, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() - 600000 + ); + } + + public void testInitializeToSpecificTimestampNoMetadataFiles() throws IOException { + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ) + ).thenReturn(new ArrayList<>()); + assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); + } + + public void testInitializeToSpecificTimestampNoMdMatchingTimestamp() throws IOException { + String metadataPrefix = "metadata__1__2__3__4__5__"; + List metadataFiles = new ArrayList<>(); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(3000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(4000)); + + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ) + ).thenReturn(metadataFiles); + assertNull(remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L)); + } + + public void testInitializeToSpecificTimestampMatchingMdFile() throws IOException { + String metadataPrefix = "metadata__1__2__3__4__5__"; + List metadataFiles = new ArrayList<>(); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(1000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(2000)); + metadataFiles.add(metadataPrefix + RemoteStoreUtils.invertLong(3000)); + + Map metadata = new HashMap<>(); + metadata.put("_0.cfe", "_0.cfe::_0.cfe__" + UUIDs.base64UUID() + "::1234::512::" + Version.LATEST.major); + metadata.put("_0.cfs", "_0.cfs::_0.cfs__" + UUIDs.base64UUID() + "::2345::1024::" + Version.LATEST.major); + + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX, + Integer.MAX_VALUE + ) + ).thenReturn(metadataFiles); + when(remoteMetadataDirectory.getBlobStream(metadataPrefix + RemoteStoreUtils.invertLong(1000))).thenReturn( + createMetadataFileBytes(metadata, indexShard.getLatestReplicationCheckpoint(), segmentInfos) + ); + + RemoteSegmentMetadata remoteSegmentMetadata = remoteSegmentStoreDirectory.initializeToSpecificTimestamp(1234L); + assertNotNull(remoteSegmentMetadata); + Map uploadedSegments = remoteSegmentStoreDirectory + .getSegmentsUploadedToRemoteStore(); + assertEquals(2, uploadedSegments.size()); + assertTrue(uploadedSegments.containsKey("_0.cfe")); + assertTrue(uploadedSegments.containsKey("_0.cfs")); + } + + public void testDeleteStaleCommitsNoPinnedTimestampMdFilesLatest() throws Exception { + metadataFilename = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 12, + 23, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() + ); + metadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 12, + 13, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() + ); + metadataFilename3 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getMetadataFilename( + 10, + 38, + 34, + 1, + 1, + "node-1", + System.currentTimeMillis() + ); + + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + eq(RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX), + anyInt() + ) + ).thenReturn(List.of(metadataFilename, metadataFilename2, metadataFilename3)); + + populateMetadata(); + remoteSegmentStoreDirectory.init(); + + // populateMetadata() adds stub to return 3 metadata files + // We are passing lastNMetadataFilesToKeep=2 here so that oldest 1 metadata file will be deleted + // But as the oldest metadata file's timestamp is within time threshold since last successful fetch, + // GC will skip deleting any data or metadata files. + remoteSegmentStoreDirectory.deleteStaleSegmentsAsync(2); + + assertBusy(() -> assertThat(remoteSegmentStoreDirectory.canDeleteStaleCommits.get(), is(true))); + verify(remoteDataDirectory, times(0)).deleteFile(any()); + verify(remoteMetadataDirectory, times(0)).deleteFile(any()); + } + + public void testDeleteStaleCommitsPinnedTimestampMdFile() throws Exception { + when( + remoteMetadataDirectory.listFilesByPrefixInLexicographicOrder( + eq(RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX), + anyInt() + ) + ).thenReturn(List.of(metadataFilename, metadataFilename2, metadataFilename3)); + + doAnswer(invocationOnMock -> { + ActionListener> actionListener = invocationOnMock.getArgument(3); + actionListener.onResponse(List.of(new PlainBlobMetadata("pinned_timestamp_123", 1000))); + return null; + }).when(blobStoreTransferService).listAllInSortedOrder(any(), any(), eq(1), any()); + + long pinnedTimestampMatchingMetadataFilename2 = RemoteSegmentStoreDirectory.MetadataFilenameUtils.getTimestamp(metadataFilename2) + 10; + when(remoteStorePinnedTimestampsBlobStore.read(any())).thenReturn(new RemotePinnedTimestamps.PinnedTimestamps(Map.of(pinnedTimestampMatchingMetadataFilename2, List.of("xyz")))); + when(remoteStorePinnedTimestampsBlobStore.getBlobPathForUpload(any())).thenReturn(new BlobPath()); + + final Map> metadataFilenameContentMapping = populateMetadata(); + final List filesToBeDeleted = metadataFilenameContentMapping.get(metadataFilename3) + .values() + .stream() + .map(metadata -> metadata.split(RemoteSegmentStoreDirectory.UploadedSegmentMetadata.SEPARATOR)[1]) + .collect(Collectors.toList()); + + updatePinnedTimstampTask.run(); + + remoteSegmentStoreDirectory.init(); + + // popluateMetadata() adds stub to return 3 metadata files + // We are passing lastNMetadataFilesToKeep=2 here so that oldest 1 metadata file will be deleted + remoteSegmentStoreDirectory.deleteStaleSegmentsAsync(1); + + for (final String file : filesToBeDeleted) { + verify(remoteDataDirectory).deleteFile(file); + } + assertBusy(() -> assertThat(remoteSegmentStoreDirectory.canDeleteStaleCommits.get(), is(true))); + verify(remoteMetadataDirectory).deleteFile(metadataFilename3); + verify(remoteMetadataDirectory, times(0)).deleteFile(metadataFilename2); + } +} From 2dc98524b2fcfe5c55ba1250b4af6ca12ac643bf Mon Sep 17 00:00:00 2001 From: Sachin Kale Date: Tue, 27 Aug 2024 08:21:11 +0530 Subject: [PATCH 4/5] Enable pinned timestamp randomly for integ tests Signed-off-by: Sachin Kale --- .../opensearch/remotestore/RemoteStoreIT.java | 20 ++++++++++++++----- .../test/OpenSearchIntegTestCase.java | 1 + 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java index 194dce5f4a57a..a327b683874f6 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java @@ -217,10 +217,15 @@ public void testStaleCommitDeletionWithInvokeFlush() throws Exception { } else { // As delete is async its possible that the file gets created before the deletion or after // deletion. - MatcherAssert.assertThat( - actualFileCount, - is(oneOf(lastNMetadataFilesToKeep - 1, lastNMetadataFilesToKeep, lastNMetadataFilesToKeep + 1)) - ); + if (RemoteStoreSettings.isPinnedTimestampsEnabled()) { + // With pinned timestamp, we also keep md files since last successful fetch + assertTrue(actualFileCount >= lastNMetadataFilesToKeep); + } else { + MatcherAssert.assertThat( + actualFileCount, + is(oneOf(lastNMetadataFilesToKeep - 1, lastNMetadataFilesToKeep, lastNMetadataFilesToKeep + 1)) + ); + } } }, 30, TimeUnit.SECONDS); } @@ -249,7 +254,12 @@ public void testStaleCommitDeletionWithMinSegmentFiles_3() throws Exception { Path indexPath = Path.of(segmentRepoPath + "/" + shardPath); int actualFileCount = getFileCount(indexPath); // We also allow (numberOfIterations + 1) as index creation also triggers refresh. - MatcherAssert.assertThat(actualFileCount, is(oneOf(4))); + if (RemoteStoreSettings.isPinnedTimestampsEnabled()) { + // With pinned timestamp, we also keep md files since last successful fetch + assertTrue(actualFileCount >= 4); + } else { + assertEquals(4, actualFileCount); + } } public void testStaleCommitDeletionWithMinSegmentFiles_Disabled() throws Exception { diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java index b86cce682c68e..911aa92340de6 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java @@ -2792,6 +2792,7 @@ private static Settings buildRemoteStoreNodeAttributes( } settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING.getKey(), randomFrom(PathType.values())); settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_METADATA.getKey(), randomBoolean()); + settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), randomBoolean()); return settings.build(); } From 1637f0f55810ffe7d2ecac060f8317b5e8561134 Mon Sep 17 00:00:00 2001 From: Sachin Kale Date: Tue, 27 Aug 2024 14:20:44 +0530 Subject: [PATCH 5/5] Check integ test status by enabling pinned timestamps Signed-off-by: Sachin Kale --- .../main/java/org/opensearch/test/OpenSearchIntegTestCase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java index 911aa92340de6..0260daa48367d 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java @@ -2792,7 +2792,7 @@ private static Settings buildRemoteStoreNodeAttributes( } settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING.getKey(), randomFrom(PathType.values())); settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_METADATA.getKey(), randomBoolean()); - settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), randomBoolean()); + settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), true); return settings.build(); }