Move RCS settings to RemoteClusterSettings class (#133092)

Relocates all Settings from RemoteClusterService to a
RemoteClusterSettings class.  Note that a
RemoteClusterSettingsTests class already exists that tests some of
these settings. This is one part in a series of refactors to use a
LinkedProjectConfig data class in RemoteClusterService and its related
classes.

Relates: ES-12656, ES-12569

Author: JeremyDahlgren

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -60,6 +60,7 @@
 import org.elasticsearch.transport.AbstractTransportRequest;
 import org.elasticsearch.transport.RemoteClusterConnection;
 import org.elasticsearch.transport.RemoteClusterService;
+import org.elasticsearch.transport.RemoteClusterSettings;
 import org.elasticsearch.transport.RemoteConnectionInfo;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.Transports;
@@ -493,7 +494,7 @@ void start(Task task, Collection<String> remotes, ActionListener<Map<String, Rem
         RemoteClusterStats makeRemoteClusterStats(String clusterAlias) {
             RemoteClusterConnection remoteConnection = remoteClusterService.getRemoteClusterConnection(clusterAlias);
             RemoteConnectionInfo remoteConnectionInfo = remoteConnection.getConnectionInfo();
-            var compression = RemoteClusterService.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace(clusterAlias).get(settings);
+            var compression = RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace(clusterAlias).get(settings);
             return new RemoteClusterStats(
                 remoteConnectionInfo.getModeInfo().modeName(),
                 remoteConnection.isSkipUnavailable(),

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -134,7 +134,7 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.ProxyConnectionStrategy;
 import org.elasticsearch.transport.RemoteClusterPortSettings;
-import org.elasticsearch.transport.RemoteClusterService;
+import org.elasticsearch.transport.RemoteClusterSettings;
 import org.elasticsearch.transport.RemoteConnectionStrategy;
 import org.elasticsearch.transport.SniffConnectionStrategy;
 import org.elasticsearch.transport.TransportService;
@@ -366,13 +366,13 @@ public void apply(Settings value, Settings current, Settings previous) {
         SearchService.DEFAULT_ALLOW_PARTIAL_SEARCH_RESULTS,
         TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,
         TransportSearchAction.DEFAULT_PRE_FILTER_SHARD_SIZE,
-        RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
+        RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
         SniffConnectionStrategy.REMOTE_CONNECTIONS_PER_CLUSTER,
-        RemoteClusterService.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING,
-        RemoteClusterService.REMOTE_NODE_ATTRIBUTE,
-        RemoteClusterService.REMOTE_CLUSTER_PING_SCHEDULE,
-        RemoteClusterService.REMOTE_CLUSTER_COMPRESS,
-        RemoteClusterService.REMOTE_CLUSTER_COMPRESSION_SCHEME,
+        RemoteClusterSettings.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING,
+        RemoteClusterSettings.REMOTE_NODE_ATTRIBUTE,
+        RemoteClusterSettings.REMOTE_CLUSTER_PING_SCHEDULE,
+        RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS,
+        RemoteClusterSettings.REMOTE_CLUSTER_COMPRESSION_SCHEME,
         RemoteConnectionStrategy.REMOTE_CONNECTION_MODE,
         ProxyConnectionStrategy.PROXY_ADDRESS,
         ProxyConnectionStrategy.REMOTE_SOCKET_CONNECTIONS,
@@ -605,7 +605,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         SimulatePipelineTransportAction.INGEST_NODE_TRANSPORT_ACTION_TIMEOUT,
         WriteAckDelay.WRITE_ACK_DELAY_INTERVAL,
         WriteAckDelay.WRITE_ACK_DELAY_RANDOMNESS_BOUND,
-        RemoteClusterService.REMOTE_CLUSTER_CREDENTIALS,
+        RemoteClusterSettings.REMOTE_CLUSTER_CREDENTIALS,
         RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED,
         RemoteClusterPortSettings.HOST,
         RemoteClusterPortSettings.PUBLISH_HOST,

server/src/main/java/org/elasticsearch/transport/RemoteClusterAware.java

@@ -229,8 +229,8 @@ void validateAndUpdateRemoteCluster(String clusterAlias, Settings settings) {
      */
     public void listenForUpdates(ClusterSettings clusterSettings) {
         List<Setting.AffixSetting<?>> remoteClusterSettings = List.of(
-            RemoteClusterService.REMOTE_CLUSTER_COMPRESS,
-            RemoteClusterService.REMOTE_CLUSTER_PING_SCHEDULE,
+            RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS,
+            RemoteClusterSettings.REMOTE_CLUSTER_PING_SCHEDULE,
             RemoteConnectionStrategy.REMOTE_CONNECTION_MODE,
             SniffConnectionStrategy.REMOTE_CLUSTERS_PROXY,
             SniffConnectionStrategy.REMOTE_CLUSTER_SEEDS,

server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java

@@ -37,7 +37,7 @@
  * remote case we only connect to a subset of the nodes in the cluster in an uni-directional fashion.
  *
  * This class also handles the discovery of nodes from the remote cluster. The initial list of seed nodes is only used to discover all nodes
- * in the remote cluster and connects to all eligible nodes, for details see {@link RemoteClusterService#REMOTE_NODE_ATTRIBUTE}.
+ * in the remote cluster and connects to all eligible nodes, for details see {@link RemoteClusterSettings#REMOTE_NODE_ATTRIBUTE}.
  *
  * In the case of a disconnection, this class will issue a re-connect task to establish at most
  * {@link SniffConnectionStrategy#REMOTE_CONNECTIONS_PER_CLUSTER} until either all eligible nodes are exhausted or the maximum number of
@@ -83,10 +83,10 @@ public final class RemoteClusterConnection implements Closeable {
         this.connectionStrategy = RemoteConnectionStrategy.buildStrategy(clusterAlias, transportService, remoteConnectionManager, settings);
         // we register the transport service here as a listener to make sure we notify handlers on disconnect etc.
         this.remoteConnectionManager.addListener(transportService);
-        this.skipUnavailable = RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace(clusterAlias)
+        this.skipUnavailable = RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace(clusterAlias)
             .get(settings);
         this.threadPool = transportService.threadPool;
-        initialConnectionTimeout = RemoteClusterService.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
+        initialConnectionTimeout = RemoteClusterSettings.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
     }
 
     /**

server/src/main/java/org/elasticsearch/transport/RemoteClusterCredentialsManager.java

@@ -21,7 +21,7 @@
 import java.util.Map;
 import java.util.Set;
 
-import static org.elasticsearch.transport.RemoteClusterService.REMOTE_CLUSTER_CREDENTIALS;
+import static org.elasticsearch.transport.RemoteClusterSettings.REMOTE_CLUSTER_CREDENTIALS;
 
 public class RemoteClusterCredentialsManager {
     private static final Logger logger = LogManager.getLogger(RemoteClusterCredentialsManager.class);

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

@@ -28,9 +28,6 @@
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.ClusterSettings;
-import org.elasticsearch.common.settings.SecureSetting;
-import org.elasticsearch.common.settings.SecureString;
-import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
@@ -43,9 +40,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -59,9 +54,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static org.elasticsearch.common.settings.Setting.boolSetting;
-import static org.elasticsearch.common.settings.Setting.enumSetting;
-import static org.elasticsearch.common.settings.Setting.timeSetting;
 import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED;
 
 /**
@@ -75,76 +67,6 @@ public final class RemoteClusterService extends RemoteClusterAware
 
     private static final Logger logger = LogManager.getLogger(RemoteClusterService.class);
 
-    /**
-     * The initial connect timeout for remote cluster connections
-     */
-    public static final Setting<TimeValue> REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING = Setting.positiveTimeSetting(
-        "cluster.remote.initial_connect_timeout",
-        TimeValue.timeValueSeconds(30),
-        Setting.Property.NodeScope
-    );
-
-    /**
-     * The name of a node attribute to select nodes that should be connected to in the remote cluster.
-     * For instance a node can be configured with {@code node.attr.gateway: true} in order to be eligible as a gateway node between
-     * clusters. In that case {@code cluster.remote.node.attr: gateway} can be used to filter out other nodes in the remote cluster.
-     * The value of the setting is expected to be a boolean, {@code true} for nodes that can become gateways, {@code false} otherwise.
-     */
-    public static final Setting<String> REMOTE_NODE_ATTRIBUTE = Setting.simpleString(
-        "cluster.remote.node.attr",
-        Setting.Property.NodeScope
-    );
-
-    public static final Setting.AffixSetting<Boolean> REMOTE_CLUSTER_SKIP_UNAVAILABLE = Setting.affixKeySetting(
-        "cluster.remote.",
-        "skip_unavailable",
-        (ns, key) -> boolSetting(key, true, new RemoteConnectionEnabled<>(ns, key), Setting.Property.Dynamic, Setting.Property.NodeScope)
-    );
-
-    public static final Setting.AffixSetting<TimeValue> REMOTE_CLUSTER_PING_SCHEDULE = Setting.affixKeySetting(
-        "cluster.remote.",
-        "transport.ping_schedule",
-        (ns, key) -> timeSetting(
-            key,
-            TransportSettings.PING_SCHEDULE,
-            new RemoteConnectionEnabled<>(ns, key),
-            Setting.Property.Dynamic,
-            Setting.Property.NodeScope
-        )
-    );
-
-    public static final Setting.AffixSetting<Compression.Enabled> REMOTE_CLUSTER_COMPRESS = Setting.affixKeySetting(
-        "cluster.remote.",
-        "transport.compress",
-        (ns, key) -> enumSetting(
-            Compression.Enabled.class,
-            key,
-            TransportSettings.TRANSPORT_COMPRESS,
-            new RemoteConnectionEnabled<>(ns, key),
-            Setting.Property.Dynamic,
-            Setting.Property.NodeScope
-        )
-    );
-
-    public static final Setting.AffixSetting<Compression.Scheme> REMOTE_CLUSTER_COMPRESSION_SCHEME = Setting.affixKeySetting(
-        "cluster.remote.",
-        "transport.compression_scheme",
-        (ns, key) -> enumSetting(
-            Compression.Scheme.class,
-            key,
-            TransportSettings.TRANSPORT_COMPRESSION_SCHEME,
-            new RemoteConnectionEnabled<>(ns, key),
-            Setting.Property.Dynamic,
-            Setting.Property.NodeScope
-        )
-    );
-
-    public static final Setting.AffixSetting<SecureString> REMOTE_CLUSTER_CREDENTIALS = Setting.affixKeySetting(
-        "cluster.remote.",
-        "credentials",
-        key -> SecureSetting.secureString(key, null)
-    );
-
     public static final String REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME = "cluster:internal/remote_cluster/handshake";
 
     private final boolean isRemoteClusterClient;
@@ -343,7 +265,11 @@ public RemoteClusterConnection getRemoteClusterConnection(String cluster) {
     @Override
     public void listenForUpdates(ClusterSettings clusterSettings) {
         super.listenForUpdates(clusterSettings);
-        clusterSettings.addAffixUpdateConsumer(REMOTE_CLUSTER_SKIP_UNAVAILABLE, this::updateSkipUnavailable, (alias, value) -> {});
+        clusterSettings.addAffixUpdateConsumer(
+            RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
+            this::updateSkipUnavailable,
+            (alias, value) -> {}
+        );
     }
 
     private synchronized void updateSkipUnavailable(String clusterAlias, Boolean skipUnavailable) {
@@ -535,7 +461,7 @@ enum RemoteClusterConnectionStatus {
     void initializeRemoteClusters() {
         @FixForMultiProject(description = "Refactor for initializing connections to linked projects for each origin project supported.")
         final var projectId = projectResolver.getProjectId();
-        final TimeValue timeValue = REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
+        final TimeValue timeValue = RemoteClusterSettings.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
         final PlainActionFuture<Void> future = new PlainActionFuture<>();
         Set<String> enabledClusters = RemoteClusterAware.getEnabledRemoteClusters(settings);
 
@@ -630,9 +556,9 @@ public enum DisconnectedStrategy {
         FAIL_IF_DISCONNECTED,
 
         /**
-         * Behave according to the {@link #REMOTE_CLUSTER_SKIP_UNAVAILABLE} setting for this remote cluster: if this setting is
-         * {@code false} (the default) then behave like {@link #RECONNECT_IF_DISCONNECTED}, but if it is {@code true} then behave like
-         * {@link #FAIL_IF_DISCONNECTED}.
+         * Behave according to the {@link RemoteClusterSettings#REMOTE_CLUSTER_SKIP_UNAVAILABLE} setting for this remote cluster: if this
+         * setting is {@code false} (the default) then behave like {@link #RECONNECT_IF_DISCONNECTED}, but if it is {@code true} then behave
+         * like {@link #FAIL_IF_DISCONNECTED}.
          */
         RECONNECT_UNLESS_SKIP_UNAVAILABLE
     }
@@ -737,39 +663,4 @@ private Map<String, RemoteClusterConnection> getConnectionsMapForProject(Project
         assert ProjectId.DEFAULT.equals(projectId) : "Only the default project ID should be used when multiple projects are not supported";
         return remoteClusters.get(projectId);
     }
-
-    private static class RemoteConnectionEnabled<T> implements Setting.Validator<T> {
-
-        private final String clusterAlias;
-        private final String key;
-
-        private RemoteConnectionEnabled(String clusterAlias, String key) {
-            this.clusterAlias = clusterAlias;
-            this.key = key;
-        }
-
-        @Override
-        public void validate(T value) {}
-
-        @Override
-        public void validate(T value, Map<Setting<?>, Object> settings, boolean isPresent) {
-            if (isPresent && RemoteConnectionStrategy.isConnectionEnabled(clusterAlias, settings) == false) {
-                throw new IllegalArgumentException("Cannot configure setting [" + key + "] if remote cluster is not enabled.");
-            }
-        }
-
-        @Override
-        public Iterator<Setting<?>> settings() {
-            return Stream.concat(
-                Stream.of(RemoteConnectionStrategy.REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace(clusterAlias)),
-                settingsStream()
-            ).iterator();
-        }
-
-        private Stream<Setting<?>> settingsStream() {
-            return Arrays.stream(RemoteConnectionStrategy.ConnectionStrategy.values())
-                .flatMap(strategy -> strategy.getEnablementSettings().get())
-                .map(as -> as.getConcreteSettingForNamespace(clusterAlias));
-        }
-    };
 }