Skip to content

Commit

Permalink
Integrate IO Usage Tracker to the Resource Usage Collector Service an…
Browse files Browse the repository at this point in the history
…d Emit IO Usage Stats

Signed-off-by: Ajay Kumar Movva <[email protected]>
  • Loading branch information
Ajay Kumar Movva committed Jan 14, 2024
1 parent 988dea8 commit 842d3d4
Show file tree
Hide file tree
Showing 14 changed files with 289 additions and 14 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
- GHA to verify checklist items completion in PR descriptions ([#10800](https://github.com/opensearch-project/OpenSearch/pull/10800))
- Allow to pass the list settings through environment variables (like [], ["a", "b", "c"], ...) ([#10625](https://github.com/opensearch-project/OpenSearch/pull/10625))
- [Admission Control] Integrate CPU AC with ResourceUsageCollector and add CPU AC stats to nodes/stats ([#10887](https://github.com/opensearch-project/OpenSearch/pull/10887))
- [Admission Control] Integrate IO Usage Tracker to the Resource Usage Collector Service and Emit IO Usage Stats ([#11880](https://github.com/opensearch-project/OpenSearch/pull/11880))

### Dependencies
- Bump `log4j-core` from 2.18.0 to 2.19.0
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -666,6 +666,7 @@ public void apply(Settings value, Settings current, Settings previous) {
// Settings related to resource trackers
ResourceTrackerSettings.GLOBAL_CPU_USAGE_AC_WINDOW_DURATION_SETTING,
ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING,
ResourceTrackerSettings.GLOBAL_IO_USAGE_AC_WINDOW_DURATION_SETTING,

// Settings related to Searchable Snapshots
Node.NODE_SEARCH_CACHE_SIZE_SETTING,
Expand Down
70 changes: 70 additions & 0 deletions server/src/main/java/org/opensearch/node/IoUsageStats.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
/*
* 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 org.opensearch.core.xcontent.ToXContentFragment;
import org.opensearch.core.xcontent.XContentBuilder;

import java.io.IOException;
import java.util.Locale;

/**
* This class is to store tne IO Usage Stats and used to return in node stats API.
*/
public class IoUsageStats implements Writeable, ToXContentFragment {

private double ioUtilisationPercent;

public IoUsageStats(double ioUtilisationPercent) {
this.ioUtilisationPercent = ioUtilisationPercent;
}

public IoUsageStats(StreamInput in) throws IOException {
this.ioUtilisationPercent = in.readDouble();
}

/**
* Write this into the {@linkplain StreamOutput}.
*
* @param out
*/
@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeDouble(this.ioUtilisationPercent);
}

public double getIoUtilisationPercent() {
return ioUtilisationPercent;
}

public void setIoUtilisationPercent(double ioUtilisationPercent) {
this.ioUtilisationPercent = ioUtilisationPercent;
}

/**
* @param builder
* @param params
* @return
* @throws IOException
*/
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.field("io_utilization_percent", this.ioUtilisationPercent);
return builder.endObject();
}

@Override
public String toString() {
return ", IO utilization percent: " + String.format(Locale.ROOT, "%.1f", this.ioUtilisationPercent);
}
}
1 change: 1 addition & 0 deletions server/src/main/java/org/opensearch/node/Node.java
Original file line number Diff line number Diff line change
Expand Up @@ -912,6 +912,7 @@ protected Node(
final RestController restController = actionModule.getRestController();

final NodeResourceUsageTracker nodeResourceUsageTracker = new NodeResourceUsageTracker(
monitorService.fsService(),
threadPool,
settings,
clusterService.getClusterSettings()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,19 +24,28 @@ public class NodeResourceUsageStats implements Writeable {
long timestamp;
double cpuUtilizationPercent;
double memoryUtilizationPercent;
IoUsageStats ioUsageStats;

public NodeResourceUsageStats(String nodeId, long timestamp, double memoryUtilizationPercent, double cpuUtilizationPercent) {
public NodeResourceUsageStats(
String nodeId,
long timestamp,
double memoryUtilizationPercent,
double cpuUtilizationPercent,
IoUsageStats ioUsageStats
) {
this.nodeId = nodeId;
this.timestamp = timestamp;
this.cpuUtilizationPercent = cpuUtilizationPercent;
this.memoryUtilizationPercent = memoryUtilizationPercent;
this.ioUsageStats = ioUsageStats;
}

public NodeResourceUsageStats(StreamInput in) throws IOException {
this.nodeId = in.readString();
this.timestamp = in.readLong();
this.cpuUtilizationPercent = in.readDouble();
this.memoryUtilizationPercent = in.readDouble();
this.ioUsageStats = new IoUsageStats(in);
}

@Override
Expand All @@ -45,6 +54,7 @@ public void writeTo(StreamOutput out) throws IOException {
out.writeLong(this.timestamp);
out.writeDouble(this.cpuUtilizationPercent);
out.writeDouble(this.memoryUtilizationPercent);
this.ioUsageStats.writeTo(out);
}

@Override
Expand All @@ -54,6 +64,7 @@ public String toString() {
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(this.ioUsageStats.toString());
sb.append(")");
return sb.toString();
}
Expand All @@ -63,7 +74,8 @@ public String toString() {
nodeResourceUsageStats.nodeId,
nodeResourceUsageStats.timestamp,
nodeResourceUsageStats.memoryUtilizationPercent,
nodeResourceUsageStats.cpuUtilizationPercent
nodeResourceUsageStats.cpuUtilizationPercent,
nodeResourceUsageStats.ioUsageStats
);
}

Expand All @@ -75,6 +87,10 @@ public double getCpuUtilizationPercent() {
return cpuUtilizationPercent;
}

public IoUsageStats getIoUsageStats() {
return ioUsageStats;
}

public long getTimestamp() {
return timestamp;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
"memory_utilization_percent",
String.format(Locale.ROOT, "%.1f", resourceUsageStats.memoryUtilizationPercent)
);
builder.field("io_usage_stats", resourceUsageStats.ioUsageStats);
}
builder.endObject();
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -78,14 +78,16 @@ public void collectNodeResourceUsageStats(
String nodeId,
long timestamp,
double memoryUtilizationPercent,
double cpuUtilizationPercent
double cpuUtilizationPercent,
IoUsageStats ioUsageStats
) {
nodeIdToResourceUsageStats.compute(nodeId, (id, resourceUsageStats) -> {
if (resourceUsageStats == null) {
return new NodeResourceUsageStats(nodeId, timestamp, memoryUtilizationPercent, cpuUtilizationPercent);
return new NodeResourceUsageStats(nodeId, timestamp, memoryUtilizationPercent, cpuUtilizationPercent, ioUsageStats);
} else {
resourceUsageStats.cpuUtilizationPercent = cpuUtilizationPercent;
resourceUsageStats.memoryUtilizationPercent = memoryUtilizationPercent;
resourceUsageStats.ioUsageStats = ioUsageStats;
resourceUsageStats.timestamp = timestamp;
return resourceUsageStats;
}
Expand Down Expand Up @@ -129,7 +131,8 @@ private void collectLocalNodeResourceUsageStats() {
clusterService.state().nodes().getLocalNodeId(),
System.currentTimeMillis(),
nodeResourceUsageTracker.getMemoryUtilizationPercent(),
nodeResourceUsageTracker.getCpuUtilizationPercent()
nodeResourceUsageTracker.getCpuUtilizationPercent(),
nodeResourceUsageTracker.getIoUsageStats()
);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,12 @@
public abstract class AbstractAverageUsageTracker extends AbstractLifecycleComponent {
private static final Logger LOGGER = LogManager.getLogger(AbstractAverageUsageTracker.class);

private final ThreadPool threadPool;
private final TimeValue pollingInterval;
protected final ThreadPool threadPool;
protected final TimeValue pollingInterval;
private TimeValue windowDuration;
private final AtomicReference<MovingAverage> observations = new AtomicReference<>();

private volatile Scheduler.Cancellable scheduledFuture;
protected volatile Scheduler.Cancellable scheduledFuture;

public AbstractAverageUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration) {
this.threadPool = threadPool;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
/*
* 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.resource.tracker;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.opensearch.common.unit.TimeValue;
import org.opensearch.monitor.fs.FsService;
import org.opensearch.node.IoUsageStats;
import org.opensearch.threadpool.ThreadPool;

/**
* AverageIoUsageTracker tracks the IO usage by polling the FS Stats for IO metrics every (pollingInterval)
* and keeping track of the rolling average over a defined time window (windowDuration).
*/
public class AverageIoUsageTracker extends AbstractAverageUsageTracker {

private static final Logger LOGGER = LogManager.getLogger(AverageIoUsageTracker.class);
private final FsService fsService;
private long prevIoTimeMillis;
private long prevTimeMillis;
private final IoUsageStats ioUsageStats;

public AverageIoUsageTracker(FsService fsService, ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration) {
super(threadPool, pollingInterval, windowDuration);
this.fsService = fsService;
this.prevIoTimeMillis = -1;
this.prevTimeMillis = -1;
this.ioUsageStats = new IoUsageStats(0);
}

/**
* Get current IO usage percentage calculated using fs stats
*/
@Override
public long getUsage() {
long usage = 0;
if (this.preValidateFsStats()) {
return usage;
}
long currentIoTimeMillis = fsService.stats().getIoStats().getTotalIOTimeMillis();
long ioDevicesCount = fsService.stats().getIoStats().getDevicesStats().length;
long currentTimeMillis = fsService.stats().getTimestamp();
if (prevTimeMillis > 0 && (currentTimeMillis - this.prevTimeMillis > 0)) {
long averageIoTime = (currentIoTimeMillis - this.prevIoTimeMillis) / ioDevicesCount;
usage = averageIoTime * 100 / (currentTimeMillis - this.prevTimeMillis);
}
this.prevTimeMillis = currentTimeMillis;
this.prevIoTimeMillis = currentIoTimeMillis;
return usage;
}

@Override
protected void doStart() {
scheduledFuture = threadPool.scheduleWithFixedDelay(() -> {
long usage = getUsage();
recordUsage(usage);
updateIoUsageStats();
}, pollingInterval, ThreadPool.Names.GENERIC);
}

private boolean preValidateFsStats() {
return fsService == null
|| fsService.stats() == null
|| fsService.stats().getIoStats() == null
|| fsService.stats().getIoStats().getDevicesStats() == null;
}

private void updateIoUsageStats() {
this.ioUsageStats.setIoUtilisationPercent(this.isReady() ? this.getAverage() : 0);
}

public IoUsageStats getIoUsageStats() {
return this.ioUsageStats;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import org.opensearch.common.settings.ClusterSettings;
import org.opensearch.common.settings.Settings;
import org.opensearch.common.unit.TimeValue;
import org.opensearch.monitor.fs.FsService;
import org.opensearch.node.IoUsageStats;
import org.opensearch.threadpool.ThreadPool;

/**
Expand All @@ -22,10 +24,14 @@ public class NodeResourceUsageTracker extends AbstractLifecycleComponent {
private final ClusterSettings clusterSettings;
private AverageCpuUsageTracker cpuUsageTracker;
private AverageMemoryUsageTracker memoryUsageTracker;
private AverageIoUsageTracker ioUsageTracker;

private ResourceTrackerSettings resourceTrackerSettings;

public NodeResourceUsageTracker(ThreadPool threadPool, Settings settings, ClusterSettings clusterSettings) {
private FsService fsService;

public NodeResourceUsageTracker(FsService fsService, ThreadPool threadPool, Settings settings, ClusterSettings clusterSettings) {
this.fsService = fsService;
this.threadPool = threadPool;
this.clusterSettings = clusterSettings;
this.resourceTrackerSettings = new ResourceTrackerSettings(settings);
Expand All @@ -52,6 +58,13 @@ public double getMemoryUtilizationPercent() {
return 0.0;
}

/**
* Return io stats average if we have enough datapoints, otherwise return 0
*/
public IoUsageStats getIoUsageStats() {
return ioUsageTracker.getIoUsageStats();
}

/**
* Checks if all of the resource usage trackers are ready
*/
Expand Down Expand Up @@ -79,6 +92,17 @@ void initialize() {
ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING,
this::setMemoryWindowDuration
);

ioUsageTracker = new AverageIoUsageTracker(
fsService,
threadPool,
resourceTrackerSettings.getIoPollingInterval(),
resourceTrackerSettings.getIoWindowDuration()
);
clusterSettings.addSettingsUpdateConsumer(
ResourceTrackerSettings.GLOBAL_IO_USAGE_AC_WINDOW_DURATION_SETTING,
this::setIoWindowDuration
);
}

private void setMemoryWindowDuration(TimeValue windowDuration) {
Expand All @@ -91,6 +115,11 @@ private void setCpuWindowDuration(TimeValue windowDuration) {
resourceTrackerSettings.setCpuWindowDuration(windowDuration);
}

private void setIoWindowDuration(TimeValue windowDuration) {
ioUsageTracker.setWindowSize(windowDuration);
resourceTrackerSettings.setIoWindowDuration(windowDuration);
}

/**
* Visible for testing
*/
Expand All @@ -102,17 +131,20 @@ ResourceTrackerSettings getResourceTrackerSettings() {
protected void doStart() {
cpuUsageTracker.doStart();
memoryUsageTracker.doStart();
ioUsageTracker.doStart();
}

@Override
protected void doStop() {
cpuUsageTracker.doStop();
memoryUsageTracker.doStop();
ioUsageTracker.doStop();
}

@Override
protected void doClose() {
cpuUsageTracker.doClose();
memoryUsageTracker.doClose();
ioUsageTracker.doClose();
}
}
Loading

0 comments on commit 842d3d4

Please sign in to comment.