[ML] Propagate xpack.ml.min_disk_space_off_heap to native process (#64733)

This change is the Java-side companion to elastic/ml-cpp#1556.
It ensures that customisations to xpack.ml.min_disk_space_off_heap
are reflected in the disk space check done in the C++ code as well
as in the disk space check done in the Java code.  The check done
in the C++ code runs later than the check in the Java code, and
only requires that 80% of the disk space required in the Java code
is available.  (The defaults are 5GB and 4GB.)

Backport of #64720

Author: David Roberts

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/ForecastIT.java

@@ -214,10 +214,7 @@ public void testMemoryStatus() {
 
     public void testOverflowToDisk() throws Exception {
         assumeFalse("https://github.com/elastic/elasticsearch/issues/44609", Constants.WINDOWS);
-        // This test repeatedly fails in encryption-at-rest (EAR) builds, and
-        // the only way to detect such a build appears to be the CI job name
-        assumeFalse("https://github.com/elastic/elasticsearch/issues/58806",
-            System.getenv().getOrDefault("JOB_NAME", "not a CI build").contains("EAR"));
+
         Detector.Builder detector = new Detector.Builder("mean", "value");
         detector.setByFieldName("clientIP");
 

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportForecastJobAction.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -24,6 +25,7 @@
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.job.results.ForecastRequestStats;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.ml.MachineLearning;
 import org.elasticsearch.xpack.ml.job.JobManager;
 import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider;
 import org.elasticsearch.xpack.ml.job.process.autodetect.AutodetectProcessManager;
@@ -48,20 +50,23 @@ public class TransportForecastJobAction extends TransportJobTaskAction<ForecastJ
     private final JobManager jobManager;
     private final NativeStorageProvider nativeStorageProvider;
     private final AnomalyDetectionAuditor auditor;
+    private final long cppMinAvailableDiskSpaceBytes;
 
     @Inject
-    public TransportForecastJobAction(TransportService transportService,
+    public TransportForecastJobAction(TransportService transportService, Settings settings,
                                       ClusterService clusterService, ActionFilters actionFilters,
                                       JobResultsProvider jobResultsProvider, AutodetectProcessManager processManager,
                                       JobManager jobManager, NativeStorageProvider nativeStorageProvider, AnomalyDetectionAuditor auditor) {
         super(ForecastJobAction.NAME, clusterService, transportService, actionFilters,
             ForecastJobAction.Request::new, ForecastJobAction.Response::new,
-                ThreadPool.Names.SAME, processManager);
+            // ThreadPool.Names.SAME, because operations is executed by autodetect worker thread
+            ThreadPool.Names.SAME, processManager);
         this.jobResultsProvider = jobResultsProvider;
         this.jobManager = jobManager;
         this.nativeStorageProvider = nativeStorageProvider;
         this.auditor = auditor;
-        // ThreadPool.Names.SAME, because operations is executed by autodetect worker thread
+        // The C++ enforces 80% of the free disk space that the Java enforces
+        this.cppMinAvailableDiskSpaceBytes = MachineLearning.MIN_DISK_SPACE_OFF_HEAP.get(settings).getBytes() / 5 * 4;
     }
 
     @Override
@@ -92,6 +97,10 @@ protected void taskOperation(ForecastJobAction.Request request, JobTask task, Ac
                         paramsBuilder.tmpStorage(tmpStorage.toString());
                     }
 
+                    if (cppMinAvailableDiskSpaceBytes >= 0) {
+                        paramsBuilder.minAvailableDiskSpace(cppMinAvailableDiskSpaceBytes);
+                    }
+
                     ForecastParams params = paramsBuilder.build();
                     processManager.forecastJob(task, params, e -> {
                         if (e == null) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/params/ForecastParams.java

@@ -18,14 +18,17 @@ public class ForecastParams {
     private final long expiresIn;
     private final String tmpStorage;
     private final Long maxModelMemory;
+    private final Long minAvailableDiskSpace;
 
-    private ForecastParams(String forecastId, long createTime, long duration, long expiresIn, String tmpStorage, Long maxModelMemory) {
+    private ForecastParams(String forecastId, long createTime, long duration, long expiresIn, String tmpStorage, Long maxModelMemory,
+                           Long minAvailableDiskSpace) {
         this.forecastId = forecastId;
         this.createTime = createTime;
         this.duration = duration;
         this.expiresIn = expiresIn;
         this.tmpStorage = tmpStorage;
         this.maxModelMemory = maxModelMemory;
+        this.minAvailableDiskSpace = minAvailableDiskSpace;
     }
 
     public String getForecastId() {
@@ -69,9 +72,13 @@ public Long getMaxModelMemory() {
         return maxModelMemory;
     }
 
+    public Long getMinAvailableDiskSpace() {
+        return minAvailableDiskSpace;
+    }
+
     @Override
     public int hashCode() {
-        return Objects.hash(forecastId, createTime, duration, expiresIn, tmpStorage, maxModelMemory);
+        return Objects.hash(forecastId, createTime, duration, expiresIn, tmpStorage, maxModelMemory, minAvailableDiskSpace);
     }
 
     @Override
@@ -88,7 +95,8 @@ public boolean equals(Object obj) {
                 && Objects.equals(duration, other.duration)
                 && Objects.equals(expiresIn, other.expiresIn)
                 && Objects.equals(tmpStorage, other.tmpStorage)
-                && Objects.equals(maxModelMemory, other.maxModelMemory);
+                && Objects.equals(maxModelMemory, other.maxModelMemory)
+                && Objects.equals(minAvailableDiskSpace, other.minAvailableDiskSpace);
     }
 
     public static Builder builder() {
@@ -101,6 +109,7 @@ public static class Builder {
         private long durationSecs;
         private long expiresInSecs;
         private Long maxModelMemory;
+        private Long minAvailableDiskSpace;
         private String tmpStorage;
 
         private Builder() {
@@ -132,8 +141,14 @@ public Builder maxModelMemory(long maxModelMemory) {
             return this;
         }
 
+        public Builder minAvailableDiskSpace(long minAvailableDiskSpace) {
+            this.minAvailableDiskSpace = minAvailableDiskSpace;
+            return this;
+        }
+
         public ForecastParams build() {
-            return new ForecastParams(forecastId, createTimeEpochSecs, durationSecs, expiresInSecs, tmpStorage, maxModelMemory);
+            return new ForecastParams(forecastId, createTimeEpochSecs, durationSecs, expiresInSecs, tmpStorage, maxModelMemory,
+                minAvailableDiskSpace);
         }
     }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/writer/AutodetectControlMsgWriter.java

@@ -166,6 +166,9 @@ public void writeForecastMessage(ForecastParams params) throws IOException {
         if (params.getMaxModelMemory() != null) {
             builder.field("max_model_memory", params.getMaxModelMemory());
         }
+        if (params.getMinAvailableDiskSpace() != null) {
+            builder.field("min_available_disk_space", params.getMinAvailableDiskSpace());
+        }
         builder.endObject();
 
         writeMessage(FORECAST_MESSAGE_CODE + Strings.toString(builder));

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/params/ForecastParamsTests.java

@@ -16,7 +16,7 @@
 
 public class ForecastParamsTests extends ESTestCase {
 
-    private static ParseField DURATION = new ParseField("duration");
+    private static final ParseField DURATION = new ParseField("duration");
 
     public void testForecastIdsAreUnique() {
         Set<String> ids = new HashSet<>();

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/writer/AutodetectControlMsgWriterTests.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.xpack.core.ml.job.config.RuleCondition;
 import org.elasticsearch.xpack.ml.job.process.autodetect.params.DataLoadParams;
 import org.elasticsearch.xpack.ml.job.process.autodetect.params.FlushJobParams;
+import org.elasticsearch.xpack.ml.job.process.autodetect.params.ForecastParams;
 import org.elasticsearch.xpack.ml.job.process.autodetect.params.TimeRange;
 import org.elasticsearch.xpack.ml.process.writer.LengthEncodedWriter;
 import org.junit.Before;
@@ -30,7 +31,9 @@
 import java.util.Optional;
 import java.util.stream.IntStream;
 
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.startsWith;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -286,6 +289,39 @@ public void testWriteStartBackgroundPersistMessage() throws IOException {
         verifyNoMoreInteractions(lengthEncodedWriter);
     }
 
+    public void testWriteForecastParamsMessage() throws IOException {
+        AutodetectControlMsgWriter writer = new AutodetectControlMsgWriter(lengthEncodedWriter, 2);
+
+        ForecastParams params = ForecastParams.builder()
+            .duration(TimeValue.timeValueHours(3))
+            .expiresIn(TimeValue.timeValueDays(4))
+            .tmpStorage("/my_temp_dir")
+            .maxModelMemory(12345)
+            .minAvailableDiskSpace(98765)
+            .build();
+
+        writer.writeForecastMessage(params);
+
+        InOrder inOrder = inOrder(lengthEncodedWriter);
+        inOrder.verify(lengthEncodedWriter).writeNumFields(2);
+        inOrder.verify(lengthEncodedWriter).writeField("");
+        ArgumentCaptor<String> capturedMessage = ArgumentCaptor.forClass(String.class);
+        inOrder.verify(lengthEncodedWriter).writeField(capturedMessage.capture());
+
+        assertThat(capturedMessage.getValue(), startsWith("p{\"forecast_id\":\""));
+        assertThat(capturedMessage.getValue(), endsWith("\"duration\":10800,\"expires_in\":345600,\"tmp_storage\":\"/my_temp_dir\","
+            +"\"max_model_memory\":12345,\"min_available_disk_space\":98765}"));
+
+        inOrder.verify(lengthEncodedWriter).writeNumFields(2);
+        inOrder.verify(lengthEncodedWriter).writeField("");
+        StringBuilder spaces = new StringBuilder();
+        IntStream.rangeClosed(1, AutodetectControlMsgWriter.FLUSH_SPACES_LENGTH).forEach(i -> spaces.append(' '));
+        inOrder.verify(lengthEncodedWriter).writeField(spaces.toString());
+        inOrder.verify(lengthEncodedWriter).flush();
+
+        verifyNoMoreInteractions(lengthEncodedWriter);
+    }
+
     private static List<RuleCondition> createRule(double value) {
         return Collections.singletonList(new RuleCondition(RuleCondition.AppliesTo.ACTUAL, Operator.GT, value));
     }