Use LinkedProjectConfig in RemoteClusterService

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

@@ -132,11 +132,10 @@
 import org.elasticsearch.snapshots.SnapshotShutdownProgressTracker;
 import org.elasticsearch.snapshots.SnapshotsService;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.ProxyConnectionStrategy;
 import org.elasticsearch.transport.RemoteClusterPortSettings;
 import org.elasticsearch.transport.RemoteClusterSettings;
-import org.elasticsearch.transport.RemoteConnectionStrategy;
-import org.elasticsearch.transport.SniffConnectionStrategy;
+import org.elasticsearch.transport.RemoteClusterSettings.ProxyConnectionStrategySettings;
+import org.elasticsearch.transport.RemoteClusterSettings.SniffConnectionStrategySettings;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.TransportSettings;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -367,19 +366,19 @@ public void apply(Settings value, Settings current, Settings previous) {
         TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,
         TransportSearchAction.DEFAULT_PRE_FILTER_SHARD_SIZE,
         RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
-        SniffConnectionStrategy.REMOTE_CONNECTIONS_PER_CLUSTER,
         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,
-        ProxyConnectionStrategy.SERVER_NAME,
-        SniffConnectionStrategy.REMOTE_CLUSTERS_PROXY,
-        SniffConnectionStrategy.REMOTE_CLUSTER_SEEDS,
-        SniffConnectionStrategy.REMOTE_NODE_CONNECTIONS,
+        RemoteClusterSettings.REMOTE_CONNECTION_MODE,
+        ProxyConnectionStrategySettings.PROXY_ADDRESS,
+        ProxyConnectionStrategySettings.REMOTE_SOCKET_CONNECTIONS,
+        ProxyConnectionStrategySettings.SERVER_NAME,
+        SniffConnectionStrategySettings.REMOTE_CLUSTERS_PROXY,
+        SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS,
+        SniffConnectionStrategySettings.REMOTE_CONNECTIONS_PER_CLUSTER,
+        SniffConnectionStrategySettings.REMOTE_NODE_CONNECTIONS,
         TransportCloseIndexAction.CLUSTER_INDICES_CLOSE_ENABLE_SETTING,
         ShardsLimitAllocationDecider.CLUSTER_TOTAL_SHARDS_PER_NODE_SETTING,
         SnapshotShutdownProgressTracker.SNAPSHOT_PROGRESS_DURING_SHUTDOWN_LOG_INTERVAL_SETTING,

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

@@ -0,0 +1,165 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.transport;
+
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.core.TimeValue;
+
+import java.util.List;
+import java.util.function.Predicate;
+
+import static org.elasticsearch.transport.RemoteConnectionStrategy.ConnectionStrategy;
+
+public record LinkedProjectConfig(
+    ProjectId originProjectId,
+    ProjectId linkedProjectId,
+    String linkedProjectAlias,
+    TimeValue transportConnectTimeout,
+    Compression.Enabled connectionCompression,
+    Compression.Scheme connectionCompressionScheme,
+    TimeValue clusterPingSchedule,
+    TimeValue initialConnectionTimeout,
+    boolean skipUnavailable,
+    ConnectionStrategy connectionStrategy,
+    int maxNumConnections,
+    String proxyAddress,
+    String proxyServerName,
+    Predicate<DiscoveryNode> sniffNodePredicate,
+    List<String> sniffSeedNodes,
+    int maxPendingConnectionListeners // only used in tests
+) {
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        private ProjectId originProjectId;
+        private ProjectId linkedProjectId;
+        private String linkedProjectAlias;
+        private TimeValue transportConnectTimeout;
+        private Compression.Enabled connectionCompression;
+        private Compression.Scheme connectionCompressionScheme;
+        private TimeValue clusterPingSchedule;
+        private TimeValue initialConnectionTimeout;
+        private boolean skipUnavailable;
+        private ConnectionStrategy connectionStrategy;
+        private int maxNumConnections;
+        private String proxyAddress;
+        private String proxyServerName;
+        private Predicate<DiscoveryNode> sniffNodePredicate;
+        private List<String> sniffSeedNodes;
+        private int maxPendingConnectionListeners = 1000;
+
+        private Builder() {}
+
+        public Builder originProjectId(ProjectId originProjectId) {
+            this.originProjectId = originProjectId;
+            return this;
+        }
+
+        public Builder linkedProjectId(ProjectId linkedProjectId) {
+            this.linkedProjectId = linkedProjectId;
+            return this;
+        }
+
+        public Builder linkedProjectAlias(String linkedProjectAlias) {
+            this.linkedProjectAlias = linkedProjectAlias;
+            return this;
+        }
+
+        public Builder transportConnectTimeout(TimeValue transportConnectTimeout) {
+            this.transportConnectTimeout = transportConnectTimeout;
+            return this;
+        }
+
+        public Builder connectionCompression(Compression.Enabled connectionCompression) {
+            this.connectionCompression = connectionCompression;
+            return this;
+        }
+
+        public Builder connectionCompressionScheme(Compression.Scheme connectionCompressionScheme) {
+            this.connectionCompressionScheme = connectionCompressionScheme;
+            return this;
+        }
+
+        public Builder clusterPingSchedule(TimeValue clusterPingSchedule) {
+            this.clusterPingSchedule = clusterPingSchedule;
+            return this;
+        }
+
+        public Builder initialConnectionTimeout(TimeValue initialConnectionTimeout) {
+            this.initialConnectionTimeout = initialConnectionTimeout;
+            return this;
+        }
+
+        public Builder skipUnavailable(boolean skipUnavailable) {
+            this.skipUnavailable = skipUnavailable;
+            return this;
+        }
+
+        public Builder connectionStrategy(ConnectionStrategy connectionStrategy) {
+            this.connectionStrategy = connectionStrategy;
+            return this;
+        }
+
+        public Builder maxNumConnections(int maxNumConnections) {
+            this.maxNumConnections = maxNumConnections;
+            return this;
+        }
+
+        public Builder proxyAddress(String proxyAddress) {
+            this.proxyAddress = proxyAddress;
+            return this;
+        }
+
+        public Builder proxyServerName(String proxyServerName) {
+            this.proxyServerName = proxyServerName;
+            return this;
+        }
+
+        public Builder sniffNodePredicate(Predicate<DiscoveryNode> sniffNodePredicate) {
+            this.sniffNodePredicate = sniffNodePredicate;
+            return this;
+        }
+
+        public Builder sniffSeedNodes(List<String> sniffSeedNodes) {
+            this.sniffSeedNodes = sniffSeedNodes;
+            return this;
+        }
+
+        public Builder maxPendingConnectionListeners(int maxPendingConnectionListeners) {
+            this.maxPendingConnectionListeners = maxPendingConnectionListeners;
+            return this;
+        }
+
+        public LinkedProjectConfig build() {
+            return new LinkedProjectConfig(
+                originProjectId,
+                linkedProjectId,
+                linkedProjectAlias,
+                transportConnectTimeout,
+                connectionCompression,
+                connectionCompressionScheme,
+                clusterPingSchedule,
+                initialConnectionTimeout,
+                skipUnavailable,
+                connectionStrategy,
+                maxNumConnections,
+                proxyAddress,
+                proxyServerName,
+                sniffNodePredicate,
+                sniffSeedNodes,
+                maxPendingConnectionListeners
+            );
+        }
+    }
+}

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

