ML expired data cleanup cannot run on the client thread

Author: davidkyle

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

@@ -15,6 +15,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.transport.Transports;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.action.DeleteExpiredDataAction;
 import org.elasticsearch.xpack.ml.MachineLearning;
@@ -66,10 +67,12 @@ private void deleteExpiredData(ActionListener<DeleteExpiredDataAction.Response>
 
     private void deleteExpiredData(Iterator<MlDataRemover> mlDataRemoversIterator,
                                    ActionListener<DeleteExpiredDataAction.Response> listener) {
+        Transports.assertNotTransportThread("ML Daily Maintenance");
         if (mlDataRemoversIterator.hasNext()) {
             MlDataRemover remover = mlDataRemoversIterator.next();
             remover.remove(ActionListener.wrap(
-                    booleanResponse -> deleteExpiredData(mlDataRemoversIterator, listener),
+                    booleanResponse -> threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(() ->
+                            deleteExpiredData(mlDataRemoversIterator, listener)),
                     listener::onFailure));
         } else {
             logger.info("Completed deletion of expired data");

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/BatchedDocumentsIterator.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.sort.SortBuilders;
+import org.elasticsearch.transport.Transports;
 import org.elasticsearch.xpack.core.ml.job.persistence.ElasticsearchMappings;
 import org.elasticsearch.xpack.core.ml.utils.MlIndicesUtils;
 
@@ -89,6 +90,8 @@ public Deque<T> next() {
     private SearchResponse initScroll() {
         LOGGER.trace("ES API CALL: search index {}", index);
 
+        Transports.assertNotTransportThread("BatchedDocumentsIterator makes blocking calls");
+
         isScrollInitialised = true;
 
         SearchRequest searchRequest = new SearchRequest(index);

x-pack/qa/ml-native-multi-node-tests/build.gradle

@@ -0,0 +1,85 @@
+import org.elasticsearch.gradle.LoggedExec
+
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply plugin: 'elasticsearch.rest-test'
+
+dependencies {
+    testCompile project(path: xpackModule('core'), configuration: 'runtime')
+    testCompile project(path: xpackModule('core'), configuration: 'testArtifacts')
+    testCompile project(path: xpackModule('ml'), configuration: 'runtime')
+    testCompile project(path: xpackModule('ml'), configuration: 'testArtifacts')
+}
+
+integTestRunner {
+    /*
+     * We have to disable setting the number of available processors as tests in the same JVM randomize processors and will step on each
+     * other if we allow them to set the number of available processors as it's set-once in Netty.
+     */
+    systemProperty 'es.set.netty.runtime.available.processors', 'false'
+}
+
+// location of generated keystores and certificates
+File keystoreDir = new File(project.buildDir, 'keystore')
+
+// Generate the node's keystore
+File nodeKeystore = new File(keystoreDir, 'test-node.jks')
+task createNodeKeyStore(type: LoggedExec) {
+    doFirst {
+        if (nodeKeystore.parentFile.exists() == false) {
+            nodeKeystore.parentFile.mkdirs()
+        }
+        if (nodeKeystore.exists()) {
+            delete nodeKeystore
+        }
+    }
+    executable = new File(project.runtimeJavaHome, 'bin/keytool')
+    standardInput = new ByteArrayInputStream('FirstName LastName\nUnit\nOrganization\nCity\nState\nNL\nyes\n\n'.getBytes('UTF-8'))
+    args '-genkey',
+            '-alias', 'test-node',
+            '-keystore', nodeKeystore,
+            '-keyalg', 'RSA',
+            '-keysize', '2048',
+            '-validity', '712',
+            '-dname', 'CN=smoke-test-plugins-ssl',
+            '-keypass', 'keypass',
+            '-storepass', 'keypass'
+}
+
+// Add keystores to test classpath: it expects it there
+sourceSets.test.resources.srcDir(keystoreDir)
+processTestResources.dependsOn(createNodeKeyStore)
+
+integTestCluster {
+    dependsOn createNodeKeyStore
+    setting 'xpack.security.enabled', 'true'
+    setting 'xpack.ml.enabled', 'true'
+    setting 'logger.org.elasticsearch.xpack.ml.datafeed', 'TRACE'
+    setting 'xpack.monitoring.enabled', 'false'
+    setting 'xpack.security.authc.token.enabled', 'true'
+    setting 'xpack.security.transport.ssl.enabled', 'true'
+    setting 'xpack.security.transport.ssl.keystore.path', nodeKeystore.name
+    setting 'xpack.security.transport.ssl.verification_mode', 'certificate'
+    setting 'xpack.security.audit.enabled', 'true'
+    setting 'xpack.license.self_generated.type', 'trial'
+
+    keystoreSetting 'bootstrap.password', 'x-pack-test-password'
+    keystoreSetting 'xpack.security.transport.ssl.keystore.secure_password', 'keypass'
+
+    numNodes = 3
+
+    setupCommand 'setupDummyUser',
+            'bin/elasticsearch-users', 'useradd', 'x_pack_rest_user', '-p', 'x-pack-test-password', '-r', 'superuser'
+
+    extraConfigFile nodeKeystore.name, nodeKeystore
+
+    waitCondition = { node, ant ->
+        File tmpFile = new File(node.cwd, 'wait.success')
+        ant.get(src: "http://${node.httpUri()}/_cluster/health?wait_for_nodes=>=${numNodes}&wait_for_status=yellow",
+                dest: tmpFile.toString(),
+                username: 'x_pack_rest_user',
+                password: 'x-pack-test-password',
+                ignoreerrors: true,
+                retries: 10)
+        return tmpFile.exists()
+    }
+}

x-pack/qa/ml-native-tests/src/test/java/org/elasticsearch/xpack/ml/integration/DeleteExpiredDataIT.java renamed to x-pack/qa/ml-native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/DeleteExpiredDataIT.java

@@ -37,7 +37,6 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.Matchers.equalTo;