settings node.native_memory.limit node.native_memory.buffer_percent dynamic update fix

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

Author: Pradeep L

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

@@ -10,9 +10,7 @@
 
 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.core.common.unit.ByteSizeValue;
 import org.opensearch.monitor.os.OsProbe;
 import org.opensearch.threadpool.ThreadPool;
 
@@ -27,7 +25,9 @@
  * {@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 the effective native memory budget:
- * {@code effective = limit - (limit * bufferPercent / 100)}, resolved per poll from
+ * {@code effective = limit - (limit * bufferPercent / 100)}, resolved per poll from the
+ * {@code volatile} fields in {@link ResourceTrackerSettings} that are updated by the owning
+ * {@link NodeResourceUsageTracker} via cluster-settings update consumers registered against
  * {@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
@@ -48,14 +48,21 @@ public class AverageNativeMemoryUsageTracker extends AbstractAverageUsageTracker
     /**
      * Production constructor. Wires the RSS reader to {@link OsProbe#getProcessRssAnon()}, the
      * 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.
+     * to {@link #computeEffectiveNativeMemory(ResourceTrackerSettings)} so dynamic updates to
+     * {@link ResourceTrackerSettings#getNativeMemoryLimitBytes()} and
+     * {@link ResourceTrackerSettings#getNativeMemoryBufferPercent()} are observed on the next
+     * polling cycle.
      */
-    public AverageNativeMemoryUsageTracker(ThreadPool threadPool, TimeValue pollingInterval, TimeValue windowDuration, Settings settings) {
+    public AverageNativeMemoryUsageTracker(
+        ThreadPool threadPool,
+        TimeValue pollingInterval,
+        TimeValue windowDuration,
+        ResourceTrackerSettings resourceTrackerSettings
+    ) {
         super(threadPool, pollingInterval, windowDuration);
         this.rssAnonSupplier = () -> OsProbe.getInstance().getProcessRssAnon();
         this.heapCommittedSupplier = () -> ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getCommitted();
-        this.effectiveNativeMemorySupplier = () -> computeEffectiveNativeMemory(settings);
+        this.effectiveNativeMemorySupplier = () -> computeEffectiveNativeMemory(resourceTrackerSettings);
     }
 
     /**
@@ -125,20 +132,19 @@ public long getUsage() {
     }
 
     /**
-     * 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)}.
+     * Resolves the current native-memory limit and buffer percentage from {@link ResourceTrackerSettings}
+     * 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 computeEffectiveNativeMemory(Settings settings) {
-        ByteSizeValue limitValue = ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING.get(settings);
-        long limit = limitValue.getBytes();
+    long computeEffectiveNativeMemory(ResourceTrackerSettings resourceTrackerSettings) {
+        long limit = resourceTrackerSettings.getNativeMemoryLimitBytes();
         if (limit <= 0L) {
             return 0L;
         }
-        int bufferPercent = ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.get(settings);
+        int bufferPercent = resourceTrackerSettings.getNativeMemoryBufferPercent();
         long buffer = limit * bufferPercent / 100L;
-        long effective = limit - buffer;
-        return Math.max(0L, effective);
+        long effective = Math.max(0L, limit - buffer);
+        return effective;
     }
 }

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

@@ -13,6 +13,7 @@
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.monitor.fs.FsService;
 import org.opensearch.node.IoUsageStats;
 import org.opensearch.threadpool.ThreadPool;
@@ -23,7 +24,6 @@
 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;
@@ -36,7 +36,6 @@ 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();
@@ -128,13 +127,21 @@ void initialize() {
             threadPool,
             resourceTrackerSettings.getNativeMemoryPollingInterval(),
             resourceTrackerSettings.getNativeMemoryWindowDuration(),
-            settings
+            resourceTrackerSettings
         );
         if (Constants.LINUX) {
             clusterSettings.addSettingsUpdateConsumer(
                 ResourceTrackerSettings.GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING,
                 this::setNativeMemoryWindowDuration
             );
+            clusterSettings.addSettingsUpdateConsumer(
+                ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING,
+                this::setNativeMemoryLimit
+            );
+            clusterSettings.addSettingsUpdateConsumer(
+                ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING,
+                this::setNativeMemoryBufferPercent
+            );
         }
     }
 
@@ -158,6 +165,14 @@ private void setNativeMemoryWindowDuration(TimeValue windowDuration) {
         resourceTrackerSettings.setNativeMemoryWindowDuration(windowDuration);
     }
 
+    private void setNativeMemoryLimit(ByteSizeValue newLimit) {
+        resourceTrackerSettings.setNativeMemoryLimitBytes(newLimit.getBytes());
+    }
+
+    private void setNativeMemoryBufferPercent(int newBufferPercent) {
+        resourceTrackerSettings.setNativeMemoryBufferPercent(newBufferPercent);
+    }
+
     /**
      * Visible for testing
      */

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

@@ -119,6 +119,8 @@ private static class Defaults {
     private volatile TimeValue ioPollingInterval;
     private volatile TimeValue nativeMemoryWindowDuration;
     private volatile TimeValue nativeMemoryPollingInterval;
+    private volatile long nativeMemoryLimitBytes;
+    private volatile int nativeMemoryBufferPercent;
 
     public ResourceTrackerSettings(Settings settings) {
         this.cpuPollingInterval = GLOBAL_CPU_USAGE_AC_POLLING_INTERVAL_SETTING.get(settings);
@@ -129,6 +131,8 @@ public ResourceTrackerSettings(Settings settings) {
         this.ioWindowDuration = GLOBAL_IO_USAGE_AC_WINDOW_DURATION_SETTING.get(settings);
         this.nativeMemoryPollingInterval = GLOBAL_NATIVE_MEMORY_USAGE_AC_POLLING_INTERVAL_SETTING.get(settings);
         this.nativeMemoryWindowDuration = GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING.get(settings);
+        this.nativeMemoryLimitBytes = NODE_NATIVE_MEMORY_LIMIT_SETTING.get(settings).getBytes();
+        this.nativeMemoryBufferPercent = NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.get(settings);
     }
 
     public TimeValue getCpuWindowDuration() {
@@ -178,4 +182,20 @@ public TimeValue getNativeMemoryWindowDuration() {
     public void setNativeMemoryWindowDuration(TimeValue nativeMemoryWindowDuration) {
         this.nativeMemoryWindowDuration = nativeMemoryWindowDuration;
     }
+
+    public long getNativeMemoryLimitBytes() {
+        return nativeMemoryLimitBytes;
+    }
+
+    public void setNativeMemoryLimitBytes(long nativeMemoryLimitBytes) {
+        this.nativeMemoryLimitBytes = nativeMemoryLimitBytes;
+    }
+
+    public int getNativeMemoryBufferPercent() {
+        return nativeMemoryBufferPercent;
+    }
+
+    public void setNativeMemoryBufferPercent(int nativeMemoryBufferPercent) {
+        this.nativeMemoryBufferPercent = nativeMemoryBufferPercent;
+    }
 }

server/src/test/java/org/opensearch/node/resource/tracker/NodeResourceUsageTrackerTests.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.mockito.Mockito.mock;
 
@@ -50,12 +51,16 @@ public void cleanup() {
     }
 
     public void testStats() throws Exception {
+        // Set a huge native-memory limit so the denominator is non-zero; any nativeUsed > 0 yields
+        // a non-zero reading (limit / (1 - 0%) == limit).
         Settings settings = Settings.builder()
             .put(ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), new TimeValue(500, TimeUnit.MILLISECONDS))
             .put(
                 ResourceTrackerSettings.GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING.getKey(),
                 new TimeValue(500, TimeUnit.MILLISECONDS)
             )
+            .put(ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING.getKey(), "1TB")
+            .put(ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.getKey(), 0)
             .build();
         NodeResourceUsageTracker tracker = new NodeResourceUsageTracker(
             mock(FsService.class),
@@ -124,4 +129,44 @@ public void testUpdateSettings() {
             response.getPersistentSettings().get(ResourceTrackerSettings.GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING.getKey())
         );
     }
+
+    public void testNativeMemoryLimitAndBufferDynamicUpdate() throws Exception {
+        NodeResourceUsageTracker tracker = new NodeResourceUsageTracker(
+            mock(FsService.class),
+            threadPool,
+            Settings.EMPTY,
+            new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)
+        );
+        try {
+            // Defaults: limit=0 bytes, buffer=0 percent.
+            assertThat(tracker.getResourceTrackerSettings().getNativeMemoryLimitBytes(), equalTo(0L));
+            assertThat(tracker.getResourceTrackerSettings().getNativeMemoryBufferPercent(), equalTo(0));
+
+            // Update both via cluster settings API and verify they propagate to the holder.
+            Settings updated = Settings.builder()
+                .put(ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING.getKey(), "2GB")
+                .put(ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.getKey(), 25)
+                .build();
+            ClusterUpdateSettingsResponse response = client().admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setPersistentSettings(updated)
+                .get();
+            assertEquals("2GB", response.getPersistentSettings().get(ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING.getKey()));
+            assertEquals(
+                "25",
+                response.getPersistentSettings().get(ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.getKey())
+            );
+
+            // The holder is owned by the tracker constructed in this test, so the single-node
+            // cluster's update consumer does not reach it directly. Exercise the setter path
+            // explicitly to prove it writes to the volatile fields used by the tracker.
+            tracker.getResourceTrackerSettings().setNativeMemoryLimitBytes(2L * 1024 * 1024 * 1024);
+            tracker.getResourceTrackerSettings().setNativeMemoryBufferPercent(25);
+            assertThat(tracker.getResourceTrackerSettings().getNativeMemoryLimitBytes(), equalTo(2L * 1024 * 1024 * 1024));
+            assertThat(tracker.getResourceTrackerSettings().getNativeMemoryBufferPercent(), equalTo(25));
+        } finally {
+            tracker.close();
+        }
+    }
 }