Add remote clusters xpack usage report (#94862)

This PR adds a new remote_clusters section to the xpack usage response
to report stats of remote cluster connections including total number,
mode and security model.

It also adds a new remote_cluster_server sub-section under the existing
security section.

Relates: #94817

docs/build.gradle

@@ -116,6 +116,15 @@ tasks.named("yamlRestTest").configure {
   }
 }
 
+// TODO: Remove the following when RCS feature is released
+// The get-builtin-privileges doc test does not include the new cluster privilege for RCS
+// So we disable the test if the build is a snapshot where unreleased feature is enabled by default
+tasks.named("yamlRestTest").configure {
+  if (BuildParams.isSnapshotBuild()) {
+    systemProperty 'tests.rest.blacklist', 'reference/rest-api/usage/*'
+  }
+}
+
 tasks.named("forbiddenPatterns").configure {
   exclude '**/*.mmdb'
 }

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

@@ -93,6 +93,10 @@ public final class RemoteConnectionInfo implements ToXContentFragment, Writeable
         return skipUnavailable;
     }
 
+    public boolean hasClusterCredentials() {
+        return hasClusterCredentials;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(TransportVersion.V_7_6_0)) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/RemoteClusterFeatureSetUsage.java

@@ -0,0 +1,73 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.transport.RemoteClusterPortSettings;
+import org.elasticsearch.transport.RemoteConnectionInfo;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.List;
+
+public class RemoteClusterFeatureSetUsage extends XPackFeatureSet.Usage {
+
+    private final List<RemoteConnectionInfo> remoteConnectionInfos;
+
+    public RemoteClusterFeatureSetUsage(StreamInput in) throws IOException {
+        super(in);
+        this.remoteConnectionInfos = in.readImmutableList(RemoteConnectionInfo::new);
+    }
+
+    public RemoteClusterFeatureSetUsage(List<RemoteConnectionInfo> remoteConnectionInfos) {
+        super(XPackField.REMOTE_CLUSTERS, true, true);
+        this.remoteConnectionInfos = remoteConnectionInfos;
+    }
+
+    @Override
+    public TransportVersion getMinimalSupportedVersion() {
+        return RemoteClusterPortSettings.TRANSPORT_VERSION_REMOTE_CLUSTER_SECURITY;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeList(remoteConnectionInfos);
+    }
+
+    @Override
+    protected void innerXContent(XContentBuilder builder, Params params) throws IOException {
+        final int size = remoteConnectionInfos.size();
+        builder.field("size", size);
+
+        int numberOfSniffModes = 0;
+        int numberOfConfigurableModels = 0;
+        for (var info : remoteConnectionInfos) {
+            if ("sniff".equals(info.getModeInfo().modeName())) {
+                numberOfSniffModes += 1;
+            } else {
+                assert "proxy".equals(info.getModeInfo().modeName());
+            }
+            if (info.hasClusterCredentials()) {
+                numberOfConfigurableModels += 1;
+            }
+        }
+
+        builder.startObject("mode");
+        builder.field("proxy", size - numberOfSniffModes);
+        builder.field("sniff", numberOfSniffModes);
+        builder.endObject();
+
+        builder.startObject("security");
+        builder.field("basic", size - numberOfConfigurableModels);
+        builder.field("configurable", numberOfConfigurableModels);
+        builder.endObject();
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/RemoteClusterUsageTransportAction.java

@@ -0,0 +1,58 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.protocol.xpack.XPackUsageRequest;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.RemoteClusterService;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureResponse;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureTransportAction;
+
+public class RemoteClusterUsageTransportAction extends XPackUsageFeatureTransportAction {
+
+    @Inject
+    public RemoteClusterUsageTransportAction(
+        TransportService transportService,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ActionFilters actionFilters,
+        IndexNameExpressionResolver indexNameExpressionResolver
+    ) {
+        super(
+            XPackUsageFeatureAction.REMOTE_CLUSTERS.name(),
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            indexNameExpressionResolver
+        );
+    }
+
+    @Override
+    protected void masterOperation(
+        Task task,
+        XPackUsageRequest request,
+        ClusterState state,
+        ActionListener<XPackUsageFeatureResponse> listener
+    ) throws Exception {
+        final RemoteClusterService remoteClusterService = transportService.getRemoteClusterService();
+
+        listener.onResponse(
+            new XPackUsageFeatureResponse(new RemoteClusterFeatureSetUsage(remoteClusterService.getRemoteConnectionInfos().toList()))
+        );
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -555,7 +555,9 @@ public class XPackClientPlugin extends Plugin implements ActionPlugin, NetworkPl
             // TSDB Downsampling
             new NamedWriteableRegistry.Entry(LifecycleAction.class, DownsampleAction.NAME, DownsampleAction::new),
             // Health API usage
-            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.HEALTH_API, HealthApiFeatureSetUsage::new)
+            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.HEALTH_API, HealthApiFeatureSetUsage::new),
+            // Remote cluster usage
+            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.REMOTE_CLUSTERS, RemoteClusterFeatureSetUsage::new)
         );
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackField.java

