[ML] Switch ML native QA tests to use a 3 node cluster

x-pack/qa/ml-native-tests/src/test/java/org/elasticsearch/xpack/ml/integration/MlJobIT.java → x-pack/qa/ml-native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/MlJobIT.java

@@ -185,23 +185,32 @@ public void testCreateJobsWithIndexNameOption() throws Exception {
                         + "anomaly_detectors/" + jobId2, Collections.emptyMap(), new StringEntity(jobConfig, ContentType.APPLICATION_JSON));
         assertEquals(200, response.getStatusLine().getStatusCode());
 
-        response = client().performRequest("get", "_aliases");
-        assertEquals(200, response.getStatusLine().getStatusCode());
-        String responseAsString = responseEntityToString(response);
+        // With security enabled GET _aliases throws an index_not_found_exception
+        // if no aliases have been created. In multi-node tests the alias may not
+        // appear immediately so wait here.
+        assertBusy(() -> {
+            try {
+                Response aliasesResponse = client().performRequest("get", "_aliases");
+                assertEquals(200, aliasesResponse.getStatusLine().getStatusCode());
+                String responseAsString = responseEntityToString(aliasesResponse);
+                assertThat(responseAsString,
+                        containsString("\"" + AnomalyDetectorsIndex.jobResultsAliasedName("custom-" + indexName) + "\":{\"aliases\":{"));
+                assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.jobResultsAliasedName(jobId1)
+                        + "\":{\"filter\":{\"term\":{\"job_id\":{\"value\":\"" + jobId1 + "\",\"boost\":1.0}}}}"));
+                assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.resultsWriteAlias(jobId1) + "\":{}"));
+                assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.jobResultsAliasedName(jobId2)
+                        + "\":{\"filter\":{\"term\":{\"job_id\":{\"value\":\"" + jobId2 + "\",\"boost\":1.0}}}}"));
+                assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.resultsWriteAlias(jobId2) + "\":{}"));
+            } catch (ResponseException e) {
+                throw new AssertionError(e);
+            }
+        });
 
+        Response indicesResponse = client().performRequest("get", "_cat/indices");
+        assertEquals(200, indicesResponse.getStatusLine().getStatusCode());
+        String responseAsString = responseEntityToString(indicesResponse);
         assertThat(responseAsString,
-                containsString("\"" + AnomalyDetectorsIndex.jobResultsAliasedName("custom-" + indexName) + "\":{\"aliases\":{"));
-        assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.jobResultsAliasedName(jobId1)
-                + "\":{\"filter\":{\"term\":{\"job_id\":{\"value\":\"" + jobId1 + "\",\"boost\":1.0}}}}"));
-        assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.resultsWriteAlias(jobId1) + "\":{}"));
-        assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.jobResultsAliasedName(jobId2)
-                + "\":{\"filter\":{\"term\":{\"job_id\":{\"value\":\"" + jobId2 + "\",\"boost\":1.0}}}}"));
-        assertThat(responseAsString, containsString("\"" + AnomalyDetectorsIndex.resultsWriteAlias(jobId2) + "\":{}"));
-
-        response = client().performRequest("get", "_cat/indices");
-        assertEquals(200, response.getStatusLine().getStatusCode());
-        responseAsString = responseEntityToString(response);
-        assertThat(responseAsString, containsString(AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + "custom-" + indexName));
+                containsString(AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + "custom-" + indexName));
         assertThat(responseAsString, not(containsString(AnomalyDetectorsIndex.jobResultsAliasedName(jobId1))));
         assertThat(responseAsString, not(containsString(AnomalyDetectorsIndex.jobResultsAliasedName(jobId2))));
 
