Skip to content

Commit 83ef0cf

Browse files
JeremyDahlgrenJeremyDahlgren
authored
Consolidate settings consumers for RemoteClusterService (#135722)
Moves the REMOTE_CLUSTER_SKIP_UNAVAILABLE setting into the main AffixSettings list in ClusterSettingsLinkedProjectConfigService, eliminating the need for an extra skip_unavailable specific update method on the LinkedProjectConfigListener interface, simplifying the design. Note that it is not possible to just set REMOTE_CLUSTER_SKIP_UNAVAILABLE in a settings update, the RemoteConnectionEnabled settings validator checks that the required settings for an enabled connection are also present. Resolves: ES-12860
1 parent e3a2a38 commit 83ef0cf

File tree

6 files changed

+75
-107
lines changed

6 files changed

+75
-107
lines changed

‎server/src/main/java/org/elasticsearch/transport/AbstractLinkedProjectConfigService.java‎

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,6 @@
99

1010
package org.elasticsearch.transport;
1111

12-
import org.elasticsearch.cluster.metadata.ProjectId;
13-
1412
import java.util.List;
1513
import java.util.concurrent.CopyOnWriteArrayList;
1614

@@ -29,15 +27,4 @@ public void register(LinkedProjectConfigListener listener) {
2927
protected void handleUpdate(LinkedProjectConfig config) {
3028
listeners.forEach(listener -> listener.updateLinkedProject(config));
3129
}
32-
33-
protected void handleSkipUnavailableChanged(
34-
ProjectId originProjectId,
35-
ProjectId linkedProjectId,
36-
String linkedProjectAlias,
37-
boolean skipUnavailable
38-
) {
39-
listeners.forEach(
40-
listener -> listener.skipUnavailableChanged(originProjectId, linkedProjectId, linkedProjectAlias, skipUnavailable)
41-
);
42-
}
4330
}

‎server/src/main/java/org/elasticsearch/transport/ClusterSettingsLinkedProjectConfigService.java‎

Lines changed: 1 addition & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ public ClusterSettingsLinkedProjectConfigService(
4848
RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS,
4949
RemoteClusterSettings.REMOTE_CLUSTER_PING_SCHEDULE,
5050
RemoteClusterSettings.REMOTE_CONNECTION_MODE,
51+
RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
5152
RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTERS_PROXY,
5253
RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS,
5354
RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_NODE_CONNECTIONS,
@@ -56,11 +57,6 @@ public ClusterSettingsLinkedProjectConfigService(
5657
RemoteClusterSettings.ProxyConnectionStrategySettings.SERVER_NAME
5758
);
5859
clusterSettings.addAffixGroupUpdateConsumer(remoteClusterSettings, this::settingsChangedCallback);
59-
clusterSettings.addAffixUpdateConsumer(
60-
RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
61-
this::skipUnavailableChangedCallback,
62-
(alias, value) -> {}
63-
);
6460
}
6561
}
6662

@@ -79,9 +75,4 @@ private void settingsChangedCallback(String clusterAlias, Settings newSettings)
7975
final var config = RemoteClusterSettings.toConfig(projectResolver.getProjectId(), ProjectId.DEFAULT, clusterAlias, mergedSettings);
8076
handleUpdate(config);
8177
}
82-
83-
@FixForMultiProject(description = "Refactor to add the linked project ID associated with the alias.")
84-
private void skipUnavailableChangedCallback(String clusterAlias, Boolean skipUnavailable) {
85-
handleSkipUnavailableChanged(projectResolver.getProjectId(), ProjectId.DEFAULT, clusterAlias, skipUnavailable);
86-
}
8778
}

‎server/src/main/java/org/elasticsearch/transport/LinkedProjectConfigService.java‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,6 @@
99

1010
package org.elasticsearch.transport;
1111

12-
import org.elasticsearch.cluster.metadata.ProjectId;
13-
1412
import java.util.Collection;
1513

1614
/**
@@ -29,22 +27,6 @@ interface LinkedProjectConfigListener {
2927
* @param config The updated {@link LinkedProjectConfig}.
3028
*/
3129
void updateLinkedProject(LinkedProjectConfig config);
32-
33-
/**
34-
* Called when the boolean skip_unavailable setting has changed for a linked project configuration.
35-
* Note that skip_unavailable may not be supported in all contexts where linked projects are used.
36-
*
37-
* @param originProjectId The {@link ProjectId} of the owning project that has the linked project configuration.
38-
* @param linkedProjectId The {@link ProjectId} of the linked project.
39-
* @param linkedProjectAlias The alias used for the linked project.
40-
* @param skipUnavailable The new value of the skip_unavailable setting.
41-
*/
42-
default void skipUnavailableChanged(
43-
ProjectId originProjectId,
44-
ProjectId linkedProjectId,
45-
String linkedProjectAlias,
46-
boolean skipUnavailable
47-
) {}
4830
}
4931

