From 928d24bffbd40643b1d77ec4e3bf2ed5aa992f80 Mon Sep 17 00:00:00 2001
From: Dimitris Athanasiou <dimitris@elastic.co>
Date: Thu, 9 Jan 2020 15:58:25 +0200
Subject: [PATCH 1/3] [ML] Reuse SourceDestValidator for data frame analytics

This commit removes validation logic of source and dest indices
for data frame analytics and replaces it with using the common
`SourceDestValidator` class which is already used by transforms.
This way the validations and their messages become consistent
while we reduce code.

This means that where these validations fail the error messages
will be slightly different for data frame analytics.
---
 .../validation/SourceDestValidator.java       |  36 ++--
 .../action/PutDataFrameAnalyticsAction.java   |  21 +-
 .../ml/dataframe/DataFrameAnalyticsDest.java  |  13 --
 .../validation/SourceDestValidatorTests.java  |  94 ++++-----
 .../DataFrameAnalyticsDestTests.java          |  19 --
 .../ml/qa/ml-with-security/build.gradle       |   2 +
 .../TransportPutDataFrameAnalyticsAction.java |  44 ++--
 ...ransportStartDataFrameAnalyticsAction.java |  47 +++--
 .../ml/dataframe/SourceDestValidator.java     |  68 -------
 .../dataframe/SourceDestValidatorTests.java   | 188 ------------------
 .../test/ml/data_frame_analytics_crud.yml     |  46 ++++-
 .../test/ml/start_data_frame_analytics.yml    |   2 +-
 .../TransportPreviewTransformAction.java      |   3 +-
 .../action/TransportPutTransformAction.java   |   3 +-
 .../action/TransportStartTransformAction.java |   3 +-
 .../TransportUpdateTransformAction.java       |   3 +-
 .../utils/SourceDestValidations.java          |  40 ++++
 17 files changed, 237 insertions(+), 395 deletions(-)
 delete mode 100644 x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidator.java
 delete mode 100644 x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidatorTests.java
 create mode 100644 x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/utils/SourceDestValidations.java

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java
index ef28fed54cf48..f8171884131b1 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java
@@ -46,7 +46,6 @@ public final class SourceDestValidator {
 
     // messages
     public static final String SOURCE_INDEX_MISSING = "Source index [{0}] does not exist";
-    public static final String SOURCE_LOWERCASE = "Source index [{0}] must be lowercase";
     public static final String DEST_IN_SOURCE = "Destination index [{0}] is included in source expression [{1}]";
     public static final String DEST_LOWERCASE = "Destination index [{0}] must be lowercase";
     public static final String NEEDS_REMOTE_CLUSTER_SEARCH = "Source index is configured with a remote index pattern(s) [{0}]"
@@ -216,7 +215,7 @@ private void resolveLocalAndRemoteSource() {
         }
     }
 
