native memory based admission control

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

@@ -145,6 +145,9 @@ 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 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/-
 ReadOnlyPaths=/sys/fs/cgroup/cpuacct /sys/fs/cgroup/cpuacct/- /sys/fs/cgroup/memory /sys/fs/cgroup/memory/-

qa/systemd-test/src/test/java/org/opensearch/systemdinteg/SystemdIntegTests.java

@@ -97,7 +97,7 @@ private static String executeCommand(String command, String errorMessage) throws
     public void testReadOnlyPaths() throws IOException, InterruptedException {
         String[] readOnlyPaths = {
             "/etc/os-release", "/usr/lib/os-release", "/etc/system-release",
-            "/proc/self/mountinfo", "/proc/diskstats",
+            "/proc/self/mountinfo", "/proc/diskstats", "/proc/meminfo",
             "/proc/self/cgroup", "/sys/fs/cgroup/cpu", "/sys/fs/cgroup/cpu/-",
             "/sys/fs/cgroup/cpuacct", "/sys/fs/cgroup/cpuacct/-",
             "/sys/fs/cgroup/memory", "/sys/fs/cgroup/memory/-"

server/src/internalClusterTest/java/org/opensearch/cluster/ClusterInfoServiceIT.java

@@ -260,7 +260,8 @@ public void testClusterInfoServiceCollectsNodeResourceStatsInformation() throws
         Settings.Builder settingsBuilder = Settings.builder()
             .put(ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueSeconds(1))
             .put(ResourceTrackerSettings.GLOBAL_IO_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueMillis(5000))
-            .put(ResourceTrackerSettings.GLOBAL_CPU_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueSeconds(1));
+            .put(ResourceTrackerSettings.GLOBAL_CPU_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueSeconds(1))
+            .put(ResourceTrackerSettings.GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueMillis(1000));
 
         internalCluster().startClusterManagerOnlyNode(settingsBuilder.build());
         internalCluster().startWarmOnlyNodes(1, settingsBuilder.build());
@@ -292,6 +293,7 @@ public void testClusterInfoServiceInformationClearOnError() throws InterruptedEx
                 .put(ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueMillis(500))
                 .put(ResourceTrackerSettings.GLOBAL_IO_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueMillis(5000))
                 .put(ResourceTrackerSettings.GLOBAL_CPU_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueMillis(500))
+                .put(ResourceTrackerSettings.GLOBAL_NATIVE_MEMORY_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueMillis(500))
                 .build()
         );
         prepareCreate("test").setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)).get();

server/src/internalClusterTest/java/org/opensearch/ratelimitting/admissioncontrol/AdmissionForClusterManagerIT.java

@@ -10,6 +10,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.lucene.util.Constants;
 import org.opensearch.action.admin.indices.alias.get.GetAliasesRequest;
 import org.opensearch.action.admin.indices.alias.get.GetAliasesResponse;
 import org.opensearch.action.support.clustermanager.term.GetTermVersionAction;
@@ -25,6 +26,7 @@
 import org.opensearch.node.resource.tracker.ResourceTrackerSettings;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.ratelimitting.admissioncontrol.controllers.CpuBasedAdmissionController;
+import org.opensearch.ratelimitting.admissioncontrol.controllers.NativeMemoryBasedAdmissionController;
 import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlActionType;
 import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlMode;
 import org.opensearch.ratelimitting.admissioncontrol.stats.AdmissionControllerStats;
@@ -47,6 +49,8 @@
 
 import static org.opensearch.ratelimitting.admissioncontrol.AdmissionControlSettings.ADMISSION_CONTROL_TRANSPORT_LAYER_MODE;
 import static org.opensearch.ratelimitting.admissioncontrol.settings.CpuBasedAdmissionControllerSettings.CLUSTER_ADMIN_CPU_USAGE_LIMIT;
