changing source from meminfo to status

Signed-off-by: Pradeep L <spradeel@amazon.com>

Author: Pradeep L

server/src/main/java/org/opensearch/monitor/os/OsProbe.java

@@ -237,63 +237,79 @@ String readProcLoadavg() throws IOException {
         return readSingleLine(PathUtils.get("/proc/loadavg"));
     }
 
-    public short getSystemCpuPercent() {
-        return Probes.getLoadAndScaleToPercent(getSystemCpuLoad, osMxBean);
-    }
-
     /**
-     * Reads a file containing a single line.
+     * Reads the {@code RssAnon} field (anonymous resident memory) of the current process from
+     * {@code /proc/self/status} and returns it in bytes. Returns {@code -1L} when the host is
+     * not Linux, when {@code /proc/self/status} cannot be read, or when the {@code RssAnon:}
+     * line is missing or malformed. Failure paths are logged at debug level; callers that want
+     * user-visible warnings must wrap this method themselves.
      *
-     * @param path path to the file to read
-     * @return the single line
-     * @throws IOException if an I/O exception occurs reading the file
+     * @return the {@code RssAnon} value in bytes, or {@code -1L} if unavailable
      */
-    private String readSingleLine(final Path path) throws IOException {
-        final List<String> lines = Files.readAllLines(path);
-        assert lines.size() == 1 : String.join("\n", lines);
-        return lines.get(0);
-    }
-
-    /**
-     * Returns the available memory in bytes on Linux by reading {@code MemAvailable} from {@code /proc/meminfo}.
-     * Available memory is a better estimate than free memory as it accounts for reclaimable page cache and slab memory.
-     * Returns -1 if not on Linux or if the value cannot be read.
-     */
-    public long getAvailableMemorySize() {
-        if (!Constants.LINUX) {
-            return -1;
+    public long getProcessRssAnon() {
+        if (Constants.LINUX == false) {
+            return -1L;
         }
         try {
-            return readAvailableMemoryFromProcMeminfo();
-        } catch (Exception e) {
-            logger.warn("error reading available memory from /proc/meminfo", e);
-            return -1;
+            return readRssAnonFromProcSelfStatus();
+        } catch (IOException e) {
+            logger.debug("failed to read /proc/self/status", e);
+            return -1L;
         }
     }
 
     /**
-     * Reads {@code MemAvailable} from {@code /proc/meminfo}.
+     * Reads the {@code RssAnon} field from {@code /proc/self/status}. Package-private so tests
+     * can override it with canned file contents.
      *
-     * @return the available memory in bytes, or -1 if not found
-     * @throws IOException if an I/O exception occurs reading {@code /proc/meminfo}
+     * @return the {@code RssAnon} value in bytes, or {@code -1L} when the line is missing or
+     *         the value is unparseable/negative
+     * @throws IOException if the file cannot be opened or read
      */
-    @SuppressForbidden(reason = "access /proc/meminfo")
-    long readAvailableMemoryFromProcMeminfo() throws IOException {
-        try (BufferedReader reader = Files.newBufferedReader(PathUtils.get("/proc/meminfo"))) {
+    @SuppressForbidden(reason = "access /proc/self/status")
+    long readRssAnonFromProcSelfStatus() throws IOException {
+        try (BufferedReader reader = Files.newBufferedReader(PathUtils.get("/proc/self/status"))) {
             String line;
             while ((line = reader.readLine()) != null) {
-                if (line.startsWith("MemAvailable:")) {
-                    final String[] parts = line.split("\\s+");
+                if (line.startsWith("RssAnon:")) {
+                    // Format: "RssAnon:\t 12345 kB"
+                    String[] parts = line.split("\\s+");
                     if (parts.length >= 2) {
-                        // Value in /proc/meminfo is in kB
-                        return Long.parseLong(parts[1]) * 1024;
-                    } else {
-                        return -1;
+                        try {
+                            long kb = Long.parseLong(parts[1]);
+                            if (kb < 0L) {
+                                return -1L;
+                            }
+                            return kb * 1024L;
+                        } catch (NumberFormatException nfe) {
+                            logger.debug("malformed RssAnon value in /proc/self/status", nfe);
+                            return -1L;
+                        }
                     }
+                    logger.debug("RssAnon line has unexpected shape: [{}]", line);
+                    return -1L;
                 }
             }
+            logger.debug("RssAnon line not found in /proc/self/status");
+            return -1L;
         }
-        return -1;
+    }
+
+    public short getSystemCpuPercent() {
+        return Probes.getLoadAndScaleToPercent(getSystemCpuLoad, osMxBean);
+    }
+
+    /**
+     * Reads a file containing a single line.
+     *
+     * @param path path to the file to read
+     * @return the single line
+     * @throws IOException if an I/O exception occurs reading the file
+     */
+    private String readSingleLine(final Path path) throws IOException {
+        final List<String> lines = Files.readAllLines(path);
+        assert lines.size() == 1 : String.join("\n", lines);
+        return lines.get(0);
     }
 
     // this property is to support a hack to workaround an issue with Docker containers mounting the cgroups hierarchy inconsistently with

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

@@ -10,52 +10,174 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.monitor.os.OsProbe;
 import org.opensearch.threadpool.ThreadPool;
 
+import java.lang.management.ManagementFactory;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
 /**
- * AverageNativeMemoryUsageTracker tracks the average native (physical) memory usage on the node
- * by polling the OS-level memory stats every (pollingInterval) and keeping track of the rolling
- * average over a defined time window (windowDuration).
+ * AverageNativeMemoryUsageTracker reports this OpenSearch process's off-heap native memory
+ * utilization as a percentage of the budget configured for the native analytics engines,
+ * averaged over a rolling window.
+ *
+ * <p>On Linux, each polling cycle computes
+ * {@code usage = max(0, RssAnon - HeapCommitted)} where {@code RssAnon} comes from
+ * {@link OsProbe#getProcessRssAnon()} and {@code HeapCommitted} comes from the JVM memory MX
+ * bean. The denominator is a 20% padded sum of the two plugin budgets:
+ * {@code cap = 1.2 * (datafusion.memory_pool_limit_bytes + resolved(parquet.max_native_allocation))}.
+ * When either plugin setting is absent, zero, negative, or unparseable, its contribution is 0;
+ * when both contribute 0 the cap is 0 and {@code getUsage()} returns 0 without dividing.
  *
- * On Linux, it uses available memory from /proc/meminfo (MemAvailable) which accounts for
- * reclaimable page cache and slab memory, giving a more accurate picture of actual memory
- * pressure. On other platforms, it falls back to free physical memory.
+ * <p>Activation is already gated to Linux in {@link NodeResourceUsageTracker}; on non-Linux
+ * platforms the polling loop is not started and {@link OsProbe#getProcessRssAnon()} returns
+ * {@code -1} without touching the filesystem.
  */
 public class AverageNativeMemoryUsageTracker extends AbstractAverageUsageTracker {
 
     private static final Logger LOGGER = LogManager.getLogger(AverageNativeMemoryUsageTracker.class);
 
-    public AverageNativeMemoryUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration) {
+    static final String DATAFUSION_MEMORY_POOL_LIMIT_KEY = "datafusion.memory_pool_limit_bytes";
+    static final String PARQUET_MAX_NATIVE_ALLOCATION_KEY = "parquet.max_native_allocation";
+    static final double HEADROOM_FACTOR = 1.2;
+
+    private final LongSupplier rssAnonSupplier;
+    private final LongSupplier heapCommittedSupplier;
+    private final LongSupplier nativeMemoryCapSupplier;
+
+    /**
+     * Production constructor. Wires the RSS reader to {@link OsProbe#getProcessRssAnon()}, the
+     * heap-committed supplier to the JVM memory MX bean, and the native-memory cap supplier to
+     * {@link #computeNativeMemoryCap(Settings)} so dynamic updates to
+     * {@code datafusion.memory_pool_limit_bytes} are observed on the next polling cycle.
+     */
+    public AverageNativeMemoryUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration, Settings settings) {
         super(threadPool, pollingInterval, windowDuration);
+        this.rssAnonSupplier = () -> OsProbe.getInstance().getProcessRssAnon();
+        this.heapCommittedSupplier = () -> ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getCommitted();
+        this.nativeMemoryCapSupplier = () -> computeNativeMemoryCap(settings);
     }
 
     /**
-     * Get current native memory usage percentage using OS-level physical memory stats.
-     * Prefers available memory (MemAvailable) over free memory (MemFree) as it provides
-     * a more accurate measure of actual memory pressure on the node.
+     * Package-private test constructor. Accepts three {@link LongSupplier}s in place of the
+     * production defaults so tests can drive {@link #getUsage()} deterministically without
+     * reading {@code /proc/self/status}, the JVM memory MX bean, or plugin settings.
      */
+    AverageNativeMemoryUsageTracker(
+        ThreadPool threadPool,
+        TimeValue pollingInterval,
+        TimeValue windowDuration,
+        LongSupplier rssAnonSupplier,
+        LongSupplier heapCommittedSupplier,
+        LongSupplier nativeMemoryCapSupplier
+    ) {
+        super(threadPool, pollingInterval, windowDuration);
+        this.rssAnonSupplier = rssAnonSupplier;
+        this.heapCommittedSupplier = heapCommittedSupplier;
+        this.nativeMemoryCapSupplier = nativeMemoryCapSupplier;
+    }
+
     @Override
     public long getUsage() {
-        OsProbe osProbe = OsProbe.getInstance();
-        long totalMemory = osProbe.getTotalPhysicalMemorySize();
-        if (totalMemory <= 0) {
-            LOGGER.warn("Unable to retrieve total physical memory size");
-            return 0;
-        }
-        long availableMemory = osProbe.getAvailableMemorySize();
-        long unusedMemory;
-        if (availableMemory >= 0) {
-            // Use available memory (includes reclaimable cache) for a more accurate picture
-            unusedMemory = availableMemory;
-        } else {
-            LOGGER.warn("unable to retrieve available memory");
-            return 0;
-        }
-        long usedMemory = totalMemory - unusedMemory;
-        long usage = usedMemory * 100 / totalMemory;
-        LOGGER.debug("Recording native memory usage: {}%", usage);
-        return usage;
+
+        long rssAnon = rssAnonSupplier.getAsLong();
+        if (rssAnon < 0L) {
+            LOGGER.debug("Native memory poll skipped: RssAnon unavailable from /proc/self/status");
+            return 0L;
+        }
+
+        long heapCommitted = heapCommittedSupplier.getAsLong();
+        long usage = Math.max(0L, rssAnon - heapCommitted);
+
+        long cap = nativeMemoryCapSupplier.getAsLong();
+        if (cap <= 0L) {
+            LOGGER.debug("Native memory poll: rssAnon={} heapCommitted={} usage={} cap=0 -> 0%", rssAnon, heapCommitted, usage);
+            return 0L;
+        }
+
+        long percent = (usage * 100L) / cap;
+        if (percent > 100L) {
+            percent = 100L;
+        }
+        if (percent < 0L) {
+            percent = 0L;
+        }
+
+        LOGGER.debug("Native memory poll: rssAnon={} heapCommitted={} usage={} cap={} pct={}", rssAnon, heapCommitted, usage, cap, percent);
+        return percent;
+    }
+
+
+    /**
+     * Computes {@code Non_Heap_Base = max(0, totalPhysicalMemory - Runtime.maxMemory())}, the
+     * reference value against which {@code parquet.max_native_allocation} percentages resolve.
+     */
+    long computeNonHeapBase() {
+        return Math.max(0L, OsProbe.getInstance().getTotalPhysicalMemorySize() - Runtime.getRuntime().maxMemory());
+    }
+
+    /**
+     * Resolves the DataFusion native-pool contribution. Absent, zero, negative, or unparseable
+     * values all collapse to {@code 0L}.
+     */
+    long resolveDataFusionContribution(Settings settings) {
+        try {
+            long value = settings.getAsLong(DATAFUSION_MEMORY_POOL_LIMIT_KEY, 0L);
+            return value > 0L ? value : 0L;
+        } catch (IllegalArgumentException unparseable) {
+            return 0L;
+        }
+    }
+
+    /**
+     * Resolves the Parquet native-allocation contribution from the percentage-string setting.
+     * Absent, empty, missing trailing {@code %}, unparseable, NaN, or non-positive values all
+     * collapse to {@code 0L}. Percentages above 100 are defensively clamped to 100.
+     */
+    long resolveParquetContribution(Settings settings, long nonHeapBase) {
+        if (nonHeapBase <= 0L) {
+            return 0L;
+        }
+        String raw = settings.get(PARQUET_MAX_NATIVE_ALLOCATION_KEY);
+        if (raw == null) {
+            return 0L;
+        }
+        String trimmed = raw.trim();
+        if (trimmed.isEmpty() || trimmed.endsWith("%") == false) {
+            return 0L;
+        }
+        String pctStr = trimmed.substring(0, trimmed.length() - 1).trim();
+        double pct;
+        try {
+            pct = Double.parseDouble(pctStr);
+        } catch (NumberFormatException nfe) {
+            return 0L;
+        }
+        if (Double.isNaN(pct) || pct <= 0.0) {
+            return 0L;
+        }
+        if (pct > 100.0) {
+            pct = 100.0;
+        }
+        return (long) Math.floor(nonHeapBase * pct / 100.0);
+    }
+
+    /**
+     * Resolves both plugin settings to byte values and returns the 20% headroom-padded sum.
+     * Returns {@code 0L} when neither plugin setting contributes a positive value, so
+     * {@link #getUsage()} can short-circuit without dividing.
+     */
+    long computeNativeMemoryCap(Settings settings) {
+        long df = resolveDataFusionContribution(settings);
+        long base = computeNonHeapBase();
+        long pq = resolveParquetContribution(settings, base);
+        long sum = df + pq;
+        if (sum <= 0L) {
+            return 0L;
+        }
+        return (long) Math.floor(HEADROOM_FACTOR * (double) sum);
     }
 }

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

@@ -23,6 +23,7 @@
 public class NodeResourceUsageTracker extends AbstractLifecycleComponent {
     private ThreadPool threadPool;
     private final ClusterSettings clusterSettings;
+    private final Settings settings;
     private AverageCpuUsageTracker cpuUsageTracker;
     private AverageMemoryUsageTracker memoryUsageTracker;
     private AverageIoUsageTracker ioUsageTracker;
@@ -35,6 +36,7 @@ public class NodeResourceUsageTracker extends AbstractLifecycleComponent {
     public NodeResourceUsageTracker(FsService fsService, ThreadPool threadPool, Settings settings, ClusterSettings clusterSettings) {
         this.fsService = fsService;
         this.threadPool = threadPool;
+        this.settings = settings;
         this.clusterSettings = clusterSettings;
         this.resourceTrackerSettings = new ResourceTrackerSettings(settings);
         initialize();
@@ -125,7 +127,8 @@ void initialize() {
         nativeMemoryUsageTracker = new AverageNativeMemoryUsageTracker(
             threadPool,
             resourceTrackerSettings.getNativeMemoryPollingInterval(),
-            resourceTrackerSettings.getNativeMemoryWindowDuration()
+            resourceTrackerSettings.getNativeMemoryWindowDuration(),
+            settings
         );
         if (Constants.LINUX) {
             clusterSettings.addSettingsUpdateConsumer(

server/src/test/java/org/opensearch/monitor/os/OsProbeTests.java

@@ -310,39 +310,6 @@ List<String> readSysFsCgroupCpuAcctCpuStat(String controlGroup) throws IOExcepti
         verify(logger, never()).warn(anyString());
     }
 
-    public void testGetAvailableMemorySizeLinux() {
-        assumeTrue("test runs on Linux only", Constants.LINUX);
-
-        final OsProbe probe = new OsProbe() {
-            @Override
-            long readAvailableMemoryFromProcMeminfo() {
-                return 8388608L; // 8 MB in bytes
-            }
-        };
-
-        assertThat(probe.getAvailableMemorySize(), equalTo(8388608L));
-    }
-
-    public void testGetAvailableMemorySizeNonLinux() {
-        assumeFalse("test does not run on Linux", Constants.LINUX);
-
-        final OsProbe probe = new OsProbe();
-        assertThat(probe.getAvailableMemorySize(), equalTo(-1L));
-    }
-
-    public void testGetAvailableMemorySizeWhenReadThrowsException() {
-        assumeTrue("test runs on Linux only", Constants.LINUX);
-
-        final OsProbe probe = new OsProbe() {
-            @Override
-            long readAvailableMemoryFromProcMeminfo() throws IOException {
-                throw new IOException("simulated /proc/meminfo read failure");
-            }
-        };
-
-        assertThat(probe.getAvailableMemorySize(), equalTo(-1L));
-    }
-
     private static List<String> getProcSelfGroupLines(String hierarchy) {
         return Arrays.asList(
             "10:freezer:/",