Split config into sniff and proxy types

Author: JeremyDahlgren

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

@@ -11,38 +11,126 @@
 
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Strings;
 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;
 
-// FIXME: Refactor into a class heirarchy with subclasses for each connection strategy
-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() {
+public 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();
+
+    boolean isConnectionEnabled();
+
+    RemoteConnectionStrategy buildConnectionStrategy(TransportService transportService, RemoteConnectionManager connectionManager);
+
+    record ProxyConnectionStrategyConfig(
+        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 buildConnectionStrategy(
+            TransportService transportService,
+            RemoteConnectionManager connectionManager
+        ) {
+            return new ProxyConnectionStrategy(this, transportService, connectionManager);
+        }
+    }
+
+    record SniffConnectionStrategyConfig(
+        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 buildConnectionStrategy(
+            TransportService transportService,
+            RemoteConnectionManager connectionManager
+        ) {
+            return new SniffConnectionStrategy(this, transportService, connectionManager);
+        }
+    }
+
+    static Builder builder() {
         return new Builder();
     }
 
-    public static class Builder {
+    static Builder buildForAlias(String linkedProjectAlias) {
+        return new Builder().linkedProjectAlias(linkedProjectAlias);
+    }
+
+    static Builder buildForLinkedProject(ProjectId originProjectId, ProjectId linkedProjectId, String linkedProjectAlias) {
+        return new Builder(originProjectId, linkedProjectId, linkedProjectAlias);
+    }
+
+    class Builder {
         private ProjectId originProjectId;
         private ProjectId linkedProjectId;
         private String linkedProjectAlias;
@@ -62,13 +150,19 @@ public static class Builder {
 
         private Builder() {}
 
+        private Builder(ProjectId originProjectId, ProjectId linkedProjectId, String linkedProjectAlias) {
+            originProjectId(originProjectId);
+            linkedProjectId(linkedProjectId);
+            linkedProjectAlias(linkedProjectAlias);
+        }
+
         public Builder originProjectId(ProjectId originProjectId) {
-            this.originProjectId = originProjectId;
+            this.originProjectId = Objects.requireNonNull(originProjectId);
             return this;
         }
 
         public Builder linkedProjectId(ProjectId linkedProjectId) {
-            this.linkedProjectId = linkedProjectId;
+            this.linkedProjectId = Objects.requireNonNull(linkedProjectId);
             return this;
         }
 
@@ -81,27 +175,27 @@ public Builder linkedProjectAlias(String linkedProjectAlias) {
         }
 
         public Builder transportConnectTimeout(TimeValue transportConnectTimeout) {
-            this.transportConnectTimeout = transportConnectTimeout;
+            this.transportConnectTimeout = Objects.requireNonNull(transportConnectTimeout);
             return this;
         }
 
         public Builder connectionCompression(Compression.Enabled connectionCompression) {
-            this.connectionCompression = connectionCompression;
+            this.connectionCompression = Objects.requireNonNull(connectionCompression);
             return this;
         }
 
         public Builder connectionCompressionScheme(Compression.Scheme connectionCompressionScheme) {
-            this.connectionCompressionScheme = connectionCompressionScheme;
+            this.connectionCompressionScheme = Objects.requireNonNull(connectionCompressionScheme);
             return this;
         }
 
         public Builder clusterPingSchedule(TimeValue clusterPingSchedule) {
-            this.clusterPingSchedule = clusterPingSchedule;
+            this.clusterPingSchedule = Objects.requireNonNull(clusterPingSchedule);
             return this;
         }
 
         public Builder initialConnectionTimeout(TimeValue initialConnectionTimeout) {
-            this.initialConnectionTimeout = initialConnectionTimeout;
+            this.initialConnectionTimeout = Objects.requireNonNull(initialConnectionTimeout);
             return this;
         }
 
@@ -111,7 +205,7 @@ public Builder skipUnavailable(boolean skipUnavailable) {
         }
 
         public Builder connectionStrategy(ConnectionStrategy connectionStrategy) {
-            this.connectionStrategy = connectionStrategy;
+            this.connectionStrategy = Objects.requireNonNull(connectionStrategy);
             return this;
         }
 
@@ -131,12 +225,12 @@ public Builder proxyServerName(String proxyServerName) {
         }
 
         public Builder sniffNodePredicate(Predicate<DiscoveryNode> sniffNodePredicate) {
-            this.sniffNodePredicate = sniffNodePredicate;
+            this.sniffNodePredicate = Objects.requireNonNull(sniffNodePredicate);
             return this;
         }
 
         public Builder sniffSeedNodes(List<String> sniffSeedNodes) {
-            this.sniffSeedNodes = sniffSeedNodes;
+            this.sniffSeedNodes = Objects.requireNonNull(sniffSeedNodes);
             return this;
         }
 
@@ -146,7 +240,16 @@ public Builder maxPendingConnectionListeners(int maxPendingConnectionListeners)
         }
 
         public LinkedProjectConfig build() {
-            return new LinkedProjectConfig(
+            assert connectionStrategy != null : "ConnectionStrategy must be set";
+            return switch (connectionStrategy) {
+                case PROXY -> buildProxyConnectionStrategyConfig();
+                case SNIFF -> buildSniffConnectionStrategyConfig();
+            };
+        }
+
+        public ProxyConnectionStrategyConfig buildProxyConnectionStrategyConfig() {
+            assert ConnectionStrategy.PROXY.equals(connectionStrategy) : "ConnectionStrategy must be PROXY";
+            return new ProxyConnectionStrategyConfig(
                 originProjectId,
                 linkedProjectId,
                 linkedProjectAlias,
@@ -156,13 +259,30 @@ public LinkedProjectConfig build() {
                 clusterPingSchedule,
                 initialConnectionTimeout,
                 skipUnavailable,
-                connectionStrategy,
+                maxPendingConnectionListeners,
                 maxNumConnections,
                 proxyAddress,
-                proxyServerName,
+                proxyServerName
+            );
+        }
+
+        public SniffConnectionStrategyConfig buildSniffConnectionStrategyConfig() {
+            assert ConnectionStrategy.SNIFF.equals(connectionStrategy) : "ConnectionStrategy must be SNIFF";
+            return new SniffConnectionStrategyConfig(
+                originProjectId,
+                linkedProjectId,
+                linkedProjectAlias,
+                transportConnectTimeout,
+                connectionCompression,
+                connectionCompressionScheme,
+                clusterPingSchedule,
+                initialConnectionTimeout,
+                skipUnavailable,
+                maxPendingConnectionListeners,
+                maxNumConnections,
                 sniffNodePredicate,
                 sniffSeedNodes,
-                maxPendingConnectionListeners
+                proxyAddress
             );
         }
     }

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

@@ -38,6 +38,7 @@
 import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
+import static org.elasticsearch.transport.LinkedProjectConfig.ProxyConnectionStrategyConfig;
 
 public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
 
@@ -52,20 +53,24 @@ public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
     private final AtomicReference<ClusterName> remoteClusterName = new AtomicReference<>();
     private final ConnectionManager.ConnectionValidator clusterNameValidator;
 
-    ProxyConnectionStrategy(LinkedProjectConfig config, TransportService transportService, RemoteConnectionManager connectionManager) {
+    ProxyConnectionStrategy(
+        ProxyConnectionStrategyConfig config,
+        TransportService transportService,
+        RemoteConnectionManager connectionManager
+    ) {
         this(config, () -> resolveAddress(config.proxyAddress()), transportService, connectionManager);
     }
 
     ProxyConnectionStrategy(
-        LinkedProjectConfig config,
+        ProxyConnectionStrategyConfig config,
         Supplier<TransportAddress> address,
         TransportService transportService,
         RemoteConnectionManager connectionManager
     ) {
         super(config, transportService, connectionManager);
         this.maxNumConnections = config.maxNumConnections();
         this.configuredAddress = config.proxyAddress();
-        this.configuredServerName = config.proxyServerName();
+        this.configuredServerName = config.serverName();
         assert Strings.isEmpty(configuredAddress) == false : "Cannot use proxy connection strategy with no configured addresses";
         this.address = address;
         this.clusterNameValidator = (newConnection, actualProfile, listener) -> {
@@ -106,9 +111,11 @@ protected boolean shouldOpenMoreConnections() {
 
     @Override
     protected boolean strategyMustBeRebuilt(LinkedProjectConfig config) {
-        return config.maxNumConnections() != maxNumConnections
-            || configuredAddress.equals(config.proxyAddress()) == false
-            || Objects.equals(config.proxyServerName(), configuredServerName) == false;
+        assert config instanceof ProxyConnectionStrategyConfig : "expected config to be of type " + ProxyConnectionStrategy.class;
+        final var proxyConfig = (ProxyConnectionStrategyConfig) config;
+        return proxyConfig.maxNumConnections() != maxNumConnections
+            || configuredAddress.equals(proxyConfig.proxyAddress()) == false
+            || Objects.equals(proxyConfig.serverName(), configuredServerName) == false;
     }
 
     @Override

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

@@ -77,7 +77,7 @@ public final class RemoteClusterConnection implements Closeable {
             credentialsManager,
             createConnectionManager(profile, transportService)
         );
-        this.connectionStrategy = RemoteConnectionStrategy.buildStrategy(config, transportService, remoteConnectionManager);
+        this.connectionStrategy = config.buildConnectionStrategy(transportService, remoteConnectionManager);
         // we register the transport service here as a listener to make sure we notify handlers on disconnect etc.
         this.remoteConnectionManager.addListener(transportService);
         this.skipUnavailable = config.skipUnavailable();

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

@@ -435,10 +435,7 @@ private synchronized void updateRemoteCluster(
             throw new IllegalArgumentException("remote clusters must not have the empty string as its key");
         }
         final var mergedSettings = Settings.builder().put(settings, false).put(newSettings, false).build();
-        final var linkedProjectConfig = RemoteClusterSettings.toConfigBuilder(clusterAlias, mergedSettings)
-            .originProjectId(projectId)
-            .linkedProjectAlias(clusterAlias)
-            .build();
+        final var linkedProjectConfig = RemoteClusterSettings.toConfig(projectId, clusterAlias, mergedSettings);
         updateRemoteCluster(linkedProjectConfig, forceRebuild, listener);
     }
 
@@ -451,7 +448,7 @@ private synchronized void updateRemoteCluster(
         final var clusterAlias = config.linkedProjectAlias();
         final var connectionMap = getConnectionsMapForProject(projectId);
         RemoteClusterConnection remote = connectionMap.get(clusterAlias);
-        if (RemoteConnectionStrategy.isConnectionEnabled(config) == false) {
+        if (config.isConnectionEnabled() == false) {
             try {
                 IOUtils.close(remote);
             } catch (IOException e) {
@@ -508,8 +505,8 @@ void initializeRemoteClusters() {
 
         CountDownActionListener listener = new CountDownActionListener(enabledClusters.size(), future);
         for (String clusterAlias : enabledClusters) {
-            final var config = RemoteClusterSettings.toConfigBuilder(clusterAlias, settings).originProjectId(projectId).build();
-            updateRemoteCluster(config, false, listener.map(ignored -> null));
+            final var linkedProjectConfig = RemoteClusterSettings.toConfig(projectId, clusterAlias, settings);
+            updateRemoteCluster(linkedProjectConfig, false, listener.map(ignored -> null));
         }
 
         if (enabledClusters.isEmpty()) {