+import static org.opensearch.ratelimitting.admissioncontrol.settings.NativeMemoryBasedAdmissionControllerSettings.CLUSTER_ADMIN_NATIVE_MEMORY_USAGE_LIMIT;
+import static org.opensearch.ratelimitting.admissioncontrol.settings.NativeMemoryBasedAdmissionControllerSettings.NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -63,6 +67,7 @@ public class AdmissionForClusterManagerIT extends OpenSearchIntegTestCase {
 
     private static final Settings DISABLE_ADMISSION_CONTROL = Settings.builder()
         .put(ADMISSION_CONTROL_TRANSPORT_LAYER_MODE.getKey(), AdmissionControlMode.DISABLED.getMode())
+        .put(NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(), AdmissionControlMode.DISABLED.getMode())
         .build();
 
     private static final Settings ENFORCE_ADMISSION_CONTROL = Settings.builder()
@@ -114,7 +119,14 @@ public void init() {
     }
 
     public void testAdmissionControlEnforced() throws Exception {
-        cMResourceCollector.collectNodeResourceUsageStats(clusterManagerNodeId, System.currentTimeMillis(), 97, 99, new IoUsageStats(98));
+        cMResourceCollector.collectNodeResourceUsageStats(
+            clusterManagerNodeId,
+            System.currentTimeMillis(),
+            97,
+            99,
+            new IoUsageStats(98),
+            90
+        );
 
         // Write API on ClusterManager
         assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}"));
@@ -149,7 +161,14 @@ public void testAdmissionControlEnforced() throws Exception {
 
     public void testAdmissionControlEnabledOnNoBreach() throws InterruptedException {
         // CPU usage is less than threshold 50%
-        cMResourceCollector.collectNodeResourceUsageStats(clusterManagerNodeId, System.currentTimeMillis(), 97, 35, new IoUsageStats(98));
+        cMResourceCollector.collectNodeResourceUsageStats(
+            clusterManagerNodeId,
+            System.currentTimeMillis(),
+            97,
+            35,
+            new IoUsageStats(98),
+            90
+        );
 
         // Write API on ClusterManager
         assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}").execute().actionGet());
@@ -174,7 +193,14 @@ public void testAdmissionControlDisabledOnBreach() throws InterruptedException {
     public void admissionControlDisabledOnBreach(Settings admission) throws InterruptedException {
         client().admin().cluster().prepareUpdateSettings().setTransientSettings(admission).execute().actionGet();
 
-        cMResourceCollector.collectNodeResourceUsageStats(clusterManagerNodeId, System.currentTimeMillis(), 97, 97, new IoUsageStats(98));
+        cMResourceCollector.collectNodeResourceUsageStats(
+            clusterManagerNodeId,
+            System.currentTimeMillis(),
+            97,
+            97,
+            new IoUsageStats(98),
+            90
+        );
 
         // Write API on ClusterManager
         assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}").execute().actionGet());
@@ -187,8 +213,105 @@ public void admissionControlDisabledOnBreach(Settings admission) throws Interrup
 
     }
 
+    public void testAdmissionControlEnforcedNativeMemory() throws Exception {
+        assumeTrue("native memory controller is Linux-only", Constants.LINUX);
+        Settings enforceNativeMemory = Settings.builder()
+            .put(NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(), AdmissionControlMode.ENFORCED.getMode())
+            .put(CLUSTER_ADMIN_NATIVE_MEMORY_USAGE_LIMIT.getKey(), 50)
+            .build();
+        client().admin().cluster().prepareUpdateSettings().setTransientSettings(enforceNativeMemory).execute().actionGet();
+
+        cMResourceCollector.collectNodeResourceUsageStats(
+            clusterManagerNodeId,
+            System.currentTimeMillis(),
+            97,
+            35,
+            new IoUsageStats(98),
+            90
+        );
+
+        // Write API on ClusterManager
+        assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}"));
+
+        // Read API on ClusterManager
+        GetAliasesRequest aliasesRequest = new GetAliasesRequest();
+        aliasesRequest.aliases("alias1");
+        try {
+            dataNodeClient().admin().indices().getAliases(aliasesRequest).actionGet();
+            fail("expected failure");
+        } catch (Exception e) {
+            assertTrue(e instanceof OpenSearchRejectedExecutionException);
+            assertTrue(e.getMessage().contains("Memory usage admission controller rejected the request"));
+            assertTrue(e.getMessage().contains("[indices:admin/aliases/get]"));
+            assertTrue(e.getMessage().contains("action-type [CLUSTER_ADMIN]"));
+        }
+
+        client().admin().cluster().prepareUpdateSettings().setTransientSettings(DISABLE_ADMISSION_CONTROL).execute().actionGet();
+        GetAliasesResponse getAliasesResponse = dataNodeClient().admin().indices().getAliases(aliasesRequest).actionGet();
+        assertThat(getAliasesResponse.getAliases().get("test").size(), equalTo(1));
+
+        AdmissionControlService admissionControlServiceCM = internalCluster().getClusterManagerNodeInstance(AdmissionControlService.class);
+
+        AdmissionControllerStats admissionStats = getAdmissionControlStats(admissionControlServiceCM).get(
+            NativeMemoryBasedAdmissionController.NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER
+        );
+
+        assertEquals(admissionStats.rejectionCount.get(AdmissionControlActionType.CLUSTER_ADMIN.getType()).longValue(), 1);
+        assertNull(admissionStats.rejectionCount.get(AdmissionControlActionType.SEARCH.getType()));
+        assertNull(admissionStats.rejectionCount.get(AdmissionControlActionType.INDEXING.getType()));
+    }
+
+    public void testAdmissionControlMonitorOnBreachNativeMemory() throws InterruptedException {
+        assumeTrue("native memory controller is Linux-only", Constants.LINUX);
+        admissionControlDisabledOnBreachNativeMemory(
+            Settings.builder()
+                .put(NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(), AdmissionControlMode.MONITOR.getMode())
+                .put(CLUSTER_ADMIN_NATIVE_MEMORY_USAGE_LIMIT.getKey(), 50)
+                .build()
+        );
+    }
+
+    public void testAdmissionControlDisabledOnBreachNativeMemory() throws InterruptedException {
+        assumeTrue("native memory controller is Linux-only", Constants.LINUX);
+        admissionControlDisabledOnBreachNativeMemory(
+            Settings.builder()
+                .put(NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE.getKey(), AdmissionControlMode.DISABLED.getMode())
+                .put(CLUSTER_ADMIN_NATIVE_MEMORY_USAGE_LIMIT.getKey(), 30)
+                .build()
+        );
+    }
+
+    public void admissionControlDisabledOnBreachNativeMemory(Settings admission) throws InterruptedException {
+        client().admin().cluster().prepareUpdateSettings().setTransientSettings(admission).execute().actionGet();
+
+        cMResourceCollector.collectNodeResourceUsageStats(
+            clusterManagerNodeId,
+            System.currentTimeMillis(),
+            97,
+            35,
+            new IoUsageStats(98),
+            90
+        );
+
+        // Write API on ClusterManager
+        assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}").execute().actionGet());
+
+        // Read API on ClusterManager
+        GetAliasesRequest aliasesRequest = new GetAliasesRequest();
+        aliasesRequest.aliases("alias1");
+        GetAliasesResponse getAliasesResponse = dataNodeClient().admin().indices().getAliases(aliasesRequest).actionGet();
+        assertThat(getAliasesResponse.getAliases().get("test").size(), equalTo(1));
+    }
+
     public void testAdmissionControlResponseStatus() throws Exception {
-        cMResourceCollector.collectNodeResourceUsageStats(clusterManagerNodeId, System.currentTimeMillis(), 97, 99, new IoUsageStats(98));
+        cMResourceCollector.collectNodeResourceUsageStats(
+            clusterManagerNodeId,
+            System.currentTimeMillis(),
+            97,
+            99,
+            new IoUsageStats(98),
+            90
+        );
 
         // Write API on ClusterManager
         assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}"));

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

