Integrate IO Based Usage Tracker and Stats for the Resource Usage Ser…

…vice

Signed-off-by: Ajay Kumar Movva <movvaam@amazon.com>

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

@@ -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,

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

@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+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);
+    }
+}

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

@@ -912,6 +912,7 @@ protected Node(
             final RestController restController = actionModule.getRestController();
 
             final NodeResourceUsageTracker nodeResourceUsageTracker = new NodeResourceUsageTracker(
+                monitorService.fsService(),
                 threadPool,
                 settings,
                 clusterService.getClusterSettings()

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

@@ -24,19 +24,22 @@ 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
@@ -45,6 +48,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
@@ -54,6 +58,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();
     }
@@ -63,7 +68,8 @@ public String toString() {
             nodeResourceUsageStats.nodeId,
             nodeResourceUsageStats.timestamp,
             nodeResourceUsageStats.memoryUtilizationPercent,
-            nodeResourceUsageStats.cpuUtilizationPercent
+            nodeResourceUsageStats.cpuUtilizationPercent,
+            nodeResourceUsageStats.ioUsageStats
         );
     }
 
@@ -75,6 +81,10 @@ public double getCpuUtilizationPercent() {
         return cpuUtilizationPercent;
     }
 
+    public IoUsageStats getIoUsageStats() {
+        return ioUsageStats;
+    }
+
     public long getTimestamp() {
         return timestamp;
     }

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

@@ -60,6 +60,10 @@ 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();
         }

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

@@ -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;
             }
@@ -129,7 +131,8 @@ private void collectLocalNodeResourceUsageStats() {
                 clusterService.state().nodes().getLocalNodeId(),
                 System.currentTimeMillis(),
                 nodeResourceUsageTracker.getMemoryUtilizationPercent(),
-                nodeResourceUsageTracker.getCpuUtilizationPercent()
+                nodeResourceUsageTracker.getCpuUtilizationPercent(),
+                nodeResourceUsageTracker.getIoUsageStats()
             );
         }
     }

server/src/main/java/org/opensearch/node/resource/tracker/AbstractAverageUsageTracker.java

@@ -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;

server/src/main/java/org/opensearch/node/resource/tracker/AverageIoUsageTracker.java

@@ -0,0 +1,78 @@
+/*
+ * 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;
+
+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 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)) {
+            LOGGER.info("Io Time Diff: " + (currentIoTimeMillis - prevIoTimeMillis));
+            LOGGER.info("Time Diff: " + (currentTimeMillis - prevTimeMillis));
+            LOGGER.info("Count of Devices: " + ioDevicesCount);
+            long averageIoTime = (currentIoTimeMillis - this.prevIoTimeMillis) / ioDevicesCount;
+            usage = averageIoTime * 100 / (currentTimeMillis - this.prevTimeMillis);
+            LOGGER.info("Recording IO usage: {}%", usage);
+        }
+        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;
+    }
+}

server/src/main/java/org/opensearch/node/resource/tracker/NodeResourceUsageTracker.java

@@ -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;
 
 /**
@@ -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);
@@ -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
      */
@@ -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) {
@@ -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
      */
@@ -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();
     }
 }