-    interface SourceDestValidation {
+    public interface SourceDestValidation {
         void validate(Context context, ActionListener<Context> listener);
     }
 
@@ -228,18 +227,7 @@ interface SourceDestValidation {
     public static final SourceDestValidation REMOTE_SOURCE_VALIDATION = new RemoteSourceEnabledAndRemoteLicenseValidation();
     public static final SourceDestValidation DESTINATION_IN_SOURCE_VALIDATION = new DestinationInSourceValidation();
     public static final SourceDestValidation DESTINATION_SINGLE_INDEX_VALIDATION = new DestinationSingleIndexValidation();
-
-    // set of default validation collections, if you want to automatically benefit from new validators, use those
-    public static final List<SourceDestValidation> PREVIEW_VALIDATIONS = Arrays.asList(SOURCE_MISSING_VALIDATION, REMOTE_SOURCE_VALIDATION);
-
-    public static final List<SourceDestValidation> ALL_VALIDATIONS = Arrays.asList(
-        SOURCE_MISSING_VALIDATION,
-        REMOTE_SOURCE_VALIDATION,
-        DESTINATION_IN_SOURCE_VALIDATION,
-        DESTINATION_SINGLE_INDEX_VALIDATION
-    );
-
-    public static final List<SourceDestValidation> NON_DEFERABLE_VALIDATIONS = Arrays.asList(DESTINATION_SINGLE_INDEX_VALIDATION);
+    public static final SourceDestValidation REMOTE_SOURCE_NOT_SUPPORTED_VALIDATION = new RemoteSourceNotSupportedValidation();
 
     /**
      * Create a new Source Dest Validator
@@ -299,10 +287,11 @@ public void validate(
             }
         }, listener::onFailure);
 
+        // We traverse the validations in reverse order as we chain the listeners from back to front
         for (int i = validations.size() - 1; i >= 0; i--) {
-            final SourceDestValidation validation = validations.get(i);
+            SourceDestValidation validation = validations.get(i);
             final ActionListener<Context> previousValidationListener = validationListener;
-            validationListener = ActionListener.wrap(c -> { validation.validate(c, previousValidationListener); }, listener::onFailure);
+            validationListener = ActionListener.wrap(c -> validation.validate(c, previousValidationListener), listener::onFailure);
         }
 
         validationListener.onResponse(context);
@@ -427,13 +416,13 @@ public void validate(Context context, ActionListener<Context> listener) {
                 return;
             }
 
-            if (context.resolvedSource.contains(destIndex)) {
+            if (context.resolveSource().contains(destIndex)) {
                 context.addValidationError(DEST_IN_SOURCE, destIndex, Strings.arrayToCommaDelimitedString(context.getSource()));
                 listener.onResponse(context);
                 return;
             }
 
-            if (context.resolvedSource.contains(context.resolveDest())) {
+            if (context.resolveSource().contains(context.resolveDest())) {
                 context.addValidationError(
                     DEST_IN_SOURCE,
                     context.resolveDest(),
@@ -454,6 +443,17 @@ public void validate(Context context, ActionListener<Context> listener) {
         }
     }
 
+    static class RemoteSourceNotSupportedValidation implements SourceDestValidation {
+
+        @Override
+        public void validate(Context context, ActionListener<Context> listener) {
+            if (context.resolveRemoteSource().isEmpty() == false) {
+                context.addValidationError("remote source indices are not supported");
+            }
+            listener.onResponse(context);
+        }
+    }
+
     private static String getMessage(String message, Object... args) {
         return new MessageFormat(message, Locale.ROOT).format(args);
     }
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/PutDataFrameAnalyticsAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/PutDataFrameAnalyticsAction.java
index 4f4ddc388aed7..7ce5be3b980bd 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/PutDataFrameAnalyticsAction.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/PutDataFrameAnalyticsAction.java
@@ -18,9 +18,11 @@
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsSource;
 import org.elasticsearch.xpack.core.ml.job.messages.Messages;
+import org.elasticsearch.xpack.core.ml.utils.MlStrings;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -90,14 +92,29 @@ public DataFrameAnalyticsConfig getConfig() {
         @Override
         public ActionRequestValidationException validate() {
             ActionRequestValidationException error = null;
+            error = checkConfigIdIsValid(config, error);
+            error = SourceDestValidator.validateRequest(error, config.getDest().getIndex());
             error = checkNoIncludedAnalyzedFieldsAreExcludedBySourceFiltering(config, error);
             return error;
         }
 
+        private ActionRequestValidationException checkConfigIdIsValid(DataFrameAnalyticsConfig config,
+                                                                      ActionRequestValidationException error) {
+            if (MlStrings.isValidId(config.getId()) == false) {
+                error = ValidateActions.addValidationError(Messages.getMessage(Messages.INVALID_ID, DataFrameAnalyticsConfig.ID,
+                    config.getId()), error);
+            }
+            if (!MlStrings.hasValidLengthForId(config.getId())) {
+                error = ValidateActions.addValidationError(Messages.getMessage(Messages.ID_TOO_LONG, DataFrameAnalyticsConfig.ID,
+                    config.getId(), MlStrings.ID_LENGTH_LIMIT), error);
+            }
+            return error;
+        }
+
         private ActionRequestValidationException checkNoIncludedAnalyzedFieldsAreExcludedBySourceFiltering(
                 DataFrameAnalyticsConfig config, ActionRequestValidationException error) {
             if (config.getAnalyzedFields() == null) {
-                return null;
+                return error;
             }
             for (String analyzedInclude : config.getAnalyzedFields().includes()) {
                 if (config.getSource().isFieldExcluded(analyzedInclude)) {
@@ -107,7 +124,7 @@ private ActionRequestValidationException checkNoIncludedAnalyzedFieldsAreExclude
                         + DataFrameAnalyticsSource._SOURCE.getPreferredName() + "]", error);
                 }
             }
-            return null;
+            return error;
         }
 
         @Override
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDest.java
index 3bc435336f062..98f1bdeb19189 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDest.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDest.java
@@ -13,15 +13,11 @@
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.indices.InvalidIndexNameException;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 
 import java.io.IOException;
-import java.util.Locale;
 import java.util.Objects;
 
-import static org.elasticsearch.cluster.metadata.MetaDataCreateIndexService.validateIndexOrAliasName;
-
 public class DataFrameAnalyticsDest implements Writeable, ToXContentObject {
 
     public static final ParseField INDEX = new ParseField("index");
@@ -94,13 +90,4 @@ public String getIndex() {
     public String getResultsField() {
         return resultsField;
     }
-
-    public void validate() {
-        if (index != null) {
-            validateIndexOrAliasName(index, InvalidIndexNameException::new);
-            if (index.toLowerCase(Locale.ROOT).equals(index) == false) {
-                throw new InvalidIndexNameException(index, "dest.index must be lowercase");
-            }
-        }
-    }
 }
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java
index 566925258c879..109148f5696d7 100644
--- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java
+++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java
@@ -41,7 +41,9 @@
 import org.junit.After;
 import org.junit.Before;
 
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -52,6 +54,10 @@
 import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
 import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_VERSION_CREATED;
 import static org.elasticsearch.mock.orig.Mockito.when;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_IN_SOURCE_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_SINGLE_INDEX_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.REMOTE_SOURCE_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.SOURCE_MISSING_VALIDATION;
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Mockito.spy;
 
@@ -67,6 +73,14 @@ public class SourceDestValidatorTests extends ESTestCase {
     private static final String REMOTE_PLATINUM = "remote-platinum";
 
     private static final ClusterState CLUSTER_STATE;
+
+    private static final List<SourceDestValidator.SourceDestValidation> TEST_VALIDATIONS = Arrays.asList(
+        SOURCE_MISSING_VALIDATION,
+        DESTINATION_IN_SOURCE_VALIDATION,
+        DESTINATION_SINGLE_INDEX_VALIDATION,
+        REMOTE_SOURCE_VALIDATION
+    );
+
     private Client clientWithBasicLicense;
     private Client clientWithExpiredBasicLicense;
     private Client clientWithPlatinumLicense;
@@ -184,13 +198,13 @@ public void closeComponents() throws Exception {
         ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS);
     }
 
-    public void testCheck_GivenSimpleSourceIndexAndValidDestIndex() throws InterruptedException {
+    public void testValidate_GivenSimpleSourceIndexAndValidDestIndex() throws InterruptedException {
         assertValidation(
             listener -> simpleNonRemoteValidator.validate(
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 "dest",
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             true,
@@ -204,7 +218,7 @@ public void testCheck_GivenNoSourceIndexAndValidDestIndex() throws InterruptedEx
                 CLUSTER_STATE,
                 new String[] {},
                 "dest",
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -221,7 +235,7 @@ public void testCheck_GivenMissingConcreteSourceIndex() throws InterruptedExcept
                 CLUSTER_STATE,
                 new String[] { "missing" },
                 "dest",
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -236,7 +250,7 @@ public void testCheck_GivenMissingConcreteSourceIndex() throws InterruptedExcept
                 CLUSTER_STATE,
                 new String[] { "missing" },
                 "dest",
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
             true,
@@ -250,7 +264,7 @@ public void testCheck_GivenMixedMissingAndExistingConcreteSourceIndex() throws I
                 CLUSTER_STATE,
                 new String[] { SOURCE_1, "missing" },
                 "dest",
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -260,17 +274,17 @@ public void testCheck_GivenMixedMissingAndExistingConcreteSourceIndex() throws I
             }
         );
 
-        assertValidation(
-            listener -> simpleNonRemoteValidator.validate(
-                CLUSTER_STATE,
-                new String[] { SOURCE_1, "missing" },
-                "dest",
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
-                listener
-            ),
-            true,
-            null
-        );
+//        assertValidation(
+//            listener -> simpleNonRemoteValidator.validate(
+//                CLUSTER_STATE,
+//                new String[] { SOURCE_1, "missing" },
+//                "dest",
+//                Collections.emptyList(),
+//                listener
+//            ),
+//            true,
+//            null
+//        );
     }
 
     public void testCheck_GivenMixedMissingWildcardExistingConcreteSourceIndex() throws InterruptedException {
@@ -279,7 +293,7 @@ public void testCheck_GivenMixedMissingWildcardExistingConcreteSourceIndex() thr
                 CLUSTER_STATE,
                 new String[] { SOURCE_1, "wildcard*", "missing" },
                 "dest",
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -294,7 +308,7 @@ public void testCheck_GivenMixedMissingWildcardExistingConcreteSourceIndex() thr
                 CLUSTER_STATE,
                 new String[] { SOURCE_1, "wildcard*", "missing" },
                 "dest",
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
             true,
@@ -308,7 +322,7 @@ public void testCheck_GivenWildcardSourceIndex() throws InterruptedException {
                 CLUSTER_STATE,
                 new String[] { "wildcard*" },
                 "dest",
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             true,
@@ -322,7 +336,7 @@ public void testCheck_GivenDestIndexSameAsSourceIndex() throws InterruptedExcept
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 SOURCE_1,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -340,7 +354,7 @@ public void testCheck_GivenDestIndexSameAsSourceIndex() throws InterruptedExcept
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 SOURCE_1,
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
             true,
@@ -354,7 +368,7 @@ public void testCheck_GivenDestIndexMatchesSourceIndex() throws InterruptedExcep
                 CLUSTER_STATE,
                 new String[] { "source-*" },
                 SOURCE_2,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -372,7 +386,7 @@ public void testCheck_GivenDestIndexMatchesSourceIndex() throws InterruptedExcep
                 CLUSTER_STATE,
                 new String[] { "source-*" },
                 SOURCE_2,
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
             true,
@@ -386,7 +400,7 @@ public void testCheck_GivenDestIndexMatchesOneOfSourceIndices() throws Interrupt
                 CLUSTER_STATE,
                 new String[] { "source-1", "source-*" },
                 SOURCE_2,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -404,7 +418,7 @@ public void testCheck_GivenDestIndexMatchesOneOfSourceIndices() throws Interrupt
                 CLUSTER_STATE,
                 new String[] { "source-1", "source-*" },
                 SOURCE_2,
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
             true,
@@ -418,7 +432,7 @@ public void testCheck_GivenDestIndexMatchesMultipleSourceIndices() throws Interr
                 CLUSTER_STATE,
                 new String[] { "source-1", "source-*", "sou*" },
                 SOURCE_2,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -442,7 +456,7 @@ public void testCheck_GivenDestIndexIsAliasThatMatchesMultipleIndices() throws I
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 DEST_ALIAS,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -464,21 +478,11 @@ public void testCheck_GivenDestIndexIsAliasThatMatchesMultipleIndices() throws I
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 DEST_ALIAS,
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
-            (Boolean) null,
-            e -> {
-                assertEquals(1, e.validationErrors().size());
-                assertThat(
-                    e.validationErrors().get(0),
-                    equalTo(
-                        "no write index is defined for alias [dest-alias]. "
-                            + "The write index may be explicitly disabled using is_write_index=false or the alias points "
-                            + "to multiple indices without one being designated as a write index"
-                    )
-                );
-            }
+            true,
+            null
         );
     }
 
@@ -488,7 +492,7 @@ public void testCheck_GivenDestIndexIsAliasThatMatchesMultipleIndicesButHasSingl
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 ALIAS_READ_WRITE_DEST,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -514,7 +518,7 @@ public void testCheck_GivenDestIndexIsAliasThatIsIncludedInSource() throws Inter
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 SOURCE_1_ALIAS,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
@@ -532,7 +536,7 @@ public void testCheck_GivenDestIndexIsAliasThatIsIncludedInSource() throws Inter
                 CLUSTER_STATE,
                 new String[] { SOURCE_1 },
                 SOURCE_1_ALIAS,
-                SourceDestValidator.NON_DEFERABLE_VALIDATIONS,
+                Collections.emptyList(),
                 listener
             ),
             true,
@@ -546,7 +550,7 @@ public void testCheck_MultipleValidationErrors() throws InterruptedException {
                 CLUSTER_STATE,
                 new String[] { SOURCE_1, "missing" },
                 SOURCE_1_ALIAS,
-                SourceDestValidator.ALL_VALIDATIONS,
+                TEST_VALIDATIONS,
                 listener
             ),
             (Boolean) null,
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDestTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDestTests.java
index bf8ce4c8a99b0..7332687723805 100644
--- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDestTests.java
+++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDestTests.java
@@ -7,14 +7,10 @@
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.XContentParser;
-import org.elasticsearch.indices.InvalidIndexNameException;
-import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.AbstractSerializingTestCase;
 
 import java.io.IOException;
 
-import static org.hamcrest.Matchers.equalTo;
-
 public class DataFrameAnalyticsDestTests extends AbstractSerializingTestCase<DataFrameAnalyticsDest> {
 
     @Override
@@ -37,19 +33,4 @@ public static DataFrameAnalyticsDest createRandom() {
     protected Writeable.Reader<DataFrameAnalyticsDest> instanceReader() {
         return DataFrameAnalyticsDest::new;
     }
-
-    public void testValidate_GivenIndexWithFunkyChars() {
-        expectThrows(InvalidIndexNameException.class, () -> new DataFrameAnalyticsDest("<script>foo", null).validate());
-    }
-
-    public void testValidate_GivenIndexWithUppercaseChars() {
-        InvalidIndexNameException e = expectThrows(InvalidIndexNameException.class,
-            () -> new DataFrameAnalyticsDest("Foo", null).validate());
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(), equalTo("Invalid index name [Foo], dest.index must be lowercase"));
-    }
-
-    public void testValidate_GivenValidIndexName() {
-        new DataFrameAnalyticsDest("foo_bar_42", null).validate();
-    }
 }
diff --git a/x-pack/plugin/ml/qa/ml-with-security/build.gradle b/x-pack/plugin/ml/qa/ml-with-security/build.gradle
index fe53f1ca39ff5..8c9ba6df7f009 100644
--- a/x-pack/plugin/ml/qa/ml-with-security/build.gradle
+++ b/x-pack/plugin/ml/qa/ml-with-security/build.gradle
@@ -50,6 +50,7 @@ integTest.runner {
     'ml/data_frame_analytics_crud/Test put config with dest index same as source index',
     'ml/data_frame_analytics_crud/Test put config with dest index matching multiple indices',
     'ml/data_frame_analytics_crud/Test put config with dest index included in source via alias',
+    'ml/data_frame_analytics_crud/Test put config with remote source index',
     'ml/data_frame_analytics_crud/Test put config with unknown top level field',
     'ml/data_frame_analytics_crud/Test put config with unknown field in outlier detection analysis',
     'ml/data_frame_analytics_crud/Test put config given analyzed_fields include field excluded by source',
@@ -58,6 +59,7 @@ integTest.runner {
     'ml/data_frame_analytics_crud/Test put config given source with empty string in index array',
     'ml/data_frame_analytics_crud/Test put config given source without index',
     'ml/data_frame_analytics_crud/Test put config given missing dest',
+    'ml/data_frame_analytics_crud/Test put config given dest index contains uppercase chars',
     'ml/data_frame_analytics_crud/Test put config given dest with empty index',
     'ml/data_frame_analytics_crud/Test put config given dest without index',
     'ml/data_frame_analytics_crud/Test put config given missing analysis',
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPutDataFrameAnalyticsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPutDataFrameAnalyticsAction.java
index 1cbed7ed76613..c35a5c9a9d4a9 100644
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPutDataFrameAnalyticsAction.java
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPutDataFrameAnalyticsAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.license.License;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.tasks.Task;
@@ -32,14 +33,13 @@
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.XPackField;
 import org.elasticsearch.xpack.core.XPackSettings;
+import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
 import org.elasticsearch.xpack.core.ml.MachineLearningField;
 import org.elasticsearch.xpack.core.ml.action.PutDataFrameAnalyticsAction;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
 import org.elasticsearch.xpack.core.ml.job.messages.Messages;
 import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
 import org.elasticsearch.xpack.core.ml.job.persistence.ElasticsearchMappings;
-import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
-import org.elasticsearch.xpack.core.ml.utils.MlStrings;
 import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.core.security.action.user.HasPrivilegesAction;
 import org.elasticsearch.xpack.core.security.action.user.HasPrivilegesRequest;
@@ -47,7 +47,7 @@
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.permission.ResourcePrivileges;
 import org.elasticsearch.xpack.core.security.support.Exceptions;
-import org.elasticsearch.xpack.ml.dataframe.SourceDestValidator;
+import org.elasticsearch.xpack.ml.dataframe.SourceDestValidations;
 import org.elasticsearch.xpack.ml.dataframe.persistence.DataFrameAnalyticsConfigProvider;
 import org.elasticsearch.xpack.ml.notifications.DataFrameAnalyticsAuditor;
 
@@ -66,6 +66,7 @@ public class TransportPutDataFrameAnalyticsAction
     private final SecurityContext securityContext;
     private final Client client;
     private final DataFrameAnalyticsAuditor auditor;
+    private final SourceDestValidator sourceDestValidator;
 
     private volatile ByteSizeValue maxModelMemoryLimit;
 
@@ -86,6 +87,14 @@ public TransportPutDataFrameAnalyticsAction(Settings settings, TransportService
         maxModelMemoryLimit = MachineLearningField.MAX_MODEL_MEMORY_LIMIT.get(settings);
         clusterService.getClusterSettings()
             .addSettingsUpdateConsumer(MachineLearningField.MAX_MODEL_MEMORY_LIMIT, this::setMaxModelMemoryLimit);
+
+        this.sourceDestValidator = new SourceDestValidator(
+            indexNameExpressionResolver,
+            transportService.getRemoteClusterService(),
+            null,
+            clusterService.getNodeName(),
+            License.OperationMode.PLATINUM.description()
+        );
     }
 
     private void setMaxModelMemoryLimit(ByteSizeValue maxModelMemoryLimit) {
@@ -110,9 +119,21 @@ protected ClusterBlockException checkBlock(PutDataFrameAnalyticsAction.Request r
     @Override
     protected void masterOperation(Task task, PutDataFrameAnalyticsAction.Request request, ClusterState state,
                                    ActionListener<PutDataFrameAnalyticsAction.Response> listener) {
-        validateConfig(request.getConfig());
+
+        final DataFrameAnalyticsConfig config = request.getConfig();
+
+        ActionListener<Boolean> sourceDestValidationListener = ActionListener.wrap(
+            aBoolean -> putValidatedConfig(config, listener),
+            listener::onFailure
+        );
+
+        sourceDestValidator.validate(clusterService.state(), config.getSource().getIndex(), config.getDest().getIndex(),
+            SourceDestValidations.ALL_VALIDATIONS, sourceDestValidationListener);
+    }
+
+    private void putValidatedConfig(DataFrameAnalyticsConfig config, ActionListener<PutDataFrameAnalyticsAction.Response> listener) {
         DataFrameAnalyticsConfig preparedForPutConfig =
-            new DataFrameAnalyticsConfig.Builder(request.getConfig(), maxModelMemoryLimit)
+            new DataFrameAnalyticsConfig.Builder(config, maxModelMemoryLimit)
                 .setCreateTime(Instant.now())
                 .setVersion(Version.CURRENT)
                 .build();
@@ -202,19 +223,6 @@ private void updateDocMappingAndPutConfig(DataFrameAnalyticsConfig config,
                 listener::onFailure));
     }
 
-    private void validateConfig(DataFrameAnalyticsConfig config) {
-        if (MlStrings.isValidId(config.getId()) == false) {
-            throw ExceptionsHelper.badRequestException(Messages.getMessage(Messages.INVALID_ID, DataFrameAnalyticsConfig.ID,
-                config.getId()));
-        }
-        if (!MlStrings.hasValidLengthForId(config.getId())) {
-            throw ExceptionsHelper.badRequestException("id [{}] is too long; must not contain more than {} characters", config.getId(),
-                MlStrings.ID_LENGTH_LIMIT);
-        }
-        config.getDest().validate();
-        new SourceDestValidator(clusterService.state(), indexNameExpressionResolver).check(config);
-    }
-
     @Override
     protected void doExecute(Task task, PutDataFrameAnalyticsAction.Request request,
                              ActionListener<PutDataFrameAnalyticsAction.Response> listener) {
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDataFrameAnalyticsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDataFrameAnalyticsAction.java
index 14130402a0baa..b203ea95630a1 100644
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDataFrameAnalyticsAction.java
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDataFrameAnalyticsAction.java
@@ -32,6 +32,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.license.License;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.persistent.AllocatedPersistentTask;
@@ -47,6 +48,7 @@
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.XPackField;
+import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
 import org.elasticsearch.xpack.core.ml.MlTasks;
 import org.elasticsearch.xpack.core.ml.action.ExplainDataFrameAnalyticsAction;
@@ -64,7 +66,7 @@
 import org.elasticsearch.xpack.ml.dataframe.DataFrameAnalyticsManager;
 import org.elasticsearch.xpack.ml.dataframe.DataFrameAnalyticsTask;
 import org.elasticsearch.xpack.ml.dataframe.MappingsMerger;
-import org.elasticsearch.xpack.ml.dataframe.SourceDestValidator;
+import org.elasticsearch.xpack.ml.dataframe.SourceDestValidations;
 import org.elasticsearch.xpack.ml.dataframe.extractor.DataFrameDataExtractorFactory;
 import org.elasticsearch.xpack.ml.dataframe.extractor.ExtractedFieldsDetectorFactory;
 import org.elasticsearch.xpack.ml.dataframe.persistence.DataFrameAnalyticsConfigProvider;
@@ -101,6 +103,7 @@ public class TransportStartDataFrameAnalyticsAction
     private final DataFrameAnalyticsConfigProvider configProvider;
     private final MlMemoryTracker memoryTracker;
     private final DataFrameAnalyticsAuditor auditor;
+    private final SourceDestValidator sourceDestValidator;
 
     @Inject
     public TransportStartDataFrameAnalyticsAction(TransportService transportService, Client client, ClusterService clusterService,
@@ -117,6 +120,14 @@ public TransportStartDataFrameAnalyticsAction(TransportService transportService,
         this.configProvider = configProvider;
         this.memoryTracker = memoryTracker;
         this.auditor = Objects.requireNonNull(auditor);
+
+        this.sourceDestValidator = new SourceDestValidator(
+            indexNameExpressionResolver,
+            transportService.getRemoteClusterService(),
+            null,
+            clusterService.getNodeName(),
+            License.OperationMode.PLATINUM.description()
+        );
     }
 
     @Override
@@ -228,13 +239,13 @@ private void estimateMemoryUsageAndUpdateMemoryTracker(StartContext startContext
 
     private void getStartContext(String id, ActionListener<StartContext> finalListener) {
 
-        // Step 6. Validate that there are analyzable data in the source index
+        // Step 7. Validate that there are analyzable data in the source index
         ActionListener<StartContext> validateMappingsMergeListener = ActionListener.wrap(
             startContext -> validateSourceIndexHasRows(startContext, finalListener),
             finalListener::onFailure
         );
 
-        // Step 5. Validate mappings can be merged
+        // Step 6. Validate mappings can be merged
         ActionListener<StartContext> toValidateMappingsListener = ActionListener.wrap(
             startContext -> MappingsMerger.mergeMappings(client, startContext.config.getHeaders(),
                 startContext.config.getSource(), ActionListener.wrap(
@@ -242,7 +253,7 @@ private void getStartContext(String id, ActionListener<StartContext> finalListen
             finalListener::onFailure
         );
 
-        // Step 4. Validate dest index is empty if task is starting for first time
+        // Step 5. Validate dest index is empty if task is starting for first time
         ActionListener<StartContext> toValidateDestEmptyListener = ActionListener.wrap(
             startContext -> {
                 switch (startContext.startingState) {
@@ -266,22 +277,30 @@ private void getStartContext(String id, ActionListener<StartContext> finalListen
             finalListener::onFailure
         );
 
-        // Step 3. Validate source and dest; check data extraction is possible
-        ActionListener<StartContext> startContextListener = ActionListener.wrap(
+        // Step 4. Check data extraction is possible
+        ActionListener<StartContext> toValidateExtractionPossibleListener = ActionListener.wrap(
             startContext -> {
-                // Validate the query parses
-                startContext.config.getSource().getParsedQuery();
-
-                // Validate source/dest are valid
-                new SourceDestValidator(clusterService.state(), indexNameExpressionResolver).check(startContext.config);
-
-                // Validate extraction is possible
                 new ExtractedFieldsDetectorFactory(client).createFromSource(startContext.config, ActionListener.wrap(
                     extractedFieldsDetector -> {
                         startContext.extractedFields = extractedFieldsDetector.detect().v1();
                         toValidateDestEmptyListener.onResponse(startContext);
                     },
-                    finalListener::onFailure));
+                    finalListener::onFailure)
+                );
+            },
+            finalListener::onFailure
+        );
+
+        // Step 3. Validate source and dest
+        ActionListener<StartContext> startContextListener = ActionListener.wrap(
+            startContext -> {
+                // Validate the query parses
+                startContext.config.getSource().getParsedQuery();
+
+                // Validate source/dest are valid
+                sourceDestValidator.validate(clusterService.state(), startContext.config.getSource().getIndex(),
+                    startContext.config.getDest().getIndex(), SourceDestValidations.ALL_VALIDATIONS, ActionListener.wrap(
+                        aBoolean -> toValidateExtractionPossibleListener.onResponse(startContext), finalListener::onFailure));
             },
             finalListener::onFailure
         );
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidator.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidator.java
deleted file mode 100644
index 01803dc4359d5..0000000000000
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidator.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License;
- * you may not use this file except in compliance with the Elastic License.
- */
-package org.elasticsearch.xpack.ml.dataframe;
-
-import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
-import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.regex.Regex;
-import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
-import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Objects;
-import java.util.Set;
-
-public class SourceDestValidator {
-
-    private final ClusterState clusterState;
-    private final IndexNameExpressionResolver indexNameExpressionResolver;
-
-    public SourceDestValidator(ClusterState clusterState, IndexNameExpressionResolver indexNameExpressionResolver) {
-        this.clusterState = Objects.requireNonNull(clusterState);
-        this.indexNameExpressionResolver = Objects.requireNonNull(indexNameExpressionResolver);
-    }
-
-    public void check(DataFrameAnalyticsConfig config) {
-        String[] sourceIndex = config.getSource().getIndex();
-        String destIndex = config.getDest().getIndex();
-
-        String[] sourceExpressions = Arrays.stream(sourceIndex)
-            .map(index -> Strings.tokenizeToStringArray(index, ","))
-            .flatMap(Arrays::stream)
-            .toArray(String[]::new);
-
-        for (String sourceExpression : sourceExpressions) {
-            if (Regex.simpleMatch(sourceExpression, destIndex)) {
-                throw ExceptionsHelper.badRequestException("Destination index [{}] must not be included in source index [{}]",
-                    destIndex, sourceExpression);
-            }
-        }
-
-        Set<String> concreteSourceIndexNames = new HashSet<>(Arrays.asList(indexNameExpressionResolver.concreteIndexNames(clusterState,
-            IndicesOptions.lenientExpandOpen(), sourceExpressions)));
-
-        if (concreteSourceIndexNames.isEmpty()) {
-            throw ExceptionsHelper.badRequestException("No index matches source index {}", Arrays.toString(sourceIndex));
-        }
-
-        final String[] concreteDestIndexNames = indexNameExpressionResolver.concreteIndexNames(clusterState,
-            IndicesOptions.lenientExpandOpen(), destIndex);
-
-        if (concreteDestIndexNames.length > 1) {
-            // In case it is an alias, it may match multiple indices
-            throw ExceptionsHelper.badRequestException("Destination index [{}] should match a single index; matches {}", destIndex,
-                Arrays.toString(concreteDestIndexNames));
-        }
-        if (concreteDestIndexNames.length == 1 && concreteSourceIndexNames.contains(concreteDestIndexNames[0])) {
-            // In case the dest index is an alias, we need to check the concrete index is not matched by source
-            throw ExceptionsHelper.badRequestException("Destination index [{}], which is an alias for [{}], " +
-                    "must not be included in source index {}", destIndex, concreteDestIndexNames[0], Arrays.toString(sourceIndex));
-        }
-    }
-}
diff --git a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidatorTests.java b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidatorTests.java
deleted file mode 100644
index ec8b97942ac0a..0000000000000
--- a/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidatorTests.java
+++ /dev/null
@@ -1,188 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License;
- * you may not use this file except in compliance with the Elastic License.
- */
-package org.elasticsearch.xpack.ml.dataframe;
-
-import org.elasticsearch.ElasticsearchStatusException;
-import org.elasticsearch.Version;
-import org.elasticsearch.cluster.ClusterName;
-import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.metadata.AliasMetaData;
-import org.elasticsearch.cluster.metadata.IndexMetaData;
-import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
-import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
-import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsDest;
-import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsSource;
-import org.elasticsearch.xpack.core.ml.dataframe.analyses.OutlierDetection;
-
-import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_CREATION_DATE;
-import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
-import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
-import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_VERSION_CREATED;
-import static org.hamcrest.Matchers.equalTo;
-
-public class SourceDestValidatorTests extends ESTestCase {
-
-    private static final String SOURCE_1 = "source-1";
-    private static final String SOURCE_2 = "source-2";
-    private static final String ALIASED_DEST = "aliased-dest";
-
-    private static final ClusterState CLUSTER_STATE;
-
-    static {
-        IndexMetaData source1 = IndexMetaData.builder(SOURCE_1).settings(Settings.builder()
-            .put(SETTING_VERSION_CREATED, Version.CURRENT)
-            .put(SETTING_NUMBER_OF_SHARDS, 1).put(SETTING_NUMBER_OF_REPLICAS, 0)
-            .put(SETTING_CREATION_DATE, System.currentTimeMillis()))
-            .putAlias(AliasMetaData.builder("source-1-alias").build())
-            .build();
-        IndexMetaData source2 = IndexMetaData.builder(SOURCE_2).settings(Settings.builder()
-            .put(SETTING_VERSION_CREATED, Version.CURRENT)
-            .put(SETTING_NUMBER_OF_SHARDS, 1).put(SETTING_NUMBER_OF_REPLICAS, 0)
-            .put(SETTING_CREATION_DATE, System.currentTimeMillis()))
-            .putAlias(AliasMetaData.builder("dest-alias").build())
-            .build();
-        IndexMetaData aliasedDest = IndexMetaData.builder(ALIASED_DEST).settings(Settings.builder()
-            .put(SETTING_VERSION_CREATED, Version.CURRENT)
-            .put(SETTING_NUMBER_OF_SHARDS, 1).put(SETTING_NUMBER_OF_REPLICAS, 0)
-            .put(SETTING_CREATION_DATE, System.currentTimeMillis()))
-            .putAlias(AliasMetaData.builder("dest-alias").build())
-            .build();
-        ClusterState.Builder state = ClusterState.builder(new ClusterName("test"));
-        state.metaData(MetaData.builder()
-            .put(IndexMetaData.builder(source1).build(), false)
-            .put(IndexMetaData.builder(source2).build(), false)
-            .put(IndexMetaData.builder(aliasedDest).build(), false));
-        CLUSTER_STATE = state.build();
-    }
-
-    public void testCheck_GivenSimpleSourceIndexAndValidDestIndex() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("source-1"))
-            .setDest(new DataFrameAnalyticsDest("dest", null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-        validator.check(config);
-    }
-
-    public void testCheck_GivenMissingConcreteSourceIndex() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("missing"))
-            .setDest(new DataFrameAnalyticsDest("dest", null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(), equalTo("No index matches source index [missing]"));
-    }
-
-    public void testCheck_GivenMissingWildcardSourceIndex() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("missing*"))
-            .setDest(new DataFrameAnalyticsDest("dest", null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(), equalTo("No index matches source index [missing*]"));
-    }
-
-    public void testCheck_GivenDestIndexSameAsSourceIndex() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("source-1"))
-            .setDest(new DataFrameAnalyticsDest("source-1", null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(), equalTo("Destination index [source-1] must not be included in source index [source-1]"));
-    }
-
-    public void testCheck_GivenDestIndexMatchesSourceIndex() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("source-*"))
-            .setDest(new DataFrameAnalyticsDest(SOURCE_2, null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(), equalTo("Destination index [source-2] must not be included in source index [source-*]"));
-    }
-
-    public void testCheck_GivenDestIndexMatchesOneOfSourceIndices() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("source-1,source-*"))
-            .setDest(new DataFrameAnalyticsDest(SOURCE_2, null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(), equalTo("Destination index [source-2] must not be included in source index [source-*]"));
-    }
-
-    public void testCheck_GivenDestIndexIsAliasThatMatchesMultipleIndices() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource(SOURCE_1))
-            .setDest(new DataFrameAnalyticsDest("dest-alias", null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(),
-            equalTo("Destination index [dest-alias] should match a single index; matches [source-2, aliased-dest]"));
-    }
-
-    public void testCheck_GivenDestIndexIsAliasThatIsIncludedInSource() {
-        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder()
-            .setId("test")
-            .setSource(createSource("source-1"))
-            .setDest(new DataFrameAnalyticsDest("source-1-alias", null))
-            .setAnalysis(new OutlierDetection.Builder().build())
-            .build();
-
-        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
-
-        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class, () -> validator.check(config));
-        assertThat(e.status(), equalTo(RestStatus.BAD_REQUEST));
-        assertThat(e.getMessage(),
-            equalTo("Destination index [source-1-alias], which is an alias for [source-1], " +
-                "must not be included in source index [source-1]"));
-    }
-
-    private static DataFrameAnalyticsSource createSource(String... index) {
-        return new DataFrameAnalyticsSource(index, null, null);
-    }
-}
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml
index 4335a50382a94..6dc8c800bd414 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml
@@ -268,6 +268,24 @@ setup:
             "analysis": {"outlier_detection":{}}
           }
 