5032
/**

‎server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java‎

Lines changed: 5 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -293,21 +293,6 @@ public RemoteClusterConnection getRemoteClusterConnection(String cluster) {
293293
return connection;
294294
}
295295

296-
@Override
297-
public void skipUnavailableChanged(
298-
ProjectId originProjectId,
299-
ProjectId linkedProjectId,
300-
String linkedProjectAlias,
301-
boolean skipUnavailable
302-
) {
303-
assert crossProjectEnabled == false
304-
: "Cannot configure setting [" + RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getKey() + "] in CPS environments.";
305-
final var remote = getConnectionsMapForProject(originProjectId).get(linkedProjectAlias);
306-
if (remote != null) {
307-
remote.setSkipUnavailable(skipUnavailable);
308-
}
309-
}
310-
311296
@Override
312297
public void updateLinkedProject(LinkedProjectConfig config) {
313298
final var projectId = config.originProjectId();
@@ -385,6 +370,9 @@ public synchronized void updateRemoteCluster(
385370
remote = new RemoteClusterConnection(config, transportService, remoteClusterCredentialsManager, crossProjectEnabled);
386371
connectionMap.put(clusterAlias, remote);
387372
remote.ensureConnected(listener.map(ignored -> RemoteClusterConnectionStatus.RECONNECTED));
373+
} else if (remote.isSkipUnavailable() != config.skipUnavailable()) {
374+
remote.setSkipUnavailable(config.skipUnavailable());
375+
listener.onResponse(RemoteClusterConnectionStatus.UPDATED);
388376
} else {
389377
// No changes to connection configuration.
390378
listener.onResponse(RemoteClusterConnectionStatus.UNCHANGED);
@@ -395,7 +383,8 @@ public enum RemoteClusterConnectionStatus {
395383
CONNECTED,
396384
DISCONNECTED,
397385
RECONNECTED,
398-
UNCHANGED
386+
UNCHANGED,
387+
UPDATED
399388
}
400389

401390
/**

‎server/src/test/java/org/elasticsearch/transport/ClusterSettingsLinkedProjectConfigServiceTests.java‎

Lines changed: 9 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@
99

1010
package org.elasticsearch.transport;
1111

12-
import org.elasticsearch.cluster.metadata.ProjectId;
1312
import org.elasticsearch.cluster.project.DefaultProjectResolver;
1413
import org.elasticsearch.common.settings.ClusterSettings;
1514
import org.elasticsearch.common.settings.Settings;
@@ -25,35 +24,16 @@
2524

2625
public class ClusterSettingsLinkedProjectConfigServiceTests extends ESTestCase {
2726

28-
private record SkipUnavailableUpdate(
29-
ProjectId originProjectId,
30-
ProjectId linkedProjectId,
31-
String linkedProjectAlias,
32-
boolean skipUnavailable
33-
) {}
34-
3527
private static class StubLinkedProjectConfigListener implements LinkedProjectConfigListener {
3628
LinkedProjectConfig updatedConfig;
37-
SkipUnavailableUpdate skipUnavailableUpdate;
3829

3930
@Override
4031
public void updateLinkedProject(LinkedProjectConfig config) {
4132
updatedConfig = config;
4233
}
4334

44-
@Override
45-
public void skipUnavailableChanged(
46-
ProjectId originProjectId,
47-
ProjectId linkedProjectId,
48-
String linkedProjectAlias,
49-
boolean skipUnavailable
50-
) {
51-
skipUnavailableUpdate = new SkipUnavailableUpdate(originProjectId, linkedProjectId, linkedProjectAlias, skipUnavailable);
52-
}
53-
5435
void reset() {
5536
updatedConfig = null;
56-
skipUnavailableUpdate = null;
5737
}
5838
}
5939

@@ -66,17 +46,21 @@ public void testListenersReceiveUpdates() {
6646
final var alias = randomAlphaOfLength(10);
6747

6848
final var initialProxyAddress = "localhost:9400";
49+
final var initialSkipUnavailable = randomBoolean();
6950
final var initialSettings = Settings.builder()
7051
.put("cluster.remote." + alias + ".mode", "proxy")
7152
.put("cluster.remote." + alias + ".proxy_address", initialProxyAddress)
53+
.put("cluster.remote." + alias + ".skip_unavailable", initialSkipUnavailable)
7254
.build();
7355
final var clusterSettings = ClusterSettings.createBuiltInClusterSettings(initialSettings);
7456
final var service = new ClusterSettingsLinkedProjectConfigService(
7557
initialSettings,
7658
clusterSettings,
7759
DefaultProjectResolver.INSTANCE
7860
);
79-
final var config = new ProxyLinkedProjectConfigBuilder(alias).proxyAddress(initialProxyAddress).build();
61+
final var config = new ProxyLinkedProjectConfigBuilder(alias).proxyAddress(initialProxyAddress)
62+
.skipUnavailable(initialSkipUnavailable)
63+
.build();
8064

8165
// Verify we can get the linked projects on startup.
8266
assertThat(service.getInitialLinkedProjectConfigs(), equalTo(List.of(config)));
@@ -92,48 +76,23 @@ public void testListenersReceiveUpdates() {
9276
clusterSettings.applySettings(initialSettings);
9377
for (int i = 0; i < numListeners; ++i) {
9478
assertThat(listeners.get(i).updatedConfig, sameInstance(null));
95-
assertThat(listeners.get(i).skipUnavailableUpdate, sameInstance(null));
9679
listeners.get(i).reset();
9780
}
9881

99-
// Change the skip_unavailable, leave the other settings alone, we should get the skip_unavailable update only.
100-
var expectedSkipUnavailableUpdate = new SkipUnavailableUpdate(
101-
config.originProjectId(),
102-
config.linkedProjectId(),
103-
config.linkedProjectAlias(),
104-
config.skipUnavailable() == false
105-
);
106-
clusterSettings.applySettings(
107-
Settings.builder()
108-
.put(initialSettings)
109-
.put("cluster.remote." + alias + ".skip_unavailable", expectedSkipUnavailableUpdate.skipUnavailable)
110-
.build()
111-
);
112-
for (int i = 0; i < numListeners; ++i) {
113-
assertThat(listeners.get(i).updatedConfig, sameInstance(null));
114-
assertThat(listeners.get(i).skipUnavailableUpdate, equalTo(expectedSkipUnavailableUpdate));
115-
listeners.get(i).reset();
116-
}
117-
118-
// Change the proxy address, and set skip_unavailable back to original value, we should get both updates.
119-
expectedSkipUnavailableUpdate = new SkipUnavailableUpdate(
120-
config.originProjectId(),
121-
config.linkedProjectId(),
122-
config.linkedProjectAlias(),
123-
config.skipUnavailable()
124-
);
82+
// Change the proxy address and skip_unavailable values.
12583
final var newProxyAddress = "localhost:9401";
84+
final var newSkipUnavailable = config.skipUnavailable() == false;
12685
clusterSettings.applySettings(
12786
Settings.builder()
12887
.put(initialSettings)
12988
.put("cluster.remote." + alias + ".proxy_address", newProxyAddress)
130-
.put("cluster.remote." + alias + ".skip_unavailable", expectedSkipUnavailableUpdate.skipUnavailable)
89+
.put("cluster.remote." + alias + ".skip_unavailable", newSkipUnavailable)
13190
.build()
13291
);
13392
for (int i = 0; i < numListeners; ++i) {
13493
assertNotNull("expected non-null updatedConfig for listener " + i, listeners.get(i).updatedConfig);
13594
assertThat(listeners.get(i).updatedConfig.proxyAddress(), equalTo(newProxyAddress));
136-
assertThat(listeners.get(i).skipUnavailableUpdate, equalTo(expectedSkipUnavailableUpdate));
95+
assertThat(listeners.get(i).updatedConfig.skipUnavailable(), equalTo(newSkipUnavailable));
13796
}
13897
}
13998
}

‎server/src/test/java/org/elasticsearch/transport/RemoteClusterServiceTests.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import java.util.concurrent.TimeUnit;
5454
import java.util.concurrent.atomic.AtomicReference;
5555
import java.util.function.BiFunction;
56+
import java.util.function.Consumer;
5657

5758
import static org.elasticsearch.test.MockLog.assertThatLogger;
5859
import static org.elasticsearch.test.NodeRoles.masterOnlyNode;
@@ -112,6 +113,10 @@ private MockTransportService startTransport(
112113
return RemoteClusterConnectionTests.startTransport(id, knownNodes, version, transportVersion, threadPool, settings);
113114
}
114115

116+
private MockTransportService startTransport(final String id) {
117+
return startTransport(id, List.of(), VersionInformation.CURRENT, TransportVersion.current(), Settings.EMPTY);
118+
}
119+
115120
public void testSettingsAreRegistered() {
116121
assertTrue(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.contains(RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE));
117122
assertTrue(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.contains(RemoteClusterSettings.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING));
@@ -1877,6 +1882,61 @@ public void testLogsConnectionResult() throws IOException {
18771882
}
18781883
}
18791884

1885+
public void testSetSkipUnavailable() throws IOException {
1886+
final var skipUnavailableProperty = RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace("remote")
1887+
.getKey();
1888+
final var seedNodeProperty = SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS.getConcreteSettingForNamespace("remote").getKey();
1889+
final var clusterSettings = ClusterSettings.createBuiltInClusterSettings();
1890+
1891+
try (
1892+
var remote1Transport = startTransport("remote1");
1893+
var remote2Transport = startTransport("remote2");
1894+
var local = startTransport("local");
1895+
var remoteClusterService = createRemoteClusterService(Settings.EMPTY, clusterSettings, local)
1896+
) {
1897+
linkedProjectConfigService.register(remoteClusterService);
1898+
1899+
record SkipUnavailableTestConfig(
1900+
boolean skipUnavailable,
1901+
MockTransportService seedNodeTransportService,
1902+
boolean isNewConnection,
1903+
boolean rebuildExpected
1904+
) {}
1905+
1906+
final Consumer<SkipUnavailableTestConfig> applySettingsAndVerify = (cfg) -> {
1907+
final var currentConnection = cfg.isNewConnection ? null : remoteClusterService.getRemoteClusterConnection("remote");
1908+
clusterSettings.applySettings(
1909+
Settings.builder()
1910+
.put(skipUnavailableProperty, cfg.skipUnavailable)
1911+
.put(seedNodeProperty, cfg.seedNodeTransportService.getLocalNode().getAddress().toString())
1912+
.build()
1913+
);
1914+
assertTrue(isRemoteClusterRegistered(remoteClusterService, "remote"));
1915+
assertEquals(cfg.skipUnavailable, remoteClusterService.isSkipUnavailable("remote").orElseThrow());
1916+
if (cfg.rebuildExpected) {
1917+
assertNotSame(currentConnection, remoteClusterService.getRemoteClusterConnection("remote"));
1918+
} else if (cfg.isNewConnection == false) {
1919+
assertSame(currentConnection, remoteClusterService.getRemoteClusterConnection("remote"));
1920+
}
1921+
};
1922+
1923+
// Apply the initial settings and verify the new connection is built.
1924+
var skipUnavailable = randomBoolean();
1925+
applySettingsAndVerify.accept(new SkipUnavailableTestConfig(skipUnavailable, remote1Transport, true, true));
1926+
1927+
// Change skip_unavailable value, but not seed node, connection should not be rebuilt, but skip_unavailable should be modified.
1928+
skipUnavailable = skipUnavailable == false;
1929+
applySettingsAndVerify.accept(new SkipUnavailableTestConfig(skipUnavailable, remote1Transport, false, false));
1930+
1931+
// Change the seed node but not skip_unavailable, connection should be rebuilt and skip_unavailable should stay the same.
1932+
applySettingsAndVerify.accept(new SkipUnavailableTestConfig(skipUnavailable, remote2Transport, false, true));
1933+
1934+
// Change skip_unavailable value and the seed node, connection should be rebuilt and skip_unavailable should also be modified.
1935+
skipUnavailable = skipUnavailable == false;
1936+
applySettingsAndVerify.accept(new SkipUnavailableTestConfig(skipUnavailable, remote1Transport, false, true));
1937+
}
1938+
}
1939+
18801940
@FixForMultiProject(description = "Refactor to add the linked project ID associated with the alias.")
18811941
private LinkedProjectConfig buildLinkedProjectConfig(String alias, Settings staticSettings, Settings newSettings) {
18821942
final var mergedSettings = Settings.builder().put(staticSettings, false).put(newSettings, false).build();

0 commit comments

Comments
 (0)