Add resource usage trackers and resource usage collector service

CHANGELOG.md

@@ -11,6 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add events correlation engine plugin ([#6854](https://github.com/opensearch-project/OpenSearch/issues/6854))
 - Introduce new dynamic cluster setting to control slice computation for concurrent segment search ([#9107](https://github.com/opensearch-project/OpenSearch/pull/9107))
 - Implement on behalf of token passing for extensions ([#8679](https://github.com/opensearch-project/OpenSearch/pull/8679))
+- [Admission control] Add Performance collector service and node performance stats tracker  ([#9890](https://github.com/opensearch-project/OpenSearch/pull/9890))
 - Implement Visitor Design pattern in QueryBuilder to enable the capability to traverse through the complex QueryBuilder tree. ([#10110](https://github.com/opensearch-project/OpenSearch/pull/10110))
 - Provide service accounts tokens to extensions ([#9618](https://github.com/opensearch-project/OpenSearch/pull/9618))
 - Configurable merge policy for index with an option to choose from LogByteSize and Tiered merge policy ([#9992](https://github.com/opensearch-project/OpenSearch/pull/9992))

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

@@ -56,6 +56,7 @@
 import org.opensearch.monitor.os.OsStats;
 import org.opensearch.monitor.process.ProcessStats;
 import org.opensearch.node.AdaptiveSelectionStats;
+import org.opensearch.node.NodesPerformanceStats;
 import org.opensearch.script.ScriptCacheStats;
 import org.opensearch.script.ScriptStats;
 import org.opensearch.search.backpressure.stats.SearchBackpressureStats;
@@ -142,6 +143,9 @@ public class NodeStats extends BaseNodeResponse implements ToXContentFragment {
     @Nullable
     private SearchPipelineStats searchPipelineStats;
 
+    @Nullable
+    private NodesPerformanceStats nodesPerformanceStats;
+
     public NodeStats(StreamInput in) throws IOException {
         super(in);
         timestamp = in.readVLong();
@@ -198,6 +202,11 @@ public NodeStats(StreamInput in) throws IOException {
         } else {
             searchPipelineStats = null;
         }
+        if (in.getVersion().onOrAfter(Version.V_3_0_0)) { // make it 2.12 when we backport
+            nodesPerformanceStats = in.readOptionalWriteable(NodesPerformanceStats::new);
+        } else {
+            nodesPerformanceStats = null;
+        }
     }
 
     public NodeStats(
@@ -216,6 +225,7 @@ public NodeStats(
         @Nullable DiscoveryStats discoveryStats,
         @Nullable IngestStats ingestStats,
         @Nullable AdaptiveSelectionStats adaptiveSelectionStats,
+        @Nullable NodesPerformanceStats nodesPerformanceStats,
         @Nullable ScriptCacheStats scriptCacheStats,
         @Nullable IndexingPressureStats indexingPressureStats,
         @Nullable ShardIndexingPressureStats shardIndexingPressureStats,
@@ -241,6 +251,7 @@ public NodeStats(
         this.discoveryStats = discoveryStats;
         this.ingestStats = ingestStats;
         this.adaptiveSelectionStats = adaptiveSelectionStats;
+        this.nodesPerformanceStats = nodesPerformanceStats;
         this.scriptCacheStats = scriptCacheStats;
         this.indexingPressureStats = indexingPressureStats;
         this.shardIndexingPressureStats = shardIndexingPressureStats;
@@ -344,6 +355,11 @@ public AdaptiveSelectionStats getAdaptiveSelectionStats() {
         return adaptiveSelectionStats;
     }
 
+    @Nullable
+    public NodesPerformanceStats getNodesPerformanceStats() {
+        return nodesPerformanceStats;
+    }
+
     @Nullable
     public ScriptCacheStats getScriptCacheStats() {
         return scriptCacheStats;
@@ -430,6 +446,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_2_9_0)) {
             out.writeOptionalWriteable(searchPipelineStats);
         }
+        if (out.getVersion().onOrAfter(Version.V_3_0_0)) { // make it 2.12 when we backport
+            out.writeOptionalWriteable(nodesPerformanceStats);
+        }
     }
 
     @Override
@@ -520,7 +539,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (getSearchPipelineStats() != null) {
             getSearchPipelineStats().toXContent(builder, params);
         }
-
+        if (getNodesPerformanceStats() != null) {
+            getNodesPerformanceStats().toXContent(builder, params);
+        }
         return builder;
     }
 }

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

@@ -213,7 +213,8 @@ public enum Metric {
         WEIGHTED_ROUTING_STATS("weighted_routing"),
         FILE_CACHE_STATS("file_cache"),
         TASK_CANCELLATION("task_cancellation"),
-        SEARCH_PIPELINE("search_pipeline");
+        SEARCH_PIPELINE("search_pipeline"),
+        PERFORMANCE_STATS("performance_stats");
 
         private String metricName;
 

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

@@ -124,7 +124,8 @@ protected NodeStats nodeOperation(NodeStatsRequest nodeStatsRequest) {
             NodesStatsRequest.Metric.WEIGHTED_ROUTING_STATS.containedIn(metrics),
             NodesStatsRequest.Metric.FILE_CACHE_STATS.containedIn(metrics),
             NodesStatsRequest.Metric.TASK_CANCELLATION.containedIn(metrics),
-            NodesStatsRequest.Metric.SEARCH_PIPELINE.containedIn(metrics)
+            NodesStatsRequest.Metric.SEARCH_PIPELINE.containedIn(metrics),
+            NodesStatsRequest.Metric.PERFORMANCE_STATS.containedIn(metrics)
         );
     }
 

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

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

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -133,6 +133,7 @@
 import org.opensearch.persistent.PersistentTasksClusterService;
 import org.opensearch.persistent.decider.EnableAssignmentDecider;
 import org.opensearch.plugins.PluginsService;
+import org.opensearch.ratelimiting.tracker.PerformanceTrackerSettings;
 import org.opensearch.repositories.fs.FsRepository;
 import org.opensearch.rest.BaseRestHandler;
 import org.opensearch.script.ScriptService;
@@ -654,6 +655,10 @@ public void apply(Settings value, Settings current, Settings previous) {
                 SegmentReplicationPressureService.MAX_REPLICATION_LIMIT_STALE_REPLICA_SETTING,
                 SegmentReplicationPressureService.MAX_ALLOWED_STALE_SHARDS,
 
+                // Settings related to admission control
+                PerformanceTrackerSettings.GLOBAL_CPU_USAGE_AC_WINDOW_DURATION_SETTING,
+                PerformanceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING,
+
                 // Settings related to Searchable Snapshots
                 Node.NODE_SEARCH_CACHE_SIZE_SETTING,
                 FileCache.DATA_TO_FILE_CACHE_SIZE_RATIO_SETTING,

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

@@ -195,6 +195,7 @@
 import org.opensearch.plugins.SearchPlugin;
 import org.opensearch.plugins.SystemIndexPlugin;
 import org.opensearch.plugins.TelemetryPlugin;
+import org.opensearch.ratelimiting.tracker.NodePerformanceTracker;
 import org.opensearch.repositories.RepositoriesModule;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.rest.RestController;
@@ -805,7 +806,6 @@ protected Node(
                 searchRequestStats,
                 remoteStoreStatsTrackerFactory
             );
-
             final AliasValidator aliasValidator = new AliasValidator();
 
             final ShardLimitValidator shardLimitValidator = new ShardLimitValidator(settings, clusterService, systemIndices);
@@ -1070,6 +1070,16 @@ protected Node(
                 transportService.getTaskManager(),
                 taskCancellationMonitoringSettings
             );
+            final NodePerformanceTracker nodePerformanceTracker = new NodePerformanceTracker(
+                threadPool,
+                settings,
+                clusterService.getClusterSettings()
+            );
+            final PerformanceCollectorService performanceCollectorService = new PerformanceCollectorService(
+                nodePerformanceTracker,
+                clusterService,
+                threadPool
+            );
             this.nodeService = new NodeService(
                 settings,
                 threadPool,
@@ -1091,7 +1101,8 @@ protected Node(
                 searchBackpressureService,
                 searchPipelineService,
                 fileCache,
-                taskCancellationMonitoringService
+                taskCancellationMonitoringService,
+                performanceCollectorService
             );
 
             final SearchService searchService = newSearchService(
@@ -1212,6 +1223,8 @@ protected Node(
                 b.bind(RerouteService.class).toInstance(rerouteService);
                 b.bind(ShardLimitValidator.class).toInstance(shardLimitValidator);
                 b.bind(FsHealthService.class).toInstance(fsHealthService);
+                b.bind(NodePerformanceTracker.class).toInstance(nodePerformanceTracker);
+                b.bind(PerformanceCollectorService.class).toInstance(performanceCollectorService);
                 b.bind(SystemIndices.class).toInstance(systemIndices);
                 b.bind(IdentityService.class).toInstance(identityService);
                 b.bind(Tracer.class).toInstance(tracer);
@@ -1328,6 +1341,8 @@ public Node start() throws NodeValidationException {
         injector.getInstance(RepositoriesService.class).start();
         injector.getInstance(SearchService.class).start();
         injector.getInstance(FsHealthService.class).start();
+        injector.getInstance(NodePerformanceTracker.class).start();
+        injector.getInstance(PerformanceCollectorService.class).start();
         nodeService.getMonitorService().start();
         nodeService.getSearchBackpressureService().start();
         nodeService.getTaskCancellationMonitoringService().start();
@@ -1490,6 +1505,8 @@ private Node stop() {
         injector.getInstance(ClusterService.class).stop();
         injector.getInstance(NodeConnectionsService.class).stop();
         injector.getInstance(FsHealthService.class).stop();
+        injector.getInstance(NodePerformanceTracker.class).stop();
+        injector.getInstance(PerformanceCollectorService.class).stop();
         nodeService.getMonitorService().stop();
         nodeService.getSearchBackpressureService().stop();
         injector.getInstance(GatewayService.class).stop();
@@ -1553,6 +1570,10 @@ public synchronized void close() throws IOException {
         toClose.add(nodeService.getSearchBackpressureService());
         toClose.add(() -> stopWatch.stop().start("fsHealth"));
         toClose.add(injector.getInstance(FsHealthService.class));
+        toClose.add(() -> stopWatch.stop().start("node_performance_tracker"));
+        toClose.add(injector.getInstance(NodePerformanceTracker.class));
+        toClose.add(() -> stopWatch.stop().start("performance_collector"));
+        toClose.add(injector.getInstance(PerformanceCollectorService.class));
         toClose.add(() -> stopWatch.stop().start("gateway"));
         toClose.add(injector.getInstance(GatewayService.class));
         toClose.add(() -> stopWatch.stop().start("search"));

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

@@ -0,0 +1,81 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.node;
+
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.Writeable;
+
+import java.io.IOException;
+import java.util.Locale;
+
+/**
+ * This represents the performance stats of a node along with the timestamp at which the stats object was created
+ * in the respective node
+ */
+public class NodePerformanceStats implements Writeable {
+    final String nodeId;
+    long timestamp;
+    double cpuUtilizationPercent;
+    double memoryUtilizationPercent;
+
+    public NodePerformanceStats(String nodeId, long timestamp, double memoryUtilizationPercent, double cpuUtilizationPercent) {
+        this.nodeId = nodeId;
+        this.timestamp = timestamp;
+        this.cpuUtilizationPercent = cpuUtilizationPercent;
+        this.memoryUtilizationPercent = memoryUtilizationPercent;
+    }
+
+    public NodePerformanceStats(StreamInput in) throws IOException {
+        this.nodeId = in.readString();
+        this.timestamp = in.readLong();
+        this.cpuUtilizationPercent = in.readDouble();
+        this.memoryUtilizationPercent = in.readDouble();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(this.nodeId);
+        out.writeLong(this.timestamp);
+        out.writeDouble(this.cpuUtilizationPercent);
+        out.writeDouble(this.memoryUtilizationPercent);
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder("NodePerformanceStatistics[");
+        sb.append(nodeId).append("](");
+        sb.append("Timestamp: ").append(timestamp);
+        sb.append(", CPU utilization percent: ").append(String.format(Locale.ROOT, "%.1f", cpuUtilizationPercent));
+        sb.append(", Memory utilization percent: ").append(String.format(Locale.ROOT, "%.1f", memoryUtilizationPercent));
+        sb.append(")");
+        return sb.toString();
+    }
+
+    NodePerformanceStats(NodePerformanceStats nodePerformanceStats) {
+        this(
+            nodePerformanceStats.nodeId,
+            nodePerformanceStats.timestamp,
+            nodePerformanceStats.memoryUtilizationPercent,
+            nodePerformanceStats.cpuUtilizationPercent
+        );
+    }
+
+    public double getMemoryUtilizationPercent() {
+        return memoryUtilizationPercent;
+    }
+
+    public double getCpuUtilizationPercent() {
+        return cpuUtilizationPercent;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+}

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

@@ -83,6 +83,7 @@ public class NodeService implements Closeable {
     private final ScriptService scriptService;
     private final HttpServerTransport httpServerTransport;
     private final ResponseCollectorService responseCollectorService;
+    private final PerformanceCollectorService performanceCollectorService;
     private final SearchTransportService searchTransportService;
     private final IndexingPressureService indexingPressureService;
     private final AggregationUsageService aggregationUsageService;
@@ -114,7 +115,8 @@ public class NodeService implements Closeable {
         SearchBackpressureService searchBackpressureService,
         SearchPipelineService searchPipelineService,
         FileCache fileCache,
-        TaskCancellationMonitoringService taskCancellationMonitoringService
+        TaskCancellationMonitoringService taskCancellationMonitoringService,
+        PerformanceCollectorService performanceCollectorService
     ) {
         this.settings = settings;
         this.threadPool = threadPool;
@@ -137,6 +139,7 @@ public class NodeService implements Closeable {
         this.clusterService = clusterService;
         this.fileCache = fileCache;
         this.taskCancellationMonitoringService = taskCancellationMonitoringService;
+        this.performanceCollectorService = performanceCollectorService;
         clusterService.addStateApplier(ingestService);
         clusterService.addStateApplier(searchPipelineService);
     }
@@ -217,7 +220,8 @@ public NodeStats stats(
         boolean weightedRoutingStats,
         boolean fileCacheStats,
         boolean taskCancellation,
-        boolean searchPipelineStats
+        boolean searchPipelineStats,
+        boolean nodesPerfStats
     ) {
         // 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)
@@ -237,6 +241,7 @@ public NodeStats stats(
             discoveryStats ? discovery.stats() : null,
             ingest ? ingestService.stats() : null,
             adaptiveSelection ? responseCollectorService.getAdaptiveStats(searchTransportService.getPendingSearchRequests()) : null,
+            nodesPerfStats ? performanceCollectorService.stats() : null,
             scriptCache ? scriptService.cacheStats() : null,
             indexingPressure ? this.indexingPressureService.nodeStats() : null,
             shardIndexingPressure ? this.indexingPressureService.shardStats(indices) : null,