CPS usage telemetry support

Author: smalyshev

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

@@ -300,7 +300,11 @@ public void add(CCSTelemetrySnapshot stats) {
         successCount += stats.successCount;
         skippedRemotes += stats.skippedRemotes;
         stats.failureReasons.forEach((k, v) -> failureReasons.merge(k, v, Long::sum));
-        stats.featureCounts.forEach((k, v) -> featureCounts.merge(k, v, Long::sum));
+        stats.featureCounts.forEach((k, v) -> {
+            if (useMRT || k.equals(CCSUsageTelemetry.MRT_FEATURE) == false) {
+                featureCounts.merge(k, v, Long::sum);
+            }
+        });
         stats.clientCounts.forEach((k, v) -> clientCounts.merge(k, v, Long::sum));
         took.add(stats.took);
         if (useMRT) {

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

@@ -28,6 +28,11 @@ public class ClusterStatsRequest extends BaseNodesRequest {
      * Return stripped down stats for remote clusters.
      */
     private boolean remoteStats;
+    /**
+     * Are we dealing with CPS environment?
+     * In CPS, we do not display MRT stats.
+     */
+    private boolean isCPS = false;
 
     /**
      * Get stats from nodes based on the nodes ids specified. If none are passed, stats
@@ -43,13 +48,19 @@ public ClusterStatsRequest(boolean doRemotes, String... nodesIds) {
         this.remoteStats = false;
     }
 
+    public static ClusterStatsRequest newServerlessRequest(String[] nodeIds) {
+        final ClusterStatsRequest request = new ClusterStatsRequest(false, nodeIds);
+        request.isCPS = true;
+        return request;
+    }
+
     @Override
     public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
         return new CancellableTask(id, type, action, "", parentTaskId, headers);
     }
 
     public static ClusterStatsRequest newRemoteClusterStatsRequest() {
-        final var request = new ClusterStatsRequest();
+        final ClusterStatsRequest request = new ClusterStatsRequest();
         request.remoteStats = true;
         return request;
     }
@@ -67,4 +78,8 @@ public boolean doRemotes() {
     public boolean isRemoteStats() {
         return remoteStats;
     }
+
+    public boolean isCPS() {
+        return isCPS;
+    }
 }

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

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.xcontent.ToXContentFragment;
 import org.elasticsearch.xcontent.XContentBuilder;
 
@@ -55,14 +56,15 @@ public ClusterStatsResponse(
         AnalysisStats analysisStats,
         VersionStats versionStats,
         ClusterSnapshotStats clusterSnapshotStats,
-        Map<String, RemoteClusterStats> remoteClustersStats
+        Map<String, RemoteClusterStats> remoteClustersStats,
+        boolean skipMRT
     ) {
         super(clusterName, nodes, failures);
         this.clusterUUID = clusterUUID;
         this.timestamp = timestamp;
         nodesStats = new ClusterStatsNodes(nodes);
         indicesStats = new ClusterStatsIndices(nodes, mappingStats, analysisStats, versionStats);
-        ccsMetrics = new CCSTelemetrySnapshot();
+        ccsMetrics = new CCSTelemetrySnapshot().setUseMRT(skipMRT == false);
         esqlMetrics = new CCSTelemetrySnapshot().setUseMRT(false);
         ClusterHealthStatus status = null;
         for (ClusterStatsNodeResponse response : nodes) {

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

@@ -197,7 +197,8 @@ protected void newResponseAsync(
                     null,
                     null,
                     null,
-                    Map.of()
+                    Map.of(),
+                    false
                 )
                 : new ClusterStatsResponse(
                     System.currentTimeMillis(),
@@ -209,7 +210,8 @@ protected void newResponseAsync(
                     additionalStats.analysisStats(),
                     VersionStats.of(clusterService.state().metadata(), responses),
                     additionalStats.clusterSnapshotStats(),
-                    additionalStats.getRemoteStats()
+                    additionalStats.getRemoteStats(),
+                    request.isCPS()
                 )
         ).addListener(listener);
     }
@@ -440,7 +442,7 @@ public Map<String, RemoteClusterStats> getRemoteStats() {
     }
 
     private boolean doRemotes(ClusterStatsRequest request) {
-        return SearchService.CCS_COLLECT_TELEMETRY.get(settings) && request.doRemotes();
+        return SearchService.CCS_COLLECT_TELEMETRY.get(settings) && request.doRemotes() && request.isCPS() == false;
     }
 
     private class RemoteStatsFanout extends CancellableFanOut<String, RemoteClusterStatsResponse, Map<String, RemoteClusterStats>> {

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStatsAction.java

@@ -56,10 +56,9 @@ public Set<String> supportedQueryParameters() {
 
     @Override
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
-        ClusterStatsRequest clusterStatsRequest = new ClusterStatsRequest(
-            request.paramAsBoolean("include_remotes", false),
-            request.paramAsStringArray("nodeId", null)
-        );
+        ClusterStatsRequest clusterStatsRequest = request.isServerlessRequest()
+            ? ClusterStatsRequest.newServerlessRequest(request.paramAsStringArray("nodeId", null))
+            : new ClusterStatsRequest(request.paramAsBoolean("include_remotes", false), request.paramAsStringArray("nodeId", null));
         clusterStatsRequest.setTimeout(getTimeout(request));
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .cluster()

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineLimits.java

@@ -92,7 +92,7 @@ public LogicalPlan rule(Limit limit, LogicalOptimizerContext ctx) {
             // We use withLocal = false because if we have a remote join it will be forced into the fragment by the mapper anyway,
             // And the verifier checks that there are no non-synthetic limits before the join.
             // TODO: However, this means that the non-remote join will be always forced on the coordinator. We may want to revisit this.
-            return duplicateLimitAsFirstGrandchild(limit, false);
+            return duplicateLimitAsFirstGrandchild(limit, join.isRemote());
         }
         return limit;
     }

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDocTests.java

@@ -434,7 +434,8 @@ public void testToXContent() throws IOException {
             AnalysisStats.of(metadata, () -> {}),
             VersionStats.of(metadata, singletonList(mockNodeResponse)),
             ClusterSnapshotStats.EMPTY,
-            null
+            null,
+            false
         );
 
         final MonitoringDoc.Node node = new MonitoringDoc.Node("_uuid", "_host", "_addr", "_ip", "_name", 1504169190855L);