Add node-level native memory tracker budget

Replace the datafusion/parquet-derived cap in
AverageNativeMemoryUsageTracker with two explicit node-scope dynamic
settings:

  node.native_memory.limit         (ByteSizeValue, default 0b)
  node.native_memory.buffer_percent (int 0-99, default 0)

The tracker now divides observed native-memory use
(max(0, RssAnon - HeapCommitted)) by
limit - (limit * bufferPercent / 100). When the limit is unset or the
buffer fully consumes it, getUsage() returns 0 without dividing.

Also: allow read access to /proc/self/status in the Java security
policy and systemd unit so OsProbe.getProcessRssAnon() works in
hardened installs.

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

Author: Pradeep L

distribution/packages/src/common/systemd/opensearch.service

@@ -145,8 +145,8 @@ ReadOnlyPaths=-/etc/os-release -/usr/lib/os-release -/etc/system-release
 ## Allow read access to Linux IO stats
 ReadOnlyPaths=/proc/self/mountinfo /proc/diskstats
 
-## Allow read access to Linux memory stats
-ReadOnlyPaths=/proc/meminfo
+## Allow read access to native memory stats (RssAnon via /proc/self/status)
+ReadOnlyPaths=/proc/self/status
 
 ## Allow read access to control group stats
 ReadOnlyPaths=/proc/self/cgroup /sys/fs/cgroup/cpu /sys/fs/cgroup/cpu/-

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

@@ -758,6 +758,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING,
                 ResourceTrackerSettings.GLOBAL_IO_USAGE_AC_WINDOW_DURATION_SETTING,
                 ResourceTrackerSettings.GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING,
+                ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING,
+                ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING,
 
                 // Settings related to Searchable Snapshots
                 Node.NODE_SEARCH_CACHE_SIZE_SETTING,

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

@@ -253,7 +253,7 @@ public long getProcessRssAnon() {
         try {
             return readRssAnonFromProcSelfStatus();
         } catch (IOException e) {
-            logger.debug("failed to read /proc/self/status", e);
+            logger.warn("failed to read /proc/self/status", e);
             return -1L;
         }
     }
@@ -282,15 +282,15 @@ long readRssAnonFromProcSelfStatus() throws IOException {
                             }
                             return kb * 1024L;
                         } catch (NumberFormatException nfe) {
-                            logger.debug("malformed RssAnon value in /proc/self/status", nfe);
+                            logger.warn("malformed RssAnon value in /proc/self/status", nfe);
                             return -1L;
                         }
                     }
-                    logger.debug("RssAnon line has unexpected shape: [{}]", line);
+                    logger.warn("RssAnon line has unexpected shape: [{}]", line);
                     return -1L;
                 }
             }
-            logger.debug("RssAnon line not found in /proc/self/status");
+            logger.warn("RssAnon line not found in /proc/self/status");
             return -1L;
         }
     }

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

@@ -12,25 +12,26 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.unit.ByteSizeValue;
 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 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.