@@ -75,6 +75,7 @@ public final class XPackField {
     public static final String ARCHIVE = "archive";
     /** Name constant for the health api feature. */
     public static final String HEALTH_API = "health_api";
+    public static final String REMOTE_CLUSTERS = "remote_clusters";
 
     private XPackField() {}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackPlugin.java

@@ -342,6 +342,7 @@ public class XPackPlugin extends XPackClientPlugin
         actions.add(new ActionHandler<>(XPackUsageFeatureAction.DATA_STREAMS, DataStreamUsageTransportAction.class));
         actions.add(new ActionHandler<>(XPackInfoFeatureAction.DATA_STREAMS, DataStreamInfoTransportAction.class));
         actions.add(new ActionHandler<>(XPackUsageFeatureAction.HEALTH, HealthApiUsageTransportAction.class));
+        actions.add(new ActionHandler<>(XPackUsageFeatureAction.REMOTE_CLUSTERS, RemoteClusterUsageTransportAction.class));
         return actions;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/XPackUsageFeatureAction.java

@@ -7,9 +7,12 @@
 package org.elasticsearch.xpack.core.action;
 
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.xpack.core.XPackField;
 
 import java.util.List;
+import java.util.Objects;
+import java.util.stream.Stream;
 
 /**
  * A base action for usage of a feature plugin.
@@ -46,8 +49,9 @@ public class XPackUsageFeatureAction extends ActionType<XPackUsageFeatureRespons
     public static final XPackUsageFeatureAction AGGREGATE_METRIC = new XPackUsageFeatureAction(XPackField.AGGREGATE_METRIC);
     public static final XPackUsageFeatureAction ARCHIVE = new XPackUsageFeatureAction(XPackField.ARCHIVE);
     public static final XPackUsageFeatureAction HEALTH = new XPackUsageFeatureAction(XPackField.HEALTH_API);
+    public static final XPackUsageFeatureAction REMOTE_CLUSTERS = new XPackUsageFeatureAction(XPackField.REMOTE_CLUSTERS);
 
-    static final List<XPackUsageFeatureAction> ALL = List.of(
+    static final List<XPackUsageFeatureAction> ALL = Stream.of(
         AGGREGATE_METRIC,
         ANALYTICS,
         CCR,
@@ -70,8 +74,9 @@ public class XPackUsageFeatureAction extends ActionType<XPackUsageFeatureRespons
         VOTING_ONLY,
         WATCHER,
         ARCHIVE,
-        HEALTH
-    );
+        HEALTH,
+        TcpTransport.isUntrustedRemoteClusterEnabled() ? REMOTE_CLUSTERS : null
+    ).filter(Objects::nonNull).toList();
 
     // public for testing
     public XPackUsageFeatureAction(String name) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/SecurityFeatureSetUsage.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.core.security;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.transport.RemoteClusterPortSettings;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.core.XPackFeatureSet;
 import org.elasticsearch.xpack.core.XPackField;
@@ -32,6 +33,7 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
     private static final String OPERATOR_PRIVILEGES_XFIELD = XPackField.OPERATOR_PRIVILEGES;
     private static final String DOMAINS_XFIELD = "domains";
     private static final String USER_PROFILE_XFIELD = "user_profile";
+    private static final String REMOTE_CLUSTER_SERVER_XFIELD = "remote_cluster_server";
 
     private Map<String, Object> realmsUsage;
     private Map<String, Object> rolesStoreUsage;
@@ -46,6 +48,7 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
     private Map<String, Object> operatorPrivilegesUsage;
     private Map<String, Object> domainsUsage;
     private Map<String, Object> userProfileUsage;
+    private Map<String, Object> remoteClusterServerUsage;
 
     public SecurityFeatureSetUsage(StreamInput in) throws IOException {
         super(in);
@@ -72,6 +75,9 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
         if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_5_0)) {
             userProfileUsage = in.readMap();
         }
+        if (in.getTransportVersion().onOrAfter(RemoteClusterPortSettings.TRANSPORT_VERSION_REMOTE_CLUSTER_SECURITY)) {
+            remoteClusterServerUsage = in.readMap();
+        }
     }
 
     public SecurityFeatureSetUsage(
@@ -88,7 +94,8 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
         Map<String, Object> fips140Usage,
         Map<String, Object> operatorPrivilegesUsage,
         Map<String, Object> domainsUsage,
-        Map<String, Object> userProfileUsage
+        Map<String, Object> userProfileUsage,
+        Map<String, Object> remoteClusterServerUsage
     ) {
         super(XPackField.SECURITY, true, enabled);
         this.realmsUsage = realmsUsage;
@@ -104,6 +111,7 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
         this.operatorPrivilegesUsage = operatorPrivilegesUsage;
         this.domainsUsage = domainsUsage;
         this.userProfileUsage = userProfileUsage;
+        this.remoteClusterServerUsage = remoteClusterServerUsage;
     }
 
     @Override
@@ -137,6 +145,9 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
         if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_5_0)) {
             out.writeGenericMap(userProfileUsage);
         }
+        if (out.getTransportVersion().onOrAfter(RemoteClusterPortSettings.TRANSPORT_VERSION_REMOTE_CLUSTER_SECURITY)) {
+            out.writeGenericMap(remoteClusterServerUsage);
+        }
     }
 
     @Override
@@ -160,6 +171,9 @@ public class SecurityFeatureSetUsage extends XPackFeatureSet.Usage {
             if (userProfileUsage != null && false == userProfileUsage.isEmpty()) {
                 builder.field(USER_PROFILE_XFIELD, userProfileUsage);
             }
+            if (remoteClusterServerUsage != null && false == remoteClusterServerUsage.isEmpty()) {
+                builder.field(REMOTE_CLUSTER_SERVER_XFIELD, remoteClusterServerUsage);
+            }
         } else if (sslUsage.isEmpty() == false) {
             // A trial (or basic) license can have SSL without security.
             // This is because security defaults to disabled on that license, but that dynamic-default does not disable SSL.

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/RemoteClusterFeatureSetUsageTests.java

@@ -0,0 +1,87 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core;
+
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.ProxyConnectionStrategy;
+import org.elasticsearch.transport.RemoteConnectionInfo;
+import org.elasticsearch.transport.SniffConnectionStrategy;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.json.JsonXContent;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class RemoteClusterFeatureSetUsageTests extends ESTestCase {
+
+    public void testToXContent() throws IOException {
+        final int numberOfRemoteClusters = randomIntBetween(0, 10);
+        int numberOfSniffModes = 0;
+        int numberOfConfigurableModels = 0;
+        final List<RemoteConnectionInfo> infos = new ArrayList<>();
+        for (int i = 0; i < numberOfRemoteClusters; i++) {
+            final boolean hasCredentials = randomBoolean();
+            if (hasCredentials) {
+                numberOfConfigurableModels += 1;
+            }
+            final RemoteConnectionInfo.ModeInfo modeInfo;
+            if (randomBoolean()) {
+                modeInfo = new SniffConnectionStrategy.SniffModeInfo(List.of(), randomIntBetween(1, 8), randomIntBetween(1, 8));
+                numberOfSniffModes += 1;
+            } else {
+                modeInfo = new ProxyConnectionStrategy.ProxyModeInfo(
+                    randomAlphaOfLengthBetween(3, 8),
+                    randomAlphaOfLengthBetween(3, 8),
+                    randomIntBetween(1, 8),
+                    randomIntBetween(1, 8)
+                );
+            }
+            infos.add(
+                new RemoteConnectionInfo(randomAlphaOfLengthBetween(3, 8), modeInfo, TimeValue.ZERO, randomBoolean(), hasCredentials)
+            );
+        }
+
+        try (XContentBuilder builder = JsonXContent.contentBuilder()) {
+            new RemoteClusterFeatureSetUsage(infos).toXContent(builder, ToXContent.EMPTY_PARAMS);
+            assertThat(
+                Strings.toString(builder),
+                equalTo(
+                    XContentHelper.stripWhitespace(
+                        Strings.format(
+                            """
+                                {
+                                  "size": %s,
+                                  "mode": {
+                                    "proxy": %s,
+                                    "sniff": %s
+                                  },
+                                  "security": {
+                                     "basic": %s,
+                                     "configurable": %s
+                                  }
+                                }""",
+                            numberOfRemoteClusters,
+                            numberOfRemoteClusters - numberOfSniffModes,
+                            numberOfSniffModes,
+                            numberOfRemoteClusters - numberOfConfigurableModels,
+                            numberOfConfigurableModels
+                        )
+                    )
+                )
+            );
+        }
+
+    }
+}

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/Constants.java

@@ -354,6 +354,7 @@ public class Constants {
         "cluster:monitor/xpack/usage/logstash",
         "cluster:monitor/xpack/usage/ml",
         "cluster:monitor/xpack/usage/monitoring",
+        "cluster:monitor/xpack/usage/remote_clusters",
         "cluster:monitor/xpack/usage/rollup",
         "cluster:monitor/xpack/usage/searchable_snapshots",
         "cluster:monitor/xpack/usage/security",

x-pack/plugin/security/qa/security-trial/src/javaRestTest/java/org/elasticsearch/xpack/security/crossclusteraccess/CrossClusterAccessHeadersForCcsRestIT.java

@@ -18,12 +18,17 @@ import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsAction;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsGroup;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsRequest;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsResponse;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.action.search.SearchAction;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.SecureString;
@@ -35,6 +40,7 @@ import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
 import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.internal.InternalSearchResponse;
+import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.rest.ObjectPath;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.threadpool.TestThreadPool;
@@ -76,8 +82,10 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.emptyOrNullString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 
 public class CrossClusterAccessHeadersForCcsRestIT extends SecurityOnTrialLicenseRestTestCase {
     @BeforeClass
@@ -705,6 +713,52 @@ public class CrossClusterAccessHeadersForCcsRestIT extends SecurityOnTrialLicens
         }
     }
 
+    public void testRemoteClustersXPackUsage() throws IOException {
+        try (MockTransportService remoteTransport = startTransport("remoteNodeA", threadPool, ConcurrentCollections.newBlockingQueue())) {
+            final TransportAddress transportPortAddress = remoteTransport.getOriginalTransport().boundAddress().publishAddress();
+            final TransportAddress remoteClusterServerPortAddress = remoteTransport.getOriginalTransport()
+                .boundRemoteIngressAddress()
+                .publishAddress();
+            final int numberOfRemoteClusters = randomIntBetween(0, 5);
+            final int numberOfConfigurables = randomIntBetween(0, Math.min(2, numberOfRemoteClusters));
+            final int numberOfBasics = numberOfRemoteClusters - numberOfConfigurables;
+            final List<Boolean> useProxyModes = randomList(numberOfRemoteClusters, numberOfRemoteClusters, ESTestCase::randomBoolean);
+
+            // Remote clusters with new configurable model
+            switch (numberOfConfigurables) {
+                case 0 -> {}
+                case 1 -> setupClusterSettings(CLUSTER_A, remoteClusterServerPortAddress, useProxyModes.get(0));
+                case 2 -> {
+                    setupClusterSettings(CLUSTER_A, remoteClusterServerPortAddress, useProxyModes.get(0));
+                    setupClusterSettings(CLUSTER_B, remoteClusterServerPortAddress, useProxyModes.get(1));
+                }
+                default -> throw new IllegalArgumentException("invalid number of configurable remote clusters");
+            }
+
+            // Remote clusters with basic model
+            for (int i = 0; i < numberOfBasics; i++) {
+                setupClusterSettings("basic_cluster_" + i, transportPortAddress, useProxyModes.get(i + numberOfConfigurables));
+            }
+
+            final Request xPackUsageRequest = new Request("GET", "/_xpack/usage");
+            final Response xPackUsageResponse = adminClient().performRequest(xPackUsageRequest);
+            assertOK(xPackUsageResponse);
+            final ObjectPath path = ObjectPath.createFromResponse(xPackUsageResponse);
+
+            assertThat(path.evaluate("remote_clusters.size"), equalTo(numberOfRemoteClusters));
+            final int numberOfProxyModes = (int) useProxyModes.stream().filter(e -> e).count();
+            assertThat(path.evaluate("remote_clusters.mode.proxy"), equalTo(numberOfProxyModes));
+            assertThat(path.evaluate("remote_clusters.mode.sniff"), equalTo(numberOfRemoteClusters - numberOfProxyModes));
+            assertThat(path.evaluate("remote_clusters.security.basic"), equalTo(numberOfBasics));
+            assertThat(path.evaluate("remote_clusters.security.configurable"), equalTo(numberOfConfigurables));
+
+            assertThat(path.evaluate("security.remote_cluster_server.available"), is(true));
+            assertThat(path.evaluate("security.remote_cluster_server.enabled"), is(false));
+            assertThat(path.evaluate("security.ssl.remote_cluster_server.enabled"), nullValue());
+            assertThat(path.evaluate("security.ssl.remote_cluster_client.enabled"), is(false));
+        }
+    }
+
     private void testCcsWithApiKeyCrossClusterAccessAuthenticationAgainstSingleCluster(
         String cluster,
         String apiKeyEncoded,
@@ -980,6 +1034,19 @@ public class CrossClusterAccessHeadersForCcsRestIT extends SecurityOnTrialLicens
             null
         );
         try {
+            service.registerRequestHandler(
+                ClusterStateAction.NAME,
+                ThreadPool.Names.SAME,
+                ClusterStateRequest::new,
+                (request, channel, task) -> {
+                    capturedHeaders.add(
+                        new CapturedActionWithHeaders(task.getAction(), Map.copyOf(threadPool.getThreadContext().getHeaders()))
+                    );
+                    channel.sendResponse(
+                        new ClusterStateResponse(ClusterName.DEFAULT, ClusterState.builder(ClusterName.DEFAULT).build(), false)
+                    );
+                }
+            );
             service.registerRequestHandler(
                 RemoteClusterNodesAction.NAME,
                 ThreadPool.Names.SAME,

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/SecurityUsageTransportAction.java

@@ -21,6 +21,7 @@ import org.elasticsearch.protocol.xpack.XPackUsageRequest;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteClusterPortSettings;
+import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
@@ -45,9 +46,11 @@ import static java.util.Collections.singletonMap;
 import static org.elasticsearch.xpack.core.XPackSettings.API_KEY_SERVICE_ENABLED_SETTING;
 import static org.elasticsearch.xpack.core.XPackSettings.FIPS_MODE_ENABLED;
 import static org.elasticsearch.xpack.core.XPackSettings.HTTP_SSL_ENABLED;
+import static org.elasticsearch.xpack.core.XPackSettings.REMOTE_CLUSTER_CLIENT_SSL_ENABLED;
 import static org.elasticsearch.xpack.core.XPackSettings.REMOTE_CLUSTER_SERVER_SSL_ENABLED;
 import static org.elasticsearch.xpack.core.XPackSettings.TOKEN_SERVICE_ENABLED_SETTING;
 import static org.elasticsearch.xpack.core.XPackSettings.TRANSPORT_SSL_ENABLED;
+import static org.elasticsearch.xpack.security.Security.CONFIGURABLE_CROSS_CLUSTER_ACCESS_FEATURE;
 
 public class SecurityUsageTransportAction extends XPackUsageFeatureTransportAction {
 
@@ -132,7 +135,8 @@ public class SecurityUsageTransportAction extends XPackUsageFeatureTransportActi
                     fips140Usage,
                     operatorPrivilegesUsage,
                     domainsUsageRef.get(),
-                    userProfileUsageRef.get()
+                    userProfileUsageRef.get(),
+                    remoteClusterServerUsage()
                 );
                 listener.onResponse(new XPackUsageFeatureResponse(usage));
             }
@@ -183,6 +187,19 @@ public class SecurityUsageTransportAction extends XPackUsageFeatureTransportActi
         }
     }
 
+    private Map<String, Object> remoteClusterServerUsage() {
+        if (TcpTransport.isUntrustedRemoteClusterEnabled() && XPackSettings.SECURITY_ENABLED.get(settings)) {
+            return Map.of(
+                "available",
+                CONFIGURABLE_CROSS_CLUSTER_ACCESS_FEATURE.checkWithoutTracking(licenseState),
+                "enabled",
+                RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED.get(settings)
+            );
+        } else {
+            return Map.of();
+        }
+    }
+
     static Map<String, Object> sslUsage(Settings settings) {
         // If security has been explicitly disabled in the settings, then SSL is also explicitly disabled, and we don't want to report
         // these http/transport settings as they would be misleading (they could report `true` even though they were ignored)
@@ -190,8 +207,11 @@ public class SecurityUsageTransportAction extends XPackUsageFeatureTransportActi
             Map<String, Object> map = Maps.newMapWithExpectedSize(2);
             map.put("http", singletonMap("enabled", HTTP_SSL_ENABLED.get(settings)));
             map.put("transport", singletonMap("enabled", TRANSPORT_SSL_ENABLED.get(settings)));
-            if (RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED.get(settings)) {
-                map.put("remote_cluster_server", singletonMap("enabled", REMOTE_CLUSTER_SERVER_SSL_ENABLED.get(settings)));
+            if (TcpTransport.isUntrustedRemoteClusterEnabled()) {
+                if (RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED.get(settings)) {
+                    map.put("remote_cluster_server", singletonMap("enabled", REMOTE_CLUSTER_SERVER_SSL_ENABLED.get(settings)));
+                }
+                map.put("remote_cluster_client", singletonMap("enabled", REMOTE_CLUSTER_CLIENT_SSL_ENABLED.get(settings)));
             }
             return map;
         } else {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityInfoTransportActionTests.java

@@ -14,6 +14,7 @@ import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.license.MockLicenseState;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -31,7 +32,6 @@ import org.elasticsearch.xpack.security.authc.support.mapper.NativeRoleMappingSt
 import org.elasticsearch.xpack.security.authz.store.CompositeRolesStore;
 import org.elasticsearch.xpack.security.profile.ProfileService;
 import org.elasticsearch.xpack.security.transport.filter.IPFilter;
-import org.hamcrest.Matchers;
 import org.junit.Before;
 
 import java.io.IOException;
@@ -102,6 +102,8 @@ public class SecurityInfoTransportActionTests extends ESTestCase {
         final boolean enabled = explicitlyDisabled == false;
         final boolean operatorPrivilegesAvailable = randomBoolean();
         when(licenseState.isAllowed(Security.OPERATOR_PRIVILEGES_FEATURE)).thenReturn(operatorPrivilegesAvailable);
+        final boolean remoteClusterServerAvailable = randomBoolean();
+        when(licenseState.isAllowed(Security.CONFIGURABLE_CROSS_CLUSTER_ACCESS_FEATURE)).thenReturn(remoteClusterServerAvailable);
 
         Settings.Builder settings = Settings.builder().put(this.settings);
 
@@ -114,10 +116,12 @@ public class SecurityInfoTransportActionTests extends ESTestCase {
         settings.put("xpack.security.transport.ssl.enabled", transportSSLEnabled);
 
         // Remote cluster server requires security to be enabled
-        final boolean remoteClusterPortEnabled = explicitlyDisabled ? false : randomBoolean();
-        settings.put("remote_cluster_server.enabled", remoteClusterPortEnabled);
-        final boolean remoteClusterSslEnabled = randomBoolean();
-        settings.put("xpack.security.remote_cluster_server.ssl.enabled", remoteClusterSslEnabled);
+        final boolean remoteClusterServerEnabled = explicitlyDisabled ? false : randomBoolean();
+        settings.put("remote_cluster_server.enabled", remoteClusterServerEnabled);
+        final boolean remoteClusterServerSslEnabled = randomBoolean();
+        settings.put("xpack.security.remote_cluster_server.ssl.enabled", remoteClusterServerSslEnabled);
+        final boolean remoteClusterClientSslEnabled = randomBoolean();
+        settings.put("xpack.security.remote_cluster_client.ssl.enabled", remoteClusterClientSslEnabled);
 
         boolean configureEnabledFlagForTokenService = randomBoolean();
         final boolean tokenServiceEnabled;
@@ -265,18 +269,21 @@ public class SecurityInfoTransportActionTests extends ESTestCase {
                 assertThat(source.getValue("user_profile.total"), equalTo(userProfileUsage.get("total")));
                 assertThat(source.getValue("user_profile.enabled"), equalTo(userProfileUsage.get("enabled")));
                 assertThat(source.getValue("user_profile.recent"), equalTo(userProfileUsage.get("recent")));
-            } else {
-                if (explicitlyDisabled) {
-                    assertThat(source.getValue("ssl"), is(nullValue()));
-                } else {
-                    assertThat(source.getValue("ssl.http.enabled"), is(httpSSLEnabled));
-                    assertThat(source.getValue("ssl.transport.enabled"), is(transportSSLEnabled));
-                    if (remoteClusterPortEnabled) {
-                        assertThat(source.getValue("ssl.remote_cluster_server.enabled"), is(remoteClusterSslEnabled));
+
+                assertThat(source.getValue("ssl.http.enabled"), is(httpSSLEnabled));
+                assertThat(source.getValue("ssl.transport.enabled"), is(transportSSLEnabled));
+                if (TcpTransport.isUntrustedRemoteClusterEnabled()) {
+                    if (remoteClusterServerEnabled) {
+                        assertThat(source.getValue("ssl.remote_cluster_server.enabled"), is(remoteClusterServerSslEnabled));
                     } else {
                         assertThat(source.getValue("ssl.remote_cluster_server.enabled"), nullValue());
                     }
+                    assertThat(source.getValue("ssl.remote_cluster_client.enabled"), is(remoteClusterClientSslEnabled));
+                    assertThat(source.getValue("remote_cluster_server.available"), is(remoteClusterServerAvailable));
+                    assertThat(source.getValue("remote_cluster_server.enabled"), is(remoteClusterServerEnabled));
                 }
+            } else {
+                assertThat(source.getValue("ssl"), is(nullValue()));
                 assertThat(source.getValue("realms"), is(nullValue()));
                 assertThat(source.getValue("token_service"), is(nullValue()));
                 assertThat(source.getValue("api_key_service"), is(nullValue()));
@@ -285,7 +292,8 @@ public class SecurityInfoTransportActionTests extends ESTestCase {
                 assertThat(source.getValue("ipfilter"), is(nullValue()));
                 assertThat(source.getValue("roles"), is(nullValue()));
                 assertThat(source.getValue("operator_privileges"), is(nullValue()));
-                assertThat(source.getValue("user_profile"), is(Matchers.nullValue()));
+                assertThat(source.getValue("user_profile"), is(nullValue()));
+                assertThat(source.getValue("remote_cluster_server"), is(nullValue()));
             }
         }
     }