[Test] Reenable Monitoring Bulk tests (elastic/x-pack-elasticsearch#908)

This commit reenables the Monitoring Bulk Api REST tests. The XPackRestIT
now enables/disables the local default exporter before executing the monitoring
 tests, and also waits for the monitoring service to be started before executing
 the test.

Original commit: elastic/x-pack-elasticsearch@10b696198c

dev-tools/checkstyle_suppressions.xml

@@ -1236,8 +1236,6 @@
   <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]ssl[/\\]TestsSSLService.java" checks="LineLength" />
   <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]support[/\\]clock[/\\]ClockMock.java" checks="LineLength" />
   <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]template[/\\]TemplateUtilsTests.java" checks="LineLength" />
-  <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]test[/\\]rest[/\\]XPackRestIT.java" checks="LineLength" />
-  <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]test[/\\]rest[/\\]XPackRestTestCase.java" checks="LineLength" />
   <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]watcher[/\\]OldWatcherIndicesBackwardsCompatibilityTests.java" checks="LineLength" />
   <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]watcher[/\\]WatcherF.java" checks="LineLength" />
   <suppress files="plugin[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]xpack[/\\]watcher[/\\]WatcherFeatureSetTests.java" checks="LineLength" />

plugin/build.gradle

@@ -308,12 +308,16 @@ project.afterEvaluate {
 
 integTestRunner {
   // TODO: fix this rest test to not depend on a hardcoded port!
-  systemProperty 'tests.rest.blacklist', 'getting_started/10_monitor_cluster_health/*,bulk/10_basic/*'
+  systemProperty 'tests.rest.blacklist', 'getting_started/10_monitor_cluster_health/*'
 }
 
 integTestCluster {
   setting 'xpack.ml.enabled', 'true'
-  setting 'xpack.monitoring.collection.interval', '3s'
+  // Integration tests are supposed to enable/disable exporters before/after each test
+  setting 'xpack.monitoring.exporters._local.type', 'local'
+  setting 'xpack.monitoring.exporters._local.enabled', 'false'
+  setting 'xpack.monitoring.collection.interval', '-1'
+
   waitCondition = { NodeInfo node, AntBuilder ant ->
       File tmpFile = new File(node.cwd, 'wait.success')
       for (int i = 0; i < 10; i++) {

plugin/src/test/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporterIT.java

@@ -31,9 +31,7 @@ import org.elasticsearch.test.http.MockResponse;
 import org.elasticsearch.test.http.MockWebServer;
 import org.elasticsearch.xpack.monitoring.MonitoredSystem;
 import org.elasticsearch.xpack.monitoring.MonitoringSettings;
-import org.elasticsearch.xpack.monitoring.collector.cluster.ClusterStateCollector;
 import org.elasticsearch.xpack.monitoring.collector.cluster.ClusterStateMonitoringDoc;
-import org.elasticsearch.xpack.monitoring.collector.indices.IndexRecoveryCollector;
 import org.elasticsearch.xpack.monitoring.collector.indices.IndexRecoveryMonitoringDoc;
 import org.elasticsearch.xpack.monitoring.exporter.Exporter;
 import org.elasticsearch.xpack.monitoring.exporter.Exporters;

plugin/src/test/java/org/elasticsearch/xpack/test/rest/XPackRestIT.java

@@ -6,9 +6,14 @@
 package org.elasticsearch.xpack.test.rest;
 
 import org.apache.http.HttpStatus;
-import org.elasticsearch.common.Strings;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.CheckedFunction;
+import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestResponse;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestResponseException;
 import org.elasticsearch.xpack.ml.MachineLearningTemplateRegistry;
 import org.elasticsearch.xpack.ml.integration.MlRestTestStateCleaner;
 import org.elasticsearch.xpack.security.SecurityLifecycleService;
@@ -16,11 +21,17 @@ import org.junit.After;
 import org.junit.Before;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
+import static java.util.Collections.singletonList;
 import static java.util.Collections.singletonMap;
 
 /** Runs rest tests against external cluster */
@@ -41,22 +52,114 @@ public class XPackRestIT extends XPackRestTestCase {
      */
     @Before
     public void waitForTemplates() throws Exception {
-        waitForTemplate(SecurityLifecycleService.SECURITY_TEMPLATE_NAME);
+        List<String> templates = new ArrayList<>();
-        waitForTemplate(Strings.arrayToCommaDelimitedString(MachineLearningTemplateRegistry.TEMPLATE_NAMES));
+        templates.add(SecurityLifecycleService.SECURITY_TEMPLATE_NAME);
+        templates.addAll(Arrays.asList(MachineLearningTemplateRegistry.TEMPLATE_NAMES));
+
+        for (String template : templates) {
+            awaitCallApi("indices.exists_template", singletonMap("name", template), emptyList(),
+                    response -> true,
+                    () -> "Exception when waiting for [" + template + "] template to be created");
+        }
     }
 
-    private void waitForTemplate(String templateName) throws Exception {
+    /**
-        Map<String, String> params = singletonMap("name", templateName);
+     * Enable monitoring and waits for monitoring documents to be collected and indexed in
+     * monitoring indices.This is the signal that the local exporter is started and ready
+     * for the tests.
+     */
+    @Before
+    public void enableMonitoring() throws Exception {
+        if (isMonitoringTest()) {
+            final Map<String, Object> settings = new HashMap<>();
+            settings.put("xpack.monitoring.collection.interval", "3s");
+            settings.put("xpack.monitoring.exporters._local.enabled", true);
+
+            awaitCallApi("cluster.put_settings", emptyMap(),
+                    singletonList(singletonMap("transient", settings)),
+                    response -> {
+                        Object acknowledged = response.evaluate("acknowledged");
+                        return acknowledged != null && (Boolean) acknowledged;
+                    },
+                    () -> "Exception when enabling monitoring");
+            awaitCallApi("search", singletonMap("index", ".monitoring-*"), emptyList(),
+                    response -> ((Number) response.evaluate("hits.total")).intValue() > 0,
+                    () -> "Exception when waiting for monitoring documents to be indexed");
+        }
+    }
+
+    /**
+     * Disable monitoring
+     */
+    @After
+    public void disableMonitoring() throws Exception {
+        if (isMonitoringTest()) {
+            final Map<String, Object> settings = new HashMap<>();
+            settings.put("xpack.monitoring.collection.interval", (String) null);
+            settings.put("xpack.monitoring.exporters._local.enabled", (String) null);
+
+            awaitCallApi("cluster.put_settings", emptyMap(),
+                    singletonList(singletonMap("transient", settings)),
+                    response -> {
+                        Object acknowledged = response.evaluate("acknowledged");
+                        return acknowledged != null && (Boolean) acknowledged;
+                    },
+                    () -> "Exception when disabling monitoring");
+
+            // Now the local exporter is disabled, we try to check if the monitoring indices are
+            // re created by an inflight bulk request. We try this 10 times or 10 seconds.
+            final CountDown retries = new CountDown(10);
+            awaitBusy(() -> {
+                try {
+                    Map<String, String> params = new HashMap<>();
+                    params.put("index", ".monitoring-*");
+                    params.put("allow_no_indices", "false");
+
+                    ClientYamlTestResponse response =
+                            callApi("indices.exists", params, emptyList());
+                    if (response.getStatusCode() == HttpStatus.SC_OK) {
+                        params = singletonMap("index", ".monitoring-*");
+                        callApi("indices.delete", params, emptyList());
+                        return false;
+                    }
+                } catch (ClientYamlTestResponseException e) {
+                    ResponseException exception = e.getResponseException();
+                    if (exception != null) {
+                        Response response = exception.getResponse();
+                        if (response != null) {
+                            int responseCode = response.getStatusLine().getStatusCode();
+                            if (responseCode == HttpStatus.SC_NOT_FOUND) {
+                                return retries.countDown();
+                            }
+                        }
+                    }
+                    throw new ElasticsearchException("Failed to delete monitoring indices: ", e);
+                } catch (IOException e) {
+                    throw new ElasticsearchException("Failed to delete monitoring indices: ", e);
+                }
+                return retries.countDown();
+            });
+        }
+    }
+
+    /**
+     * Executes an API call using the admin context, waiting for it to succeed.
+     */
+    private void awaitCallApi(String apiName,
+                              Map<String, String> params,
+                              List<Map<String, Object>> bodies,
+                              CheckedFunction<ClientYamlTestResponse, Boolean, IOException> success,
+                              Supplier<String> error) throws Exception {
+
         AtomicReference<IOException> exceptionHolder = new AtomicReference<>();
         awaitBusy(() -> {
             try {
-                ClientYamlTestResponse response = getAdminExecutionContext().callApi("indices.exists_template",
+                ClientYamlTestResponse response = callApi(apiName, params, bodies);
-                        params, emptyList(), emptyMap());
-                // We don't check the version of the template - it is the right one when testing documentation.
                 if (response.getStatusCode() == HttpStatus.SC_OK) {
                     exceptionHolder.set(null);
-                    return true;
+                    return success.apply(response);
                 }
+                return false;
             } catch (IOException e) {
                 exceptionHolder.set(e);
             }
@@ -65,7 +168,18 @@ public class XPackRestIT extends XPackRestTestCase {
 
         IOException exception = exceptionHolder.get();
         if (exception != null) {
-            throw new IllegalStateException("Exception when waiting for [" + templateName + "] template to be created", exception);
+            throw new IllegalStateException(error.get(), exception);
         }
     }
+
+    private ClientYamlTestResponse callApi(String apiName,
+                                           Map<String, String> params,
+                                           List<Map<String, Object>> bodies) throws IOException {
+        return getAdminExecutionContext().callApi(apiName, params, bodies, emptyMap());
+    }
+
+    private boolean isMonitoringTest() {
+        String testName = getTestName();
+        return testName != null && testName.contains("=monitoring/");
+    }
 }

plugin/src/test/java/org/elasticsearch/xpack/test/rest/XPackRestTestCase.java

@@ -21,7 +21,8 @@ import static org.elasticsearch.xpack.security.authc.support.UsernamePasswordTok
 
 public abstract class XPackRestTestCase extends ESClientYamlSuiteTestCase {
 
-    private static final String BASIC_AUTH_VALUE = basicAuthHeaderValue("elastic", new SecuredString("changeme".toCharArray()));
+    private static final String BASIC_AUTH_VALUE =
+            basicAuthHeaderValue("elastic", new SecuredString("changeme".toCharArray()));
 
     public XPackRestTestCase(@Name("yaml") ClientYamlTestCandidate testCandidate) {
         super(testCandidate);

plugin/src/test/resources/rest-api-spec/test/monitoring/bulk/10_basic.yaml

@@ -1,17 +1,3 @@
----
-setup:
-
-  - do:
-      cluster.health:
-        wait_for_status: yellow
-  - do:
-      # Waits for the monitoring data index to be available:
-      # it indicates that the local exporter is ready
-      cluster.health:
-          index: ".monitoring-data-*"
-          wait_for_active_shards: 1
-          timeout: 60s
-
 ---
 "Bulk indexing of monitoring data":