CPS S2D4: do not expose skip_unavailable in API responses (#133335)

S2D4 requires that we not expose the `skip_unavailable`
value in the responses to the user. This commit makes
it unavailable for the `_cluster/stats` call.

Author: pawankartik-elastic

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsRemoteIT.java

@@ -97,8 +97,8 @@ public void testRemoteClusterStats() throws ExecutionException, InterruptedExcep
             assertThat(remoteStats.get(clusterAlias).clusterUUID(), not(equalTo("")));
             assertThat(remoteStats.get(clusterAlias).mode(), oneOf("sniff", "proxy"));
         }
-        assertFalse(remoteStats.get(REMOTE1).skipUnavailable());
-        assertTrue(remoteStats.get(REMOTE2).skipUnavailable());
+        assertFalse(remoteStats.get(REMOTE1).skipUnavailable().get());
+        assertTrue(remoteStats.get(REMOTE2).skipUnavailable().get());
     }
 
     private void setupClusters() {

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/stats/ServerlessClusterStatsShouldNotExposeSkipUnavailableIT.java

@@ -0,0 +1,74 @@
+/*
+ * 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.action.admin.cluster.stats;
+
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.plugins.ClusterPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.AbstractMultiClustersTestCase;
+import org.hamcrest.Matchers;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
+
+// TODO: Move this test to the Serverless repo once the IT framework is ready there.
+public class ServerlessClusterStatsShouldNotExposeSkipUnavailableIT extends AbstractMultiClustersTestCase {
+    private static final String LINKED_CLUSTER_1 = "cluster-a";
+
+    public static class CpsPlugin extends Plugin implements ClusterPlugin {
+        @Override
+        public List<Setting<?>> getSettings() {
+            return List.of(CpsEnableSetting);
+        }
+    }
+
+    private static final Setting<String> CpsEnableSetting = Setting.simpleString(
+        "serverless.cross_project.enabled",
+        Setting.Property.NodeScope
+    );
+
+    @Override
+    protected List<String> remoteClusterAlias() {
+        return List.of(LINKED_CLUSTER_1);
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
+        return CollectionUtils.appendToCopy(super.nodePlugins(clusterAlias), CpsPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings() {
+        return Settings.builder().put(super.nodeSettings()).put("serverless.cross_project.enabled", "true").build();
+    }
+
+    @Override
+    protected Map<String, Boolean> skipUnavailableForRemoteClusters() {
+        return Map.of(LINKED_CLUSTER_1, randomBoolean());
+    }
+
+    public void testSkipUnavailableShouldNotBeExposed() throws Exception {
+        assertResponse(
+            client().execute(TransportClusterStatsAction.TYPE, new ClusterStatsRequest(/* Do include remotes */ true)),
+            result -> {
+                // In the Serverless environment, skip_unavailable should map to `Optional.empty()`.
+                assertThat(result.getRemoteClustersStats().get(LINKED_CLUSTER_1).skipUnavailable(), Matchers.is(Optional.empty()));
+                // When this result is serialised to JSON, it should not mention `skip_unavailable`.
+                assertThat(result.toString().contains("skip_unavailable"), Matchers.is(false));
+            }
+        );
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsResponse.java

@@ -26,6 +26,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 public class ClusterStatsResponse extends BaseNodesResponse<ClusterStatsNodeResponse> implements ToXContentFragment {
@@ -181,7 +182,7 @@ public String toString() {
     public record RemoteClusterStats(
         String clusterUUID,
         String mode,
-        boolean skipUnavailable,
+        Optional<Boolean> skipUnavailable,
         String transportCompress,
         Set<String> versions,
         String status,
@@ -192,7 +193,7 @@ public record RemoteClusterStats(
         long heapBytes,
         long memBytes
     ) implements ToXContentFragment {
-        public RemoteClusterStats(String mode, boolean skipUnavailable, String transportCompress) {
+        public RemoteClusterStats(String mode, Optional<Boolean> skipUnavailable, String transportCompress) {
             this(
                 "unavailable",
                 mode,
@@ -231,7 +232,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.startObject();
             builder.field("cluster_uuid", clusterUUID);
             builder.field("mode", mode);
-            builder.field("skip_unavailable", skipUnavailable);
+            if (skipUnavailable.isPresent()) {
+                builder.field("skip_unavailable", skipUnavailable.get());
+            }
             builder.field("transport.compress", transportCompress);
             builder.field("status", status);
             builder.field("version", versions);

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -497,7 +497,7 @@ RemoteClusterStats makeRemoteClusterStats(String clusterAlias) {
             var compression = RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace(clusterAlias).get(settings);
             return new RemoteClusterStats(
                 remoteConnectionInfo.getModeInfo().modeName(),
-                remoteConnection.isSkipUnavailable(),
+                remoteClusterService.isSkipUnavailable(clusterAlias),
                 compression.toString()
             );
         }