+---
+"Test put config given dest index contains uppercase chars":
+
+  - do:
+      catch: /.*reason=Validation Failed.* Destination index \[Foo\] must be lowercase;.*/
+      ml.put_data_frame_analytics:
+        id: "config-given-dest-index-uppercase"
+        body: >
+          {
+            "source": {
+              "index": "index-source"
+            },
+            "dest": {
+              "index": "Foo"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
 ---
 "Test put config with pattern dest index name":
 
@@ -290,7 +308,7 @@ setup:
 "Test put config with missing concrete source index":
 
   - do:
-      catch: /No index matches source index \[missing]/
+      catch: /.*reason=Validation Failed.* no such index \[missing\]/
       ml.put_data_frame_analytics:
         id: "config-with-missing-concrete-source-index"
         body: >
@@ -308,7 +326,6 @@ setup:
 "Test put config with missing wildcard source index":
 
   - do:
-      catch: /No index matches source index \[missing\*\]/
       ml.put_data_frame_analytics:
         id: "config-with-missing-wildcard-source-index"
         body: >
@@ -321,12 +338,13 @@ setup:
             },
             "analysis": {"outlier_detection":{}}
           }
+  - match: { id: "config-with-missing-wildcard-source-index" }
 
 ---
 "Test put config with dest index same as source index":
 
   - do:
