Skip to content

Commit 71a447e

Browse files
rajiv-kvrajiv-kv
committed
adding test for remote-cluster-service
1 parent d5ec82d commit 71a447e

File tree

8 files changed

+90
-74
lines changed

8 files changed

+90
-74
lines changed

‎server/src/main/java/org/opensearch/action/admin/cluster/state/TransportClusterStateAction.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@
5151
import org.opensearch.common.unit.TimeValue;
5252
import org.opensearch.core.action.ActionListener;
5353
import org.opensearch.core.common.io.stream.StreamInput;
54-
import org.opensearch.discovery.Discovery;
5554
import org.opensearch.gateway.remote.RemoteClusterStateService;
5655
import org.opensearch.node.NodeClosedException;
5756
import org.opensearch.threadpool.ThreadPool;
@@ -84,7 +83,8 @@ public TransportClusterStateAction(
8483
ThreadPool threadPool,
8584
ActionFilters actionFilters,
8685
IndexNameExpressionResolver indexNameExpressionResolver,
87-
@Nullable RemoteClusterStateService remoteClusterStateService) {
86+
@Nullable RemoteClusterStateService remoteClusterStateService
87+
) {
8888
super(
8989
ClusterStateAction.NAME,
9090
false,

‎server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,6 @@
5050
import org.opensearch.cluster.NotClusterManagerException;
5151
import org.opensearch.cluster.block.ClusterBlockException;
5252
import org.opensearch.cluster.coordination.ClusterStateTermVersion;
53-
import org.opensearch.cluster.coordination.Coordinator;
5453
import org.opensearch.cluster.coordination.FailedToCommitClusterStateException;
5554
import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
5655
import org.opensearch.cluster.metadata.ProcessClusterEventTimeoutException;
@@ -65,7 +64,6 @@
6564
import org.opensearch.core.common.io.stream.StreamInput;
6665
import org.opensearch.core.common.io.stream.Writeable;
6766
import org.opensearch.discovery.ClusterManagerNotDiscoveredException;
68-
import org.opensearch.discovery.Discovery;
6967
import org.opensearch.gateway.remote.ClusterMetadataManifest;
7068
import org.opensearch.gateway.remote.RemoteClusterStateService;
7169
import org.opensearch.gateway.remote.RemoteManifestManager;
@@ -439,16 +437,18 @@ public ClusterState getStateFromLocalNode(ClusterStateTermVersion termVersion) {
439437
termVersion.getTerm(),
440438
termVersion.getVersion()
441439
);
442-
ClusterMetadataManifest clusterMetadataManifest = remoteClusterStateService
443-
.getClusterMetadataManifestByFileName(appliedState.stateUUID(), manifestFile);
440+
ClusterMetadataManifest clusterMetadataManifest = remoteClusterStateService.getClusterMetadataManifestByFileName(
441+
appliedState.stateUUID(),
442+
manifestFile
443+
);
444444
ClusterState clusterStateFromRemote = remoteClusterStateService.getClusterStateForManifest(
445445
appliedState.getClusterName().value(),
446446
clusterMetadataManifest,
447447
appliedState.nodes().getLocalNode().getId(),
448448
true
449449
);
450450

451-
if(clusterStateFromRemote!=null) {
451+
if (clusterStateFromRemote != null) {
452452
logger.trace("Using the remote cluster-state fetched from local node, ClusterStateTermVersion {}", termVersion);
453453
return clusterStateFromRemote;
454454
}

‎server/src/main/java/org/opensearch/action/support/clustermanager/term/TransportGetTermVersionAction.java

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,11 @@
1515
import org.opensearch.cluster.ClusterState;
1616
import org.opensearch.cluster.block.ClusterBlockException;
1717
import org.opensearch.cluster.coordination.ClusterStateTermVersion;
18-
import org.opensearch.cluster.coordination.Coordinator;
1918
import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
2019
import org.opensearch.cluster.service.ClusterService;
2120
import org.opensearch.common.inject.Inject;
2221
import org.opensearch.core.action.ActionListener;
2322
import org.opensearch.core.common.io.stream.StreamInput;
24-
import org.opensearch.discovery.Discovery;
2523
import org.opensearch.threadpool.ThreadPool;
2624
import org.opensearch.transport.TransportService;
2725

@@ -42,7 +40,8 @@ public TransportGetTermVersionAction(
4240
ClusterService clusterService,
4341
ThreadPool threadPool,
4442
ActionFilters actionFilters,
45-
IndexNameExpressionResolver indexNameExpressionResolver) {
43+
IndexNameExpressionResolver indexNameExpressionResolver
44+
) {
4645
super(
4746
GetTermVersionAction.NAME,
4847
false,

‎server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -527,16 +527,15 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
527527
reposToSkip.add(joiningNodeRepoName);
528528
}
529529
}
530-
//if non-or1
531-
//mix of remote-state (enabled and disabled)
532-
//all of them settings
530+
// if non-or1
531+
// mix of remote-state (enabled and disabled)
532+
// all of them settings
533533

534-
// publishes to all nodes
534+
// publishes to all nodes
535535

536-
//commits
537-
538-
//30s still commit []
536+
// commits
539537

538+
// 30s still commit []
540539

541540
if (STRICT.equals(remoteStoreCompatibilityMode)) {
542541
DiscoveryNode existingNode = remoteRoutingTableNode.orElseGet(() -> existingNodes.get(0));

‎server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java

Lines changed: 30 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,11 @@
1313

1414
package org.opensearch.action.support.clustermanager;
1515

16-
import org.mockito.Mock;
17-
import org.mockito.Mockito;
1816
import org.opensearch.OpenSearchException;
1917
import org.opensearch.Version;
2018
import org.opensearch.action.ActionRequestValidationException;
2119
import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
2220
import org.opensearch.action.admin.cluster.settings.TransportClusterUpdateSettingsAction;
23-
import org.opensearch.action.admin.cluster.state.TransportClusterStateAction;
2421
import org.opensearch.action.support.ActionFilters;
2522
import org.opensearch.action.support.PlainActionFuture;
2623
import org.opensearch.action.support.ThreadedActionListener;
@@ -92,10 +89,8 @@
9289
import java.util.concurrent.TimeUnit;
9390
import java.util.concurrent.atomic.AtomicBoolean;
9491

95-
import static java.util.Collections.emptySet;
96-
import static org.mockito.ArgumentMatchers.any;
97-
import static org.mockito.ArgumentMatchers.eq;
98-
import static org.mockito.Mockito.when;
92+
import org.mockito.Mockito;
93+
9994
import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdaterTests.createIndexMetadataWithRemoteStoreSettings;
10095
import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY;
10196
import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY;
@@ -105,6 +100,9 @@
105100
import static org.hamcrest.Matchers.containsString;
106101
import static org.hamcrest.Matchers.equalTo;
107102
import static org.hamcrest.Matchers.instanceOf;
103+
import static org.mockito.ArgumentMatchers.any;
104+
import static org.mockito.ArgumentMatchers.eq;
105+
import static org.mockito.Mockito.when;
108106

109107
public class TransportClusterManagerNodeActionTests extends OpenSearchTestCase {
110108
private static ThreadPool threadPool;
@@ -220,7 +218,8 @@ public void writeTo(StreamOutput out) throws IOException {
220218
}
221219

222220
class Action extends TransportClusterManagerNodeAction<Request, Response> {
223-
private boolean localExecuteSupported = false;
221+
private boolean localExecuteSupported = false;
222+
224223
Action(String actionName, TransportService transportService, ClusterService clusterService, ThreadPool threadPool) {
225224
super(
226225
actionName,
@@ -233,11 +232,17 @@ class Action extends TransportClusterManagerNodeAction<Request, Response> {
233232
);
234233
}
235234

236-
237235
protected boolean localExecuteSupportedByAction() {
238236
return localExecuteSupported;
239237
}
240-
Action(String actionName, TransportService transportService, ClusterService clusterService, ThreadPool threadPool, RemoteClusterStateService clusterStateService) {
238+
239+
Action(
240+
String actionName,
241+
TransportService transportService,
242+
ClusterService clusterService,
243+
ThreadPool threadPool,
244+
RemoteClusterStateService clusterStateService
245+
) {
241246
this(actionName, transportService, clusterService, threadPool);
242247
this.remoteClusterStateService = clusterStateService;
243248
this.localExecuteSupported = true;
@@ -743,34 +748,32 @@ public void testFetchFromRemoteStore() throws InterruptedException, BrokenBarrie
743748
ClusterState state = clusterService.state();
744749
RemoteClusterStateService clusterStateService = Mockito.mock(RemoteClusterStateService.class);
745750
Request request = new Request();
746-
ClusterMetadataManifest manifest = ClusterMetadataManifest.builder().clusterTerm(state.term() + 1).stateVersion(state.version() + 1).build();
751+
ClusterMetadataManifest manifest = ClusterMetadataManifest.builder()
752+
.clusterTerm(state.term() + 1)
753+
.stateVersion(state.version() + 1)
754+
.build();
747755
when(clusterStateService.getClusterMetadataManifestByFileName(eq(state.stateUUID()), any())).thenReturn(manifest);
748756

749757
when(clusterStateService.getClusterStateForManifest(state.getClusterName().value(), manifest, localNode.getId(), true)).thenReturn(
750-
buildClusterState(state, state.term() + 1, state.version() + 1));
758+
buildClusterState(state, state.term() + 1, state.version() + 1)
759+
);
751760

752761
PlainActionFuture<Response> listener = new PlainActionFuture<>();
753762
Action action = new Action("internal:testAction", transportService, clusterService, threadPool, clusterStateService);
754763
action.execute(request, listener);
755764

756-
757-
CapturingTransport.CapturedRequest capturedRequest = transport.capturedRequests()[0];
758-
// mismatch term and version
759-
GetTermVersionResponse termResp = new GetTermVersionResponse(
760-
new ClusterStateTermVersion(
761-
state.getClusterName(),
762-
state.metadata().clusterUUID(),
763-
state.term() + 1,
764-
state.version() + 1
765-
)
766-
);
767-
transport.handleResponse(capturedRequest.requestId, termResp);
768-
//no more transport calls
769-
assertThat(transport.capturedRequests().length, equalTo(1));
765+
CapturingTransport.CapturedRequest capturedRequest = transport.capturedRequests()[0];
766+
// mismatch term and version
767+
GetTermVersionResponse termResp = new GetTermVersionResponse(
768+
new ClusterStateTermVersion(state.getClusterName(), state.metadata().clusterUUID(), state.term() + 1, state.version() + 1)
769+
);
770+
transport.handleResponse(capturedRequest.requestId, termResp);
771+
// no more transport calls
772+
assertThat(transport.capturedRequests().length, equalTo(1));
770773
assertTrue(listener.isDone());
771774
}
772775

773-
private ClusterState buildClusterState(ClusterState state , long term, long version) {
776+
private ClusterState buildClusterState(ClusterState state, long term, long version) {
774777
CoordinationMetadata.Builder coordMetadataBuilder = CoordinationMetadata.builder().term(term);
775778
Metadata newMetadata = Metadata.builder().coordinationMetadata(coordMetadataBuilder.build()).build();
776779
return ClusterState.builder(state).version(version).metadata(newMetadata).build();

‎server/src/test/java/org/opensearch/action/support/clustermanager/term/ClusterTermVersionIT.java

Lines changed: 30 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,10 @@
88

99
package org.opensearch.action.support.clustermanager.term;
1010

11-
import java.util.*;
12-
import java.util.concurrent.ExecutionException;
13-
import java.util.concurrent.TimeUnit;
1411
import org.opensearch.action.admin.cluster.state.ClusterStateAction;
1512
import org.opensearch.action.admin.cluster.state.ClusterStateRequest;
1613
import org.opensearch.action.admin.cluster.state.ClusterStateResponse;
1714
import org.opensearch.action.admin.indices.create.CreateIndexResponse;
18-
import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest;
19-
import org.opensearch.action.support.ActiveShardCount;
2015
import org.opensearch.client.Client;
2116
import org.opensearch.cluster.ClusterChangedEvent;
2217
import org.opensearch.cluster.ClusterName;
@@ -33,6 +28,10 @@
3328
import org.opensearch.test.transport.MockTransportService;
3429
import org.opensearch.transport.TransportService;
3530

31+
import java.util.Collection;
32+
import java.util.List;
33+
import java.util.Map;
34+
import java.util.concurrent.TimeUnit;
3635
import java.util.concurrent.atomic.AtomicBoolean;
3736
import java.util.concurrent.atomic.AtomicInteger;
3837

@@ -124,7 +123,6 @@ public void testDatanodeWithSlowClusterApplierFallbackToPublish() throws Excepti
124123

125124
String master = internalCluster().getClusterManagerName();
126125

127-
128126
AtomicBoolean processState = new AtomicBoolean();
129127
ClusterService cmClsService = internalCluster().getInstance(ClusterService.class, datas.get(0));
130128
cmClsService.addStateApplier(new ClusterStateApplier() {
@@ -145,38 +143,47 @@ public void applyClusterState(ClusterChangedEvent event) {
145143

146144
ensureGreen();
147145

148-
GetTermVersionResponse respBeforeUpdate = internalCluster().getInstance(Client.class, master).execute(GetTermVersionAction.INSTANCE, new GetTermVersionRequest()).get();
146+
GetTermVersionResponse respBeforeUpdate = internalCluster().getInstance(Client.class, master)
147+
.execute(GetTermVersionAction.INSTANCE, new GetTermVersionRequest())
148+
.get();
149149

150150
processState.set(true);
151151
String index = "index_1";
152-
ActionFuture<CreateIndexResponse> startCreateIndex1 = prepareCreate(index)
153-
.setSettings(Settings.builder()
152+
ActionFuture<CreateIndexResponse> startCreateIndex1 = prepareCreate(index).setSettings(
153+
Settings.builder()
154154
.put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
155155
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
156156
.put(MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING.getKey(), Long.MAX_VALUE)
157-
.build())
158-
.execute();
159-
157+
.build()
158+
).execute();
160159

161-
ActionFuture<CreateIndexResponse> startCreateIndex2 = prepareCreate("index_2").setSettings(Settings.builder()
162-
.put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
163-
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
164-
.put(MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING.getKey(), Long.MAX_VALUE)
165-
.build()).execute();
160+
ActionFuture<CreateIndexResponse> startCreateIndex2 = prepareCreate("index_2").setSettings(
161+
Settings.builder()
162+
.put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
163+
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
164+
.put(MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING.getKey(), Long.MAX_VALUE)
165+
.build()
166+
).execute();
166167

167-
//wait for cluster-manager to publish new state
168+
// wait for cluster-manager to publish new state
168169
waitUntil(() -> {
169170
try {
170-
//node is yet to ack commit to cluster-manager , only the state-update corresponding to index_1 should have been published
171-
GetTermVersionResponse respAfterUpdate = internalCluster().getInstance(Client.class, master).execute(GetTermVersionAction.INSTANCE, new GetTermVersionRequest()).get();
172-
logger.info("data has latest , {} , {}", respAfterUpdate.getClusterStateTermVersion().getTerm(), respAfterUpdate.getClusterStateTermVersion().getVersion());
173-
return respBeforeUpdate.getClusterStateTermVersion().getVersion() + 1 == respAfterUpdate.getClusterStateTermVersion().getVersion();
171+
// node is yet to ack commit to cluster-manager , only the state-update corresponding to index_1 should have been published
172+
GetTermVersionResponse respAfterUpdate = internalCluster().getInstance(Client.class, master)
173+
.execute(GetTermVersionAction.INSTANCE, new GetTermVersionRequest())
174+
.get();
175+
logger.info(
176+
"data has latest , {} , {}",
177+
respAfterUpdate.getClusterStateTermVersion().getTerm(),
178+
respAfterUpdate.getClusterStateTermVersion().getVersion()
179+
);
180+
return respBeforeUpdate.getClusterStateTermVersion().getVersion() + 1 == respAfterUpdate.getClusterStateTermVersion()
181+
.getVersion();
174182
} catch (Exception e) {
175183
throw new RuntimeException(e);
176184
}
177185
}, 100, TimeUnit.SECONDS);
178186

179-
180187
addCallCountInterceptor(master, callCounters);
181188
ClusterStateResponse stateResponseD = internalCluster().getInstance(Client.class, datas.get(0))
182189
.admin()

‎server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,6 @@
4646
import org.opensearch.common.settings.ClusterSettings;
4747
import org.opensearch.common.settings.Settings;
4848
import org.opensearch.common.util.FeatureFlags;
49-
import org.opensearch.core.ParseField;
5049
import org.opensearch.core.action.ActionListener;
5150
import org.opensearch.core.common.bytes.BytesArray;
5251
import org.opensearch.core.common.bytes.BytesReference;
@@ -2335,8 +2334,9 @@ public void testReadLatestMetadataManifestSuccess() throws IOException {
23352334
}
23362335

23372336
public void testReadGlobalMetadata() throws IOException {
2338-
// when(blobStoreRepository.getNamedXContentRegistry()).thenReturn(new NamedXContentRegistry(
2339-
// List.of(new NamedXContentRegistry.Entry(Metadata.Custom.class, new ParseField(IndexGraveyard.TYPE), IndexGraveyard::fromXContent))));
2337+
// when(blobStoreRepository.getNamedXContentRegistry()).thenReturn(new NamedXContentRegistry(
2338+
// List.of(new NamedXContentRegistry.Entry(Metadata.Custom.class, new ParseField(IndexGraveyard.TYPE),
2339+
// IndexGraveyard::fromXContent))));
23402340
final ClusterState clusterState = generateClusterStateWithGlobalMetadata().nodes(nodesWithLocalNodeClusterManager()).build();
23412341
remoteClusterStateService.start();
23422342

@@ -2351,15 +2351,21 @@ public void testReadGlobalMetadata() throws IOException {
23512351
.coordinationMetadata(new ClusterMetadataManifest.UploadedMetadataAttribute(COORDINATION_METADATA, "mock-coordination-file"))
23522352
.settingMetadata(new ClusterMetadataManifest.UploadedMetadataAttribute(SETTING_METADATA, "mock-setting-file"))
23532353
.templatesMetadata(new ClusterMetadataManifest.UploadedMetadataAttribute(TEMPLATES_METADATA, "mock-templates-file"))
2354-
.put(IndexGraveyard.TYPE, new ClusterMetadataManifest.UploadedMetadataAttribute(IndexGraveyard.TYPE, "mock-custom-" +IndexGraveyard.TYPE+ "-file"))
2354+
.put(
2355+
IndexGraveyard.TYPE,
2356+
new ClusterMetadataManifest.UploadedMetadataAttribute(IndexGraveyard.TYPE, "mock-custom-" + IndexGraveyard.TYPE + "-file")
2357+
)
23552358
.nodeId("nodeA")
23562359
.opensearchVersion(VersionUtils.randomOpenSearchVersion(random()))
23572360
.previousClusterUUID("prev-cluster-uuid")
23582361
.routingTableVersion(1)
23592362
.indicesRouting(List.of())
23602363
.build();
23612364

2362-
Metadata expectedMetadata = Metadata.builder().clusterUUID("cluster-uuid").persistentSettings(Settings.builder().put("readonly", true).build()).build();
2365+
Metadata expectedMetadata = Metadata.builder()
2366+
.clusterUUID("cluster-uuid")
2367+
.persistentSettings(Settings.builder().put("readonly", true).build())
2368+
.build();
23632369
mockBlobContainerForGlobalMetadata(mockBlobStoreObjects(), expectedManifest, expectedMetadata);
23642370

23652371
ClusterState newClusterState = remoteClusterStateService.getLatestClusterState(

‎server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2413,7 +2413,8 @@ public void onFailure(final Exception e) {
24132413
threadPool,
24142414
actionFilters,
24152415
indexNameExpressionResolver,
2416-
null)
2416+
null
2417+
)
24172418
);
24182419
actions.put(
24192420
IndicesShardStoresAction.INSTANCE,
@@ -2454,7 +2455,8 @@ public void onFailure(final Exception e) {
24542455
clusterService,
24552456
threadPool,
24562457
actionFilters,
2457-
indexNameExpressionResolver)
2458+
indexNameExpressionResolver
2459+
)
24582460
);
24592461

24602462
DynamicActionRegistry dynamicActionRegistry = new DynamicActionRegistry();

0 commit comments

Comments
 (0)