Implement CCS telemetry export as part of _cluster/stats

Author: smalyshev

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -202,6 +202,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion REPOSITORIES_TELEMETRY = def(8_732_00_0);
     public static final TransportVersion ML_INFERENCE_ALIBABACLOUD_SEARCH_ADDED = def(8_733_00_0);
 
+
+    public static final TransportVersion CCS_TELEMETRY_STATS = def(8_733_00_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

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

@@ -277,7 +277,7 @@ public int hashCode() {
      */
     public void add(CCSTelemetrySnapshot stats) {
         // This should be called in ClusterStatsResponse ctor only, so we don't need to worry about concurrency
-        if (stats.totalCount == 0) {
+        if (stats == null || stats.totalCount == 0) {
             // Just ignore the empty stats.
             // This could happen if the node is brand new or if the stats are not available, e.g. because it runs an old version.
             return;
@@ -315,7 +315,7 @@ public void add(CCSTelemetrySnapshot stats) {
      *      "p90": 2570
      * }
      */
-    public static void publishLatency(XContentBuilder builder, String name, LongMetricValue took) throws IOException {
+    private static void publishLatency(XContentBuilder builder, String name, LongMetricValue took) throws IOException {
         builder.startObject(name);
         {
             builder.field("max", took.max());

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

@@ -30,6 +30,7 @@ public class ClusterStatsNodeResponse extends BaseNodeResponse {
     private final ClusterHealthStatus clusterStatus;
     private final SearchUsageStats searchUsageStats;
     private final RepositoryUsageStats repositoryUsageStats;
+    private final CCSTelemetrySnapshot ccsMetrics;
 
     public ClusterStatsNodeResponse(StreamInput in) throws IOException {
         super(in);
@@ -47,6 +48,11 @@ public ClusterStatsNodeResponse(StreamInput in) throws IOException {
         } else {
             repositoryUsageStats = RepositoryUsageStats.EMPTY;
         }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.CCS_TELEMETRY_STATS)) {
+            ccsMetrics = new CCSTelemetrySnapshot(in);
+        } else {
+            ccsMetrics = new CCSTelemetrySnapshot();
+        }
     }
 
     public ClusterStatsNodeResponse(
@@ -56,7 +62,8 @@ public ClusterStatsNodeResponse(
         NodeStats nodeStats,
         ShardStats[] shardsStats,
         SearchUsageStats searchUsageStats,
-        RepositoryUsageStats repositoryUsageStats
+        RepositoryUsageStats repositoryUsageStats,
+        CCSTelemetrySnapshot ccsTelemetrySnapshot
     ) {
         super(node);
         this.nodeInfo = nodeInfo;
@@ -65,6 +72,7 @@ public ClusterStatsNodeResponse(
         this.clusterStatus = clusterStatus;
         this.searchUsageStats = Objects.requireNonNull(searchUsageStats);
         this.repositoryUsageStats = Objects.requireNonNull(repositoryUsageStats);
+        this.ccsMetrics = ccsTelemetrySnapshot;
     }
 
     public NodeInfo nodeInfo() {
@@ -95,6 +103,10 @@ public RepositoryUsageStats repositoryUsageStats() {
         return repositoryUsageStats;
     }
 
+    public CCSTelemetrySnapshot getCcsMetrics() {
+        return ccsMetrics;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -108,5 +120,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(TransportVersions.REPOSITORIES_TELEMETRY)) {
             repositoryUsageStats.writeTo(out);
         } // else just drop these stats, ok for bwc
+        if (out.getTransportVersion().onOrAfter(TransportVersions.CCS_TELEMETRY_STATS)) {
+            ccsMetrics.writeTo(out);
+        }
     }
+
 }

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

@@ -24,13 +24,17 @@
 import java.util.List;
 import java.util.Locale;
 
+import static org.elasticsearch.action.search.TransportSearchAction.CCS_TELEMETRY_FEATURE_FLAG;
+
 public class ClusterStatsResponse extends BaseNodesResponse<ClusterStatsNodeResponse> implements ToXContentFragment {
 
     final ClusterStatsNodes nodesStats;
     final ClusterStatsIndices indicesStats;
     final ClusterHealthStatus status;
     final ClusterSnapshotStats clusterSnapshotStats;
     final RepositoryUsageStats repositoryUsageStats;
+
+    final CCSTelemetrySnapshot ccsMetrics;
     final long timestamp;
     final String clusterUUID;
 
@@ -50,6 +54,7 @@ public ClusterStatsResponse(
         this.timestamp = timestamp;
         nodesStats = new ClusterStatsNodes(nodes);
         indicesStats = new ClusterStatsIndices(nodes, mappingStats, analysisStats, versionStats);
+        ccsMetrics = new CCSTelemetrySnapshot();
         ClusterHealthStatus status = null;
         for (ClusterStatsNodeResponse response : nodes) {
             // only the master node populates the status
@@ -58,6 +63,7 @@ public ClusterStatsResponse(
                 break;
             }
         }
+        nodes.forEach(node -> ccsMetrics.add(node.getCcsMetrics()));
         this.status = status;
         this.clusterSnapshotStats = clusterSnapshotStats;
 
@@ -90,6 +96,10 @@ public ClusterStatsIndices getIndicesStats() {
         return indicesStats;
     }
 
+    public CCSTelemetrySnapshot getCcsMetrics() {
+        return ccsMetrics;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         TransportAction.localOnly();
@@ -125,6 +135,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.field("repositories");
         repositoryUsageStats.toXContent(builder, params);
 
+        if (CCS_TELEMETRY_FEATURE_FLAG.isEnabled()) {
+            ccsMetrics.toXContent(builder, params);
+        }
+
         return builder;
     }
 

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

@@ -81,6 +81,7 @@ public class TransportClusterStatsAction extends TransportNodesAction<
     private final IndicesService indicesService;
     private final RepositoriesService repositoriesService;
     private final SearchUsageHolder searchUsageHolder;
+    private final CCSUsageTelemetry ccsUsageHolder;
 
     private final MetadataStatsCache<MappingStats> mappingStatsCache;
     private final MetadataStatsCache<AnalysisStats> analysisStatsCache;
@@ -108,6 +109,7 @@ public TransportClusterStatsAction(
         this.indicesService = indicesService;
         this.repositoriesService = repositoriesService;
         this.searchUsageHolder = usageService.getSearchUsageHolder();
+        this.ccsUsageHolder = usageService.getCcsUsageHolder();
         this.mappingStatsCache = new MetadataStatsCache<>(threadPool.getThreadContext(), MappingStats::of);
         this.analysisStatsCache = new MetadataStatsCache<>(threadPool.getThreadContext(), AnalysisStats::of);
     }
@@ -249,6 +251,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
         final SearchUsageStats searchUsageStats = searchUsageHolder.getSearchUsageStats();
 
         final RepositoryUsageStats repositoryUsageStats = repositoriesService.getUsageStats();
+        final CCSTelemetrySnapshot ccsUsage = ccsUsageHolder.getCCSTelemetrySnapshot();
 
         return new ClusterStatsNodeResponse(
             nodeInfo.getNode(),
@@ -257,7 +260,8 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
             nodeStats,
             shardsStats.toArray(new ShardStats[shardsStats.size()]),
             searchUsageStats,
-            repositoryUsageStats
+            repositoryUsageStats,
+            ccsUsage
         );
     }
 

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -127,7 +127,7 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
     public static final String FROZEN_INDICES_DEPRECATION_MESSAGE = "Searching frozen indices [{}] is deprecated."
         + " Consider cold or frozen tiers in place of frozen indices. The frozen feature will be removed in a feature release.";
 
-    private static final FeatureFlag CCS_TELEMETRY_FEATURE_FLAG = new FeatureFlag("ccs_telemetry");
+    public static final FeatureFlag CCS_TELEMETRY_FEATURE_FLAG = new FeatureFlag("ccs_telemetry");
 
     /** The maximum number of shards for a single search request. */
     public static final Setting<Long> SHARD_COUNT_LIMIT_SETTING = Setting.longSetting(

server/src/test/java/org/elasticsearch/action/admin/cluster/stats/VersionStatsTests.java

@@ -128,7 +128,8 @@ public void testCreation() {
             null,
             new ShardStats[] { shardStats },
             new SearchUsageStats(),
-            RepositoryUsageStats.EMPTY
+            RepositoryUsageStats.EMPTY,
+            null
         );
 
         stats = VersionStats.of(metadata, Collections.singletonList(nodeResponse));