-      catch: /Destination index \[index-source\] must not be included in source index \[index-source\]/
+      catch: /.*reason=Validation Failed.* Destination index \[index-source\] is included in source expression \[index-source\]/
       ml.put_data_frame_analytics:
         id: "config-with-same-source-dest-index"
         body: >
@@ -362,7 +380,7 @@ setup:
         name: multiple-dest-index
 
   - do:
-      catch: /Destination index \[multiple-dest-index\] should match a single index; matches \[multiple-dest-index-[12], multiple-dest-index-[12]\]/
+      catch: /.*reason=Validation Failed.* no write index is defined for alias \[multiple-dest-index\].*/
       ml.put_data_frame_analytics:
         id: "config-with-dest-index-matching-multiple-indices"
         body: >
@@ -389,7 +407,7 @@ setup:
         name: dest-alias
 
   - do:
-      catch: /Destination index \[dest-alias\], which is an alias for \[another-source-index\], must not be included in source index \[another-source-index\]/
+      catch: /.*reason=Validation Failed.* Destination index \[another-source-index\] is included in source expression \[another-source-index\]/
       ml.put_data_frame_analytics:
         id: "config-with-dest-index-included-in-source-via-alias"
         body: >
@@ -403,6 +421,24 @@ setup:
             "analysis": {"outlier_detection":{}}
           }
 
