forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add node performance trackers and performance collector service
Signed-off-by: Bharathwaj G <[email protected]>
- Loading branch information
1 parent
d58943d
commit ae316e7
Showing
12 changed files
with
740 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
121 changes: 121 additions & 0 deletions
121
server/src/main/java/org/opensearch/node/PerformanceCollectorService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| /* | ||
| * 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.cluster.ClusterChangedEvent; | ||
| import org.opensearch.cluster.ClusterStateListener; | ||
| import org.opensearch.cluster.node.DiscoveryNode; | ||
| import org.opensearch.cluster.service.ClusterService; | ||
| import org.opensearch.common.util.concurrent.ConcurrentCollections; | ||
| import org.opensearch.core.common.io.stream.StreamOutput; | ||
| import org.opensearch.core.common.io.stream.Writeable; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.ConcurrentMap; | ||
|
|
||
| /** | ||
| * This collects node level performance statistics such as cpu, memory, IO of each node and makes it available for | ||
| * coordinator node to aid in throttling, ranking etc | ||
| */ | ||
| public class PerformanceCollectorService implements ClusterStateListener { | ||
| private final ConcurrentMap<String, PerformanceCollectorService.NodePerformanceStatistics> nodeIdToPerfStats = ConcurrentCollections | ||
| .newConcurrentMap(); | ||
|
|
||
| public PerformanceCollectorService(ClusterService clusterService) { | ||
| clusterService.addListener(this); | ||
| } | ||
|
|
||
| @Override | ||
| public void clusterChanged(ClusterChangedEvent event) { | ||
| if (event.nodesRemoved()) { | ||
| for (DiscoveryNode removedNode : event.nodesDelta().removedNodes()) { | ||
| removeNode(removedNode.getId()); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| void removeNode(String nodeId) { | ||
| nodeIdToPerfStats.remove(nodeId); | ||
| } | ||
|
|
||
| public void addNodePerfStatistics(String nodeId, double cpuUsage, double ioUtilization, double memoryUsage, long timestamp) { | ||
| nodeIdToPerfStats.compute(nodeId, (id, ns) -> { | ||
| if (ns == null) { | ||
| return new PerformanceCollectorService.NodePerformanceStatistics(nodeId, cpuUsage, ioUtilization, memoryUsage, timestamp); | ||
| } else { | ||
| ns.cpuPercent = cpuUsage; | ||
| ns.memoryPercent = memoryUsage; | ||
| ns.ioUtilizationPercent = ioUtilization; | ||
| ns.timestamp = timestamp; | ||
| return ns; | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Get all node statistics which will be used for node stats | ||
| */ | ||
| public Map<String, PerformanceCollectorService.NodePerformanceStatistics> getAllNodeStatistics() { | ||
| Map<String, NodePerformanceStatistics> nodeStats = new HashMap<>(nodeIdToPerfStats.size()); | ||
| nodeIdToPerfStats.forEach((k, v) -> { nodeStats.put(k, new PerformanceCollectorService.NodePerformanceStatistics(v)); }); | ||
| return nodeStats; | ||
| } | ||
|
|
||
| /** | ||
| * Optionally return a {@code NodePerformanceStatistics} for the given nodeid, if | ||
| * performance stats information exists for the given node. Returns an empty | ||
| * {@code Optional} if the node was not found. | ||
| */ | ||
| public Optional<NodePerformanceStatistics> getNodeStatistics(final String nodeId) { | ||
| return Optional.ofNullable(nodeIdToPerfStats.get(nodeId)).map(ns -> new NodePerformanceStatistics(ns)); | ||
| } | ||
|
|
||
| /** | ||
| * This represents the performance stats of a node along with the timestamp at which the stats object was created | ||
| * in the respective node | ||
| */ | ||
| public static class NodePerformanceStatistics implements Writeable { | ||
| final String nodeId; | ||
| long timestamp; | ||
| double cpuPercent; | ||
| double ioUtilizationPercent; | ||
| double memoryPercent; | ||
|
|
||
| public NodePerformanceStatistics( | ||
| String nodeId, | ||
| double cpuPercent, | ||
| double ioUtilizationPercent, | ||
| double memoryPercent, | ||
| long timestamp | ||
| ) { | ||
| this.nodeId = nodeId; | ||
| this.cpuPercent = cpuPercent; | ||
| this.ioUtilizationPercent = ioUtilizationPercent; | ||
| this.memoryPercent = memoryPercent; | ||
| this.timestamp = timestamp; | ||
| } | ||
|
|
||
| NodePerformanceStatistics(NodePerformanceStatistics nodeStats) { | ||
| this(nodeStats.nodeId, nodeStats.cpuPercent, nodeStats.ioUtilizationPercent, nodeStats.memoryPercent, nodeStats.timestamp); | ||
| } | ||
|
|
||
| public double getMemoryPercent() { | ||
| return memoryPercent; | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
|
|
||
| } | ||
| } | ||
|
|
||
| } |
69 changes: 69 additions & 0 deletions
69
server/src/main/java/org/opensearch/throttling/tracker/AbstractAverageUsageTracker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| * 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.throttling.tracker; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.opensearch.common.lifecycle.AbstractLifecycleComponent; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.common.util.MovingAverage; | ||
| import org.opensearch.threadpool.Scheduler; | ||
| import org.opensearch.threadpool.ThreadPool; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.atomic.AtomicReference; | ||
|
|
||
| /** | ||
| * Base class for sliding window resource usage trackers | ||
| */ | ||
| public abstract class AbstractAverageUsageTracker extends AbstractLifecycleComponent { | ||
| private static final Logger LOGGER = LogManager.getLogger(AbstractAverageUsageTracker.class); | ||
|
|
||
| private final ThreadPool threadPool; | ||
| private final TimeValue pollingInterval; | ||
| private final AtomicReference<MovingAverage> observations = new AtomicReference<>(); | ||
|
|
||
| private volatile Scheduler.Cancellable scheduledFuture; | ||
|
|
||
| public AbstractAverageUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration) { | ||
| this.threadPool = threadPool; | ||
| this.pollingInterval = pollingInterval; | ||
| this.setWindowDuration(windowDuration); | ||
| } | ||
|
|
||
| public abstract long getUsage(); | ||
|
|
||
| public double getAverage() { | ||
| return observations.get().getAverage(); | ||
| } | ||
|
|
||
| public void setWindowDuration(TimeValue windowDuration) { | ||
| int windowSize = (int) (windowDuration.nanos() / pollingInterval.nanos()); | ||
| LOGGER.debug("updated window size: {}", windowSize); | ||
| observations.set(new MovingAverage(windowSize)); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStart() { | ||
| scheduledFuture = threadPool.scheduleWithFixedDelay(() -> { | ||
| long usage = getUsage(); | ||
| observations.get().record(usage); | ||
| }, pollingInterval, ThreadPool.Names.GENERIC); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doStop() { | ||
| if (scheduledFuture != null) { | ||
| scheduledFuture.cancel(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected void doClose() throws IOException {} | ||
| } |
35 changes: 35 additions & 0 deletions
35
server/src/main/java/org/opensearch/throttling/tracker/AverageCpuUsageTracker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| /* | ||
| * 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.throttling.tracker; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.monitor.process.ProcessProbe; | ||
| import org.opensearch.threadpool.ThreadPool; | ||
|
|
||
| /** | ||
| * AverageCpuUsageTracker tracks the average CPU usage by polling the CPU usage every (pollingInterval) | ||
| * and keeping track of the rolling average over a defined time window (windowDuration). | ||
| */ | ||
| public class AverageCpuUsageTracker extends AbstractAverageUsageTracker { | ||
| private static final Logger LOGGER = LogManager.getLogger(AverageCpuUsageTracker.class); | ||
|
|
||
| public AverageCpuUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration) { | ||
| super(threadPool, pollingInterval, windowDuration); | ||
| } | ||
|
|
||
| @Override | ||
| public long getUsage() { | ||
| long usage = ProcessProbe.getInstance().getProcessCpuPercent(); | ||
| LOGGER.debug("Recording cpu usage: {}%", usage); | ||
| return usage; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.