@@ -19,8 +19,6 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.core.Predicates;
@@ -38,56 +36,11 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
-import java.util.stream.Stream;
 
-import static org.elasticsearch.common.settings.Setting.intSetting;
 import static org.elasticsearch.core.Strings.format;
 
 public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
 
-    /**
-     * The remote address for the proxy. The connections will be opened to the configured address.
-     */
-    public static final Setting.AffixSetting<String> PROXY_ADDRESS = Setting.affixKeySetting(
-        "cluster.remote.",
-        "proxy_address",
-        (ns, key) -> Setting.simpleString(key, new StrategyValidator<>(ns, key, ConnectionStrategy.PROXY, s -> {
-            if (Strings.hasLength(s)) {
-                parsePort(s);
-            }
-        }), Setting.Property.Dynamic, Setting.Property.NodeScope)
-    );
-
-    /**
-     * The maximum number of socket connections that will be established to a remote cluster. The default is 18.
-     */
-    public static final Setting.AffixSetting<Integer> REMOTE_SOCKET_CONNECTIONS = Setting.affixKeySetting(
-        "cluster.remote.",
-        "proxy_socket_connections",
-        (ns, key) -> intSetting(
-            key,
-            18,
-            1,
-            new StrategyValidator<>(ns, key, ConnectionStrategy.PROXY),
-            Setting.Property.Dynamic,
-            Setting.Property.NodeScope
-        )
-    );
-
-    /**
-     * A configurable server_name attribute
-     */
-    public static final Setting.AffixSetting<String> SERVER_NAME = Setting.affixKeySetting(
-        "cluster.remote.",
-        "server_name",
-        (ns, key) -> Setting.simpleString(
-            key,
-            new StrategyValidator<>(ns, key, ConnectionStrategy.PROXY),
-            Setting.Property.Dynamic,
-            Setting.Property.NodeScope
-        )
-    );
-
     static final int CHANNELS_PER_CONNECTION = 1;
 
     private static final int MAX_CONNECT_ATTEMPTS_PER_RUN = 3;
