Use LinkedProjectConfig in RemoteClusterService (#133266)

Moves all remaining settings and supporting methods into
RemoteClusterSettings and introduces a configuration type
LinkedProjectConfig with concrete record types that can be
built from the settings.  Refactors RemoteClusterService and
the related classes to use the LinkedProjectConfig, with settings
extraction at higher levels in RemoteClusterService and
RemoteClusterAware and its subclasses.
This is another step towards supporting multiple origin projects
with linked project configuration built from cluster state
ProjectCustom updates.

Resolves: ES-12656, ES-12569

Author: JeremyDahlgren

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,333 @@
+/*
+ * 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.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+import static org.elasticsearch.transport.RemoteConnectionStrategy.ConnectionStrategy;
+
+/**
+ * <p>Configuration for initializing {@link RemoteClusterConnection}s to linked projects.</p>
+ *
+ * <p>The {@link ProxyLinkedProjectConfigBuilder} and {@link SniffLinkedProjectConfigBuilder} classes can be used to build concrete
+ * implementations of {@link LinkedProjectConfig}.</p>
+ *
+ * <p>The {@link RemoteClusterSettings#toConfig(String, Settings)} and
+ * {@link RemoteClusterSettings#toConfig(ProjectId, ProjectId, String, Settings)} methods
+ * can be used to read {@link RemoteClusterSettings} to build a concrete {@link LinkedProjectConfig} from {@link Settings}.</p>
+ */
+public sealed interface LinkedProjectConfig {
+    ProjectId originProjectId();
+
+    ProjectId linkedProjectId();
+
+    String linkedProjectAlias();
+
+    TimeValue transportConnectTimeout();
+
+    Compression.Enabled connectionCompression();
+
+    Compression.Scheme connectionCompressionScheme();
+
+    TimeValue clusterPingSchedule();
+
+    TimeValue initialConnectionTimeout();
+
+    boolean skipUnavailable();
+
+    int maxPendingConnectionListeners();
+
+    ConnectionStrategy connectionStrategy();
+
+    int maxNumConnections();
+
+    String proxyAddress();
+
+    boolean isConnectionEnabled();
+
+    RemoteConnectionStrategy buildRemoteConnectionStrategy(TransportService transportService, RemoteConnectionManager connectionManager);
+
+    /**
+     * Configuration for initializing {@link RemoteClusterConnection}s to linked projects using the {@link ProxyConnectionStrategy}.
+     */
+    record ProxyLinkedProjectConfig(
+        ProjectId originProjectId,
+        ProjectId linkedProjectId,
+        String linkedProjectAlias,
+        TimeValue transportConnectTimeout,
+        Compression.Enabled connectionCompression,
+        Compression.Scheme connectionCompressionScheme,
+        TimeValue clusterPingSchedule,
+        TimeValue initialConnectionTimeout,
+        boolean skipUnavailable,
+        int maxPendingConnectionListeners,
+        int maxNumConnections,
+        String proxyAddress,
+        String serverName
+    ) implements LinkedProjectConfig {
+
+        @Override
+        public ConnectionStrategy connectionStrategy() {
+            return ConnectionStrategy.PROXY;
+        }
+
+        @Override
+        public boolean isConnectionEnabled() {
+            return Strings.isEmpty(proxyAddress) == false;
+        }
+
+        @Override
+        public RemoteConnectionStrategy buildRemoteConnectionStrategy(
+            TransportService transportService,
+            RemoteConnectionManager connectionManager
+        ) {
+            return new ProxyConnectionStrategy(this, transportService, connectionManager);
+        }
+    }
+
+    /**
+     * Configuration for initializing {@link RemoteClusterConnection}s to linked projects using the {@link SniffConnectionStrategy}.
+     */
+    record SniffLinkedProjectConfig(
+        ProjectId originProjectId,
+        ProjectId linkedProjectId,
+        String linkedProjectAlias,
+        TimeValue transportConnectTimeout,
+        Compression.Enabled connectionCompression,
+        Compression.Scheme connectionCompressionScheme,
+        TimeValue clusterPingSchedule,
+        TimeValue initialConnectionTimeout,
+        boolean skipUnavailable,
+        int maxPendingConnectionListeners,
+        int maxNumConnections,
+        Predicate<DiscoveryNode> nodePredicate,
+        List<String> seedNodes,
+        String proxyAddress
+    ) implements LinkedProjectConfig {
+
+        @Override
+        public ConnectionStrategy connectionStrategy() {
+            return ConnectionStrategy.SNIFF;
+        }
+
+        @Override
+        public boolean isConnectionEnabled() {
+            return seedNodes.isEmpty() == false;
+        }
+
+        @Override
+        public RemoteConnectionStrategy buildRemoteConnectionStrategy(
+            TransportService transportService,
+            RemoteConnectionManager connectionManager
+        ) {
+            return new SniffConnectionStrategy(this, transportService, connectionManager);
+        }
+    }
+
+    abstract class Builder<B extends Builder<B>> {
+        protected final ProjectId originProjectId;
+        protected final ProjectId linkedProjectId;
+        protected final String linkedProjectAlias;
+        protected final ConnectionStrategy connectionStrategy;
+        private final B concreteBuilder;
+        protected TimeValue transportConnectTimeout = TransportSettings.DEFAULT_CONNECT_TIMEOUT;
+        protected Compression.Enabled connectionCompression = TransportSettings.DEFAULT_TRANSPORT_COMPRESS;
+        protected Compression.Scheme connectionCompressionScheme = TransportSettings.DEFAULT_TRANSPORT_COMPRESSION_SCHEME;
+        protected TimeValue clusterPingSchedule = TransportSettings.DEFAULT_PING_SCHEDULE;
+        protected TimeValue initialConnectionTimeout = RemoteClusterSettings.DEFAULT_INITIAL_CONNECTION_TIMEOUT;
+        protected boolean skipUnavailable = RemoteClusterSettings.DEFAULT_SKIP_UNAVAILABLE;
+        protected String proxyAddress = "";
+        protected int maxNumConnections;
+        protected int maxPendingConnectionListeners = RemoteClusterSettings.DEFAULT_MAX_PENDING_CONNECTION_LISTENERS;
+
+        private Builder(
+            ProjectId originProjectId,
+            ProjectId linkedProjectId,
+            String linkedProjectAlias,
+            ConnectionStrategy connectionStrategy
+        ) {
+            this.originProjectId = Objects.requireNonNull(originProjectId);
+            this.linkedProjectId = Objects.requireNonNull(linkedProjectId);
+            this.linkedProjectAlias = requireNonEmpty(linkedProjectAlias, "linkedProjectAlias");
+            this.connectionStrategy = Objects.requireNonNull(connectionStrategy);
+            this.concreteBuilder = self();
+            this.maxNumConnections = switch (connectionStrategy) {
+                case PROXY -> RemoteClusterSettings.ProxyConnectionStrategySettings.DEFAULT_REMOTE_SOCKET_CONNECTIONS;
+                case SNIFF -> RemoteClusterSettings.SniffConnectionStrategySettings.DEFAULT_REMOTE_CONNECTIONS_PER_CLUSTER;
+            };
+        }
+
+        public B transportConnectTimeout(TimeValue transportConnectTimeout) {
+            this.transportConnectTimeout = Objects.requireNonNull(transportConnectTimeout);
+            return concreteBuilder;
+        }
+
+        public B connectionCompression(Compression.Enabled connectionCompression) {
+            this.connectionCompression = Objects.requireNonNull(connectionCompression);
+            return concreteBuilder;
+        }
+
+        public B connectionCompressionScheme(Compression.Scheme connectionCompressionScheme) {
+            this.connectionCompressionScheme = Objects.requireNonNull(connectionCompressionScheme);
+            return concreteBuilder;
+        }
+
+        public B clusterPingSchedule(TimeValue clusterPingSchedule) {
+            this.clusterPingSchedule = Objects.requireNonNull(clusterPingSchedule);
+            return concreteBuilder;
+        }
+
+        public B initialConnectionTimeout(TimeValue initialConnectionTimeout) {
+            this.initialConnectionTimeout = Objects.requireNonNull(initialConnectionTimeout);
+            return concreteBuilder;
+        }
+
+        public B skipUnavailable(boolean skipUnavailable) {
+            this.skipUnavailable = skipUnavailable;
+            return concreteBuilder;
+        }
+
+        public B proxyAddress(String proxyAddress) {
+            // TODO: Eliminate leniency here allowing an empty proxy address, ES-12737.
+            if (Strings.hasLength(proxyAddress)) {
+                RemoteConnectionStrategy.parsePort(proxyAddress);
+            }
+            this.proxyAddress = proxyAddress;
+            return concreteBuilder;
+        }
+
+        public B maxNumConnections(int maxNumConnections) {
+            this.maxNumConnections = requireGreaterThanZero(maxNumConnections, "maxNumConnections");
+            return concreteBuilder;
+        }
+
+        public B maxPendingConnectionListeners(int maxPendingConnectionListeners) {
+            this.maxPendingConnectionListeners = requireGreaterThanZero(maxPendingConnectionListeners, "maxPendingConnectionListeners");
+            return concreteBuilder;
+        }
+
+        public abstract LinkedProjectConfig build();
+
+        protected abstract B self();
+
+        protected static int requireGreaterThanZero(int value, String name) {
+            if (value <= 0) {
+                throw new IllegalArgumentException("[" + name + "] must be greater than 0");
+            }
+            return value;
+        }
+
+        protected static String requireNonEmpty(String value, String name) {
+            if (Objects.requireNonNull(value).isBlank()) {
+                throw new IllegalArgumentException("[" + name + "] cannot be empty");
+            }
+            return value;
+        }
+    }
+
+    class ProxyLinkedProjectConfigBuilder extends Builder<ProxyLinkedProjectConfigBuilder> {
+        private String serverName = "";
+
+        public ProxyLinkedProjectConfigBuilder(String linkedProjectAlias) {
+            super(ProjectId.DEFAULT, ProjectId.DEFAULT, linkedProjectAlias, ConnectionStrategy.PROXY);
+        }
+
+        public ProxyLinkedProjectConfigBuilder(ProjectId originProjectId, ProjectId linkedProjectId, String linkedProjectAlias) {
+            super(originProjectId, linkedProjectId, linkedProjectAlias, ConnectionStrategy.PROXY);
+        }
+
+        public ProxyLinkedProjectConfigBuilder serverName(String serverName) {
+            this.serverName = serverName;
+            return this;
+        }
+
+        @Override
+        public ProxyLinkedProjectConfig build() {
+            return new ProxyLinkedProjectConfig(
+                originProjectId,
+                linkedProjectId,
+                linkedProjectAlias,
+                transportConnectTimeout,
+                connectionCompression,
+                connectionCompressionScheme,
+                clusterPingSchedule,
+                initialConnectionTimeout,
+                skipUnavailable,
+                maxPendingConnectionListeners,
+                maxNumConnections,
+                proxyAddress,
+                serverName
+            );
+        }
+
+        @Override
+        protected ProxyLinkedProjectConfigBuilder self() {
+            return this;
+        }
+    }
+
+    class SniffLinkedProjectConfigBuilder extends Builder<SniffLinkedProjectConfigBuilder> {
+        private Predicate<DiscoveryNode> nodePredicate = RemoteClusterSettings.SniffConnectionStrategySettings.DEFAULT_NODE_PREDICATE;
+        private List<String> seedNodes = RemoteClusterSettings.SniffConnectionStrategySettings.DEFAULT_SEED_NODES;
+
+        public SniffLinkedProjectConfigBuilder(String linkedProjectAlias) {
+            super(ProjectId.DEFAULT, ProjectId.DEFAULT, linkedProjectAlias, ConnectionStrategy.SNIFF);
+        }
+
+        public SniffLinkedProjectConfigBuilder(ProjectId originProjectId, ProjectId linkedProjectId, String linkedProjectAlias) {
+            super(originProjectId, linkedProjectId, linkedProjectAlias, ConnectionStrategy.SNIFF);
+        }
+
+        public SniffLinkedProjectConfigBuilder nodePredicate(Predicate<DiscoveryNode> nodePredicate) {
+            this.nodePredicate = Objects.requireNonNull(nodePredicate);
+            return this;
+        }
+
+        public SniffLinkedProjectConfigBuilder seedNodes(List<String> seedNodes) {
+            // TODO: Eliminate leniency here allowing an empty set of seed nodes, ES-12737.
+            Objects.requireNonNull(seedNodes).forEach(RemoteConnectionStrategy::parsePort);
+            this.seedNodes = seedNodes;
+            return this;
+        }
+
+        @Override
+        public SniffLinkedProjectConfig build() {
+            return new SniffLinkedProjectConfig(
+                originProjectId,
+                linkedProjectId,
+                linkedProjectAlias,
+                transportConnectTimeout,
+                connectionCompression,
+                connectionCompressionScheme,
+                clusterPingSchedule,
+                initialConnectionTimeout,
+                skipUnavailable,
+                maxPendingConnectionListeners,
+                maxNumConnections,
+                nodePredicate,
+                seedNodes,
+                proxyAddress
+            );
+        }
+
+        @Override
+        protected SniffLinkedProjectConfigBuilder self() {
+            return this;
+        }
+    }
+}