[ML] Validate job configs before transport (elastic/x-pack-elasticsearch#1375)

If invalid job configs are transported to the master node then the root
cause of the validation exception gets reported as a remote_transport_exception,
which is extremely confusing.

This commit moves the validation of job configurations to the first node that
handles the action.

Fixes elastic/x-pack-kibana#1172

Original commit: elastic/x-pack-elasticsearch@5ed59d2a6f

plugin/src/main/java/org/elasticsearch/xpack/ml/action/PutJobAction.java

@@ -63,29 +63,32 @@ public class PutJobAction extends Action<PutJobAction.Request, PutJobAction.Resp
     public static class Request extends AcknowledgedRequest<Request> implements ToXContent {
 
         public static Request parseRequest(String jobId, XContentParser parser) {
-            Job.Builder job = Job.PARSER.apply(parser, null);
+            Job.Builder jobBuilder = Job.PARSER.apply(parser, null);
-            if (job.getId() == null) {
+            if (jobBuilder.getId() == null) {
-                job.setId(jobId);
+                jobBuilder.setId(jobId);
-            } else if (!Strings.isNullOrEmpty(jobId) && !jobId.equals(job.getId())) {
+            } else if (!Strings.isNullOrEmpty(jobId) && !jobId.equals(jobBuilder.getId())) {
-                // If we have both URI and body job ID, they must be identical
+                // If we have both URI and body jobBuilder ID, they must be identical
                 throw new IllegalArgumentException(Messages.getMessage(Messages.INCONSISTENT_ID, Job.ID.getPreferredName(),
-                        job.getId(), jobId));
+                        jobBuilder.getId(), jobId));
             }
 
-            return new Request(job);
+            return new Request(jobBuilder);
         }
 
-        private Job.Builder job;
+        private Job.Builder jobBuilder;
 
-        public Request(Job.Builder job) {
+        public Request(Job.Builder jobBuilder) {
-            this.job = job;
+            // Validate the jobBuilder immediately so that errors can be detected prior to transportation.
+            jobBuilder.validateInputFields();
+
+            this.jobBuilder = jobBuilder;
         }
 
         Request() {
         }
 
-        public Job.Builder getJob() {
+        public Job.Builder getJobBuilder() {
-            return job;
+            return jobBuilder;
         }
 
         @Override
@@ -96,18 +99,18 @@ public class PutJobAction extends Action<PutJobAction.Request, PutJobAction.Resp
         @Override
         public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
-            job = new Job.Builder(in);
+            jobBuilder = new Job.Builder(in);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
-            job.writeTo(out);
+            jobBuilder.writeTo(out);
         }
 
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-            job.toXContent(builder, params);
+            jobBuilder.toXContent(builder, params);
             return builder;
         }
 
@@ -116,12 +119,12 @@ public class PutJobAction extends Action<PutJobAction.Request, PutJobAction.Resp
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
-            return Objects.equals(job, request.job);
+            return Objects.equals(jobBuilder, request.jobBuilder);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(job);
+            return Objects.hash(jobBuilder);
         }
 
         @Override

plugin/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java

