Add cluster manager task throttling stats in nodes stats API

CHANGELOG.md

@@ -105,6 +105,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Added support to apply index create block ([#4603](https://github.com/opensearch-project/OpenSearch/issues/4603))
 - Support request level durability for remote-backed indexes ([#5671](https://github.com/opensearch-project/OpenSearch/issues/5671))
 - Added new level to get health per awareness attribute in _cluster/health ([#5694](https://github.com/opensearch-project/OpenSearch/pull/5694))
+- Added cluster manager throttling stats in nodes/stats API ([#5790](https://github.com/opensearch-project/OpenSearch/pull/5790))
 
 ### Dependencies
 - Bumps `bcpg-fips` from 1.0.5.1 to 1.0.7.1

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodeStats.java

@@ -36,6 +36,7 @@
 import org.opensearch.action.support.nodes.BaseNodeResponse;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodeRole;
+import org.opensearch.cluster.service.ClusterManagerThrottlingStats;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.io.stream.StreamInput;
 import org.opensearch.common.io.stream.StreamOutput;
@@ -122,6 +123,9 @@ public class NodeStats extends BaseNodeResponse implements ToXContentFragment {
     @Nullable
     private SearchBackpressureStats searchBackpressureStats;
 
+    @Nullable
+    private ClusterManagerThrottlingStats clusterManagerThrottlingStats;
+
     public NodeStats(StreamInput in) throws IOException {
         super(in);
         timestamp = in.readVLong();
@@ -152,6 +156,12 @@ public NodeStats(StreamInput in) throws IOException {
         } else {
             searchBackpressureStats = null;
         }
+
+        if (in.getVersion().onOrAfter(Version.V_2_5_0)) {
+            clusterManagerThrottlingStats = in.readOptionalWriteable(ClusterManagerThrottlingStats::new);
+        } else {
+            clusterManagerThrottlingStats = null;
+        }
     }
 
     public NodeStats(
@@ -173,7 +183,8 @@ public NodeStats(
         @Nullable ScriptCacheStats scriptCacheStats,
         @Nullable IndexingPressureStats indexingPressureStats,
         @Nullable ShardIndexingPressureStats shardIndexingPressureStats,
-        @Nullable SearchBackpressureStats searchBackpressureStats
+        @Nullable SearchBackpressureStats searchBackpressureStats,
+        @Nullable ClusterManagerThrottlingStats clusterManagerThrottlingStats
     ) {
         super(node);
         this.timestamp = timestamp;
@@ -194,6 +205,7 @@ public NodeStats(
         this.indexingPressureStats = indexingPressureStats;
         this.shardIndexingPressureStats = shardIndexingPressureStats;
         this.searchBackpressureStats = searchBackpressureStats;
+        this.clusterManagerThrottlingStats = clusterManagerThrottlingStats;
     }
 
     public long getTimestamp() {
@@ -308,6 +320,11 @@ public SearchBackpressureStats getSearchBackpressureStats() {
         return searchBackpressureStats;
     }
 
+    @Nullable
+    public ClusterManagerThrottlingStats getClusterManagerThrottlingStats() {
+        return clusterManagerThrottlingStats;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -336,6 +353,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_2_4_0)) {
             out.writeOptionalWriteable(searchBackpressureStats);
         }
+        if (out.getVersion().onOrAfter(Version.V_2_5_0)) {
+            out.writeOptionalWriteable(clusterManagerThrottlingStats);
+        }
     }
 
     @Override
@@ -411,6 +431,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (getSearchBackpressureStats() != null) {
             getSearchBackpressureStats().toXContent(builder, params);
         }
+        if (getClusterManagerThrottlingStats() != null) {
+            getClusterManagerThrottlingStats().toXContent(builder, params);
+        }
         return builder;
     }
 }

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodesStatsRequest.java

@@ -207,7 +207,8 @@ public enum Metric {
         SCRIPT_CACHE("script_cache"),
         INDEXING_PRESSURE("indexing_pressure"),
         SHARD_INDEXING_PRESSURE("shard_indexing_pressure"),
-        SEARCH_BACKPRESSURE("search_backpressure");
+        SEARCH_BACKPRESSURE("search_backpressure"),
+        CLUSTER_MANAGER_THROTTLING("cluster_manager_throttling");
 
         private String metricName;
 

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -119,7 +119,8 @@ protected NodeStats nodeOperation(NodeStatsRequest nodeStatsRequest) {
             NodesStatsRequest.Metric.SCRIPT_CACHE.containedIn(metrics),
             NodesStatsRequest.Metric.INDEXING_PRESSURE.containedIn(metrics),
             NodesStatsRequest.Metric.SHARD_INDEXING_PRESSURE.containedIn(metrics),
-            NodesStatsRequest.Metric.SEARCH_BACKPRESSURE.containedIn(metrics)
+            NodesStatsRequest.Metric.SEARCH_BACKPRESSURE.containedIn(metrics),
+            NodesStatsRequest.Metric.CLUSTER_MANAGER_THROTTLING.containedIn(metrics)
         );
     }
 

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

@@ -163,6 +163,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
             false,
             false,
             false,
+            false,
             false
         );
         List<ShardStats> shardsStats = new ArrayList<>();

server/src/main/java/org/opensearch/cluster/service/ClusterManagerThrottlingStats.java

@@ -8,18 +8,29 @@
 
 package org.opensearch.cluster.service;
 
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.common.io.stream.Writeable;
 import org.opensearch.common.metrics.CounterMetric;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.ToXContentFragment;
+import org.opensearch.common.xcontent.XContentBuilder;
 
+import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Contains stats of Cluster Manager Task Throttling.
  * It stores the total cumulative count of throttled tasks per task type.
  */
-public class ClusterManagerThrottlingStats implements ClusterManagerTaskThrottlerListener {
+public class ClusterManagerThrottlingStats implements ClusterManagerTaskThrottlerListener, Writeable, ToXContentFragment {
 
-    private Map<String, CounterMetric> throttledTasksCount = new ConcurrentHashMap<>();
+    private Map<String, CounterMetric> throttledTasksCount;
+
+    public ClusterManagerThrottlingStats() {
+        throttledTasksCount = new ConcurrentHashMap<>();
+    }
 
     private void incrementThrottlingCount(String type, final int counts) {
         throttledTasksCount.computeIfAbsent(type, k -> new CounterMetric()).inc(counts);
@@ -39,4 +50,38 @@ public long getTotalThrottledTaskCount() {
     public void onThrottle(String type, int counts) {
         incrementThrottlingCount(type, counts);
     }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeInt(throttledTasksCount.size());
+        for (Map.Entry<String, CounterMetric> entry : throttledTasksCount.entrySet()) {
+            out.writeString(entry.getKey());
+            out.writeLong(entry.getValue().count());
+        }
+    }
+
+    public ClusterManagerThrottlingStats(StreamInput in) throws IOException {
+        int throttledTaskEntries = in.readInt();
+        throttledTasksCount = new ConcurrentHashMap<>();
+        for (int i = 0; i < throttledTaskEntries; i++) {
+            String taskType = in.readString();
+            long throttledTaskCount = in.readLong();
+            onThrottle(taskType, (int) throttledTaskCount);
+        }
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
+        builder.startObject("cluster_manager_throttling");
+        builder.startObject("cluster_manager_stats");
+        builder.field("TotalThrottledTasks", getTotalThrottledTaskCount());
+        builder.startObject("ThrottledTasksPerTaskType");
+        for (Map.Entry<String, CounterMetric> entry : throttledTasksCount.entrySet()) {
+            builder.field(entry.getKey(), entry.getValue().count());
+        }
+        builder.endObject();
+        builder.endObject();
+        return builder.endObject();
+    }
+
 }

server/src/main/java/org/opensearch/cluster/service/MasterService.java

@@ -606,6 +606,13 @@ public long numberOfThrottledPendingTasks() {
         return throttlingStats.getTotalThrottledTaskCount();
     }
 
+    /**
+     * Returns the stats of throttled pending tasks.
+     */
+    public ClusterManagerThrottlingStats getThrottlingStats() {
+        return throttlingStats;
+    }
+
     /**
      * Returns the min version of nodes in cluster
      */

server/src/main/java/org/opensearch/node/NodeService.java

@@ -83,6 +83,7 @@ public class NodeService implements Closeable {
     private final IndexingPressureService indexingPressureService;
     private final AggregationUsageService aggregationUsageService;
     private final SearchBackpressureService searchBackpressureService;
+    private final ClusterService clusterService;
 
     private final Discovery discovery;
 
@@ -123,6 +124,7 @@ public class NodeService implements Closeable {
         this.indexingPressureService = indexingPressureService;
         this.aggregationUsageService = aggregationUsageService;
         this.searchBackpressureService = searchBackpressureService;
+        this.clusterService = clusterService;
         clusterService.addStateApplier(ingestService);
     }
 
@@ -174,7 +176,8 @@ public NodeStats stats(
         boolean scriptCache,
         boolean indexingPressure,
         boolean shardIndexingPressure,
-        boolean searchBackpressure
+        boolean searchBackpressure,
+        boolean clusterManagerThrottling
     ) {
         // for indices stats we want to include previous allocated shards stats as well (it will
         // only be applied to the sensible ones to use, like refresh/merge/flush/indexing stats)
@@ -197,7 +200,8 @@ public NodeStats stats(
             scriptCache ? scriptService.cacheStats() : null,
             indexingPressure ? this.indexingPressureService.nodeStats() : null,
             shardIndexingPressure ? this.indexingPressureService.shardStats(indices) : null,
-            searchBackpressure ? this.searchBackpressureService.nodeStats() : null
+            searchBackpressure ? this.searchBackpressureService.nodeStats() : null,
+            clusterManagerThrottling ? this.clusterService.getClusterManagerService().getThrottlingStats() : null
         );
     }
 

test/framework/src/main/java/org/opensearch/cluster/MockInternalClusterInfoService.java

@@ -115,7 +115,8 @@ List<NodeStats> adjustNodesStats(List<NodeStats> nodesStats) {
                 nodeStats.getScriptCacheStats(),
                 nodeStats.getIndexingPressureStats(),
                 nodeStats.getShardIndexingPressureStats(),
-                nodeStats.getSearchBackpressureStats()
+                nodeStats.getSearchBackpressureStats(),
+                nodeStats.getClusterManagerThrottlingStats()
             );
         }).collect(Collectors.toList());
     }

test/framework/src/main/java/org/opensearch/test/InternalTestCluster.java

@@ -2686,6 +2686,7 @@ public void ensureEstimatedStats() {
                     false,
                     false,
                     false,
+                    false,
                     false
                 );
                 assertThat(