@@ -445,15 +454,24 @@ public void testDeleteJobAfterMissingAliases() throws Exception {
         String indexName = AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + AnomalyDetectorsIndexFields.RESULTS_INDEX_DEFAULT;
         createFarequoteJob(jobId);
 
-        Response response = client().performRequest("get", "_cat/aliases");
-        assertEquals(200, response.getStatusLine().getStatusCode());
-        String responseAsString = responseEntityToString(response);
-        assertThat(responseAsString, containsString(readAliasName));
-        assertThat(responseAsString, containsString(writeAliasName));
+        // With security enabled cat aliases throws an index_not_found_exception
+        // if no aliases have been created. In multi-node tests the alias may not
+        // appear immediately so wait here.
+        assertBusy(() -> {
+            try {
+                Response aliasesResponse = client().performRequest("get", "_cat/aliases");
+                assertEquals(200, aliasesResponse.getStatusLine().getStatusCode());
+                String responseAsString = responseEntityToString(aliasesResponse);
+                assertThat(responseAsString, containsString(readAliasName));
+                assertThat(responseAsString, containsString(writeAliasName));
+            } catch (ResponseException e) {
+                throw new AssertionError(e);
+            }
+        });
 
         // Manually delete the aliases so that we can test that deletion proceeds
         // normally anyway
-        response = client().performRequest("delete", indexName + "/_alias/" + readAliasName);
+        Response response = client().performRequest("delete", indexName + "/_alias/" + readAliasName);
         assertEquals(200, response.getStatusLine().getStatusCode());
         response = client().performRequest("delete", indexName + "/_alias/" + writeAliasName);
         assertEquals(200, response.getStatusLine().getStatusCode());

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.ml.integration;
 
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
+import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterModule;
@@ -342,21 +343,17 @@ protected void waitForecastToFinish(String jobId, String forecastId) throws Exce
     }
 
     protected ForecastRequestStats getForecastStats(String jobId, String forecastId) {
-        SearchResponse searchResponse = client().prepareSearch(AnomalyDetectorsIndex.jobResultsAliasedName(jobId))
-                .setQuery(QueryBuilders.boolQuery()
-                        .filter(QueryBuilders.termQuery(Result.RESULT_TYPE.getPreferredName(), ForecastRequestStats.RESULT_TYPE_VALUE))
-                        .filter(QueryBuilders.termQuery(Job.ID.getPreferredName(), jobId))
-                        .filter(QueryBuilders.termQuery(ForecastRequestStats.FORECAST_ID.getPreferredName(), forecastId)))
+        GetResponse getResponse = client().prepareGet()
+                .setIndex(AnomalyDetectorsIndex.jobResultsAliasedName(jobId))
+                .setId(ForecastRequestStats.documentId(jobId, forecastId))
                 .execute().actionGet();
-        SearchHits hits = searchResponse.getHits();
-        if (hits.getTotalHits() == 0) {
+
+        if (getResponse.isExists() == false) {
             return null;
         }
-        assertThat(hits.getTotalHits(), equalTo(1L));
-        try {
-            XContentParser parser = XContentFactory.xContent(XContentType.JSON).createParser(
+        try (XContentParser parser = XContentFactory.xContent(XContentType.JSON).createParser(
                     NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
-                    hits.getHits()[0].getSourceRef().streamInput());
+                    getResponse.getSourceAsBytesRef().streamInput())) {
             return ForecastRequestStats.STRICT_PARSER.apply(parser, null);
         } catch (IOException e) {
             throw new IllegalStateException(e);

x-pack/qa/ml-native-tests/src/test/java/org/elasticsearch/xpack/ml/integration/RestoreModelSnapshotIT.java → x-pack/qa/ml-native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/RestoreModelSnapshotIT.java

@@ -23,7 +23,11 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.nullValue;
 
 /**
  * This test aims to catch regressions where,
@@ -68,8 +72,14 @@ public void test() throws Exception {
         openJob(job.getId());
         String forecastId = forecast(job.getId(), TimeValue.timeValueHours(3), null);
         waitForecastToFinish(job.getId(), forecastId);
-        ForecastRequestStats forecastStats = getForecastStats(job.getId(), forecastId);
-        assertThat(forecastStats.getStatus(), equalTo(ForecastRequestStats.ForecastRequestStatus.FINISHED));
+        // In a multi-node cluster the replica may not be up to date
+        // so wait for the change
+        assertBusy(() -> {
+            ForecastRequestStats forecastStats = getForecastStats(job.getId(), forecastId);
+            assertThat(forecastStats.getMessages(), anyOf(nullValue(), empty()));
+            assertThat(forecastStats.getMemoryUsage(), greaterThan(0L));
+            assertThat(forecastStats.getRecordCount(), equalTo(3L));
+        });
 
         closeJob(job.getId());