Introduced ResourceObservers when sampling the threads. Extracted metrics calculation

src/main/java/org/opensearch/performanceanalyzer/commons/metrics_generator/linux/LinuxDiskIOMetricsGenerator.java

@@ -10,10 +10,11 @@
 import java.util.Map;
 import org.opensearch.performanceanalyzer.commons.metrics_generator.DiskIOMetricsGenerator;
 import org.opensearch.performanceanalyzer.commons.os.ThreadDiskIO;
+import org.opensearch.performanceanalyzer.commons.os.metrics.IOMetrics;
 
 public class LinuxDiskIOMetricsGenerator implements DiskIOMetricsGenerator {
 
-    private Map<String, ThreadDiskIO.IOMetrics> diskIOMetricsMap;
+    private Map<String, IOMetrics> diskIOMetricsMap;
 
     public LinuxDiskIOMetricsGenerator() {
         diskIOMetricsMap = new HashMap<>();
@@ -66,7 +67,7 @@ public void addSample() {
         ThreadDiskIO.addSample();
     }
 
-    public void setDiskIOMetrics(final String threadId, final ThreadDiskIO.IOMetrics ioMetrics) {
+    public void setDiskIOMetrics(final String threadId, final IOMetrics ioMetrics) {
         diskIOMetricsMap.put(threadId, ioMetrics);
     }
 }

src/main/java/org/opensearch/performanceanalyzer/commons/metrics_generator/linux/LinuxSchedMetricsGenerator.java

@@ -10,10 +10,11 @@
 import java.util.Map;
 import org.opensearch.performanceanalyzer.commons.metrics_generator.SchedMetricsGenerator;
 import org.opensearch.performanceanalyzer.commons.os.ThreadSched;
+import org.opensearch.performanceanalyzer.commons.os.metrics.SchedMetrics;
 
 public class LinuxSchedMetricsGenerator implements SchedMetricsGenerator {
 
-    private final Map<String, ThreadSched.SchedMetrics> schedMetricsMap;
+    private final Map<String, SchedMetrics> schedMetricsMap;
 
     public LinuxSchedMetricsGenerator() {
         schedMetricsMap = new HashMap<>();
@@ -50,7 +51,7 @@ public void addSample() {
         ThreadSched.INSTANCE.addSample();
     }
 
-    public void setSchedMetric(final String threadId, final ThreadSched.SchedMetrics schedMetrics) {
+    public void setSchedMetric(final String threadId, final SchedMetrics schedMetrics) {
 
         schedMetricsMap.put(threadId, schedMetrics);
     }

src/main/java/org/opensearch/performanceanalyzer/commons/observer/ResourceObserver.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.performanceanalyzer.commons.observer;
+
+
+import java.util.Map;
+
+/**
+ * Observers resources consumption
+ *
+ * @param <T> Type of the metric
+ */
+public interface ResourceObserver<T> {
+
+    /**
+     * Retrieves the metrics for the given thread
+     *
+     * @param threadId
+     * @return
+     */
+    T observe(String threadId);
+
+    Map<String, T> observe();
+}

src/main/java/org/opensearch/performanceanalyzer/commons/os/OSGlobals.java

@@ -83,7 +83,7 @@ private static void enumTids() {
         }
     }
 
-    static synchronized List<String> getTids() {
+    public static synchronized List<String> getTids() {
         long curtime = System.currentTimeMillis();
         if (curtime - lastUpdated > REFRESH_INTERVAL_MS) {
             enumTids();

src/main/java/org/opensearch/performanceanalyzer/commons/os/ThreadCPU.java

@@ -5,160 +5,37 @@
 
 package org.opensearch.performanceanalyzer.commons.os;
 
+import static org.opensearch.performanceanalyzer.commons.os.metrics.CPUMetricsCalculator.calculateCPUUtilization;
+import static org.opensearch.performanceanalyzer.commons.os.metrics.CPUMetricsCalculator.calculateMajorFault;
+import static org.opensearch.performanceanalyzer.commons.os.metrics.CPUMetricsCalculator.calculateMinorFault;
+import static org.opensearch.performanceanalyzer.commons.os.metrics.CPUMetricsCalculator.getResidentSetSize;
 
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.Supplier;
 import org.opensearch.performanceanalyzer.commons.metrics_generator.linux.LinuxCPUPagingActivityGenerator;
+import org.opensearch.performanceanalyzer.commons.observer.ResourceObserver;
+import org.opensearch.performanceanalyzer.commons.os.observer.impl.CPUObserver;
 
 public final class ThreadCPU {
-    private static final Logger LOGGER = LogManager.getLogger(ThreadCPU.class);
+
     public static final ThreadCPU INSTANCE = new ThreadCPU();
-    private long scClkTck = 0;
-    private String pid = null;
-    private List<String> tids = null;
+    private static final ResourceObserver cpuObserver = new CPUObserver();
     private Map<String, Map<String, Object>> tidKVMap = new HashMap<>();
     private Map<String, Map<String, Object>> oldtidKVMap = new HashMap<>();
     private long kvTimestamp = 0;
     private long oldkvTimestamp = 0;
     private LinuxCPUPagingActivityGenerator cpuPagingActivityMap =
             new LinuxCPUPagingActivityGenerator();
 
-    // these two arrays map 1-1
-    private static String[] statKeys = {
-        "pid",
-        "comm",
-        "state",
-        "ppid",
-        "pgrp",
-        "session",
-        "ttynr",
-        "tpgid",
-        "flags",
-        "minflt",
-        "cminflt",
-        "majflt",
-        "cmajflt",
-        "utime",
-        "stime",
-        "cutime",
-        "cstime",
-        "prio",
-        "nice",
-        "nthreads",
-        "itrealvalue",
-        "starttime",
-        "vsize",
-        "rss",
-        "rsslim",
-        "startcode",
-        "endcode",
-        "startstack",
-        "kstkesp",
-        "kstkeip",
-        "signal",
-        "blocked",
-        "sigignore",
-        "sigcatch",
-        "wchan",
-        "nswap",
-        "cnswap",
-        "exitsig",
-        "cpu",
-        "rtprio",
-        "schedpolicy",
-        "bio_ticks",
-        "vmtime",
-        "cvmtime"
-        // more that we ignore
-    };
-
-    private static SchemaFileParser.FieldTypes[] statTypes = {
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.STRING,
-        SchemaFileParser.FieldTypes.CHAR,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.ULONG, // 10
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG, // 20
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG, // 30
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.ULONG,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT, // 40
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT,
-        SchemaFileParser.FieldTypes.INT
-    };
-
-    private ThreadCPU() {
-        try {
-            pid = OSGlobals.getPid();
-            scClkTck = OSGlobals.getScClkTck();
-            tids = OSGlobals.getTids();
-        } catch (Exception e) {
-            LOGGER.error(
-                    (Supplier<?>)
-                            () ->
-                                    new ParameterizedMessage(
-                                            "Error In Initializing ThreadCPU: {}", e.toString()),
-                    e);
-        }
-    }
-
     public synchronized void addSample() {
-        tids = OSGlobals.getTids();
-
         oldtidKVMap.clear();
         oldtidKVMap.putAll(tidKVMap);
 
         tidKVMap.clear();
         oldkvTimestamp = kvTimestamp;
         kvTimestamp = System.currentTimeMillis();
-        for (String tid : tids) {
-            Map<String, Object> sample =
-                    // (new SchemaFileParser("/proc/"+tid+"/stat",
-                    (new SchemaFileParser(
-                                    "/proc/" + pid + "/task/" + tid + "/stat",
-                                    statKeys,
-                                    statTypes,
-                                    true))
-                            .parse();
-            tidKVMap.put(tid, sample);
-        }
+        // Retrieve the cpu metrics for all threads
+        tidKVMap.putAll(cpuObserver.observe());
 
         calculateCPUDetails();
         calculatePagingActivity();
@@ -172,18 +49,8 @@ private void calculateCPUDetails() {
         for (Map.Entry<String, Map<String, Object>> entry : tidKVMap.entrySet()) {
             Map<String, Object> v = entry.getValue();
             Map<String, Object> oldv = oldtidKVMap.get(entry.getKey());
-            if (v != null && oldv != null) {
-                if (!v.containsKey("utime") || !oldv.containsKey("utime")) {
-                    continue;
-                }
-                long diff =
-                        ((long) (v.getOrDefault("utime", 0L))
-                                        - (long) (oldv.getOrDefault("utime", 0L)))
-                                + ((long) (v.getOrDefault("stime", 0L))
-                                        - (long) (oldv.getOrDefault("stime", 0L)));
-                double util = (1.0e3 * diff / scClkTck) / (kvTimestamp - oldkvTimestamp);
-                cpuPagingActivityMap.setCPUUtilization(entry.getKey(), util);
-            }
+            double util = calculateCPUUtilization(kvTimestamp, oldkvTimestamp, v, oldv);
+            cpuPagingActivityMap.setCPUUtilization(entry.getKey(), util);
         }
     }
 
@@ -196,22 +63,12 @@ private void calculatePagingActivity() {
         for (Map.Entry<String, Map<String, Object>> entry : tidKVMap.entrySet()) {
             Map<String, Object> v = entry.getValue();
             Map<String, Object> oldv = oldtidKVMap.get(entry.getKey());
-            if (v != null && oldv != null) {
-                if (!v.containsKey("majflt") || !oldv.containsKey("majflt")) {
-                    continue;
-                }
-                double majdiff =
-                        ((long) (v.getOrDefault("majflt", 0L))
-                                - (long) (oldv.getOrDefault("majflt", 0L)));
-                majdiff /= 1.0e-3 * (kvTimestamp - oldkvTimestamp);
-                double mindiff =
-                        ((long) (v.getOrDefault("minflt", 0L))
-                                - (long) (oldv.getOrDefault("minflt", 0L)));
-                mindiff /= 1.0e-3 * (kvTimestamp - oldkvTimestamp);
 
-                Double[] fltarr = {majdiff, mindiff, (double) ((long) v.getOrDefault("rss", 0L))};
-                cpuPagingActivityMap.setPagingActivities(entry.getKey(), fltarr);
-            }
+            double majdiff = calculateMajorFault(kvTimestamp, oldkvTimestamp, v, oldv);
+            double mindiff = calculateMinorFault(kvTimestamp, oldkvTimestamp, v, oldv);
+            double rss = getResidentSetSize(v);
+            Double[] fltarr = {majdiff, mindiff, rss};
+            cpuPagingActivityMap.setPagingActivities(entry.getKey(), fltarr);
         }
     }