Refactor RemoteClusterService to be multi-project aware

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

@@ -20,8 +20,11 @@
 import org.elasticsearch.action.support.RefCountingRunnable;
 import org.elasticsearch.client.internal.RemoteClusterClient;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.cluster.project.DefaultProjectResolver;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.SecureSetting;
@@ -31,6 +34,7 @@
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.indices.IndicesExpressionGrouper;
@@ -53,6 +57,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static org.elasticsearch.common.settings.Setting.boolSetting;
@@ -155,9 +160,11 @@ public boolean isRemoteClusterServerEnabled() {
     }
 
     private final TransportService transportService;
-    private final Map<String, RemoteClusterConnection> remoteClusters = ConcurrentCollections.newConcurrentMap();
+    private final Map<ProjectId, Map<String, RemoteClusterConnection>> remoteClusters = ConcurrentCollections.newConcurrentMap();
     private final RemoteClusterCredentialsManager remoteClusterCredentialsManager;
+    private final ProjectResolver projectResolver;
 
+    @FixForMultiProject(description = "Inject the ProjectResolver instance.")
     RemoteClusterService(Settings settings, TransportService transportService) {
         super(settings);
         this.enabled = DiscoveryNode.isRemoteClusterClient(settings);
@@ -167,6 +174,21 @@ public boolean isRemoteClusterServerEnabled() {
         if (remoteClusterServerEnabled) {
             registerRemoteClusterHandshakeRequestHandler(transportService);
         }
+        this.projectResolver = DefaultProjectResolver.INSTANCE;
+    }
+
+    /**
+     * Returns the map of connections for the {@link ProjectId} currently returned by the {@link ProjectResolver}.
+     */
+    private Map<String, RemoteClusterConnection> getConnectionsMap() {
+        return getConnectionsMap(projectResolver.getProjectId());
+    }
+
+    /**
+     * Returns the map of connections for the given {@link ProjectId}.
+     */
+    private Map<String, RemoteClusterConnection> getConnectionsMap(ProjectId projectId) {
+        return remoteClusters.computeIfAbsent(projectId, unused -> ConcurrentCollections.newConcurrentMap());
     }
 
     public DiscoveryNode getLocalNode() {
@@ -176,12 +198,12 @@ public DiscoveryNode getLocalNode() {
     /**
      * Returns <code>true</code> if at least one remote cluster is configured
      */
-    public boolean isCrossClusterSearchEnabled() {
-        return remoteClusters.isEmpty() == false;
+    boolean isCrossClusterSearchEnabled() {
+        return getRegisteredRemoteClusterNames().isEmpty() == false;
     }
 
     boolean isRemoteNodeConnected(final String remoteCluster, final DiscoveryNode node) {
-        return remoteClusters.get(remoteCluster).isNodeConnected(node);
+        return getConnectionsMap().get(remoteCluster).isNodeConnected(node);
     }
 
     /**
@@ -263,14 +285,14 @@ public Set<String> getConfiguredClusters() {
      * Returns <code>true</code> iff the given cluster is configured as a remote cluster. Otherwise <code>false</code>
      */
     boolean isRemoteClusterRegistered(String clusterName) {
-        return remoteClusters.containsKey(clusterName);
+        return getConnectionsMap().containsKey(clusterName);
     }
 
     /**
      * Returns the registered remote cluster names.
      */
     public Set<String> getRegisteredRemoteClusterNames() {
-        return remoteClusters.keySet();
+        return getConnectionsMap().keySet();
     }
 
     /**
@@ -347,7 +369,7 @@ public RemoteClusterConnection getRemoteClusterConnection(String cluster) {
                 "this node does not have the " + DiscoveryNodeRole.REMOTE_CLUSTER_CLIENT_ROLE.roleName() + " role"
             );
         }
-        RemoteClusterConnection connection = remoteClusters.get(cluster);
+        RemoteClusterConnection connection = getConnectionsMap().get(cluster);
         if (connection == null) {
             throw new NoSuchRemoteClusterException(cluster);
         }
@@ -361,48 +383,63 @@ public void listenForUpdates(ClusterSettings clusterSettings) {
     }
 
     private synchronized void updateSkipUnavailable(String clusterAlias, Boolean skipUnavailable) {
-        RemoteClusterConnection remote = this.remoteClusters.get(clusterAlias);
+        RemoteClusterConnection remote = getConnectionsMap().get(clusterAlias);
         if (remote != null) {
             remote.setSkipUnavailable(skipUnavailable);
         }
     }
 
+    @FixForMultiProject(description = "Refactor to provide the project ID that the settings are associated with.")
     public synchronized void updateRemoteClusterCredentials(Supplier<Settings> settingsSupplier, ActionListener<Void> listener) {
+        final var projectId = projectResolver.getProjectId();
         final Settings settings = settingsSupplier.get();
         final UpdateRemoteClusterCredentialsResult result = remoteClusterCredentialsManager.updateClusterCredentials(settings);
         // We only need to rebuild connections when a credential was newly added or removed for a cluster alias, not if the credential
         // value was updated. Therefore, only consider added or removed aliases
         final int totalConnectionsToRebuild = result.addedClusterAliases().size() + result.removedClusterAliases().size();
         if (totalConnectionsToRebuild == 0) {
-            logger.debug("no connection rebuilding required after credentials update");
+            logger.debug("project ID [{}] no connection rebuilding required after credentials update", projectId);
             listener.onResponse(null);
             return;
         }
-        logger.info("rebuilding [{}] connections after credentials update", totalConnectionsToRebuild);
+        logger.info("project ID [{}] rebuilding [{}] connections after credentials update", projectId, totalConnectionsToRebuild);
         try (var connectionRefs = new RefCountingRunnable(() -> listener.onResponse(null))) {
             for (var clusterAlias : result.addedClusterAliases()) {
-                maybeRebuildConnectionOnCredentialsChange(clusterAlias, settings, connectionRefs);
+                maybeRebuildConnectionOnCredentialsChange(projectId, clusterAlias, settings, connectionRefs);
             }
             for (var clusterAlias : result.removedClusterAliases()) {
-                maybeRebuildConnectionOnCredentialsChange(clusterAlias, settings, connectionRefs);
+                maybeRebuildConnectionOnCredentialsChange(projectId, clusterAlias, settings, connectionRefs);
             }
         }
     }
 
-    // package-private for testing
-
-    private void maybeRebuildConnectionOnCredentialsChange(String clusterAlias, Settings settings, RefCountingRunnable connectionRefs) {
-        if (false == remoteClusters.containsKey(clusterAlias)) {
+    private void maybeRebuildConnectionOnCredentialsChange(
+        ProjectId projectId,
+        String clusterAlias,
+        Settings settings,
+        RefCountingRunnable connectionRefs
+    ) {
+        final var connectionsMap = getConnectionsMap(projectId);
+        if (false == connectionsMap.containsKey(clusterAlias)) {
             // A credential was added or removed before a remote connection was configured.
             // Without an existing connection, there is nothing to rebuild.
-            logger.info("no connection rebuild required for remote cluster [{}] after credentials change", clusterAlias);
+            logger.info(
+                "project ID [{}] no connection rebuild required for remote cluster [{}] after credentials change",
+                projectId,
+                clusterAlias
+            );
             return;
         }
 
-        updateRemoteCluster(clusterAlias, settings, true, ActionListener.releaseAfter(new ActionListener<>() {
+        updateRemoteCluster(projectId, clusterAlias, settings, true, ActionListener.releaseAfter(new ActionListener<>() {
             @Override
             public void onResponse(RemoteClusterConnectionStatus status) {
-                logger.info("remote cluster connection [{}] updated after credentials change: [{}]", clusterAlias, status);
+                logger.info(
+                    "project ID [{}] remote cluster connection [{}] updated after credentials change: [{}]",
+                    projectId,
+                    clusterAlias,
+                    status
+                );
             }
 
             @Override
@@ -411,23 +448,32 @@ public void onFailure(Exception e) {
                 // does *not* imply a failure to reload secure settings; however, that's how it would surface in the reload-settings call.
                 // Instead, we log a warning which is also consistent with how we handle remote cluster settings updates (logging instead of
                 // returning an error)
-                logger.warn(() -> "failed to update remote cluster connection [" + clusterAlias + "] after credentials change", e);
+                logger.warn(
+                    () -> "project ID ["
+                        + projectId
+                        + "] failed to update remote cluster connection ["
+                        + clusterAlias
+                        + "] after credentials change",
+                    e
+                );
             }
         }, connectionRefs.acquire()));
     }
 
+    @FixForMultiProject(description = "Refactor to provide the project ID that the linked alias and settings are associated with.")
     @Override
     protected void updateRemoteCluster(String clusterAlias, Settings settings) {
+        final var projectId = projectResolver.getProjectId();
         CountDownLatch latch = new CountDownLatch(1);
-        updateRemoteCluster(clusterAlias, settings, ActionListener.runAfter(new ActionListener<>() {
+        updateRemoteCluster(projectId, clusterAlias, settings, false, ActionListener.runAfter(new ActionListener<>() {
             @Override
             public void onResponse(RemoteClusterConnectionStatus status) {
-                logger.info("remote cluster connection [{}] updated: {}", clusterAlias, status);
+                logger.info("project ID [{}] remote cluster connection [{}] updated: {}", projectId, clusterAlias, status);
             }
 
             @Override
             public void onFailure(Exception e) {
-                logger.warn(() -> "failed to update remote cluster connection [" + clusterAlias + "]", e);
+                logger.warn(() -> "project ID [" + projectId + " failed to update remote cluster connection [" + clusterAlias + "]", e);
             }
         }, latch::countDown));
 
@@ -436,7 +482,12 @@ public void onFailure(Exception e) {
             // are on the cluster state thread and our custom future implementation will throw an
             // assertion.
             if (latch.await(10, TimeUnit.SECONDS) == false) {
-                logger.warn("failed to update remote cluster connection [{}] within {}", clusterAlias, TimeValue.timeValueSeconds(10));
+                logger.warn(
+                    "project ID [{}] failed to update remote cluster connection [{}] within {}",
+                    projectId,
+                    clusterAlias,
+                    TimeValue.timeValueSeconds(10)
+                );
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
@@ -451,10 +502,11 @@ public void onFailure(Exception e) {
      * @param listener a listener invoked once every configured cluster has been connected to
      */
     void updateRemoteCluster(String clusterAlias, Settings newSettings, ActionListener<RemoteClusterConnectionStatus> listener) {
-        updateRemoteCluster(clusterAlias, newSettings, false, listener);
+        updateRemoteCluster(projectResolver.getProjectId(), clusterAlias, newSettings, false, listener);
     }
 
     private synchronized void updateRemoteCluster(
+        ProjectId projectId,
         String clusterAlias,
         Settings newSettings,
         boolean forceRebuild,
@@ -464,14 +516,15 @@ private synchronized void updateRemoteCluster(
             throw new IllegalArgumentException("remote clusters must not have the empty string as its key");
         }
 
-        RemoteClusterConnection remote = this.remoteClusters.get(clusterAlias);
+        final var connectionMap = getConnectionsMap(projectId);
+        RemoteClusterConnection remote = connectionMap.get(clusterAlias);
         if (RemoteConnectionStrategy.isConnectionEnabled(clusterAlias, newSettings) == false) {
             try {
                 IOUtils.close(remote);
             } catch (IOException e) {
-                logger.warn("failed to close remote cluster connections for cluster: " + clusterAlias, e);
+                logger.warn("project ID [" + projectId + "] failed to close remote cluster connections for cluster: " + clusterAlias, e);
             }
-            remoteClusters.remove(clusterAlias);
+            connectionMap.remove(clusterAlias);
             listener.onResponse(RemoteClusterConnectionStatus.DISCONNECTED);
             return;
         }
@@ -480,19 +533,19 @@ private synchronized void updateRemoteCluster(
             // this is a new cluster we have to add a new representation
             Settings finalSettings = Settings.builder().put(this.settings, false).put(newSettings, false).build();
             remote = new RemoteClusterConnection(finalSettings, clusterAlias, transportService, remoteClusterCredentialsManager);
-            remoteClusters.put(clusterAlias, remote);
+            connectionMap.put(clusterAlias, remote);
             remote.ensureConnected(listener.map(ignored -> RemoteClusterConnectionStatus.CONNECTED));
         } else if (forceRebuild || remote.shouldRebuildConnection(newSettings)) {
             // Changes to connection configuration. Must tear down existing connection
             try {
                 IOUtils.close(remote);
             } catch (IOException e) {
-                logger.warn("failed to close remote cluster connections for cluster: " + clusterAlias, e);
+                logger.warn("project ID [" + projectId + "] failed to close remote cluster connections for cluster: " + clusterAlias, e);
             }
-            remoteClusters.remove(clusterAlias);
+            connectionMap.remove(clusterAlias);
             Settings finalSettings = Settings.builder().put(this.settings, false).put(newSettings, false).build();
             remote = new RemoteClusterConnection(finalSettings, clusterAlias, transportService, remoteClusterCredentialsManager);
-            remoteClusters.put(clusterAlias, remote);
+            connectionMap.put(clusterAlias, remote);
             remote.ensureConnected(listener.map(ignored -> RemoteClusterConnectionStatus.RECONNECTED));
         } else {
             // No changes to connection configuration.
@@ -511,7 +564,9 @@ enum RemoteClusterConnectionStatus {
      * Connects to all remote clusters in a blocking fashion. This should be called on node startup to establish an initial connection
      * to all configured seed nodes.
      */
+    @FixForMultiProject(description = "Refactor for initializing connections to linked projects for each origin project supported.")
     void initializeRemoteClusters() {
+        final var projectId = projectResolver.getProjectId();
         final TimeValue timeValue = REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
         final PlainActionFuture<Void> future = new PlainActionFuture<>();
         Set<String> enabledClusters = RemoteClusterAware.getEnabledRemoteClusters(settings);
@@ -522,7 +577,7 @@ void initializeRemoteClusters() {
 
         CountDownActionListener listener = new CountDownActionListener(enabledClusters.size(), future);
         for (String clusterAlias : enabledClusters) {
-            updateRemoteCluster(clusterAlias, settings, listener.map(ignored -> null));
+            updateRemoteCluster(projectId, clusterAlias, settings, false, listener.map(ignored -> null));
         }
 
         if (enabledClusters.isEmpty()) {
@@ -534,19 +589,19 @@ void initializeRemoteClusters() {
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
         } catch (TimeoutException ex) {
-            logger.warn("failed to connect to remote clusters within {}", timeValue.toString());
+            logger.warn("project ID [{}] failed to connect to remote clusters within {}", projectId, timeValue.toString());
         } catch (Exception e) {
-            logger.warn("failed to connect to remote clusters", e);
+            logger.warn("project ID [" + projectId + "] failed to connect to remote clusters", e);
         }
     }
 
     @Override
     public void close() throws IOException {
-        IOUtils.close(remoteClusters.values());
+        IOUtils.close(remoteClusters.values().stream().flatMap(map -> map.values().stream()).collect(Collectors.toList()));
     }
 
     public Stream<RemoteConnectionInfo> getRemoteConnectionInfos() {
-        return remoteClusters.values().stream().map(RemoteClusterConnection::getConnectionInfo);
+        return getConnectionsMap().values().stream().map(RemoteClusterConnection::getConnectionInfo);
     }
 
     @Override
@@ -568,8 +623,9 @@ public void collectNodes(Set<String> clusters, ActionListener<BiFunction<String,
                 "this node does not have the " + DiscoveryNodeRole.REMOTE_CLUSTER_CLIENT_ROLE.roleName() + " role"
             );
         }
+        final var projectConnectionsMap = getConnectionsMap();
         for (String cluster : clusters) {
-            if (this.remoteClusters.containsKey(cluster) == false) {
+            if (projectConnectionsMap.containsKey(cluster) == false) {
                 listener.onFailure(new NoSuchRemoteClusterException(cluster));
                 return;
             }
@@ -579,7 +635,14 @@ public void collectNodes(Set<String> clusters, ActionListener<BiFunction<String,
         CountDown countDown = new CountDown(clusters.size());
         Function<String, DiscoveryNode> nullFunction = s -> null;
         for (final String cluster : clusters) {
-            RemoteClusterConnection connection = this.remoteClusters.get(cluster);
+            RemoteClusterConnection connection = projectConnectionsMap.get(cluster);
+            // Ensure the connection is not null, it could have been removed since the containsKey() call above.
+            if (connection == null) {
+                if (countDown.fastForward()) {
+                    listener.onFailure(new NoSuchRemoteClusterException(cluster));
+                }
+                continue;
+            }
             connection.collectNodes(new ActionListener<Function<String, DiscoveryNode>>() {
                 @Override
                 public void onResponse(Function<String, DiscoveryNode> nodeLookup) {
@@ -654,7 +717,7 @@ public RemoteClusterClient getRemoteClusterClient(
     }
 
     Collection<RemoteClusterConnection> getConnections() {
-        return remoteClusters.values();
+        return getConnectionsMap().values();
     }
 
     static void registerRemoteClusterHandshakeRequestHandler(TransportService transportService) {