+---
+"Test put config with remote source index":
+
+  - do:
+      catch: /.*reason=Validation Failed.* remote source indices are not supported/
+      ml.put_data_frame_analytics:
+        id: "config-with-missing-concrete-source-index"
+        body: >
+          {
+            "source": {
+              "index": "remote_cluster:foo"
+            },
+            "dest": {
+              "index": "index-dest"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
 ---
 "Test put config with unknown top level field":
 
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/start_data_frame_analytics.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/start_data_frame_analytics.yml
index 0172a83f58ef2..8cae555a201f5 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/start_data_frame_analytics.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/start_data_frame_analytics.yml
@@ -31,7 +31,7 @@
         index: missing
 
   - do:
-      catch: /No index matches source index \[missing]/
+      catch: /.*reason=Validation Failed.* no such index \[missing\]/
       ml.start_data_frame_analytics:
         id: "missing_index"
 
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPreviewTransformAction.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPreviewTransformAction.java
index 5ae62d0890ea9..4896194279e46 100644
--- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPreviewTransformAction.java
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPreviewTransformAction.java
@@ -51,6 +51,7 @@
 import org.elasticsearch.xpack.core.transform.transforms.TransformIndexerStats;
 import org.elasticsearch.xpack.transform.transforms.pivot.AggregationResultUtils;
 import org.elasticsearch.xpack.transform.transforms.pivot.Pivot;
+import org.elasticsearch.xpack.transform.utils.SourceDestValidations;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -138,7 +139,7 @@ protected void doExecute(Task task, PreviewTransformAction.Request request, Acti
             clusterState,
             config.getSource().getIndex(),
             config.getDestination().getIndex(),
-            SourceDestValidator.PREVIEW_VALIDATIONS,
+            SourceDestValidations.PREVIEW_VALIDATIONS,
             ActionListener.wrap(r -> {
 
                 Pivot pivot = new Pivot(config.getPivotConfig());
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPutTransformAction.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPutTransformAction.java
index 67d9454c98232..23bd53e4efd30 100644
--- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPutTransformAction.java
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPutTransformAction.java
@@ -57,6 +57,7 @@
 import org.elasticsearch.xpack.transform.notifications.TransformAuditor;
 import org.elasticsearch.xpack.transform.persistence.TransformConfigManager;
 import org.elasticsearch.xpack.transform.transforms.pivot.Pivot;
+import org.elasticsearch.xpack.transform.utils.SourceDestValidations;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -232,7 +233,7 @@ protected void masterOperation(Task task, Request request, ClusterState clusterS
             clusterState,
             config.getSource().getIndex(),
             config.getDestination().getIndex(),
-            request.isDeferValidation() ? SourceDestValidator.NON_DEFERABLE_VALIDATIONS : SourceDestValidator.ALL_VALIDATIONS,
+            request.isDeferValidation() ? SourceDestValidations.NON_DEFERABLE_VALIDATIONS : SourceDestValidations.ALL_VALIDATIONS,
             ActionListener.wrap(
                 validationResponse -> {
                     // Early check to verify that the user can create the destination index and can read from the source
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportStartTransformAction.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportStartTransformAction.java
index 52337427c388f..a4204d2191417 100644
--- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportStartTransformAction.java
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportStartTransformAction.java
@@ -52,6 +52,7 @@
 import org.elasticsearch.xpack.transform.persistence.TransformConfigManager;
 import org.elasticsearch.xpack.transform.persistence.TransformIndex;
 import org.elasticsearch.xpack.transform.transforms.pivot.Pivot;
+import org.elasticsearch.xpack.transform.utils.SourceDestValidations;
 
 import java.io.IOException;
 import java.time.Clock;
@@ -279,7 +280,7 @@ protected void masterOperation(
                 clusterService.state(),
                 config.getSource().getIndex(),
                 config.getDestination().getIndex(),
-                SourceDestValidator.ALL_VALIDATIONS,
+                SourceDestValidations.ALL_VALIDATIONS,
                 validationListener
             );
         }, listener::onFailure);
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportUpdateTransformAction.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportUpdateTransformAction.java
index e98360fb453ca..9764a2ddbac5f 100644
--- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportUpdateTransformAction.java
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportUpdateTransformAction.java
@@ -56,6 +56,7 @@
 import org.elasticsearch.xpack.transform.persistence.TransformConfigManager;
 import org.elasticsearch.xpack.transform.persistence.TransformIndex;
 import org.elasticsearch.xpack.transform.transforms.pivot.Pivot;
+import org.elasticsearch.xpack.transform.utils.SourceDestValidations;
 
 import java.io.IOException;
 import java.time.Clock;
@@ -177,7 +178,7 @@ protected void masterOperation(Task task, Request request, ClusterState clusterS
                 clusterState,
                 updatedConfig.getSource().getIndex(),
                 updatedConfig.getDestination().getIndex(),
-                request.isDeferValidation() ? SourceDestValidator.NON_DEFERABLE_VALIDATIONS : SourceDestValidator.ALL_VALIDATIONS,
+                request.isDeferValidation() ? SourceDestValidations.NON_DEFERABLE_VALIDATIONS : SourceDestValidations.ALL_VALIDATIONS,
                 ActionListener.wrap(
                     validationResponse -> {
                         checkPriviledgesAndUpdateTransform(request, clusterState, updatedConfig, configAndVersion.v2(), listener);
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/utils/SourceDestValidations.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/utils/SourceDestValidations.java
new file mode 100644
index 0000000000000..0941a28fc85a2
--- /dev/null
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/utils/SourceDestValidations.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.transform.utils;
+
+import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_IN_SOURCE_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_SINGLE_INDEX_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.REMOTE_SOURCE_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.SOURCE_MISSING_VALIDATION;
+
+/**
+ * Packs together useful sets of validations in the context transforms
+ */
+public final class SourceDestValidations {
+
+    private SourceDestValidations() {}
+
+    public static final List<SourceDestValidator.SourceDestValidation> PREVIEW_VALIDATIONS = Arrays.asList(
+        SOURCE_MISSING_VALIDATION, REMOTE_SOURCE_VALIDATION);
+
+    public static final List<SourceDestValidator.SourceDestValidation> ALL_VALIDATIONS = Arrays.asList(
+        SOURCE_MISSING_VALIDATION,
+        REMOTE_SOURCE_VALIDATION,
+        DESTINATION_IN_SOURCE_VALIDATION,
+        DESTINATION_SINGLE_INDEX_VALIDATION
+    );
+
+    public static final List<SourceDestValidator.SourceDestValidation> NON_DEFERABLE_VALIDATIONS = Collections.singletonList(
+        DESTINATION_SINGLE_INDEX_VALIDATION
+    );
+}

From d21d9f02688fb11e1d9ec6e3e31876abe075131a Mon Sep 17 00:00:00 2001
From: Dimitris Athanasiou <dimitris@elastic.co>
Date: Fri, 10 Jan 2020 10:49:50 +0200
Subject: [PATCH 2/3] Add missing file and extract message into constant

---
 .../validation/SourceDestValidator.java       |  3 +-
 .../ml/dataframe/SourceDestValidations.java   | 28 +++++++++++++++++++
 2 files changed, 30 insertions(+), 1 deletion(-)
 create mode 100644 x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidations.java

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java
index f8171884131b1..1a292bbebeadd 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java
@@ -58,6 +58,7 @@ public final class SourceDestValidator {
         + "alias [{0}], at least a [{1}] license is required, found license [{2}]";
     public static final String REMOTE_CLUSTER_LICENSE_INACTIVE = "License check failed for remote cluster "
         + "alias [{0}], license is not active";
+    public static final String REMOTE_SOURCE_INDICES_NOT_SUPPORTED = "remote source indices are not supported";
 
     private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final RemoteClusterService remoteClusterService;
@@ -448,7 +449,7 @@ static class RemoteSourceNotSupportedValidation implements SourceDestValidation
         @Override
         public void validate(Context context, ActionListener<Context> listener) {
             if (context.resolveRemoteSource().isEmpty() == false) {
-                context.addValidationError("remote source indices are not supported");
+                context.addValidationError(REMOTE_SOURCE_INDICES_NOT_SUPPORTED);
             }
             listener.onResponse(context);
         }
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidations.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidations.java
new file mode 100644
index 0000000000000..2372cfeaced88
--- /dev/null
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidations.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ml.dataframe;
+
+import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_IN_SOURCE_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_SINGLE_INDEX_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.REMOTE_SOURCE_NOT_SUPPORTED_VALIDATION;
+import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.SOURCE_MISSING_VALIDATION;
+
+public final class SourceDestValidations {
+
+    private SourceDestValidations() {}
+
+    public static final List<SourceDestValidator.SourceDestValidation> ALL_VALIDATIONS = Arrays.asList(
+        SOURCE_MISSING_VALIDATION,
+        DESTINATION_SINGLE_INDEX_VALIDATION,
+        DESTINATION_IN_SOURCE_VALIDATION,
+        REMOTE_SOURCE_NOT_SUPPORTED_VALIDATION
+    );
+}

From 2f2fbccef092d659270e6e3c709016e7c816040a Mon Sep 17 00:00:00 2001
From: Dimitris Athanasiou <dimitris@elastic.co>
Date: Fri, 10 Jan 2020 12:24:05 +0200
Subject: [PATCH 3/3] Reactivate commented out part of test

---
 .../validation/SourceDestValidatorTests.java  | 22 +++++++++----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java
index 109148f5696d7..4eab68271b7e1 100644
--- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java
+++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java
@@ -274,17 +274,17 @@ public void testCheck_GivenMixedMissingAndExistingConcreteSourceIndex() throws I
             }
         );
 
-//        assertValidation(
-//            listener -> simpleNonRemoteValidator.validate(
-//                CLUSTER_STATE,
-//                new String[] { SOURCE_1, "missing" },
-//                "dest",
-//                Collections.emptyList(),
-//                listener
-//            ),
-//            true,
-//            null
-//        );
+        assertValidation(
+            listener -> simpleNonRemoteValidator.validate(
+                CLUSTER_STATE,
+                new String[] { SOURCE_1, "missing" },
+                "dest",
+                Collections.emptyList(),
+                listener
+            ),
+            true,
+            null
+        );
     }
 
     public void testCheck_GivenMixedMissingWildcardExistingConcreteSourceIndex() throws InterruptedException {