Allow total memory to be overridden

distribution/tools/launchers/src/main/java/org/elasticsearch/tools/launchers/MachineDependentHeap.java

@@ -66,15 +66,32 @@ public List<String> determineHeapSettings(Path configDir, List<String> userDefin
 
         Path config = configDir.resolve(ELASTICSEARCH_YML);
         try (InputStream in = Files.newInputStream(config)) {
-            return determineHeapSettings(in);
+            return determineHeapSettings(in, parseForcedMemoryInBytes(userDefinedJvmOptions));
         }
     }
 
-    List<String> determineHeapSettings(InputStream config) {
+    static Long parseForcedMemoryInBytes(List<String> userDefinedJvmOptions) {
+        String totalMemoryBytesOption = userDefinedJvmOptions.stream()
+            .filter(option -> option.startsWith("-Des.total_memory_bytes="))
+            .findFirst()
+            .orElse(null);
+        if (totalMemoryBytesOption == null) {
+            return null;
+        }
+        try {
+            return Long.parseLong(totalMemoryBytesOption.split("=", 2)[1]);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("Unable to parse number of bytes from [" + totalMemoryBytesOption + "]");
+        }
+    }
+
+    List<String> determineHeapSettings(InputStream config, Long availableSystemMemory) {
         MachineNodeRole nodeRole = NodeRoleParser.parse(config);
 
         try {
-            long availableSystemMemory = systemMemoryInfo.availableSystemMemory();
+            if (availableSystemMemory == null) {
+                availableSystemMemory = systemMemoryInfo.availableSystemMemory();
+            }
             return options(nodeRole.heap(availableSystemMemory));
         } catch (SystemMemoryInfo.SystemMemoryInfoException e) {
             // If unable to determine system memory (ex: incompatible jdk version) fallback to defaults

distribution/tools/launchers/src/test/java/org/elasticsearch/tools/launchers/MachineDependentHeapTests.java

@@ -22,10 +22,15 @@
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 
 public class MachineDependentHeapTests extends LaunchersTestCase {
 
+    private static final long BYTES_IN_GB = 1024L * 1024L * 1024L;
+
     public void testDefaultHeapSize() throws Exception {
         MachineDependentHeap heap = new MachineDependentHeap(systemMemoryInGigabytes(8));
         List<String> options = heap.determineHeapSettings(configPath(), Collections.emptyList());
@@ -42,38 +47,68 @@ public void testUserPassedHeapArgs() throws Exception {
     }
 
     public void testMasterOnlyOptions() {
-        List<String> options = calculateHeap(16, "master");
+        List<String> options = calculateHeap(16, null, "master");
+        assertThat(options, containsInAnyOrder("-Xmx9830m", "-Xms9830m"));
+
+        calculateHeap(20, 16 * BYTES_IN_GB, "master");
         assertThat(options, containsInAnyOrder("-Xmx9830m", "-Xms9830m"));
 
-        options = calculateHeap(64, "master");
+        options = calculateHeap(64, null, "master");
+        assertThat(options, containsInAnyOrder("-Xmx31744m", "-Xms31744m"));
+
+        options = calculateHeap(77, 64 * BYTES_IN_GB, "master");
         assertThat(options, containsInAnyOrder("-Xmx31744m", "-Xms31744m"));
     }
 
     public void testMlOnlyOptions() {
-        List<String> options = calculateHeap(1, "ml");
+        List<String> options = calculateHeap(1, null, "ml");
+        assertThat(options, containsInAnyOrder("-Xmx409m", "-Xms409m"));
+
+        options = calculateHeap(4, BYTES_IN_GB, "ml");
         assertThat(options, containsInAnyOrder("-Xmx409m", "-Xms409m"));
 
-        options = calculateHeap(4, "ml");
+        options = calculateHeap(4, null, "ml");
         assertThat(options, containsInAnyOrder("-Xmx1024m", "-Xms1024m"));
 
-        options = calculateHeap(32, "ml");
+        options = calculateHeap(6, 4 * BYTES_IN_GB, "ml");
+        assertThat(options, containsInAnyOrder("-Xmx1024m", "-Xms1024m"));
+
+        options = calculateHeap(32, null, "ml");
+        assertThat(options, containsInAnyOrder("-Xmx2048m", "-Xms2048m"));
+
+        options = calculateHeap(100, 32 * BYTES_IN_GB, "ml");
         assertThat(options, containsInAnyOrder("-Xmx2048m", "-Xms2048m"));
     }
 
     public void testDataNodeOptions() {
-        List<String> options = calculateHeap(1, "data");
+        List<String> options = calculateHeap(1, null, "data");
+        assertThat(options, containsInAnyOrder("-Xmx512m", "-Xms512m"));
+
+        options = calculateHeap(5, BYTES_IN_GB, "data");
         assertThat(options, containsInAnyOrder("-Xmx512m", "-Xms512m"));
 
-        options = calculateHeap(8, "data");
+        options = calculateHeap(8, null, "data");
         assertThat(options, containsInAnyOrder("-Xmx4096m", "-Xms4096m"));
 
-        options = calculateHeap(64, "data");
+        options = calculateHeap(42, 8 * BYTES_IN_GB, "data");
+        assertThat(options, containsInAnyOrder("-Xmx4096m", "-Xms4096m"));
+
+        options = calculateHeap(64, null, "data");
+        assertThat(options, containsInAnyOrder("-Xmx31744m", "-Xms31744m"));
+
+        options = calculateHeap(65, 64 * BYTES_IN_GB, "data");
         assertThat(options, containsInAnyOrder("-Xmx31744m", "-Xms31744m"));
 
-        options = calculateHeap(0.5, "data");
+        options = calculateHeap(0.5, null, "data");
+        assertThat(options, containsInAnyOrder("-Xmx204m", "-Xms204m"));
+
+        options = calculateHeap(3, BYTES_IN_GB / 2, "data");
         assertThat(options, containsInAnyOrder("-Xmx204m", "-Xms204m"));
 
-        options = calculateHeap(0.2, "data");
+        options = calculateHeap(0.2, null, "data");
+        assertThat(options, containsInAnyOrder("-Xmx128m", "-Xms128m"));
+
+        options = calculateHeap(1, BYTES_IN_GB / 5, "data");
         assertThat(options, containsInAnyOrder("-Xmx128m", "-Xms128m"));
     }
 
@@ -83,15 +118,30 @@ public void testFallbackOptions() throws Exception {
         assertThat(options, containsInAnyOrder("-Xmx1024m", "-Xms1024m"));
     }
 
-    private static List<String> calculateHeap(double memoryInGigabytes, String... roles) {
+    public void testParseForcedMemoryInBytes() {
+        assertThat(MachineDependentHeap.parseForcedMemoryInBytes(List.of("-Da=b", "-Dx=y")), nullValue());
+        assertThat(
+            MachineDependentHeap.parseForcedMemoryInBytes(List.of("-Da=b", "-Des.total_memory_bytes=123456789", "-Dx=y")),
+            is(123456789L)
+        );
+        assertThat(MachineDependentHeap.parseForcedMemoryInBytes(List.of("-Des.total_memory_bytes=987654321")), is(987654321L));
+        try {
+            MachineDependentHeap.parseForcedMemoryInBytes(List.of("-Des.total_memory_bytes=invalid"));
+            fail("expected parse to fail");
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(), is("Unable to parse number of bytes from [-Des.total_memory_bytes=invalid]"));
+        }
+    }
+
+    private static List<String> calculateHeap(double memoryInGigabytes, Long forcedMemoryInBytes, String... roles) {
         MachineDependentHeap machineDependentHeap = new MachineDependentHeap(systemMemoryInGigabytes(memoryInGigabytes));
         String configYaml = "node.roles: [" + String.join(",", roles) + "]";
-        return calculateHeap(machineDependentHeap, configYaml);
+        return calculateHeap(machineDependentHeap, configYaml, forcedMemoryInBytes);
     }
 
-    private static List<String> calculateHeap(MachineDependentHeap machineDependentHeap, String configYaml) {
+    private static List<String> calculateHeap(MachineDependentHeap machineDependentHeap, String configYaml, Long forcedMemoryInBytes) {
         try (InputStream in = new ByteArrayInputStream(configYaml.getBytes(StandardCharsets.UTF_8))) {
-            return machineDependentHeap.determineHeapSettings(in);
+            return machineDependentHeap.determineHeapSettings(in, forcedMemoryInBytes);
         } catch (IOException e) {
             throw new UncheckedIOException(e);
         }

docs/changelog/78750.yaml

@@ -0,0 +1,6 @@
+pr: 78750
+summary: Allow total memory to be overridden
+area: Packaging
+type: enhancement
+issues:
+ - 65905

docs/reference/cluster/nodes-stats.asciidoc

@@ -1036,6 +1036,17 @@ Total amount of physical memory.
 (integer)
 Total amount of physical memory in bytes.
 
+`total_override`::
+(<<byte-units,byte value>>)
+If the amount of physical memory has been overridden using the `es.total_memory_bytes`
+system property then this reports the overridden value. Otherwise it is not present.
+
+`total_override_in_bytes`::
+(integer)
+If the amount of physical memory has been overridden using the `es.total_memory_bytes`
+system property then this reports the overridden value in bytes. Otherwise it is not
+present.
+
 `free`::
 (<<byte-units,byte value>>)
 Amount of free physical memory.

docs/reference/cluster/stats.asciidoc

@@ -916,6 +916,18 @@ Total amount of physical memory across all selected nodes.
 (integer)
 Total amount, in bytes, of physical memory across all selected nodes.
 
+`total_override`::
+(<<byte-units,byte value>>)
+If the amount of physical memory has been overridden using the `es.total_memory_bytes`
+system property on all selected nodes then this reports the sum of the overridden
+values. Otherwise it is not present.
+
+`total_override_in_bytes`::
+(integer)
+If the amount of physical memory has been overridden using the `es.total_memory_bytes`
+system property on all selected nodes then this reports the sum of the overridden
+values in bytes. Otherwise it is not present.
+
 `free`::
 (<<byte-units, byte units>>)
 Amount of free physical memory across all selected nodes.

qa/os/src/test/java/org/elasticsearch/packaging/test/DockerTests.java

@@ -868,22 +868,45 @@ public void test140CgroupOsStatsAreAvailable() throws Exception {
      * logic sets the correct heap size, based on the container limits.
      */
     public void test150MachineDependentHeap() throws Exception {
+        final List<String> xArgs = machineDependentHeapTest("942m", List.of());
+
+        // This is roughly 0.4 * 942
+        assertThat(xArgs, hasItems("-Xms376m", "-Xmx376m"));
+    }
+
+    /**
+     * Check that when available system memory is constrained by a total memory override as well as Docker,
+     * the machine-dependant heap sizing logic sets the correct heap size, preferring the override to the
+     * container limits.
+     */
+    public void test151MachineDependentHeapWithSizeOverride() throws Exception {
+        final List<String> xArgs = machineDependentHeapTest(
+            "942m",
+            // 799014912 = 762m
+            List.of("-Des.total_memory_bytes=799014912")
+        );
+
+        // This is roughly 0.4 * 762, in particular it's NOT 0.4 * 942
+        assertThat(xArgs, hasItems("-Xms304m", "-Xmx304m"));
+    }
+
+    private List<String> machineDependentHeapTest(final String containerMemory, final List<String> extraJvmOptions) throws Exception {
         // Start by ensuring `jvm.options` doesn't define any heap options
         final Path jvmOptionsPath = tempDir.resolve("jvm.options");
         final Path containerJvmOptionsPath = installation.config("jvm.options");
         copyFromContainer(containerJvmOptionsPath, jvmOptionsPath);
 
-        final List<String> jvmOptions = Files.readAllLines(jvmOptionsPath)
-            .stream()
-            .filter(line -> (line.startsWith("-Xms") || line.startsWith("-Xmx")) == false)
-            .collect(Collectors.toList());
+        final List<String> jvmOptions = Stream.concat(
+            Files.readAllLines(jvmOptionsPath).stream().filter(line -> (line.startsWith("-Xms") || line.startsWith("-Xmx")) == false),
+            extraJvmOptions.stream()
+        ).collect(Collectors.toList());
 
         Files.writeString(jvmOptionsPath, String.join("\n", jvmOptions));
 
         // Now run the container, being explicit about the available memory
         runContainer(
             distribution(),
-            builder().memory("942m")
+            builder().memory(containerMemory)
                 .volumes(Map.of(jvmOptionsPath, containerJvmOptionsPath))
                 .envVars(Map.of("ingest.geoip.downloader.enabled", "false", "ELASTIC_PASSWORD", PASSWORD))
         );
@@ -899,12 +922,9 @@ public void test150MachineDependentHeap() throws Exception {
         final JsonNode jsonNode = new ObjectMapper().readTree(jvmArgumentsLine.get());
 
         final String argsStr = jsonNode.get("message").textValue();
-        final List<String> xArgs = Arrays.stream(argsStr.substring(1, argsStr.length() - 1).split(",\\s*"))
+        return Arrays.stream(argsStr.substring(1, argsStr.length() - 1).split(",\\s*"))
             .filter(arg -> arg.startsWith("-X"))
             .collect(Collectors.toList());
-
-        // This is roughly 0.4 * 942
-        assertThat(xArgs, hasItems("-Xms376m", "-Xmx376m"));
     }
 
     /**

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java

@@ -266,20 +266,33 @@ private OsStats(List<NodeInfo> nodeInfos, List<NodeStats> nodeStatsList) {
             this.allocatedProcessors = allocatedProcessors;
 
             long totalMemory = 0;
+            Long totalMemoryOverride = 0L;
             long freeMemory = 0;
             for (NodeStats nodeStats : nodeStatsList) {
                 if (nodeStats.getOs() != null) {
-                    long total = nodeStats.getOs().getMem().getTotal().getBytes();
+                    org.elasticsearch.monitor.os.OsStats.Mem mem = nodeStats.getOs().getMem();
+                    long total = mem.getTotal().getBytes();
                     if (total > 0) {
                         totalMemory += total;
                     }
+                    // Only report a total memory override for the whole cluster if every node has overridden total memory
+                    if (totalMemoryOverride != null) {
+                        if (mem.getTotalOverride() != null) {
+                            long totalOverride = mem.getTotalOverride().getBytes();
+                            if (totalOverride > 0) {
+                                totalMemoryOverride += totalOverride;
+                            }
+                        } else {
+                            totalMemoryOverride = null;
+                        }
+                    }
                     long free = nodeStats.getOs().getMem().getFree().getBytes();
                     if (free > 0) {
                         freeMemory += free;
                     }
                 }
             }
-            this.mem = new org.elasticsearch.monitor.os.OsStats.Mem(totalMemory, freeMemory);
+            this.mem = new org.elasticsearch.monitor.os.OsStats.Mem(totalMemory, totalMemoryOverride, freeMemory);
         }
 
         public int getAvailableProcessors() {

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

@@ -60,6 +60,10 @@ public class OsProbe {
 
     private static final OperatingSystemMXBean osMxBean = ManagementFactory.getOperatingSystemMXBean();
 
+    // This property is specified without units because it also needs to be parsed by the launcher
+    // code, which does not have access to all the utility classes of the Elasticsearch server.
+    private static final String memoryOverrideProperty = System.getProperty("es.total_memory_bytes", "-1");
+
     private static final Method getFreePhysicalMemorySize;
     private static final Method getTotalPhysicalMemorySize;
     private static final Method getFreeSwapSpaceSize;
@@ -123,6 +127,26 @@ public long getTotalPhysicalMemorySize() {
         }
     }
 
+    /**
+     * Returns the overridden total amount of physical memory in bytes.
+     * Total memory may be overridden when some other process is running
+     * that is known to consume a non-negligible amount of memory. This
+     * is read from the "es.total_memory_bytes" system property. Negative
+     * values or not set at all mean no override.
+     */
+    public Long getTotalMemoryOverride() {
+        return getTotalMemoryOverride(memoryOverrideProperty);
+    }
+
+    static Long getTotalMemoryOverride(String memoryOverrideProperty) {
+        try {
+            long memoryOverride = Long.parseLong(memoryOverrideProperty);
+            return (memoryOverride < 0) ? null : memoryOverride;
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("Invalid value for [es.total_memory_bytes]: [" + memoryOverrideProperty + "]", e);
+        }
+    }
+
     /**
      * Returns the amount of free swap space in bytes.
      */
@@ -859,7 +883,7 @@ OsStats.Cgroup getCgroup(boolean isLinux) {
 
     public OsStats osStats() {
         final OsStats.Cpu cpu = new OsStats.Cpu(getSystemCpuPercent(), getSystemLoadAverage());
-        final OsStats.Mem mem = new OsStats.Mem(getTotalPhysicalMemorySize(), getFreePhysicalMemorySize());
+        final OsStats.Mem mem = new OsStats.Mem(getTotalPhysicalMemorySize(), getTotalMemoryOverride(), getFreePhysicalMemorySize());
         final OsStats.Swap swap = new OsStats.Swap(getTotalSwapSpaceSize(), getFreeSwapSpaceSize());
         final OsStats.Cgroup cgroup = getCgroup(Constants.LINUX);
         return new OsStats(System.currentTimeMillis(), cpu, mem, swap, cgroup);