@@ -163,7 +163,7 @@ public class JobManager extends AbstractComponent {
      * Stores a job in the cluster state
      */
     public void putJob(PutJobAction.Request request, ClusterState state, ActionListener<PutJobAction.Response> actionListener) {
-        Job job = request.getJob().build(new Date());
+        Job job = request.getJobBuilder().build(new Date());
 
         jobProvider.createJobResultIndex(job, state, new ActionListener<Boolean>() {
             @Override

plugin/src/main/java/org/elasticsearch/xpack/ml/job/config/Job.java

@@ -816,22 +816,11 @@ public class Job extends AbstractDiffable<Job> implements Writeable, ToXContent
                     resultsRetentionDays, customSettings, modelSnapshotId, resultsIndexName, deleted);
         }
 
-        public Job build(Date createTime) {
+        /**
-            setCreateTime(createTime);
+         * Call this method to validate that the job JSON provided by a user is valid.
-            return build();
+         * Throws an exception if there are any problems; normal return implies valid.
-        }
+         */
-
+        public void validateInputFields() {
-        public Job build() {
-
-            Date createTime;
-            Date finishedTime;
-            Date lastDataTime;
-            String modelSnapshotId;
-
-            createTime = ExceptionsHelper.requireNonNull(this.createTime, CREATE_TIME.getPreferredName());
-            finishedTime = this.finishedTime;
-            lastDataTime = this.lastDataTime;
-            modelSnapshotId = this.modelSnapshotId;
 
             if (analysisConfig == null) {
                 throw new IllegalArgumentException(Messages.getMessage(Messages.JOB_CONFIG_MISSING_ANALYSISCONFIG));
@@ -849,11 +838,29 @@ public class Job extends AbstractDiffable<Job> implements Writeable, ToXContent
                 throw new IllegalArgumentException(Messages.getMessage(Messages.JOB_CONFIG_ID_TOO_LONG, MAX_JOB_ID_LENGTH));
             }
 
-            if (Strings.isNullOrEmpty(resultsIndexName)) {
+            // Results index name not specified in user input means use the default, so is acceptable in this validation
-                resultsIndexName = AnomalyDetectorsIndex.RESULTS_INDEX_DEFAULT;
+            if (!Strings.isNullOrEmpty(resultsIndexName) && !MlStrings.isValidId(resultsIndexName)) {
-            } else if (!MlStrings.isValidId(resultsIndexName)) {
                 throw new IllegalArgumentException(
                         Messages.getMessage(Messages.INVALID_ID, RESULTS_INDEX_NAME.getPreferredName(), resultsIndexName));
+            }
+
+            // Creation time is NOT required in user input, hence validated only on build
+        }
+
+        public Job build(Date createTime) {
+            setCreateTime(createTime);
+            return build();
+        }
+
+        public Job build() {
+
+            validateInputFields();
+
+            // Creation time is NOT required in user input, hence validated only on build
+            Date createTime = ExceptionsHelper.requireNonNull(this.createTime, CREATE_TIME.getPreferredName());
+
+            if (Strings.isNullOrEmpty(resultsIndexName)) {
+                resultsIndexName = AnomalyDetectorsIndex.RESULTS_INDEX_DEFAULT;
             } else if (!resultsIndexName.equals(AnomalyDetectorsIndex.RESULTS_INDEX_DEFAULT)) {
                 // User-defined names are prepended with "custom"
                 // Conditional guards against multiple prepending due to updates instead of first creation

qa/ml-basic-multi-node/src/test/java/org/elasticsearch/xpack/ml/integration/MlBasicMultiNodeIT.java

@@ -8,12 +8,14 @@ package org.elasticsearch.xpack.ml.integration;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.xpack.ml.MachineLearning;
 
 import java.io.IOException;
+import java.net.URLEncoder;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -34,6 +36,18 @@ public class MlBasicMultiNodeIT extends ESRestTestCase {
         assertTrue((Boolean) ml.get("enabled"));
     }
 
+    public void testInvalidJob() throws Exception {
+        // The job name is invalid because it contains a space
+        String jobId = "invalid job";
+        ResponseException e = expectThrows(ResponseException.class, () -> createFarequoteJob(jobId));
+        assertTrue(e.getMessage(), e.getMessage().contains("can contain lowercase alphanumeric (a-z and 0-9), hyphens or underscores"));
+        // If validation of the invalid job is not done until after transportation to the master node then the
+        // root cause gets reported as a remote_transport_exception.  The code in PubJobAction is supposed to
+        // validate before transportation to avoid this.  This test must be done in a multi-node cluster to have
+        // a chance of catching a problem, hence it is here rather than in the single node integration tests.
+        assertFalse(e.getMessage(), e.getMessage().contains("remote_transport_exception"));
+    }
+
     public void testMiniFarequote() throws Exception {
         String jobId = "mini-farequote-job";
         createFarequoteJob(jobId);
@@ -293,7 +307,7 @@ public class MlBasicMultiNodeIT extends ESRestTestCase {
         xContentBuilder.endObject();
         xContentBuilder.endObject();
 
-        return client().performRequest("put", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId,
+        return client().performRequest("put", MachineLearning.BASE_PATH + "anomaly_detectors/" + URLEncoder.encode(jobId, "UTF-8"),
                 Collections.emptyMap(), new StringEntity(xContentBuilder.string(), ContentType.APPLICATION_JSON));
     }