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 d48ad95
Showing
11 changed files
with
724 additions
and
2 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
106 changes: 106 additions & 0 deletions
106
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,106 @@ | ||
| /* | ||
| * 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 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; | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| 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)); | ||
| } | ||
|
|
||
| public static class NodePerformanceStatistics { | ||
| 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; | ||
| } | ||
| } | ||
|
|
||
| } |
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; | ||
| } | ||
|
|
||
| } |
39 changes: 39 additions & 0 deletions
39
server/src/main/java/org/opensearch/throttling/tracker/AverageMemoryUsageTracker.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,39 @@ | ||
| /* | ||
| * 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.threadpool.ThreadPool; | ||
|
|
||
| import java.lang.management.ManagementFactory; | ||
| import java.lang.management.MemoryMXBean; | ||
|
|
||
| /** | ||
| * AverageMemoryUsageTracker tracks the average JVM usage by polling the JVM usage every (pollingInterval) | ||
| * and keeping track of the rolling average over a defined time window (windowDuration). | ||
| */ | ||
| public class AverageMemoryUsageTracker extends AbstractAverageUsageTracker { | ||
|
|
||
| private static final Logger LOGGER = LogManager.getLogger(AverageMemoryUsageTracker.class); | ||
|
|
||
| private static final MemoryMXBean MEMORY_MX_BEAN = ManagementFactory.getMemoryMXBean(); | ||
|
|
||
| public AverageMemoryUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration) { | ||
| super(threadPool, pollingInterval, windowDuration); | ||
| } | ||
|
|
||
| @Override | ||
| public long getUsage() { | ||
| long usage = MEMORY_MX_BEAN.getHeapMemoryUsage().getUsed() * 100 / MEMORY_MX_BEAN.getHeapMemoryUsage().getMax(); | ||
| LOGGER.debug("Recording memory usage: {}%", usage); | ||
| return usage; | ||
| } | ||
| } |
Oops, something went wrong.