diff --git a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterStateCustoms.java b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterStateCustoms.java index f384908bc6b65..affbc7ba66cb8 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterStateCustoms.java +++ b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteClusterStateCustoms.java @@ -34,12 +34,12 @@ */ public class RemoteClusterStateCustoms extends AbstractRemoteWritableBlobEntity { public static final String CLUSTER_STATE_CUSTOM = "cluster-state-custom"; + public final ChecksumWritableBlobStoreFormat clusterStateCustomsFormat; private long stateVersion; private final String customType; private ClusterState.Custom custom; private final NamedWriteableRegistry namedWriteableRegistry; - private final ChecksumWritableBlobStoreFormat clusterStateCustomsFormat; public RemoteClusterStateCustoms( final ClusterState.Custom custom, diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManagerTests.java index 41e1546ead164..fe9ed57fa77b8 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManagerTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManagerTests.java @@ -17,9 +17,10 @@ import org.opensearch.cluster.DiffableUtils; import org.opensearch.cluster.block.ClusterBlocks; import org.opensearch.cluster.node.DiscoveryNodes; -import org.opensearch.common.CheckedRunnable; +import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.TestCapturingListener; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.io.stream.NamedWriteableRegistry; import org.opensearch.core.common.io.stream.StreamInput; @@ -28,6 +29,7 @@ import org.opensearch.core.compress.NoneCompressor; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.gateway.remote.model.RemoteClusterBlocks; +import org.opensearch.gateway.remote.model.RemoteClusterStateCustoms; import org.opensearch.gateway.remote.model.RemoteDiscoveryNodes; import org.opensearch.gateway.remote.model.RemoteReadResult; import org.opensearch.index.translog.transfer.BlobStoreTransferService; @@ -36,46 +38,63 @@ import org.opensearch.threadpool.TestThreadPool; import org.opensearch.threadpool.ThreadPool; import org.junit.After; -import org.junit.Assert; import org.junit.Before; import java.io.IOException; +import java.io.InputStream; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.atomic.AtomicReference; -import static java.util.Collections.emptyList; +import static org.opensearch.common.blobstore.stream.write.WritePriority.URGENT; +import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.CLUSTER_STATE_ATTRIBUTE; +import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION; import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.DISCOVERY_NODES; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.CLUSTER_STATE_EPHEMERAL_PATH_TOKEN; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.CUSTOM_DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.PATH_DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.encodeString; import static org.opensearch.gateway.remote.model.RemoteClusterBlocks.CLUSTER_BLOCKS; import static org.opensearch.gateway.remote.model.RemoteClusterBlocks.CLUSTER_BLOCKS_FORMAT; import static org.opensearch.gateway.remote.model.RemoteClusterBlocksTests.randomClusterBlocks; +import static org.opensearch.gateway.remote.model.RemoteClusterStateCustoms.CLUSTER_STATE_CUSTOM; +import static org.opensearch.gateway.remote.model.RemoteClusterStateCustomsTests.getClusterStateCustom; import static org.opensearch.gateway.remote.model.RemoteDiscoveryNodes.DISCOVERY_NODES_FORMAT; import static org.opensearch.gateway.remote.model.RemoteDiscoveryNodesTests.getDiscoveryNodes; +import static org.opensearch.index.remote.RemoteStoreUtils.invertLong; import static org.hamcrest.Matchers.is; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyIterable; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; public class RemoteClusterStateAttributesManagerTests extends OpenSearchTestCase { private RemoteClusterStateAttributesManager remoteClusterStateAttributesManager; private BlobStoreTransferService blobStoreTransferService; - private BlobStoreRepository blobStoreRepository; private Compressor compressor; - private ThreadPool threadPool = new TestThreadPool(RemoteClusterStateAttributesManagerTests.class.getName()); + private final ThreadPool threadPool = new TestThreadPool(RemoteClusterStateAttributesManagerTests.class.getName()); + private final long VERSION = 7331L; + private NamedWriteableRegistry namedWriteableRegistry; + private final String CLUSTER_NAME = "test-cluster"; + private final String CLUSTER_UUID = "test-cluster-uuid"; @Before public void setup() throws Exception { ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); - NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(emptyList()); - blobStoreRepository = mock(BlobStoreRepository.class); + namedWriteableRegistry = writableRegistry(); + BlobStoreRepository blobStoreRepository = mock(BlobStoreRepository.class); + when(blobStoreRepository.basePath()).thenReturn(new BlobPath()); blobStoreTransferService = mock(BlobStoreTransferService.class); compressor = new NoneCompressor(); remoteClusterStateAttributesManager = new RemoteClusterStateAttributesManager( - "test-cluster", + CLUSTER_NAME, blobStoreRepository, blobStoreTransferService, writableRegistry(), @@ -89,7 +108,40 @@ public void tearDown() throws Exception { threadPool.shutdown(); } - public void testGetAsyncMetadataReadAction_DiscoveryNodes() throws IOException { + public void testGetAsyncMetadataWriteAction_DiscoveryNodes() throws IOException, InterruptedException { + DiscoveryNodes discoveryNodes = getDiscoveryNodes(); + RemoteDiscoveryNodes remoteDiscoveryNodes = new RemoteDiscoveryNodes(discoveryNodes, VERSION, CLUSTER_UUID, compressor); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + final CountDownLatch latch = new CountDownLatch(1); + final TestCapturingListener listener = new TestCapturingListener<>(); + remoteClusterStateAttributesManager.getAsyncMetadataWriteAction( + DISCOVERY_NODES, + remoteDiscoveryNodes, + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(DISCOVERY_NODES, listener.getResult().getComponent()); + String uploadedFileName = listener.getResult().getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(RemoteClusterStateUtils.encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(CLUSTER_STATE_EPHEMERAL_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(DISCOVERY_NODES, splitFileName[0]); + assertEquals(invertLong(VERSION), splitFileName[1]); + assertEquals(CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } + + public void testGetAsyncMetadataReadAction_DiscoveryNodes() throws IOException, InterruptedException { DiscoveryNodes discoveryNodes = getDiscoveryNodes(); String fileName = randomAlphaOfLength(10); when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( @@ -97,29 +149,57 @@ public void testGetAsyncMetadataReadAction_DiscoveryNodes() throws IOException { ); RemoteDiscoveryNodes remoteObjForDownload = new RemoteDiscoveryNodes(fileName, "cluster-uuid", compressor); CountDownLatch latch = new CountDownLatch(1); - AtomicReference readDiscoveryNodes = new AtomicReference<>(); - LatchedActionListener assertingListener = new LatchedActionListener<>( - ActionListener.wrap(response -> readDiscoveryNodes.set((DiscoveryNodes) response.getObj()), Assert::assertNull), - latch - ); - CheckedRunnable runnable = remoteClusterStateAttributesManager.getAsyncMetadataReadAction( + TestCapturingListener listener = new TestCapturingListener<>(); + remoteClusterStateAttributesManager.getAsyncMetadataReadAction( DISCOVERY_NODES, remoteObjForDownload, - assertingListener - ); + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(CLUSTER_STATE_ATTRIBUTE, listener.getResult().getComponent()); + assertEquals(DISCOVERY_NODES, listener.getResult().getComponentName()); + DiscoveryNodes readDiscoveryNodes = (DiscoveryNodes) listener.getResult().getObj(); + assertEquals(discoveryNodes.getSize(), readDiscoveryNodes.getSize()); + discoveryNodes.getNodes().forEach((nodeId, node) -> assertEquals(readDiscoveryNodes.get(nodeId), node)); + assertEquals(discoveryNodes.getClusterManagerNodeId(), readDiscoveryNodes.getClusterManagerNodeId()); + } - try { - runnable.run(); - latch.await(); - assertEquals(discoveryNodes.getSize(), readDiscoveryNodes.get().getSize()); - discoveryNodes.getNodes().forEach((nodeId, node) -> assertEquals(readDiscoveryNodes.get().get(nodeId), node)); - assertEquals(discoveryNodes.getClusterManagerNodeId(), readDiscoveryNodes.get().getClusterManagerNodeId()); - } catch (Exception e) { - throw new RuntimeException(e); - } + public void testGetAsyncMetadataWriteAction_ClusterBlocks() throws IOException, InterruptedException { + ClusterBlocks clusterBlocks = randomClusterBlocks(); + RemoteClusterBlocks remoteClusterBlocks = new RemoteClusterBlocks(clusterBlocks, VERSION, CLUSTER_UUID, compressor); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + final CountDownLatch latch = new CountDownLatch(1); + final TestCapturingListener listener = new TestCapturingListener<>(); + remoteClusterStateAttributesManager.getAsyncMetadataWriteAction( + CLUSTER_BLOCKS, + remoteClusterBlocks, + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(CLUSTER_BLOCKS, listener.getResult().getComponent()); + String uploadedFileName = listener.getResult().getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(CLUSTER_STATE_EPHEMERAL_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(CLUSTER_BLOCKS, splitFileName[0]); + assertEquals(invertLong(VERSION), splitFileName[1]); + assertEquals(CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); } - public void testGetAsyncMetadataReadAction_ClusterBlocks() throws IOException { + public void testGetAsyncMetadataReadAction_ClusterBlocks() throws IOException, InterruptedException { ClusterBlocks clusterBlocks = randomClusterBlocks(); String fileName = randomAlphaOfLength(10); when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( @@ -127,29 +207,133 @@ public void testGetAsyncMetadataReadAction_ClusterBlocks() throws IOException { ); RemoteClusterBlocks remoteClusterBlocks = new RemoteClusterBlocks(fileName, "cluster-uuid", compressor); CountDownLatch latch = new CountDownLatch(1); - AtomicReference readClusterBlocks = new AtomicReference<>(); - LatchedActionListener assertingListener = new LatchedActionListener<>( - ActionListener.wrap(response -> readClusterBlocks.set((ClusterBlocks) response.getObj()), Assert::assertNull), - latch - ); + TestCapturingListener listener = new TestCapturingListener<>(); - CheckedRunnable runnable = remoteClusterStateAttributesManager.getAsyncMetadataReadAction( + remoteClusterStateAttributesManager.getAsyncMetadataReadAction( CLUSTER_BLOCKS, remoteClusterBlocks, - assertingListener + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(CLUSTER_STATE_ATTRIBUTE, listener.getResult().getComponent()); + assertEquals(CLUSTER_BLOCKS, listener.getResult().getComponentName()); + ClusterBlocks readClusterBlocks = (ClusterBlocks) listener.getResult().getObj(); + assertEquals(clusterBlocks.global(), readClusterBlocks.global()); + assertEquals(clusterBlocks.indices().keySet(), readClusterBlocks.indices().keySet()); + for (String index : clusterBlocks.indices().keySet()) { + assertEquals(clusterBlocks.indices().get(index), readClusterBlocks.indices().get(index)); + } + } + + public void testGetAsyncMetadataWriteAction_Custom() throws IOException, InterruptedException { + Custom custom = getClusterStateCustom(); + RemoteClusterStateCustoms remoteClusterStateCustoms = new RemoteClusterStateCustoms( + custom, + custom.getWriteableName(), + VERSION, + CLUSTER_UUID, + compressor, + namedWriteableRegistry ); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onResponse(null); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + final TestCapturingListener listener = new TestCapturingListener<>(); + final CountDownLatch latch = new CountDownLatch(1); + remoteClusterStateAttributesManager.getAsyncMetadataWriteAction( + CLUSTER_STATE_CUSTOM, + remoteClusterStateCustoms, + new LatchedActionListener<>(listener, latch) + ).run(); + latch.await(); + assertNull(listener.getFailure()); + assertNotNull(listener.getResult()); + assertEquals(String.join(CUSTOM_DELIMITER, CLUSTER_STATE_CUSTOM, custom.getWriteableName()), listener.getResult().getComponent()); + String uploadedFileName = listener.getResult().getUploadedFilename(); + String[] pathTokens = uploadedFileName.split(PATH_DELIMITER); + assertEquals(5, pathTokens.length); + assertEquals(encodeString(CLUSTER_NAME), pathTokens[0]); + assertEquals(CLUSTER_STATE_PATH_TOKEN, pathTokens[1]); + assertEquals(CLUSTER_UUID, pathTokens[2]); + assertEquals(CLUSTER_STATE_EPHEMERAL_PATH_TOKEN, pathTokens[3]); + String[] splitFileName = pathTokens[4].split(DELIMITER); + assertEquals(4, splitFileName.length); + assertEquals(String.join(CUSTOM_DELIMITER, CLUSTER_STATE_CUSTOM, custom.getWriteableName()), splitFileName[0]); + assertEquals(invertLong(VERSION), splitFileName[1]); + assertEquals(CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION, Integer.parseInt(splitFileName[3])); + } - try { - runnable.run(); - latch.await(); - assertEquals(clusterBlocks.global(), readClusterBlocks.get().global()); - assertEquals(clusterBlocks.indices().keySet(), readClusterBlocks.get().indices().keySet()); - for (String index : clusterBlocks.indices().keySet()) { - assertEquals(clusterBlocks.indices().get(index), readClusterBlocks.get().indices().get(index)); - } - } catch (Exception e) { - throw new RuntimeException(e); - } + public void testGetAsyncMetadataReadAction_Custom() throws IOException, InterruptedException { + Custom custom = getClusterStateCustom(); + String fileName = randomAlphaOfLength(10); + RemoteClusterStateCustoms remoteClusterStateCustoms = new RemoteClusterStateCustoms( + fileName, + custom.getWriteableName(), + CLUSTER_UUID, + compressor, + namedWriteableRegistry + ); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenReturn( + remoteClusterStateCustoms.clusterStateCustomsFormat.serialize(custom, fileName, compressor).streamInput() + ); + TestCapturingListener capturingListener = new TestCapturingListener<>(); + final CountDownLatch latch = new CountDownLatch(1); + remoteClusterStateAttributesManager.getAsyncMetadataReadAction( + CLUSTER_STATE_CUSTOM, + remoteClusterStateCustoms, + new LatchedActionListener<>(capturingListener, latch) + ).run(); + latch.await(); + assertNull(capturingListener.getFailure()); + assertNotNull(capturingListener.getResult()); + assertEquals(custom, capturingListener.getResult().getObj()); + assertEquals(CLUSTER_STATE_ATTRIBUTE, capturingListener.getResult().getComponent()); + assertEquals(CLUSTER_STATE_CUSTOM, capturingListener.getResult().getComponentName()); + } + + public void testGetAsyncMetadataWriteAction_Exception() throws IOException, InterruptedException { + DiscoveryNodes discoveryNodes = getDiscoveryNodes(); + RemoteDiscoveryNodes remoteDiscoveryNodes = new RemoteDiscoveryNodes(discoveryNodes, VERSION, CLUSTER_UUID, compressor); + + IOException ioException = new IOException("mock test exception"); + doAnswer(invocationOnMock -> { + invocationOnMock.getArgument(4, ActionListener.class).onFailure(ioException); + return null; + }).when(blobStoreTransferService) + .uploadBlob(any(InputStream.class), anyIterable(), anyString(), eq(URGENT), any(ActionListener.class)); + + TestCapturingListener capturingListener = new TestCapturingListener<>(); + final CountDownLatch latch = new CountDownLatch(1); + remoteClusterStateAttributesManager.getAsyncMetadataWriteAction( + DISCOVERY_NODES, + remoteDiscoveryNodes, + new LatchedActionListener<>(capturingListener, latch) + ).run(); + latch.await(); + assertNull(capturingListener.getResult()); + assertTrue(capturingListener.getFailure() instanceof RemoteStateTransferException); + assertEquals(ioException, capturingListener.getFailure().getCause()); + } + + public void testGetAsyncMetadataReadAction_Exception() throws IOException, InterruptedException { + String fileName = randomAlphaOfLength(10); + RemoteDiscoveryNodes remoteDiscoveryNodes = new RemoteDiscoveryNodes(fileName, CLUSTER_UUID, compressor); + Exception ioException = new IOException("mock test exception"); + when(blobStoreTransferService.downloadBlob(anyIterable(), anyString())).thenThrow(ioException); + CountDownLatch latch = new CountDownLatch(1); + TestCapturingListener capturingListener = new TestCapturingListener<>(); + remoteClusterStateAttributesManager.getAsyncMetadataReadAction( + DISCOVERY_NODES, + remoteDiscoveryNodes, + new LatchedActionListener<>(capturingListener, latch) + ).run(); + latch.await(); + assertNull(capturingListener.getResult()); + assertEquals(ioException, capturingListener.getFailure()); } public void testGetUpdatedCustoms() {