[ML] Yml rolling upgrade tests (elastic/x-pack-elasticsearch#1555)

* ML bwc jobs tests

* Test for starting old data feeds

Original commit: elastic/x-pack-elasticsearch@131c7b15eb

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/UpgradeClusterClientYamlTestSuiteIT.java

@@ -7,18 +7,49 @@ package org.elasticsearch.upgrades;
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import com.carrotsearch.randomizedtesting.annotations.TimeoutSuite;
+import org.apache.http.HttpStatus;
 import org.apache.lucene.util.TimeUnits;
+import org.elasticsearch.common.CheckedFunction;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestResponse;
+import org.elasticsearch.xpack.ml.MachineLearningTemplateRegistry;
 import org.elasticsearch.xpack.security.SecurityClusterClientYamlTestCase;
+import org.junit.Before;
 
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
+import java.util.Collections;
+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.singletonMap;
 
 @TimeoutSuite(millis = 5 * TimeUnits.MINUTE) // to account for slow as hell VMs
 public class UpgradeClusterClientYamlTestSuiteIT extends SecurityClusterClientYamlTestCase {
 
+    /**
+     * Waits for the Machine Learning templates to be created by {@link MachineLearningTemplateRegistry}
+     */
+    @Before
+    public void waitForTemplates() throws Exception {
+        List<String> templates = new ArrayList<>();
+        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");
+        }
+    }
+
     @Override
     protected boolean preserveIndicesUponCompletion() {
         return true;
@@ -40,4 +71,35 @@ public class UpgradeClusterClientYamlTestSuiteIT extends SecurityClusterClientYa
                 .put(ThreadContext.PREFIX + ".Authorization", token)
                 .build();
     }
+
+    /**
+     * 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(apiName, params, bodies, Collections.emptyMap()
+                );
+                if (response.getStatusCode() == HttpStatus.SC_OK) {
+                    exceptionHolder.set(null);
+                    return success.apply(response);
+                }
+                return false;
+            } catch (IOException e) {
+                exceptionHolder.set(e);
+            }
+            return false;
+        });
+
+        IOException exception = exceptionHolder.get();
+        if (exception != null) {
+            throw new IllegalStateException(error.get(), exception);
+        }
+    }
 }

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/30_ml_jobs_crud.yml

@@ -0,0 +1,86 @@
+---
+setup:
+ - do:
+     cluster.health:
+        wait_for_status: yellow
+        wait_for_nodes: 2
+        timeout: 25s
+
+---
+"Test get old cluster job":
+  - do:
+      xpack.ml.get_jobs:
+        job_id: old-cluster-job
+  - match: { count: 1 }
+
+  - do:
+      xpack.ml.get_job_stats:
+        job_id: old-cluster-job
+  - match: { jobs.0.state: "closed" }
+  - match: { jobs.0.data_counts.processed_record_count: 2 }
+  - is_true: jobs.0.model_size_stats
+  - is_false: node
+
+  - do:
+      xpack.ml.open_job:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.get_job_stats:
+        job_id: old-cluster-job
+  - match: { jobs.0.state: "opened" }
+  - match: { jobs.0.data_counts.processed_record_count: 2 }
+  - is_true: jobs.0.model_size_stats
+  - is_true: jobs.0.node
+  - is_true: jobs.0.open_time
+
+  - do:
+      xpack.ml.close_job:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.get_buckets:
+        job_id: old-cluster-job
+  - match: { count: 1 }
+
+---
+"Create a job in the mixed cluster and write some data":
+  - do:
+      xpack.ml.put_job:
+        job_id: mixed-cluster-job
+        body:  >
+          {
+            "description":"Mixed Cluster",
+            "analysis_config" : {
+                "bucket_span": "60s",
+                "detectors" :[{"function":"metric","field_name":"responsetime","by_field_name":"airline"}]
+            },
+            "data_description" : {
+                "format":"xcontent",
+                "time_field":"time",
+                "time_format":"epoch"
+            }
+          }
+  - match: { job_id: mixed-cluster-job }
+
+  - do:
+      xpack.ml.open_job:
+        job_id: mixed-cluster-job
+
+  - do:
+      xpack.ml.post_data:
+        job_id: mixed-cluster-job
+        body:
+          - airline: AAL
+            responsetime: 132.2046
+            sourcetype: post-data-job
+            time: 1403481600
+          - airline: JZA
+            responsetime: 990.4628
+            sourcetype: post-data-job
+            time: 1403481700
+  - match: { processed_record_count: 2 }
+
+  - do:
+      xpack.ml.close_job:
+        job_id: mixed-cluster-job

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/40_ml_datafeed_crud.yml

@@ -0,0 +1,58 @@
+setup:
+ - do:
+     cluster.health:
+        wait_for_status: yellow
+        wait_for_nodes: 2
+        timeout: 25s
+
+---
+"Test old cluster datafeed":
+  - do:
+      xpack.ml.get_datafeeds:
+        datafeed_id: old-cluster-datafeed
+  - match: { datafeeds.0.datafeed_id: "old-cluster-datafeed"}
+  - length: { datafeeds.0.indices: 1 }
+  - length: { datafeeds.0.types: 1 }
+  - gte: { datafeeds.0.scroll_size: 2000 }
+
+  - do:
+      xpack.ml.get_datafeed_stats:
+        datafeed_id: old-cluster-datafeed
+  - match: { datafeeds.0.state: "stopped"}
+  - is_false: datafeeds.0.node
+
+---
+"Put job and datafeed in mixed cluster":
+
+  - do:
+      xpack.ml.put_job:
+        job_id: mixed-cluster-datafeed-job
+        body:  >
+          {
+            "description":"Cluster upgrade",
+            "analysis_config" : {
+                "bucket_span": "60s",
+                "detectors" :[{"function":"count"}]
+            },
+            "data_description" : {
+                "format":"xcontent",
+                "time_field":"time"
+            }
+          }
+
+  - do:
+      xpack.ml.put_datafeed:
+        datafeed_id: mixed-cluster-datafeed
+        body:  >
+          {
+            "job_id":"mixed-cluster-datafeed-job",
+            "indices":["airline-data"],
+            "types":["response"],
+            "scroll_size": 2000
+          }
+
+  - do:
+      xpack.ml.get_datafeed_stats:
+        datafeed_id: mixed-cluster-datafeed
+  - match: { datafeeds.0.state: stopped}
+  - is_false: datafeeds.0.node

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/old_cluster/30_ml_jobs_crud.yml

@@ -0,0 +1,47 @@
+---
+"Put job on the old cluster and post some data":
+
+  - do:
+      xpack.ml.put_job:
+        job_id: old-cluster-job
+        body:  >
+          {
+            "description":"Cluster upgrade",
+            "analysis_config" : {
+                "bucket_span": "60s",
+                "detectors" :[{"function":"metric","field_name":"responsetime","by_field_name":"airline"}]
+            },
+            "data_description" : {
+                "format":"xcontent",
+                "time_field":"time",
+                "time_format":"epoch"
+            }
+          }
+  - match: { job_id: old-cluster-job }
+
+  - do:
+      xpack.ml.open_job:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.post_data:
+        job_id: old-cluster-job
+        body:
+          - airline: AAL
+            responsetime: 132.2046
+            sourcetype: post-data-job
+            time: 1403481600
+          - airline: JZA
+            responsetime: 990.4628
+            sourcetype: post-data-job
+            time: 1403481700
+  - match: { processed_record_count: 2 }
+
+  - do:
+      xpack.ml.close_job:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.get_buckets:
+        job_id: old-cluster-job
+  - match: { count: 1 }

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/old_cluster/40_ml_datafeed_crud.yml

@@ -0,0 +1,36 @@
+---
+"Put job and datafeed in old cluster":
+
+  - do:
+      xpack.ml.put_job:
+        job_id: old-cluster-datafeed-job
+        body:  >
+          {
+            "description":"Cluster upgrade",
+            "analysis_config" : {
+                "bucket_span": "60s",
+                "detectors" :[{"function":"count"}]
+            },
+            "data_description" : {
+                "format":"xcontent",
+                "time_field":"time"
+            }
+          }
+  - match: { job_id: old-cluster-datafeed-job }
+
+  - do:
+      xpack.ml.put_datafeed:
+        datafeed_id: old-cluster-datafeed
+        body:  >
+          {
+            "job_id":"old-cluster-datafeed-job",
+            "indices":["airline-data"],
+            "types":["response"],
+            "scroll_size": 2000
+          }
+
+  - do:
+      xpack.ml.get_datafeed_stats:
+        datafeed_id: old-cluster-datafeed
+  - match: { datafeeds.0.state: stopped}
+  - is_false: datafeeds.0.node

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/upgraded_cluster/30_ml_jobs_crud.yml

@@ -0,0 +1,81 @@
+---
+setup:
+ - do:
+     cluster.health:
+        wait_for_status: green
+        wait_for_nodes: 2
+        timeout: 25s
+
+---
+"Test open old jobs":
+
+  - do:
+      xpack.ml.get_jobs:
+        job_id: "_all"
+  - match: { count: 2 }
+
+  - do:
+      xpack.ml.open_job:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.get_job_stats:
+        job_id: old-cluster-job
+  - match: { jobs.0.state: "opened" }
+  - match: { jobs.0.data_counts.processed_record_count: 2 }
+  - is_true: jobs.0.model_size_stats
+  - is_true: jobs.0.node
+  - is_true: jobs.0.open_time
+
+  - do:
+      xpack.ml.open_job:
+        job_id: mixed-cluster-job
+
+  - do:
+      xpack.ml.get_job_stats:
+        job_id: mixed-cluster-job
+  - match: { jobs.0.state: "opened" }
+  - match: { jobs.0.data_counts.processed_record_count: 2 }
+  - is_true: jobs.0.model_size_stats
+  - is_true: jobs.0.node
+  - is_true: jobs.0.open_time
+
+  - do:
+      xpack.ml.close_job:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.close_job:
+        job_id: mixed-cluster-job
+
+  - do:
+      xpack.ml.get_buckets:
+        job_id: old-cluster-job
+  - match: { count: 1 }
+
+  - do:
+      xpack.ml.get_buckets:
+        job_id: mixed-cluster-job
+  - match: { count: 1 }
+
+---
+teardown:
+  - do:
+      xpack.ml.delete_job:
+        job_id: old-cluster-job
+  - match: { acknowledged: true }
+
+  - do:
+      catch: missing
+      xpack.ml.get_jobs:
+        job_id: old-cluster-job
+
+  - do:
+      xpack.ml.delete_job:
+        job_id: mixed-cluster-job
+  - match: { acknowledged: true }
+
+  - do:
+      catch: missing
+      xpack.ml.get_jobs:
+        job_id: mixed-cluster-job

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/upgraded_cluster/40_ml_datafeed_crud.yml

@@ -0,0 +1,99 @@
+setup:
+  - do:
+     cluster.health:
+        wait_for_status: green
+        wait_for_nodes: 2
+        timeout: 25s
+
+  - do:
+      indices.create:
+        index: airline-data
+        body:
+          mappings:
+            response:
+              properties:
+                time:
+                  type: date
+
+---
+"Test old and mixed cluster datafeeds":
+  - do:
+      xpack.ml.get_datafeeds:
+        datafeed_id: old-cluster-datafeed
+  - match: { datafeeds.0.datafeed_id: "old-cluster-datafeed"}
+  - length: { datafeeds.0.indices: 1 }
+  - length: { datafeeds.0.types: 1 }
+  - gte: { datafeeds.0.scroll_size: 2000 }
+
+  - do:
+      xpack.ml.get_datafeed_stats:
+        datafeed_id: old-cluster-datafeed
+  - match: { datafeeds.0.state: "stopped"}
+  - is_false: datafeeds.0.node
+
+  - do:
+      xpack.ml.get_datafeeds:
+        datafeed_id: mixed-cluster-datafeed
+  - match: { datafeeds.0.datafeed_id: "mixed-cluster-datafeed"}
+  - length: { datafeeds.0.indices: 1 }
+  - length: { datafeeds.0.types: 1 }
+  - gte: { datafeeds.0.scroll_size: 2000 }
+
+  - do:
+      xpack.ml.get_datafeed_stats:
+        datafeed_id: mixed-cluster-datafeed
+  - match: { datafeeds.0.state: "stopped"}
+  - is_false: datafeeds.0.node
+
+  - do:
+      xpack.ml.open_job:
+        job_id: old-cluster-datafeed-job
+
+  - do:
+      xpack.ml.start_datafeed:
+        datafeed_id: old-cluster-datafeed
+        start: 0
+
+  - do:
+      xpack.ml.stop_datafeed:
+        datafeed_id: old-cluster-datafeed
+
+  - do:
+      xpack.ml.close_job:
+        job_id: old-cluster-datafeed-job
+
+  - do:
+      xpack.ml.delete_datafeed:
+        datafeed_id: old-cluster-datafeed
+
+  - do:
+      xpack.ml.delete_job:
+        job_id: old-cluster-datafeed-job
+  - match: { acknowledged: true }
+
+  - do:
+      xpack.ml.open_job:
+        job_id: mixed-cluster-datafeed-job
+
+  - do:
+      xpack.ml.start_datafeed:
+        datafeed_id: mixed-cluster-datafeed
+        start: 0
+
+  - do:
+      xpack.ml.stop_datafeed:
+        datafeed_id: mixed-cluster-datafeed
+
+  - do:
+      xpack.ml.close_job:
+        job_id: mixed-cluster-datafeed-job
+
+  - do:
+      xpack.ml.delete_datafeed:
+        datafeed_id: mixed-cluster-datafeed
+
+  - do:
+      xpack.ml.delete_job:
+        job_id: mixed-cluster-datafeed-job
+  - match: { acknowledged: true }
+