+ * utilization as a percentage of a configured node-level budget, 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.
+ * bean. The denominator is the effective native memory budget:
+ * {@code effective = limit - (limit * bufferPercent / 100)}, resolved per poll from
+ * {@link ResourceTrackerSettings#NODE_NATIVE_MEMORY_LIMIT_SETTING} and
+ * {@link ResourceTrackerSettings#NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING}. When the limit is
+ * absent or non-positive, or the buffer fully consumes the limit, {@code getUsage()} returns
+ * {@code 0} without dividing.
  *
  * <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
@@ -40,144 +41,104 @@ public class AverageNativeMemoryUsageTracker extends AbstractAverageUsageTracker
 
     private static final Logger LOGGER = LogManager.getLogger(AverageNativeMemoryUsageTracker.class);
 
-    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;
+    private final LongSupplier effectiveNativeMemorySupplier;
 
     /**
      * 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.
+     * heap-committed supplier to the JVM memory MX bean, and the effective-native-memory supplier
+     * to {@link #computeEffectiveNativeMemory(Settings)} so dynamic updates to the two node
+     * settings 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);
+        this.effectiveNativeMemorySupplier = () -> computeEffectiveNativeMemory(settings);
     }
 
     /**
      * 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.
+     * reading {@code /proc/self/status}, the JVM memory MX bean, or node settings. The third
+     * supplier provides the effective native memory (limit minus buffer) directly.
      */
     AverageNativeMemoryUsageTracker(
         ThreadPool threadPool,
         TimeValue pollingInterval,
         TimeValue windowDuration,
         LongSupplier rssAnonSupplier,
         LongSupplier heapCommittedSupplier,
-        LongSupplier nativeMemoryCapSupplier
+        LongSupplier effectiveNativeMemorySupplier
     ) {
         super(threadPool, pollingInterval, windowDuration);
         this.rssAnonSupplier = rssAnonSupplier;
         this.heapCommittedSupplier = heapCommittedSupplier;
-        this.nativeMemoryCapSupplier = nativeMemoryCapSupplier;
+        this.effectiveNativeMemorySupplier = effectiveNativeMemorySupplier;
     }
 
     @Override
     public long getUsage() {
 
         long rssAnon = rssAnonSupplier.getAsLong();
         if (rssAnon < 0L) {
-            LOGGER.debug("Native memory poll skipped: RssAnon unavailable from /proc/self/status");
+            LOGGER.warn("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);
+        long nativeUsed = Math.max(0L, rssAnon - heapCommitted);
+
+        long effectiveNativeMemory = effectiveNativeMemorySupplier.getAsLong();
+        if (effectiveNativeMemory <= 0L) {
+            LOGGER.warn(
+                "Native memory poll: rssAnon={} heapCommitted={} nativeUsed={} effectiveNativeMemory=0 -> 0%",
+                rssAnon,
+                heapCommitted,
+                nativeUsed
+            );
             return 0L;
         }
 
-        long percent = (usage * 100L) / cap;
+        double utilization = (double) nativeUsed / effectiveNativeMemory * 100.0;
+        long percent = (long) utilization;
         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;
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug(
+                "Native memory poll: rssAnon={} heapCommitted={} nativeUsed={} effectiveNativeMemory={} pct={}",
+                rssAnon,
+                heapCommitted,
+                nativeUsed,
+                effectiveNativeMemory,
+                percent
+            );
         }
-    }
 
-    /**
-     * 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);
+        return percent;
     }
 
     /**
-     * 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.
+     * Resolves the configured native-memory limit and buffer percentage from the node settings and
+     * returns the effective native-memory budget in bytes: {@code limit - (limit * buffer / 100)}.
+     * Returns {@code 0L} when the limit is absent or non-positive, or when the buffer fully
+     * consumes the limit, 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) {
+    long computeEffectiveNativeMemory(Settings settings) {
+        ByteSizeValue limitValue = ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING.get(settings);
+        long limit = limitValue.getBytes();
+        if (limit <= 0L) {
             return 0L;
         }
-        return (long) Math.floor(HEADROOM_FACTOR * (double) sum);
+        int bufferPercent = ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.get(settings);
+        long buffer = limit * bufferPercent / 100L;
+        long effective = limit - buffer;
+        return Math.max(0L, effective);
     }
 }

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

@@ -11,6 +11,7 @@
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.unit.ByteSizeValue;
 
 /**
  * Settings related to resource usage trackers such as polling interval, window duration etc
@@ -86,6 +87,30 @@ private static class Defaults {
         Setting.Property.NodeScope
     );
 
+    /**
+     * Absolute native-memory budget for this node, in bytes. When the value is {@link ByteSizeValue#ZERO}
+     * (default) the tracker treats the budget as unconfigured and reports {@code 0%}.
+     */
+    public static final Setting<ByteSizeValue> NODE_NATIVE_MEMORY_LIMIT_SETTING = Setting.byteSizeSetting(
+        "node.native_memory.limit",
+        ByteSizeValue.ZERO,
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+
+    /**
+     * Percentage of the native-memory limit that is reserved as buffer (not usable). The effective
+     * native memory the tracker divides against is {@code limit - (limit * bufferPercent / 100)}.
+     */
+    public static final Setting<Integer> NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING = Setting.intSetting(
+        "node.native_memory.buffer_percent",
+        0,
+        0,
+        100,
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+
     private volatile TimeValue cpuWindowDuration;
     private volatile TimeValue cpuPollingInterval;
     private volatile TimeValue memoryWindowDuration;

server/src/main/resources/org/opensearch/bootstrap/security.policy

@@ -235,8 +235,9 @@ grant {
   permission java.io.FilePermission "/proc/self/mountinfo", "read";
   permission java.io.FilePermission "/proc/diskstats", "read";
 
-  // memory stats on Linux
-  permission java.io.FilePermission "/proc/meminfo", "read";
+  // native memory stats on Linux (RssAnon from /proc/self/status)
+  permission java.io.FilePermission "/proc/self/status", "read";
+
 
   // control group stats on Linux
   // TODO: update later when wildcard is supported in policy