[ML] add multi node integ tests for data frames (#41508)

* [ML] adding native-multi-node-integTests for data frames'

* addressing streaming issues

* formatting fixes

* Addressing PR comments

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -44,6 +44,14 @@
 import org.elasticsearch.xpack.core.ccr.CCRFeatureSet;
 import org.elasticsearch.xpack.core.dataframe.DataFrameFeatureSetUsage;
 import org.elasticsearch.xpack.core.dataframe.DataFrameField;
+import org.elasticsearch.xpack.core.dataframe.action.DeleteDataFrameTransformAction;
+import org.elasticsearch.xpack.core.dataframe.action.GetDataFrameTransformsAction;
+import org.elasticsearch.xpack.core.dataframe.action.GetDataFrameTransformsStatsAction;
+import org.elasticsearch.xpack.core.dataframe.action.PreviewDataFrameTransformAction;
+import org.elasticsearch.xpack.core.dataframe.action.PutDataFrameTransformAction;
+import org.elasticsearch.xpack.core.dataframe.action.StartDataFrameTransformAction;
+import org.elasticsearch.xpack.core.dataframe.action.StartDataFrameTransformTaskAction;
+import org.elasticsearch.xpack.core.dataframe.action.StopDataFrameTransformAction;
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransform;
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformState;
 import org.elasticsearch.xpack.core.deprecation.DeprecationInfoAction;
@@ -363,7 +371,16 @@ public List<Action<? extends ActionResponse>> getClientActions() {
                 RemoveIndexLifecyclePolicyAction.INSTANCE,
                 MoveToStepAction.INSTANCE,
                 RetryAction.INSTANCE,
-                TransportFreezeIndexAction.FreezeIndexAction.INSTANCE
+                TransportFreezeIndexAction.FreezeIndexAction.INSTANCE,
+                // Data Frame
+                PutDataFrameTransformAction.INSTANCE,
+                StartDataFrameTransformAction.INSTANCE,
+                StartDataFrameTransformTaskAction.INSTANCE,
+                StopDataFrameTransformAction.INSTANCE,
+                DeleteDataFrameTransformAction.INSTANCE,
+                GetDataFrameTransformsAction.INSTANCE,
+                GetDataFrameTransformsStatsAction.INSTANCE,
+                PreviewDataFrameTransformAction.INSTANCE
         );
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/action/GetDataFrameTransformsStatsAction.java

@@ -56,21 +56,22 @@ public static class Request extends BaseTasksRequest<Request> {
 
         public static final int MAX_SIZE_RETURN = 1000;
         // used internally to expand the queried id expression
-        private List<String> expandedIds = Collections.emptyList();
+        private List<String> expandedIds;
 
         public Request(String id) {
             if (Strings.isNullOrEmpty(id) || id.equals("*")) {
                 this.id = MetaData.ALL;
             } else {
                 this.id = id;
             }
+            this.expandedIds = Collections.singletonList(id);
         }
 
         public Request(StreamInput in) throws IOException {
             super(in);
             id = in.readString();
-            expandedIds = in.readList(StreamInput::readString);
-            pageParams = in.readOptionalWriteable(PageParams::new);
+            expandedIds = Collections.unmodifiableList(in.readStringList());
+            pageParams = new PageParams(in);
         }
 
         @Override
@@ -93,7 +94,7 @@ public void setExpandedIds(List<String> expandedIds) {
         }
 
         public final void setPageParams(PageParams pageParams) {
-            this.pageParams = pageParams;
+            this.pageParams = Objects.requireNonNull(pageParams);
         }
 
         public final PageParams getPageParams() {
@@ -105,7 +106,7 @@ public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(id);
             out.writeStringCollection(expandedIds);
-            out.writeOptionalWriteable(pageParams);
+            pageParams.writeTo(out);
         }
 
         @Override
@@ -136,7 +137,7 @@ public boolean equals(Object obj) {
         }
     }
 
-    public static class Response extends BaseTasksResponse implements Writeable, ToXContentObject {
+    public static class Response extends BaseTasksResponse implements ToXContentObject {
         private List<DataFrameTransformStateAndStats> transformsStateAndStats;
 
         public Response(List<DataFrameTransformStateAndStats> transformsStateAndStats) {
@@ -165,6 +166,11 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeList(transformsStateAndStats);
         }
 
+        @Override
+        public void readFrom(StreamInput in) {
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+        }
+
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/action/StartDataFrameTransformAction.java

@@ -95,7 +95,7 @@ public boolean equals(Object obj) {
         }
     }
 
-    public static class Response extends BaseTasksResponse implements Writeable, ToXContentObject {
+    public static class Response extends BaseTasksResponse implements ToXContentObject {
         private final boolean started;
 
         public Response(StreamInput in) throws IOException {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/action/StartDataFrameTransformTaskAction.java

@@ -93,7 +93,7 @@ public boolean equals(Object obj) {
         }
     }
 
-    public static class Response extends BaseTasksResponse implements Writeable, ToXContentObject {
+    public static class Response extends BaseTasksResponse implements ToXContentObject {
         private final boolean started;
 
         public Response(StreamInput in) throws IOException {

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/dataframe/action/GetDataFrameTransformsStatsActionResponseTests.java

@@ -6,6 +6,9 @@
 
 package org.elasticsearch.xpack.core.dataframe.action;
 
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.TaskOperationFailure;
 import org.elasticsearch.common.io.stream.Writeable.Reader;
 import org.elasticsearch.xpack.core.dataframe.action.GetDataFrameTransformsStatsAction.Response;
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformStateAndStats;
@@ -18,11 +21,18 @@ public class GetDataFrameTransformsStatsActionResponseTests extends AbstractWire
     @Override
     protected Response createTestInstance() {
         List<DataFrameTransformStateAndStats> stats = new ArrayList<>();
-        for (int i = 0; i < randomInt(10); ++i) {
+        int totalStats = randomInt(10);
+        for (int i = 0; i < totalStats; ++i) {
             stats.add(DataFrameTransformStateAndStatsTests.randomDataFrameTransformStateAndStats());
         }
-
-        return new Response(stats);
+        int totalErrors = randomInt(10);
+        List<TaskOperationFailure> taskFailures = new ArrayList<>(totalErrors);
+        List<ElasticsearchException> nodeFailures = new ArrayList<>(totalErrors);
+        for (int i = 0; i < totalErrors; i++) {
+            taskFailures.add(new TaskOperationFailure("node1", randomLongBetween(1, 10), new Exception("error")));
+            nodeFailures.add(new FailedNodeException("node1", "message", new Exception("error")));
+        }
+        return new Response(stats, taskFailures, nodeFailures);
     }
 
     @Override

x-pack/plugin/data-frame/qa/multi-node-tests/build.gradle

@@ -0,0 +1,55 @@
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply plugin: 'elasticsearch.rest-test'
+
+dependencies {
+  testCompile project(path: xpackModule('core'), configuration: 'default')
+  testCompile project(path: xpackModule('core'), configuration: 'testArtifacts')
+  testCompile project(path: xpackModule('data-frame'), configuration: 'runtime')
+}
+
+// location for keys and certificates
+File keystoreDir = new File(project.buildDir, 'keystore')
+File nodeKey = file("$keystoreDir/testnode.pem")
+File nodeCert = file("$keystoreDir/testnode.crt")
+// Add key and certs to test classpath: it expects it there
+task copyKeyCerts(type: Copy) {
+    from(project(':x-pack:plugin:core').file('src/test/resources/org/elasticsearch/xpack/security/transport/ssl/certs/simple/')) {
+        include 'testnode.crt', 'testnode.pem'
+    }
+    into keystoreDir
+}
+// Add keys and cets to test classpath: it expects it there
+sourceSets.test.resources.srcDir(keystoreDir)
+processTestResources.dependsOn(copyKeyCerts)
+
+integTestCluster {
+  dependsOn copyKeyCerts
+  setting 'xpack.security.enabled', 'true'
+  setting 'xpack.license.self_generated.type', 'trial'
+  setting 'xpack.monitoring.enabled', 'false'
+  setting 'xpack.security.authc.token.enabled', 'true'
+  setting 'xpack.security.transport.ssl.enabled', 'true'
+  setting 'xpack.security.transport.ssl.key', nodeKey.name
+  setting 'xpack.security.transport.ssl.certificate', nodeCert.name
+  setting 'xpack.security.transport.ssl.verification_mode', 'certificate'
+  setting 'xpack.security.audit.enabled', 'false'
+  setting 'xpack.license.self_generated.type', 'trial'
+  keystoreSetting 'bootstrap.password', 'x-pack-test-password'
+  keystoreSetting 'xpack.security.transport.ssl.secure_key_passphrase', 'testnode'
+  setupCommand 'setupDummyUser',
+          'bin/elasticsearch-users', 'useradd', 'x_pack_rest_user', '-p', 'x-pack-test-password', '-r', 'superuser'
+
+  numNodes = 3
+  extraConfigFile nodeKey.name, nodeKey
+  extraConfigFile nodeCert.name, nodeCert
+  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()
+  }
+}