diff --git a/CHANGELOG.md b/CHANGELOG.md index d238936748f19..475e8d8c07621 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -52,7 +52,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Remove identity-related feature flagged code from the RestController ([#15430](https://github.com/opensearch-project/OpenSearch/pull/15430)) - Remove Identity FeatureFlag ([#16024](https://github.com/opensearch-project/OpenSearch/pull/16024)) - Ensure RestHandler.Wrapper delegates all implementations to the wrapped handler ([#16154](https://github.com/opensearch-project/OpenSearch/pull/16154)) - +- Optimise clone operation for incremental full cluster snapshots ([#16296](https://github.com/opensearch-project/OpenSearch/pull/16296)) ### Deprecated diff --git a/server/src/main/java/org/opensearch/snapshots/SnapshotsService.java b/server/src/main/java/org/opensearch/snapshots/SnapshotsService.java index 50630a5ce3383..45e6d11448a80 100644 --- a/server/src/main/java/org/opensearch/snapshots/SnapshotsService.java +++ b/server/src/main/java/org/opensearch/snapshots/SnapshotsService.java @@ -1510,7 +1510,8 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS private final Set currentlyCloning = Collections.synchronizedSet(new HashSet<>()); - private void runReadyClone( + // Made to package private to be able to test the method in UTs + void runReadyClone( Snapshot target, SnapshotId sourceSnapshot, ShardSnapshotStatus shardStatusBefore, @@ -1534,69 +1535,75 @@ public void onFailure(Exception e) { @Override protected void doRun() { final String localNodeId = clusterService.localNode().getId(); - repository.getRepositoryData(ActionListener.wrap(repositoryData -> { - try { - final IndexMetadata indexMetadata = repository.getSnapshotIndexMetaData( - repositoryData, + if (remoteStoreIndexShallowCopy == false) { + executeClone(localNodeId, false); + } else { + repository.getRepositoryData(ActionListener.wrap(repositoryData -> { + try { + final IndexMetadata indexMetadata = repository.getSnapshotIndexMetaData( + repositoryData, + sourceSnapshot, + repoShardId.index() + ); + final boolean cloneRemoteStoreIndexShardSnapshot = indexMetadata.getSettings() + .getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false); + executeClone(localNodeId, cloneRemoteStoreIndexShardSnapshot); + } catch (IOException e) { + logger.warn("Failed to get index-metadata from repository data for index [{}]", repoShardId.index().getName()); + failCloneShardAndUpdateClusterState(target, sourceSnapshot, repoShardId); + } + }, this::onFailure)); + } + } + + private void executeClone(String localNodeId, boolean cloneRemoteStoreIndexShardSnapshot) { + if (currentlyCloning.add(repoShardId)) { + if (cloneRemoteStoreIndexShardSnapshot) { + repository.cloneRemoteStoreIndexShardSnapshot( sourceSnapshot, - repoShardId.index() + target.getSnapshotId(), + repoShardId, + shardStatusBefore.generation(), + remoteStoreLockManagerFactory, + getCloneCompletionListener(localNodeId) ); - final boolean cloneRemoteStoreIndexShardSnapshot = remoteStoreIndexShallowCopy - && indexMetadata.getSettings().getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false); - final SnapshotId targetSnapshot = target.getSnapshotId(); - final ActionListener listener = ActionListener.wrap( - generation -> innerUpdateSnapshotState( - new ShardSnapshotUpdate( - target, - repoShardId, - new ShardSnapshotStatus(localNodeId, ShardState.SUCCESS, generation) - ), - ActionListener.runBefore( - ActionListener.wrap( - v -> logger.trace( - "Marked [{}] as successfully cloned from [{}] to [{}]", - repoShardId, - sourceSnapshot, - targetSnapshot - ), - e -> { - logger.warn("Cluster state update after successful shard clone [{}] failed", repoShardId); - failAllListenersOnMasterFailOver(e); - } - ), - () -> currentlyCloning.remove(repoShardId) - ) - ), - e -> { - logger.warn("Exception [{}] while trying to clone shard [{}]", e, repoShardId); - failCloneShardAndUpdateClusterState(target, sourceSnapshot, repoShardId); - } + } else { + repository.cloneShardSnapshot( + sourceSnapshot, + target.getSnapshotId(), + repoShardId, + shardStatusBefore.generation(), + getCloneCompletionListener(localNodeId) ); - if (currentlyCloning.add(repoShardId)) { - if (cloneRemoteStoreIndexShardSnapshot) { - repository.cloneRemoteStoreIndexShardSnapshot( - sourceSnapshot, - targetSnapshot, + } + } + } + + private ActionListener getCloneCompletionListener(String localNodeId) { + return ActionListener.wrap( + generation -> innerUpdateSnapshotState( + new ShardSnapshotUpdate(target, repoShardId, new ShardSnapshotStatus(localNodeId, ShardState.SUCCESS, generation)), + ActionListener.runBefore( + ActionListener.wrap( + v -> logger.trace( + "Marked [{}] as successfully cloned from [{}] to [{}]", repoShardId, - shardStatusBefore.generation(), - remoteStoreLockManagerFactory, - listener - ); - } else { - repository.cloneShardSnapshot( sourceSnapshot, - targetSnapshot, - repoShardId, - shardStatusBefore.generation(), - listener - ); - } - } - } catch (IOException e) { - logger.warn("Failed to get index-metadata from repository data for index [{}]", repoShardId.index().getName()); + target.getSnapshotId() + ), + e -> { + logger.warn("Cluster state update after successful shard clone [{}] failed", repoShardId); + failAllListenersOnMasterFailOver(e); + } + ), + () -> currentlyCloning.remove(repoShardId) + ) + ), + e -> { + logger.warn("Exception [{}] while trying to clone shard [{}]", e, repoShardId); failCloneShardAndUpdateClusterState(target, sourceSnapshot, repoShardId); } - }, this::onFailure)); + ); } }); } diff --git a/server/src/test/java/org/opensearch/snapshots/SnapshotsServiceTests.java b/server/src/test/java/org/opensearch/snapshots/SnapshotsServiceTests.java index 1652ba29b8fa7..e374636f60d22 100644 --- a/server/src/test/java/org/opensearch/snapshots/SnapshotsServiceTests.java +++ b/server/src/test/java/org/opensearch/snapshots/SnapshotsServiceTests.java @@ -33,24 +33,41 @@ package org.opensearch.snapshots; import org.opensearch.Version; +import org.opensearch.action.support.ActionFilters; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.SnapshotsInProgress; import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.cluster.routing.IndexRoutingTable; import org.opensearch.cluster.routing.IndexShardRoutingTable; import org.opensearch.cluster.routing.RoutingTable; import org.opensearch.cluster.routing.ShardRoutingState; import org.opensearch.cluster.routing.TestShardRouting; +import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.UUIDs; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.concurrent.OpenSearchExecutors; +import org.opensearch.core.action.ActionListener; import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; +import org.opensearch.indices.RemoteStoreSettings; import org.opensearch.repositories.IndexId; +import org.opensearch.repositories.IndexMetaDataGenerations; +import org.opensearch.repositories.RepositoriesService; +import org.opensearch.repositories.Repository; +import org.opensearch.repositories.RepositoryData; +import org.opensearch.repositories.RepositoryException; import org.opensearch.repositories.RepositoryShardId; +import org.opensearch.repositories.ShardGenerations; import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; +import java.io.IOException; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -58,8 +75,17 @@ import java.util.stream.Collectors; import java.util.stream.StreamSupport; +import org.mockito.ArgumentCaptor; + import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_VERSION_CREATED; import static org.hamcrest.Matchers.is; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class SnapshotsServiceTests extends OpenSearchTestCase { @@ -404,6 +430,419 @@ public void testCompletedCloneStartsNextClone() throws Exception { assertIsNoop(updatedClusterState, completeShardClone); } + /** + * Tests the runReadyClone method when remote store index shallow copy is disabled. + * It verifies that: + * 1. getRepositoryData is never called + * 2. cloneShardSnapshot is called with the correct parameters + */ + public void testRunReadyCloneWithRemoteStoreIndexShallowCopyDisabled() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + try (SnapshotsService snapshotsService = createSnapshotsService()) { + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, false); + verify(mockRepository, never()).getRepositoryData(any()); + verify(mockRepository).cloneShardSnapshot( + eq(sourceSnapshot), + eq(target.getSnapshotId()), + eq(repoShardId), + eq(shardStatusBefore.generation()), + any() + ); + } + } + + /** + * Tests the runReadyClone method when remote store index shallow copy is enabled. + * It verifies that: + * 1. getRepositoryData is called + * 2. cloneRemoteStoreIndexShardSnapshot is called with the correct parameters + * This test simulates a scenario where the index has remote store enabled. + */ + + public void testRunReadyCloneWithRemoteStoreIndexShallowCopyEnabled() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + + // Create a real RepositoryData instance + RepositoryData repositoryData = new RepositoryData( + RepositoryData.EMPTY_REPO_GEN, + Collections.singletonMap(sourceSnapshot.getName(), sourceSnapshot), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap(), + ShardGenerations.EMPTY, + IndexMetaDataGenerations.EMPTY + ); + + // Mock the getRepositoryData method to use the ActionListener + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(0); + listener.onResponse(repositoryData); + return null; + }).when(mockRepository).getRepositoryData(any(ActionListener.class)); + + IndexMetadata mockIndexMetadata = mock(IndexMetadata.class); + when(mockRepository.getSnapshotIndexMetaData(eq(repositoryData), eq(sourceSnapshot), any())).thenReturn(mockIndexMetadata); + + Settings mockSettings = Settings.builder().put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, true).build(); + when(mockIndexMetadata.getSettings()).thenReturn(mockSettings); + + try (SnapshotsService snapshotsService = createSnapshotsService()) { + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, true); + + // Verify that getRepositoryData was called + verify(mockRepository).getRepositoryData(any(ActionListener.class)); + + // Verify that cloneRemoteStoreIndexShardSnapshot was called with the correct arguments + verify(mockRepository).cloneRemoteStoreIndexShardSnapshot( + eq(sourceSnapshot), + eq(target.getSnapshotId()), + eq(repoShardId), + eq(shardStatusBefore.generation()), + any(), + any() + ); + } + } + + /** + * Tests the error handling in runReadyClone when a RepositoryException occurs. + * It verifies that: + * 1. getRepositoryData is called and throws an exception + * 2. Neither cloneShardSnapshot nor cloneRemoteStoreIndexShardSnapshot are called + */ + public void testRunReadyCloneWithRepositoryException() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + + // Mock the getRepositoryData method to throw an exception + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(0); + listener.onFailure(new RepositoryException(repoName, "Test exception")); + return null; + }).when(mockRepository).getRepositoryData(any(ActionListener.class)); + + try (SnapshotsService snapshotsService = createSnapshotsService()) { + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, true); + + // Verify that getRepositoryData was called + verify(mockRepository).getRepositoryData(any(ActionListener.class)); + + // Verify that neither cloneShardSnapshot nor cloneRemoteStoreIndexShardSnapshot were called + verify(mockRepository, never()).cloneShardSnapshot(any(), any(), any(), any(), any()); + verify(mockRepository, never()).cloneRemoteStoreIndexShardSnapshot(any(), any(), any(), any(), any(), any()); + } + } + + /** + * Tests the runReadyClone method when remote store index shallow copy is globally enabled, + * but disabled for the specific index. + * It verifies that: + * 1. getRepositoryData is called + * 2. cloneShardSnapshot is called instead of cloneRemoteStoreIndexShardSnapshot + */ + public void testRunReadyCloneWithRemoteStoreIndexShallowCopyEnabledButIndexDisabled() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + + RepositoryData repositoryData = new RepositoryData( + RepositoryData.EMPTY_REPO_GEN, + Collections.singletonMap(sourceSnapshot.getName(), sourceSnapshot), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap(), + ShardGenerations.EMPTY, + IndexMetaDataGenerations.EMPTY + ); + + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(0); + listener.onResponse(repositoryData); + return null; + }).when(mockRepository).getRepositoryData(any(ActionListener.class)); + + IndexMetadata mockIndexMetadata = mock(IndexMetadata.class); + when(mockRepository.getSnapshotIndexMetaData(eq(repositoryData), eq(sourceSnapshot), any())).thenReturn(mockIndexMetadata); + + Settings mockSettings = Settings.builder().put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false).build(); + when(mockIndexMetadata.getSettings()).thenReturn(mockSettings); + + try (SnapshotsService snapshotsService = createSnapshotsService()) { + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, true); + + verify(mockRepository).getRepositoryData(any(ActionListener.class)); + verify(mockRepository).cloneShardSnapshot( + eq(sourceSnapshot), + eq(target.getSnapshotId()), + eq(repoShardId), + eq(shardStatusBefore.generation()), + any() + ); + } + } + + /** + * Tests the error handling in runReadyClone when an IOException occurs while getting snapshot index metadata. + * It verifies that: + * 1. getRepositoryData is called + * 2. getSnapshotIndexMetaData throws an IOException + * 3. Neither cloneShardSnapshot nor cloneRemoteStoreIndexShardSnapshot are called + */ + public void testRunReadyCloneWithIOException() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + + RepositoryData repositoryData = new RepositoryData( + RepositoryData.EMPTY_REPO_GEN, + Collections.singletonMap(sourceSnapshot.getName(), sourceSnapshot), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap(), + ShardGenerations.EMPTY, + IndexMetaDataGenerations.EMPTY + ); + + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(0); + listener.onResponse(repositoryData); + return null; + }).when(mockRepository).getRepositoryData(any(ActionListener.class)); + + when(mockRepository.getSnapshotIndexMetaData(eq(repositoryData), eq(sourceSnapshot), any())).thenThrow( + new IOException("Test IO Exception") + ); + + try (SnapshotsService snapshotsService = createSnapshotsService()) { + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, true); + + verify(mockRepository).getRepositoryData(any(ActionListener.class)); + verify(mockRepository).getSnapshotIndexMetaData(eq(repositoryData), eq(sourceSnapshot), any()); + verify(mockRepository, never()).cloneShardSnapshot(any(), any(), any(), any(), any()); + verify(mockRepository, never()).cloneRemoteStoreIndexShardSnapshot(any(), any(), any(), any(), any(), any()); + } + } + + /** + * Tests the completion listener functionality in runReadyClone when the operation is successful. + * It verifies that: + * 1. The clone operation completes successfully + * 2. The correct ShardSnapshotUpdate is submitted to the cluster state + */ + public void testRunReadyCloneCompletionListener() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + + RepositoryData repositoryData = new RepositoryData( + RepositoryData.EMPTY_REPO_GEN, + Collections.singletonMap(sourceSnapshot.getName(), sourceSnapshot), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap(), + ShardGenerations.EMPTY, + IndexMetaDataGenerations.EMPTY + ); + + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(0); + listener.onResponse(repositoryData); + return null; + }).when(mockRepository).getRepositoryData(any(ActionListener.class)); + + IndexMetadata mockIndexMetadata = mock(IndexMetadata.class); + when(mockRepository.getSnapshotIndexMetaData(eq(repositoryData), eq(sourceSnapshot), any())).thenReturn(mockIndexMetadata); + + Settings mockSettings = Settings.builder().put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, true).build(); + when(mockIndexMetadata.getSettings()).thenReturn(mockSettings); + + String newGeneration = "new_generation"; + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(5); + listener.onResponse(newGeneration); + return null; + }).when(mockRepository).cloneRemoteStoreIndexShardSnapshot(any(), any(), any(), any(), any(), any(ActionListener.class)); + + ClusterService mockClusterService = mock(ClusterService.class); + DiscoveryNode localNode = new DiscoveryNode("local", buildNewFakeTransportAddress(), Version.CURRENT); + when(mockClusterService.localNode()).thenReturn(localNode); + + SnapshotsService snapshotsService = createSnapshotsService(mockClusterService); + + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, true); + + verify(mockRepository).getRepositoryData(any(ActionListener.class)); + verify(mockRepository).cloneRemoteStoreIndexShardSnapshot( + eq(sourceSnapshot), + eq(target.getSnapshotId()), + eq(repoShardId), + eq(shardStatusBefore.generation()), + any(), + any() + ); + + // Verify that innerUpdateSnapshotState was called with the correct arguments + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass( + SnapshotsService.ShardSnapshotUpdate.class + ); + verify(mockClusterService).submitStateUpdateTask(eq("update snapshot state"), updateCaptor.capture(), any(), any(), any()); + + SnapshotsService.ShardSnapshotUpdate capturedUpdate = updateCaptor.getValue(); + SnapshotsInProgress.ShardSnapshotStatus expectedStatus = new SnapshotsInProgress.ShardSnapshotStatus( + localNode.getId(), + SnapshotsInProgress.ShardState.SUCCESS, + newGeneration + ); + SnapshotsService.ShardSnapshotUpdate expectedUpdate = new SnapshotsService.ShardSnapshotUpdate(target, repoShardId, expectedStatus); + assertEquals(expectedUpdate.hashCode(), capturedUpdate.hashCode()); + } + + /** + * Tests the completion listener functionality in runReadyClone when the operation fails. + * It verifies that: + * 1. The clone operation fails with an exception + * 2. The correct failed ShardSnapshotUpdate is submitted to the cluster state + */ + public void testRunReadyCloneCompletionListenerFailure() throws Exception { + String repoName = "test-repo"; + Snapshot target = snapshot(repoName, "target-snapshot"); + SnapshotId sourceSnapshot = new SnapshotId("source-snapshot", uuid()); + RepositoryShardId repoShardId = new RepositoryShardId(indexId("test-index"), 0); + SnapshotsInProgress.ShardSnapshotStatus shardStatusBefore = initShardStatus(uuid()); + + Repository mockRepository = mock(Repository.class); + + RepositoryData repositoryData = new RepositoryData( + RepositoryData.EMPTY_REPO_GEN, + Collections.singletonMap(sourceSnapshot.getName(), sourceSnapshot), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.emptyMap(), + ShardGenerations.EMPTY, + IndexMetaDataGenerations.EMPTY + ); + + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(0); + listener.onResponse(repositoryData); + return null; + }).when(mockRepository).getRepositoryData(any(ActionListener.class)); + + IndexMetadata mockIndexMetadata = mock(IndexMetadata.class); + when(mockRepository.getSnapshotIndexMetaData(eq(repositoryData), eq(sourceSnapshot), any())).thenReturn(mockIndexMetadata); + + Settings mockSettings = Settings.builder().put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, true).build(); + when(mockIndexMetadata.getSettings()).thenReturn(mockSettings); + + Exception testException = new RuntimeException("Test exception"); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(5); + listener.onFailure(testException); + return null; + }).when(mockRepository).cloneRemoteStoreIndexShardSnapshot(any(), any(), any(), any(), any(), any(ActionListener.class)); + + ClusterService mockClusterService = mock(ClusterService.class); + DiscoveryNode localNode = new DiscoveryNode("local", buildNewFakeTransportAddress(), Version.CURRENT); + when(mockClusterService.localNode()).thenReturn(localNode); + + SnapshotsService snapshotsService = createSnapshotsService(mockClusterService); + + snapshotsService.runReadyClone(target, sourceSnapshot, shardStatusBefore, repoShardId, mockRepository, true); + + verify(mockRepository).getRepositoryData(any(ActionListener.class)); + verify(mockRepository).cloneRemoteStoreIndexShardSnapshot( + eq(sourceSnapshot), + eq(target.getSnapshotId()), + eq(repoShardId), + eq(shardStatusBefore.generation()), + any(), + any() + ); + + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass( + SnapshotsService.ShardSnapshotUpdate.class + ); + verify(mockClusterService).submitStateUpdateTask(eq("update snapshot state"), updateCaptor.capture(), any(), any(), any()); + + SnapshotsService.ShardSnapshotUpdate capturedUpdate = updateCaptor.getValue(); + SnapshotsInProgress.ShardSnapshotStatus expectedStatus = new SnapshotsInProgress.ShardSnapshotStatus( + localNode.getId(), + SnapshotsInProgress.ShardState.FAILED, + "failed to clone shard snapshot", + null + ); + SnapshotsService.ShardSnapshotUpdate expectedUpdate = new SnapshotsService.ShardSnapshotUpdate(target, repoShardId, expectedStatus); + assertEquals(expectedUpdate.hashCode(), capturedUpdate.hashCode()); + } + + /** + * Helper method to create a SnapshotsService instance with a provided ClusterService. + * This method mocks all necessary dependencies for the SnapshotsService. + */ + private SnapshotsService createSnapshotsService(ClusterService mockClusterService) { + ThreadPool mockThreadPool = mock(ThreadPool.class); + when(mockThreadPool.executor(ThreadPool.Names.SNAPSHOT)).thenReturn(OpenSearchExecutors.newDirectExecutorService()); + + RepositoriesService mockRepoService = mock(RepositoriesService.class); + TransportService mockTransportService = mock(TransportService.class); + when(mockTransportService.getThreadPool()).thenReturn(mockThreadPool); + + DiscoveryNode localNode = new DiscoveryNode("local", buildNewFakeTransportAddress(), Version.CURRENT); + when(mockClusterService.localNode()).thenReturn(localNode); + + ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + when(mockClusterService.getClusterSettings()).thenReturn(clusterSettings); + + return new SnapshotsService( + Settings.EMPTY, + mockClusterService, + mock(IndexNameExpressionResolver.class), + mockRepoService, + mockTransportService, + mock(ActionFilters.class), + null, + mock(RemoteStoreSettings.class) + ); + } + + /** + * Helper method to create a SnapshotsService instance with a mocked ClusterService. + * This method is a convenience wrapper around createSnapshotsService(ClusterService). + */ + private SnapshotsService createSnapshotsService() { + ClusterService mockClusterService = mock(ClusterService.class); + return createSnapshotsService(mockClusterService); + } + private static DiscoveryNodes discoveryNodes(String localNodeId) { return DiscoveryNodes.builder().localNodeId(localNodeId).build(); }