Add LinkedProjectConfigService with ClusterSettings implementation

Adds a new interface LinkedProjectConfigService for registering
LinkedProjectConfigListener instances to receive linked project
configuration updates.  This change introduces an abstract base
class and a ClusterSettings based concrete class that provides
the same functionality that was previously implemented in
RemoteClusterAware and RemoteClusterService.

Sublcasses of RemoteClusterAware have been adjusted to receive
LinkedProjectConfig updates instead of Settings updates.

These changes allow for other linked project config implementations
to be used in the system, freeing consumers of the updates from
the details of the update mechanism used.

Resolves: ES-12730

Author: JeremyDahlgren

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -223,6 +223,8 @@
 import org.elasticsearch.threadpool.ExecutorBuilder;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.threadpool.internal.BuiltInExecutorBuilders;
+import org.elasticsearch.transport.ClusterSettingsLinkedProjectConfigService;
+import org.elasticsearch.transport.LinkedProjectConfigService;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.usage.UsageService;
@@ -969,6 +971,11 @@ public Map<String, String> queryFields() {
 
         final IndexingPressure indexingLimits = new IndexingPressure(settings);
 
+        final var linkedProjectConfigService = pluginsService.loadSingletonServiceProvider(
+            LinkedProjectConfigService.class,
+            () -> new ClusterSettingsLinkedProjectConfigService(settings, clusterService.getClusterSettings(), projectResolver)
+        );
+
         PluginServiceInstances pluginServices = new PluginServiceInstances(
             client,
             clusterService,
@@ -992,7 +999,8 @@ public Map<String, String> queryFields() {
             taskManager,
             projectResolver,
             slowLogFieldProvider,
-            indexingLimits
+            indexingLimits,
+            linkedProjectConfigService
         );
 
         Collection<?> pluginComponents = pluginsService.flatMap(plugin -> {
@@ -1122,6 +1130,7 @@ public Map<String, String> queryFields() {
             taskManager,
             telemetryProvider.getTracer(),
             nodeEnvironment.nodeId(),
+            linkedProjectConfigService,
             projectResolver
         );
         final SearchResponseMetrics searchResponseMetrics = new SearchResponseMetrics(telemetryProvider.getMeterRegistry());

server/src/main/java/org/elasticsearch/node/NodeServiceProvider.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.telemetry.tracing.Tracer;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.ClusterConnectionManager;
+import org.elasticsearch.transport.LinkedProjectConfigService;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportInterceptor;
 import org.elasticsearch.transport.TransportService;
@@ -121,6 +122,7 @@ TransportService newTransportService(
         TaskManager taskManager,
         Tracer tracer,
         String nodeId,
+        LinkedProjectConfigService linkedProjectConfigService,
         ProjectResolver projectResolver
     ) {
         return new TransportService(
@@ -132,6 +134,7 @@ TransportService newTransportService(
             clusterSettings,
             new ClusterConnectionManager(settings, transport, threadPool.getThreadContext()),
             taskManager,
+            linkedProjectConfigService,
             projectResolver
         );
     }

server/src/main/java/org/elasticsearch/node/PluginServiceInstances.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.telemetry.TelemetryProvider;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.LinkedProjectConfigService;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
@@ -57,5 +58,6 @@ public record PluginServiceInstances(
     TaskManager taskManager,
     ProjectResolver projectResolver,
     SlowLogFieldProvider slowLogFieldProvider,
-    IndexingPressure indexingPressure
+    IndexingPressure indexingPressure,
+    LinkedProjectConfigService linkedProjectConfigService
 ) implements Plugin.PluginServices {}

server/src/main/java/org/elasticsearch/plugins/Plugin.java

@@ -40,6 +40,7 @@
 import org.elasticsearch.telemetry.TelemetryProvider;
 import org.elasticsearch.threadpool.ExecutorBuilder;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.LinkedProjectConfigService;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParser;
@@ -192,6 +193,11 @@ public interface PluginServices {
          * Provider for indexing pressure
          */
         IndexingPressure indexingPressure();
+
+        /**
+         * A service for registering for linked project configuration updates.
+         */
+        LinkedProjectConfigService linkedProjectConfigService();
     }
 
     /**

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

@@ -0,0 +1,39 @@
+/*
+ * 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 java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+public abstract class AbstractLinkedProjectConfigService implements LinkedProjectConfigService {
+    private final List<LinkedProjectConfigListener> listeners = new CopyOnWriteArrayList<>();
+
+    @Override
+    public void register(LinkedProjectConfigListener listener) {
+        listeners.add(listener);
+    }
+
+    protected void handleUpdate(LinkedProjectConfig config) {
+        listeners.forEach(listener -> listener.updateLinkedProject(config));
+    }
+
+    protected void handleSkipUnavailableChanged(
+        ProjectId originProjectId,
+        ProjectId linkedProjectId,
+        String linkedProjectAlias,
+        boolean skipUnavailable
+    ) {
+        listeners.forEach(
+            listener -> listener.skipUnavailableChanged(originProjectId, linkedProjectId, linkedProjectAlias, skipUnavailable)
+        );
+    }
+}

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

@@ -0,0 +1,65 @@
+/*
+ * 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.project.ProjectResolver;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+
+import java.util.Collection;
+import java.util.List;
+
+public class ClusterSettingsLinkedProjectConfigService extends AbstractLinkedProjectConfigService {
+    private final Settings settings;
+    private final ProjectResolver projectResolver;
+
+    @SuppressWarnings("this-escape")
+    public ClusterSettingsLinkedProjectConfigService(Settings settings, ClusterSettings clusterSettings, ProjectResolver projectResolver) {
+        this.settings = settings;
+        this.projectResolver = projectResolver;
+        List<Setting.AffixSetting<?>> remoteClusterSettings = List.of(
+            RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS,
+            RemoteClusterSettings.REMOTE_CLUSTER_PING_SCHEDULE,
+            RemoteClusterSettings.REMOTE_CONNECTION_MODE,
+            RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTERS_PROXY,
+            RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS,
+            RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_NODE_CONNECTIONS,
+            RemoteClusterSettings.ProxyConnectionStrategySettings.PROXY_ADDRESS,
+            RemoteClusterSettings.ProxyConnectionStrategySettings.REMOTE_SOCKET_CONNECTIONS,
+            RemoteClusterSettings.ProxyConnectionStrategySettings.SERVER_NAME
+        );
+        clusterSettings.addAffixGroupUpdateConsumer(remoteClusterSettings, this::settingsChangedCallback);
+        clusterSettings.addAffixUpdateConsumer(
+            RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
+            this::skipUnavailableChangedCallback,
+            (alias, value) -> {}
+        );
+    }
+
+    @Override
+    public Collection<LinkedProjectConfig> loadAllLinkedProjectConfigs() {
+        return RemoteClusterSettings.getRemoteClusters(settings)
+            .stream()
+            .map(alias -> RemoteClusterSettings.toConfig(projectResolver.getProjectId(), ProjectId.DEFAULT, alias, settings))
+            .toList();
+    }
+
+    private void settingsChangedCallback(String clusterAlias, Settings newSettings) {
+        final var mergedSettings = Settings.builder().put(settings, false).put(newSettings, false).build();
+        final var config = RemoteClusterSettings.toConfig(projectResolver.getProjectId(), ProjectId.DEFAULT, clusterAlias, mergedSettings);
+        handleUpdate(config);
+    }
+
+    private void skipUnavailableChangedCallback(String clusterAlias, Boolean skipUnavailable) {
+        handleSkipUnavailableChanged(projectResolver.getProjectId(), ProjectId.DEFAULT, clusterAlias, skipUnavailable);
+    }
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Collections;
+
+public interface LinkedProjectConfigService {
+
+    interface LinkedProjectConfigListener {
+        void updateLinkedProject(LinkedProjectConfig config);
+
+        default void skipUnavailableChanged(
+            ProjectId originProjectId,
+            ProjectId linkedProjectId,
+            String linkedProjectAlias,
+            boolean skipUnavailable
+        ) {}
+    }
+
+    void register(LinkedProjectConfigListener listener);
+
+    Collection<LinkedProjectConfig> loadAllLinkedProjectConfigs();
+
+    LinkedProjectConfigService NOOP = new LinkedProjectConfigService() {
+        @Override
+        public void register(LinkedProjectConfigListener listener) {}
+
+        @Override
+        public Collection<LinkedProjectConfig> loadAllLinkedProjectConfigs() {
+            return Collections.emptyList();
+        }
+    };
+}

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

@@ -13,41 +13,39 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.SelectorResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.common.settings.ClusterSettings;
-import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.node.Node;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.elasticsearch.transport.RemoteClusterSettings.ProxyConnectionStrategySettings;
-import static org.elasticsearch.transport.RemoteClusterSettings.SniffConnectionStrategySettings;
-
 /**
  * Base class for all services and components that need up-to-date information about the registered remote clusters
  */
-public abstract class RemoteClusterAware {
+public abstract class RemoteClusterAware implements LinkedProjectConfigService.LinkedProjectConfigListener {
     public static final char REMOTE_CLUSTER_INDEX_SEPARATOR = ':';
     public static final String LOCAL_CLUSTER_GROUP_KEY = "";
 
     protected final Settings settings;
+    private final LinkedProjectConfigService linkedProjectConfigService;
     private final String nodeName;
     private final boolean isRemoteClusterClientEnabled;
 
     /**
      * Creates a new {@link RemoteClusterAware} instance
      * @param settings the nodes level settings
      */
-    protected RemoteClusterAware(Settings settings) {
+    protected RemoteClusterAware(Settings settings, LinkedProjectConfigService linkedProjectConfigService) {
         this.settings = settings;
+        this.linkedProjectConfigService = linkedProjectConfigService;
         this.nodeName = Node.NODE_NAME_SETTING.get(settings);
         this.isRemoteClusterClientEnabled = DiscoveryNode.isRemoteClusterClient(settings);
     }
@@ -57,10 +55,10 @@ protected String getNodeName() {
     }
 
     /**
-     * Returns remote clusters that are enabled in these settings
+     * Returns all known {@link LinkedProjectConfig}s.
      */
-    protected static Set<String> getEnabledRemoteClusters(final Settings settings) {
-        return RemoteClusterSettings.getRemoteClusters(settings);
+    protected Collection<LinkedProjectConfig> loadAllLinkedProjectConfigs() {
+        return linkedProjectConfigService.loadAllLinkedProjectConfigs();
     }
 
     /**
@@ -215,34 +213,11 @@ protected Map<String, List<String>> groupClusterIndices(Set<String> remoteCluste
         return perClusterIndices;
     }
 
-    void validateAndUpdateRemoteCluster(String clusterAlias, Settings settings) {
-        if (RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY.equals(clusterAlias)) {
-            throw new IllegalArgumentException("remote clusters must not have the empty string as its key");
-        }
-        updateRemoteCluster(clusterAlias, settings);
-    }
-
-    /**
-     * Subclasses must implement this to receive information about updated cluster aliases.
-     */
-    protected abstract void updateRemoteCluster(String clusterAlias, Settings settings);
-
     /**
-     * Registers this instance to listen to updates on the cluster settings.
+     * Registers this instance to listen for linked project updates.
      */
-    public void listenForUpdates(ClusterSettings clusterSettings) {
-        List<Setting.AffixSetting<?>> remoteClusterSettings = List.of(
-            RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS,
-            RemoteClusterSettings.REMOTE_CLUSTER_PING_SCHEDULE,
-            RemoteClusterSettings.REMOTE_CONNECTION_MODE,
-            SniffConnectionStrategySettings.REMOTE_CLUSTERS_PROXY,
-            SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS,
-            SniffConnectionStrategySettings.REMOTE_NODE_CONNECTIONS,
-            ProxyConnectionStrategySettings.PROXY_ADDRESS,
-            ProxyConnectionStrategySettings.REMOTE_SOCKET_CONNECTIONS,
-            ProxyConnectionStrategySettings.SERVER_NAME
-        );
-        clusterSettings.addAffixGroupUpdateConsumer(remoteClusterSettings, this::validateAndUpdateRemoteCluster);
+    public void listenForUpdates() {
+        linkedProjectConfigService.register(this);
     }
 
     public static String buildRemoteIndexName(String clusterAlias, String indexName) {