@@ -156,6 +156,7 @@
 import org.opensearch.ratelimitting.admissioncontrol.AdmissionControlSettings;
 import org.opensearch.ratelimitting.admissioncontrol.settings.CpuBasedAdmissionControllerSettings;
 import org.opensearch.ratelimitting.admissioncontrol.settings.IoBasedAdmissionControllerSettings;
+import org.opensearch.ratelimitting.admissioncontrol.settings.NativeMemoryBasedAdmissionControllerSettings;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
 import org.opensearch.repositories.fs.FsRepository;
 import org.opensearch.rest.BaseRestHandler;
@@ -755,6 +756,9 @@ public void apply(Settings value, Settings current, Settings previous) {
                 ResourceTrackerSettings.GLOBAL_CPU_USAGE_AC_WINDOW_DURATION_SETTING,
                 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,
@@ -819,6 +823,10 @@ public void apply(Settings value, Settings current, Settings previous) {
                 IoBasedAdmissionControllerSettings.IO_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE,
                 IoBasedAdmissionControllerSettings.SEARCH_IO_USAGE_LIMIT,
                 IoBasedAdmissionControllerSettings.INDEXING_IO_USAGE_LIMIT,
+                NativeMemoryBasedAdmissionControllerSettings.NATIVE_MEMORY_BASED_ADMISSION_CONTROLLER_TRANSPORT_LAYER_MODE,
+                NativeMemoryBasedAdmissionControllerSettings.SEARCH_NATIVE_MEMORY_USAGE_LIMIT,
+                NativeMemoryBasedAdmissionControllerSettings.INDEXING_NATIVE_MEMORY_USAGE_LIMIT,
+                NativeMemoryBasedAdmissionControllerSettings.CLUSTER_ADMIN_NATIVE_MEMORY_USAGE_LIMIT,
 
                 // Concurrent segment search settings
                 SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING, // deprecated

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

@@ -236,6 +236,71 @@ String readProcLoadavg() throws IOException {
         return readSingleLine(PathUtils.get("/proc/loadavg"));
     }
 
+    /**
+     * 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.
+     *
+     * @return the {@code RssAnon} value in bytes, or {@code -1L} if unavailable
+     */
+    public long getProcessRssAnon() {
+        if (Constants.LINUX == false) {
+            return -1L;
+        }
+        try {
+            return readRssAnonFromProcSelfStatus();
+        } catch (IOException e) {
+            logger.warn("failed to read /proc/self/status", e);
+            return -1L;
+        }
+    }
+
+    /**
+     * Reads the lines of {@code /proc/self/status}. Package-private so tests can override it
+     * with canned file contents.
+     *
+     * @return the lines of {@code /proc/self/status}
+     * @throws IOException if the file cannot be opened or read
+     */
+    @SuppressForbidden(reason = "access /proc/self/status")
+    List<String> readProcSelfStatus() throws IOException {
+        return Files.readAllLines(PathUtils.get("/proc/self/status"));
+    }
+
+    /**
+     * Reads the {@code RssAnon} field from {@code /proc/self/status}.
+     *
+     * @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
+     */
+    long readRssAnonFromProcSelfStatus() throws IOException {
+        for (final String line : readProcSelfStatus()) {
+            if (line.startsWith("RssAnon:")) {
+                // Format: "RssAnon:\t 12345 kB"
+                final String[] parts = line.split("\\s+");
+                if (parts.length >= 2) {
+                    try {
+                        final long kb = Long.parseLong(parts[1]);
+                        if (kb < 0L) {
+                            return -1L;
+                        }
+                        return kb * 1024L;
+                    } catch (NumberFormatException nfe) {
+                        logger.warn("malformed RssAnon value in /proc/self/status", nfe);
+                        return -1L;
+                    }
+                }
+                logger.warn("RssAnon line has unexpected shape: [{}]", line);
+                return -1L;
+            }
+        }
+        logger.warn("RssAnon line not found in /proc/self/status");
+        return -1L;
+    }
+
     public short getSystemCpuPercent() {
         return Probes.getLoadAndScaleToPercent(getSystemCpuLoad, osMxBean);
     }