@@ -99,78 +52,20 @@ public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
     private final AtomicReference<ClusterName> remoteClusterName = new AtomicReference<>();
     private final ConnectionManager.ConnectionValidator clusterNameValidator;
 
-    ProxyConnectionStrategy(
-        String clusterAlias,
-        TransportService transportService,
-        RemoteConnectionManager connectionManager,
-        Settings settings
-    ) {
-        this(
-            clusterAlias,
-            transportService,
-            connectionManager,
-            settings,
-            REMOTE_SOCKET_CONNECTIONS.getConcreteSettingForNamespace(clusterAlias).get(settings),
-            PROXY_ADDRESS.getConcreteSettingForNamespace(clusterAlias).get(settings),
-            SERVER_NAME.getConcreteSettingForNamespace(clusterAlias).get(settings)
-        );
-    }
-
-    ProxyConnectionStrategy(
-        String clusterAlias,
-        TransportService transportService,
-        RemoteConnectionManager connectionManager,
-        Settings settings,
-        int maxNumConnections,
-        String configuredAddress
-    ) {
-        this(
-            clusterAlias,
-            transportService,
-            connectionManager,
-            settings,
-            maxNumConnections,
-            configuredAddress,
-            () -> resolveAddress(configuredAddress),
-            null
-        );
-    }
-
-    ProxyConnectionStrategy(
-        String clusterAlias,
-        TransportService transportService,
-        RemoteConnectionManager connectionManager,
-        Settings settings,
-        int maxNumConnections,
-        String configuredAddress,
-        String configuredServerName
-    ) {
-        this(
-            clusterAlias,
-            transportService,
-            connectionManager,
-            settings,
-            maxNumConnections,
-            configuredAddress,
-            () -> resolveAddress(configuredAddress),
-            configuredServerName
-        );
+    ProxyConnectionStrategy(LinkedProjectConfig config, TransportService transportService, RemoteConnectionManager connectionManager) {
+        this(config, () -> resolveAddress(config.proxyAddress()), transportService, connectionManager);
     }
 
     ProxyConnectionStrategy(
-        String clusterAlias,
-        TransportService transportService,
-        RemoteConnectionManager connectionManager,
-        Settings settings,
-        int maxNumConnections,
-        String configuredAddress,
+        LinkedProjectConfig config,
         Supplier<TransportAddress> address,
-        String configuredServerName
+        TransportService transportService,
+        RemoteConnectionManager connectionManager
     ) {
-        super(clusterAlias, transportService, connectionManager, settings);
-        this.maxNumConnections = maxNumConnections;
-        this.configuredAddress = configuredAddress;
-        this.configuredServerName = configuredServerName;
+        super(config, transportService, connectionManager);
+        this.maxNumConnections = config.maxNumConnections();
+        this.configuredAddress = config.proxyAddress();
+        this.configuredServerName = config.proxyServerName();
         assert Strings.isEmpty(configuredAddress) == false : "Cannot use proxy connection strategy with no configured addresses";
         this.address = address;
         this.clusterNameValidator = (newConnection, actualProfile, listener) -> {
@@ -200,10 +95,6 @@ public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
         };
     }
 
-    static Stream<Setting.AffixSetting<?>> enablementSettings() {
-        return Stream.of(ProxyConnectionStrategy.PROXY_ADDRESS);
-    }
-
     static Writeable.Reader<RemoteConnectionInfo.ModeInfo> infoReader() {
         return ProxyModeInfo::new;
     }
@@ -214,13 +105,10 @@ protected boolean shouldOpenMoreConnections() {
     }
 
     @Override
-    protected boolean strategyMustBeRebuilt(Settings newSettings) {
-        String address = PROXY_ADDRESS.getConcreteSettingForNamespace(clusterAlias).get(newSettings);
-        int numOfSockets = REMOTE_SOCKET_CONNECTIONS.getConcreteSettingForNamespace(clusterAlias).get(newSettings);
-        String serverName = SERVER_NAME.getConcreteSettingForNamespace(clusterAlias).get(newSettings);
-        return numOfSockets != maxNumConnections
-            || configuredAddress.equals(address) == false
-            || Objects.equals(serverName, configuredServerName) == false;
+    protected boolean strategyMustBeRebuilt(LinkedProjectConfig config) {
+        return config.maxNumConnections() != maxNumConnections
+            || configuredAddress.equals(config.proxyAddress()) == false
+            || Objects.equals(config.proxyServerName(), configuredServerName) == false;
     }
 
     @Override