From b1cfad2747abf7da20c96b435343b537afe5e19c Mon Sep 17 00:00:00 2001 From: Kunal Kotwani Date: Fri, 1 Sep 2023 15:53:43 -0700 Subject: [PATCH] Add async segment file download support from remote store Signed-off-by: Kunal Kotwani --- .../org/opensearch/index/IndexService.java | 5 +- .../opensearch/index/shard/IndexShard.java | 90 +++++++++++++-- .../store/RemoteSegmentStoreDirectory.java | 26 +++++ .../org/opensearch/index/store/Store.java | 20 +++- .../RemoteStoreReplicationSource.java | 54 ++++++++- .../index/shard/IndexShardTests.java | 2 +- .../RemoteSegmentStoreDirectoryTests.java | 109 ++++++++++++++++++ .../opensearch/index/store/StoreTests.java | 23 +++- .../index/shard/IndexShardTestCase.java | 13 ++- 9 files changed, 312 insertions(+), 30 deletions(-) diff --git a/server/src/main/java/org/opensearch/index/IndexService.java b/server/src/main/java/org/opensearch/index/IndexService.java index 80ead0a333ba3..20095df355d0a 100644 --- a/server/src/main/java/org/opensearch/index/IndexService.java +++ b/server/src/main/java/org/opensearch/index/IndexService.java @@ -479,7 +479,7 @@ public synchronized IndexShard createShard( Store remoteStore = null; if (this.indexSettings.isRemoteStoreEnabled()) { Directory remoteDirectory = remoteDirectoryFactory.newDirectory(this.indexSettings, path); - remoteStore = new Store(shardId, this.indexSettings, remoteDirectory, lock, Store.OnClose.EMPTY); + remoteStore = new Store(shardId, this.indexSettings, remoteDirectory, lock, Store.OnClose.EMPTY, path); } Directory directory = directoryFactory.newDirectory(this.indexSettings, path); @@ -488,7 +488,8 @@ public synchronized IndexShard createShard( this.indexSettings, directory, lock, - new StoreCloseListener(shardId, () -> eventListener.onStoreClosed(shardId)) + new StoreCloseListener(shardId, () -> eventListener.onStoreClosed(shardId)), + path ); eventListener.onStoreCreated(shardId); indexShard = new IndexShard( diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java index 352876e54547e..a2fe8bea28872 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -59,9 +59,11 @@ import org.opensearch.ExceptionsHelper; import org.opensearch.OpenSearchException; import org.opensearch.action.ActionRunnable; +import org.opensearch.action.LatchedActionListener; import org.opensearch.action.admin.indices.flush.FlushRequest; import org.opensearch.action.admin.indices.forcemerge.ForceMergeRequest; import org.opensearch.action.admin.indices.upgrade.post.UpgradeRequest; +import org.opensearch.action.support.PlainActionFuture; import org.opensearch.action.support.replication.PendingReplicationActions; import org.opensearch.action.support.replication.ReplicationResponse; import org.opensearch.cluster.metadata.DataStream; @@ -196,6 +198,7 @@ import java.nio.channels.ClosedByInterruptException; import java.nio.charset.StandardCharsets; import java.nio.file.NoSuchFileException; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -212,6 +215,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; @@ -4806,39 +4810,101 @@ private String copySegmentFiles( Map uploadedSegments, boolean overrideLocal ) throws IOException { - List downloadedSegments = new ArrayList<>(); - List skippedSegments = new ArrayList<>(); + Set toDownloadSegments = new HashSet<>(); + Set skippedSegments = new HashSet<>(); String segmentNFile = null; + try { - Set localSegmentFiles = Sets.newHashSet(storeDirectory.listAll()); if (overrideLocal) { - for (String file : localSegmentFiles) { - storeDirectory.deleteFile(file); - } + deleteExistingSegments(storeDirectory); } + for (String file : uploadedSegments.keySet()) { long checksum = Long.parseLong(uploadedSegments.get(file).getChecksum()); if (overrideLocal || localDirectoryContains(storeDirectory, file, checksum) == false) { - storeDirectory.copyFrom(sourceRemoteDirectory, file, file, IOContext.DEFAULT); - downloadedSegments.add(file); + toDownloadSegments.add(file); } else { skippedSegments.add(file); } - if (targetRemoteDirectory != null) { - targetRemoteDirectory.copyFrom(storeDirectory, file, file, IOContext.DEFAULT); - } + if (file.startsWith(IndexFileNames.SEGMENTS)) { assert segmentNFile == null : "There should be only one SegmentInfosSnapshot file"; segmentNFile = file; } } + + if (!toDownloadSegments.isEmpty()) { + try { + CountDownLatch completionLatch = new CountDownLatch(1); + LatchedActionListener downloadsCompletionListener = new LatchedActionListener<>( + new PlainActionFuture<>(), + completionLatch + ); + + downloadSegments( + storeDirectory, + sourceRemoteDirectory, + targetRemoteDirectory, + toDownloadSegments, + downloadsCompletionListener + ); + + completionLatch.await(); + } catch (Exception e) { + throw new IOException("Error occurred when downloading segments from remote store", e); + } + } } finally { - logger.trace("Downloaded segments here: {}", downloadedSegments); + logger.trace("Downloaded segments here: {}}", toDownloadSegments); logger.trace("Skipped download for segments here: {}", skippedSegments); } + return segmentNFile; } + private void downloadSegments( + Directory storeDirectory, + RemoteSegmentStoreDirectory sourceRemoteDirectory, + RemoteSegmentStoreDirectory targetRemoteDirectory, + Set toDownloadSegments, + ActionListener completionListener + ) { + final AtomicInteger totalDownloadedSegments = new AtomicInteger(toDownloadSegments.size()); + final Path indexPath = store.shardPath() == null ? null : store.shardPath().resolveIndex(); + + final ActionListener segmentsDownloadListener = new ActionListener<>() { + @Override + public void onResponse(String fileName) { + try { + if (targetRemoteDirectory != null) { + targetRemoteDirectory.copyFrom(storeDirectory, fileName, fileName, IOContext.DEFAULT); + } + if (totalDownloadedSegments.decrementAndGet() == 0) { + completionListener.onResponse(null); + } + } catch (IOException ex) { + logger.error("Failed to download segment file {} from the remote store", fileName); + onFailure(ex); + } + } + + @Override + public void onFailure(Exception e) { + logger.error("Failed to download one of the segment files from the remote store", e); + completionListener.onFailure(e); + } + }; + + toDownloadSegments.forEach(file -> sourceRemoteDirectory.copyTo(storeDirectory, file, indexPath, segmentsDownloadListener)); + } + + private void deleteExistingSegments(Directory storeDirectory) throws IOException { + Set localSegmentFiles = Sets.newHashSet(storeDirectory.listAll()); + for (String file : localSegmentFiles) { + storeDirectory.deleteFile(file); + } + } + private boolean localDirectoryContains(Directory localDirectory, String file, long checksum) { try (IndexInput indexInput = localDirectory.openInput(file, IOContext.DEFAULT)) { if (checksum == CodecUtil.retrieveChecksum(indexInput)) { 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 0f6ca2a61b67d..55e181340be6f 100644 --- a/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java +++ b/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java @@ -24,6 +24,7 @@ import org.apache.lucene.store.IndexOutput; import org.apache.lucene.util.Version; import org.opensearch.common.UUIDs; +import org.opensearch.common.blobstore.VerifyingMultiStreamBlobContainer; import org.opensearch.common.io.VersionedCodecStreamWrapper; import org.opensearch.common.lucene.store.ByteArrayIndexInput; import org.opensearch.core.action.ActionListener; @@ -39,6 +40,7 @@ import java.io.FileNotFoundException; import java.io.IOException; import java.nio.file.NoSuchFileException; +import java.nio.file.Path; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -434,6 +436,30 @@ public void copyFrom(Directory from, String src, IOContext context, ActionListen } } + /** + * Copies the segment from the remote directory to a local directory, preferably using an async mechanism using multi stream support. + * @param storeDirectory The directory where the segment needs to be copied to + * @param src The name of the segment file + * @param segmentDirectoryPath The file path for the segment directory + * @param segmentCompletionListener Async listener which should be notified once the segment is downloaded or in case of failure + */ + public void copyTo(Directory storeDirectory, String src, Path segmentDirectoryPath, ActionListener segmentCompletionListener) { + final String blobName = getExistingRemoteFilename(src); + if (segmentDirectoryPath != null && remoteDataDirectory.getBlobContainer() instanceof VerifyingMultiStreamBlobContainer) { + VerifyingMultiStreamBlobContainer blobContainer = (VerifyingMultiStreamBlobContainer) remoteDataDirectory.getBlobContainer(); + Path destinationPath = segmentDirectoryPath.resolve(src); + blobContainer.asyncBlobDownload(blobName, destinationPath, threadPool, segmentCompletionListener); + } else { + // Fallback to older mechanism of downloading the file + try { + storeDirectory.copyFrom(this, src, src, IOContext.DEFAULT); + segmentCompletionListener.onResponse(src); + } catch (IOException e) { + segmentCompletionListener.onFailure(e); + } + } + } + /** * This acquires a lock on a given commit by creating a lock file in lock directory using {@code FileLockInfo} * diff --git a/server/src/main/java/org/opensearch/index/store/Store.java b/server/src/main/java/org/opensearch/index/store/Store.java index b3ea2cdd02e21..285241ba89996 100644 --- a/server/src/main/java/org/opensearch/index/store/Store.java +++ b/server/src/main/java/org/opensearch/index/store/Store.java @@ -66,6 +66,7 @@ import org.apache.lucene.util.Version; import org.opensearch.ExceptionsHelper; import org.opensearch.common.UUIDs; +import org.opensearch.common.annotation.InternalApi; import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.logging.Loggers; import org.opensearch.common.lucene.Lucene; @@ -92,6 +93,7 @@ import org.opensearch.index.seqno.SequenceNumbers; import org.opensearch.index.shard.AbstractIndexShardComponent; import org.opensearch.index.shard.IndexShard; +import org.opensearch.index.shard.ShardPath; import org.opensearch.index.translog.Translog; import java.io.Closeable; @@ -179,6 +181,7 @@ public class Store extends AbstractIndexShardComponent implements Closeable, Ref private final ReentrantReadWriteLock metadataLock = new ReentrantReadWriteLock(); private final ShardLock shardLock; private final OnClose onClose; + private final ShardPath shardPath; // used to ref count files when a new Reader is opened for PIT/Scroll queries // prevents segment files deletion until the PIT/Scroll expires or is discarded @@ -192,10 +195,17 @@ protected void closeInternal() { }; public Store(ShardId shardId, IndexSettings indexSettings, Directory directory, ShardLock shardLock) { - this(shardId, indexSettings, directory, shardLock, OnClose.EMPTY); + this(shardId, indexSettings, directory, shardLock, OnClose.EMPTY, null); } - public Store(ShardId shardId, IndexSettings indexSettings, Directory directory, ShardLock shardLock, OnClose onClose) { + public Store( + ShardId shardId, + IndexSettings indexSettings, + Directory directory, + ShardLock shardLock, + OnClose onClose, + ShardPath shardPath + ) { super(shardId, indexSettings); final TimeValue refreshInterval = indexSettings.getValue(INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING); logger.debug("store stats are refreshed with refresh_interval [{}]", refreshInterval); @@ -203,6 +213,7 @@ public Store(ShardId shardId, IndexSettings indexSettings, Directory directory, this.directory = new StoreDirectory(sizeCachingDir, Loggers.getLogger("index.store.deletes", shardId)); this.shardLock = shardLock; this.onClose = onClose; + this.shardPath = shardPath; assert onClose != null; assert shardLock != null; assert shardLock.getShardId().equals(shardId); @@ -213,6 +224,11 @@ public Directory directory() { return directory; } + @InternalApi + public ShardPath shardPath() { + return shardPath; + } + /** * Returns the last committed segments info for this store * diff --git a/server/src/main/java/org/opensearch/indices/replication/RemoteStoreReplicationSource.java b/server/src/main/java/org/opensearch/indices/replication/RemoteStoreReplicationSource.java index 7252fea044a02..2d76d59710d56 100644 --- a/server/src/main/java/org/opensearch/indices/replication/RemoteStoreReplicationSource.java +++ b/server/src/main/java/org/opensearch/indices/replication/RemoteStoreReplicationSource.java @@ -13,23 +13,27 @@ import org.apache.lucene.index.SegmentInfos; import org.apache.lucene.store.Directory; import org.apache.lucene.store.FilterDirectory; -import org.apache.lucene.store.IOContext; import org.apache.lucene.util.Version; +import org.opensearch.action.LatchedActionListener; import org.opensearch.common.concurrent.GatedCloseable; import org.opensearch.core.action.ActionListener; import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.IndexShardState; +import org.opensearch.index.shard.ShardPath; import org.opensearch.index.store.RemoteSegmentStoreDirectory; import org.opensearch.index.store.Store; import org.opensearch.index.store.StoreFileMetadata; import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata; import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; /** @@ -106,31 +110,69 @@ public void getSegmentFiles( logger.trace("Downloading segments files from remote store {}", filesToFetch); RemoteSegmentMetadata remoteSegmentMetadata = remoteDirectory.readLatestMetadataFile(); - List downloadedSegments = new ArrayList<>(); + List toDownloadSegments = new ArrayList<>(); Collection directoryFiles = List.of(indexShard.store().directory().listAll()); if (remoteSegmentMetadata != null) { try { indexShard.store().incRef(); indexShard.remoteStore().incRef(); final Directory storeDirectory = indexShard.store().directory(); + final ShardPath shardPath = indexShard.shardPath(); for (StoreFileMetadata fileMetadata : filesToFetch) { String file = fileMetadata.name(); assert directoryFiles.contains(file) == false : "Local store already contains the file " + file; - storeDirectory.copyFrom(remoteDirectory, file, file, IOContext.DEFAULT); - downloadedSegments.add(fileMetadata); + toDownloadSegments.add(fileMetadata); } - logger.trace("Downloaded segments from remote store {}", downloadedSegments); + + CountDownLatch countDownLatch = new CountDownLatch(1); + LatchedActionListener completionListener = new LatchedActionListener<>( + listener, + countDownLatch + ); + downloadSegments(storeDirectory, remoteDirectory, toDownloadSegments, shardPath, completionListener); + countDownLatch.await(); + + logger.trace("Downloaded segments from remote store {}", toDownloadSegments); } finally { indexShard.store().decRef(); indexShard.remoteStore().decRef(); } } - listener.onResponse(new GetSegmentFilesResponse(downloadedSegments)); } catch (Exception e) { listener.onFailure(e); } } + private void downloadSegments( + Directory storeDirectory, + RemoteSegmentStoreDirectory remoteStoreDirectory, + List toDownloadSegments, + ShardPath shardPath, + ActionListener completionListener + ) { + + final AtomicInteger totalDownloadedSegments = new AtomicInteger(toDownloadSegments.size()); + final Path indexPath = shardPath == null ? null : shardPath.resolveIndex(); + final ActionListener segmentsDownloadListener = new ActionListener<>() { + @Override + public void onResponse(String fileName) { + if (totalDownloadedSegments.decrementAndGet() == 0) { + completionListener.onResponse(new GetSegmentFilesResponse(toDownloadSegments)); + } + } + + @Override + public void onFailure(Exception e) { + logger.error("Failed to download one of the segment files from the remote store", e); + completionListener.onFailure(e); + } + }; + + toDownloadSegments.forEach( + fileMetadata -> remoteStoreDirectory.copyTo(storeDirectory, fileMetadata.name(), indexPath, segmentsDownloadListener) + ); + } + @Override public String getDescription() { return "RemoteStoreReplicationSource"; diff --git a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java index 2281c086db5d8..3c381d04b5a4d 100644 --- a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java +++ b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java @@ -1778,7 +1778,7 @@ public Set getPendingDeletions() throws IOException { } }; - try (Store store = createStore(shardId, new IndexSettings(metadata, Settings.EMPTY), directory)) { + try (Store store = createStore(shardId, new IndexSettings(metadata, Settings.EMPTY), directory, shardPath)) { IndexShard shard = newShard( shardRouting, shardPath, 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 44dfb44eb9a15..7750ba748dbdf 100644 --- a/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java @@ -45,6 +45,7 @@ import java.io.IOException; import java.nio.file.NoSuchFileException; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -60,6 +61,8 @@ 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.anyString; +import static org.mockito.ArgumentMatchers.contains; import static org.mockito.Mockito.any; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; @@ -515,6 +518,112 @@ public void onFailure(Exception e) {} storeDirectory.close(); } + public void testCopyFilesToMultipart() throws Exception { + String filename = "_0.cfe"; + populateMetadata(); + remoteSegmentStoreDirectory.init(); + + Directory storeDirectory = mock(Directory.class); + VerifyingMultiStreamBlobContainer blobContainer = mock(VerifyingMultiStreamBlobContainer.class); + when(remoteDataDirectory.getBlobContainer()).thenReturn(blobContainer); + + Mockito.doAnswer(invocation -> { + ActionListener completionListener = invocation.getArgument(3); + completionListener.onResponse(invocation.getArgument(0)); + return null; + }).when(blobContainer).asyncBlobDownload(any(), any(), any(), any()); + + CountDownLatch downloadLatch = new CountDownLatch(1); + ActionListener completionListener = new ActionListener() { + @Override + public void onResponse(String unused) { + downloadLatch.countDown(); + } + + @Override + public void onFailure(Exception e) {} + }; + Path path = createTempDir(); + remoteSegmentStoreDirectory.copyTo(storeDirectory, filename, path, completionListener); + assertTrue(downloadLatch.await(5000, TimeUnit.SECONDS)); + verify(blobContainer, times(1)).asyncBlobDownload(contains(filename), eq(path.resolve(filename)), any(), any()); + verify(storeDirectory, times(0)).copyFrom(any(), any(), any(), any()); + } + + public void testCopyFilesTo() throws Exception { + String filename = "_0.cfe"; + populateMetadata(); + remoteSegmentStoreDirectory.init(); + + Directory storeDirectory = mock(Directory.class); + CountDownLatch downloadLatch = new CountDownLatch(1); + ActionListener completionListener = new ActionListener<>() { + @Override + public void onResponse(String unused) { + downloadLatch.countDown(); + } + + @Override + public void onFailure(Exception e) {} + }; + Path path = createTempDir(); + remoteSegmentStoreDirectory.copyTo(storeDirectory, filename, path, completionListener); + assertTrue(downloadLatch.await(5000, TimeUnit.MILLISECONDS)); + verify(storeDirectory, times(1)).copyFrom(any(), eq(filename), eq(filename), eq(IOContext.DEFAULT)); + } + + public void testCopyFilesToEmptyPath() throws Exception { + String filename = "_0.cfe"; + populateMetadata(); + remoteSegmentStoreDirectory.init(); + + Directory storeDirectory = mock(Directory.class); + VerifyingMultiStreamBlobContainer blobContainer = mock(VerifyingMultiStreamBlobContainer.class); + when(remoteDataDirectory.getBlobContainer()).thenReturn(blobContainer); + + CountDownLatch downloadLatch = new CountDownLatch(1); + ActionListener completionListener = new ActionListener<>() { + @Override + public void onResponse(String unused) { + downloadLatch.countDown(); + } + + @Override + public void onFailure(Exception e) {} + }; + Path path = createTempDir(); + remoteSegmentStoreDirectory.copyTo(storeDirectory, filename, null, completionListener); + assertTrue(downloadLatch.await(5000, TimeUnit.MILLISECONDS)); + verify(storeDirectory, times(1)).copyFrom(any(), eq(filename), eq(filename), eq(IOContext.DEFAULT)); + } + + public void testCopyFilesToException() throws Exception { + String filename = "_0.cfe"; + populateMetadata(); + remoteSegmentStoreDirectory.init(); + + Directory storeDirectory = mock(Directory.class); + Mockito.doThrow(new IOException()) + .when(storeDirectory) + .copyFrom(any(Directory.class), anyString(), anyString(), any(IOContext.class)); + CountDownLatch downloadLatch = new CountDownLatch(1); + ActionListener completionListener = new ActionListener<>() { + @Override + public void onResponse(String unused) { + + } + + @Override + public void onFailure(Exception e) { + downloadLatch.countDown(); + } + }; + Path path = createTempDir(); + remoteSegmentStoreDirectory.copyTo(storeDirectory, filename, path, completionListener); + assertTrue(downloadLatch.await(5000, TimeUnit.MILLISECONDS)); + verify(storeDirectory, times(1)).copyFrom(any(), eq(filename), eq(filename), eq(IOContext.DEFAULT)); + } + public void testCopyFilesFromMultipartIOException() throws Exception { String filename = "_100.si"; VerifyingMultiStreamBlobContainer blobContainer = mock(VerifyingMultiStreamBlobContainer.class); diff --git a/server/src/test/java/org/opensearch/index/store/StoreTests.java b/server/src/test/java/org/opensearch/index/store/StoreTests.java index 8395b3e8ac08e..d7d326b325cc6 100644 --- a/server/src/test/java/org/opensearch/index/store/StoreTests.java +++ b/server/src/test/java/org/opensearch/index/store/StoreTests.java @@ -84,6 +84,7 @@ import org.opensearch.index.seqno.ReplicationTracker; import org.opensearch.index.seqno.RetentionLease; import org.opensearch.index.seqno.SequenceNumbers; +import org.opensearch.index.shard.ShardPath; import org.opensearch.index.translog.Translog; import org.opensearch.indices.replication.common.ReplicationType; import org.opensearch.indices.store.TransportNodesListShardStoreMetadata; @@ -798,7 +799,7 @@ public void testOnCloseCallback() throws IOException { assertEquals(shardId, theLock.getShardId()); assertEquals(lock, theLock); count.incrementAndGet(); - }); + }, null); assertEquals(count.get(), 0); final int iters = randomIntBetween(1, 10); @@ -809,6 +810,26 @@ public void testOnCloseCallback() throws IOException { assertEquals(count.get(), 1); } + public void testStoreShardPath() { + final ShardId shardId = new ShardId("index", "_na_", 1); + final Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, org.opensearch.Version.CURRENT) + .put(Store.INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING.getKey(), TimeValue.timeValueMinutes(0)) + .build(); + final Path path = createTempDir().resolve(shardId.getIndex().getUUID()).resolve(String.valueOf(shardId.id())); + final ShardPath shardPath = new ShardPath(false, path, path, shardId); + final Store store = new Store( + shardId, + IndexSettingsModule.newIndexSettings("index", settings), + StoreTests.newDirectory(random()), + new DummyShardLock(shardId), + Store.OnClose.EMPTY, + shardPath + ); + assertEquals(shardPath, store.shardPath()); + store.close(); + } + public void testStoreStats() throws IOException { final ShardId shardId = new ShardId("index", "_na_", 1); Settings settings = Settings.builder() diff --git a/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java b/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java index 474acc764620d..6904c0f65cc9c 100644 --- a/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java +++ b/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java @@ -271,11 +271,11 @@ public Settings threadPoolSettings() { } protected Store createStore(IndexSettings indexSettings, ShardPath shardPath) throws IOException { - return createStore(shardPath.getShardId(), indexSettings, newFSDirectory(shardPath.resolveIndex())); + return createStore(shardPath.getShardId(), indexSettings, newFSDirectory(shardPath.resolveIndex()), shardPath); } - protected Store createStore(ShardId shardId, IndexSettings indexSettings, Directory directory) throws IOException { - return new Store(shardId, indexSettings, directory, new DummyShardLock(shardId)); + protected Store createStore(ShardId shardId, IndexSettings indexSettings, Directory directory, ShardPath shardPath) throws IOException { + return new Store(shardId, indexSettings, directory, new DummyShardLock(shardId), Store.OnClose.EMPTY, shardPath); } protected Releasable acquirePrimaryOperationPermitBlockingly(IndexShard indexShard) throws ExecutionException, InterruptedException { @@ -653,7 +653,7 @@ protected IndexShard newShard( remotePath = createTempDir(); } - remoteStore = createRemoteStore(remotePath, routing, indexMetadata); + remoteStore = createRemoteStore(remotePath, routing, indexMetadata, shardPath); remoteStoreStatsTrackerFactory = new RemoteStoreStatsTrackerFactory(clusterService, indexSettings.getSettings()); BlobStoreRepository repo = createRepository(remotePath); @@ -767,11 +767,12 @@ protected RepositoriesService createRepositoriesService() { return repositoriesService; } - protected Store createRemoteStore(Path path, ShardRouting shardRouting, IndexMetadata metadata) throws IOException { + protected Store createRemoteStore(Path path, ShardRouting shardRouting, IndexMetadata metadata, ShardPath shardPath) + throws IOException { Settings nodeSettings = Settings.builder().put("node.name", shardRouting.currentNodeId()).build(); ShardId shardId = shardRouting.shardId(); RemoteSegmentStoreDirectory remoteSegmentStoreDirectory = createRemoteSegmentStoreDirectory(shardId, path); - return createStore(shardId, new IndexSettings(metadata, nodeSettings), remoteSegmentStoreDirectory); + return createStore(shardId, new IndexSettings(metadata, nodeSettings), remoteSegmentStoreDirectory, shardPath); } protected RemoteSegmentStoreDirectory createRemoteSegmentStoreDirectory(ShardId shardId, Path path) throws IOException {