diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java
index d6550964f9732..12a2fc6c93032 100644
--- a/client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java
+++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java
@@ -1216,6 +1216,8 @@ public void testPutDataFrameAnalyticsConfig() throws Exception {
             .setAnalysis(OutlierDetection.createDefault())
             .build();
 
+        createIndex("put-test-source-index", defaultMappingForTest());
+
         PutDataFrameAnalyticsResponse putDataFrameAnalyticsResponse = execute(
             new PutDataFrameAnalyticsRequest(config),
             machineLearningClient::putDataFrameAnalytics, machineLearningClient::putDataFrameAnalyticsAsync);
@@ -1243,6 +1245,8 @@ public void testGetDataFrameAnalyticsConfig_SingleConfig() throws Exception {
             .setAnalysis(OutlierDetection.createDefault())
             .build();
 
+        createIndex("get-test-source-index", defaultMappingForTest());
+
         PutDataFrameAnalyticsResponse putDataFrameAnalyticsResponse = execute(
             new PutDataFrameAnalyticsRequest(config),
             machineLearningClient::putDataFrameAnalytics, machineLearningClient::putDataFrameAnalyticsAsync);
@@ -1256,6 +1260,8 @@ public void testGetDataFrameAnalyticsConfig_SingleConfig() throws Exception {
     }
 
     public void testGetDataFrameAnalyticsConfig_MultipleConfigs() throws Exception {
+        createIndex("get-test-source-index", defaultMappingForTest());
+
         MachineLearningClient machineLearningClient = highLevelClient().machineLearning();
         String configIdPrefix = "get-test-config-";
         int numberOfConfigs = 10;
@@ -1461,6 +1467,8 @@ public void testDeleteDataFrameAnalyticsConfig() throws Exception {
             .setAnalysis(OutlierDetection.createDefault())
             .build();
 
+        createIndex("delete-test-source-index", defaultMappingForTest());
+
         GetDataFrameAnalyticsResponse getDataFrameAnalyticsResponse = execute(
             new GetDataFrameAnalyticsRequest(configId + "*"),
             machineLearningClient::getDataFrameAnalytics, machineLearningClient::getDataFrameAnalyticsAsync);
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java
index 0203a3c855d14..f33926582e3ea 100644
--- a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java
+++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java
@@ -2802,6 +2802,8 @@ public void onFailure(Exception e) {
     }
 
     public void testGetDataFrameAnalytics() throws Exception {
+        createIndex(DF_ANALYTICS_CONFIG.getSource().getIndex());
+
         RestHighLevelClient client = highLevelClient();
         client.machineLearning().putDataFrameAnalytics(new PutDataFrameAnalyticsRequest(DF_ANALYTICS_CONFIG), RequestOptions.DEFAULT);
         {
@@ -2849,6 +2851,8 @@ public void onFailure(Exception e) {
     }
 
     public void testGetDataFrameAnalyticsStats() throws Exception {
+        createIndex(DF_ANALYTICS_CONFIG.getSource().getIndex());
+
         RestHighLevelClient client = highLevelClient();
         client.machineLearning().putDataFrameAnalytics(new PutDataFrameAnalyticsRequest(DF_ANALYTICS_CONFIG), RequestOptions.DEFAULT);
         {
@@ -2897,6 +2901,8 @@ public void onFailure(Exception e) {
     }
 
     public void testPutDataFrameAnalytics() throws Exception {
+        createIndex(DF_ANALYTICS_CONFIG.getSource().getIndex());
+
         RestHighLevelClient client = highLevelClient();
         {
             // tag::put-data-frame-analytics-query-config
@@ -2988,6 +2994,8 @@ public void onFailure(Exception e) {
     }
 
     public void testDeleteDataFrameAnalytics() throws Exception {
+        createIndex(DF_ANALYTICS_CONFIG.getSource().getIndex());
+
         RestHighLevelClient client = highLevelClient();
         client.machineLearning().putDataFrameAnalytics(new PutDataFrameAnalyticsRequest(DF_ANALYTICS_CONFIG), RequestOptions.DEFAULT);
         {
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 98f1bdeb19189..3bc435336f062 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,11 +13,15 @@
 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");
@@ -90,4 +94,13 @@ 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/ml/dataframe/DataFrameAnalyticsDestTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsDestTests.java
index 7332687723805..bf8ce4c8a99b0 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,10 +7,14 @@
 
 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
@@ -33,4 +37,19 @@ 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 6435ccc4bfb01..a6e1f14aee22a 100644
--- a/x-pack/plugin/ml/qa/ml-with-security/build.gradle
+++ b/x-pack/plugin/ml/qa/ml-with-security/build.gradle
@@ -40,6 +40,13 @@ integTestRunner {
     'ml/data_frame_analytics_crud/Test put config with security headers in the body',
     'ml/data_frame_analytics_crud/Test put config with inconsistent body/param ids',
     'ml/data_frame_analytics_crud/Test put config with invalid id',
+    'ml/data_frame_analytics_crud/Test put config with invalid dest index name',
+    'ml/data_frame_analytics_crud/Test put config with pattern dest index name',
+    'ml/data_frame_analytics_crud/Test put config with missing concrete source index',
+    'ml/data_frame_analytics_crud/Test put config with missing wildcard source index',
+    '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 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 missing source',
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 372fb74edd198..0f709b4e16680 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
@@ -9,6 +9,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
@@ -36,9 +37,11 @@
 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.persistence.DataFrameAnalyticsConfigProvider;
 
 import java.io.IOException;
+import java.util.Objects;
 import java.util.function.Supplier;
 
 public class TransportPutDataFrameAnalyticsAction
@@ -49,13 +52,16 @@ public class TransportPutDataFrameAnalyticsAction
     private final ThreadPool threadPool;
     private final SecurityContext securityContext;
     private final Client client;
+    private final ClusterService clusterService;
+    private final IndexNameExpressionResolver indexNameExpressionResolver;
 
     private volatile ByteSizeValue maxModelMemoryLimit;
 
     @Inject
     public TransportPutDataFrameAnalyticsAction(Settings settings, TransportService transportService, ActionFilters actionFilters,
                                                 XPackLicenseState licenseState, Client client, ThreadPool threadPool,
-                                                ClusterService clusterService, DataFrameAnalyticsConfigProvider configProvider) {
+                                                ClusterService clusterService, IndexNameExpressionResolver indexNameExpressionResolver,
+                                                DataFrameAnalyticsConfigProvider configProvider) {
         super(PutDataFrameAnalyticsAction.NAME, transportService, actionFilters,
             (Supplier<PutDataFrameAnalyticsAction.Request>) PutDataFrameAnalyticsAction.Request::new);
         this.licenseState = licenseState;
@@ -64,6 +70,8 @@ public TransportPutDataFrameAnalyticsAction(Settings settings, TransportService
         this.securityContext = XPackSettings.SECURITY_ENABLED.get(settings) ?
             new SecurityContext(settings, threadPool.getThreadContext()) : null;
         this.client = client;
+        this.clusterService = clusterService;
+        this.indexNameExpressionResolver = Objects.requireNonNull(indexNameExpressionResolver);
 
         maxModelMemoryLimit = MachineLearningField.MAX_MODEL_MEMORY_LIMIT.get(settings);
         clusterService.getClusterSettings()
@@ -146,5 +154,7 @@ private void validateConfig(DataFrameAnalyticsConfig config) {
             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);
     }
 }
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 3d195f09b26d5..6650e8000959b 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
@@ -50,6 +50,7 @@
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.ml.MachineLearning;
 import org.elasticsearch.xpack.ml.dataframe.DataFrameAnalyticsManager;
+import org.elasticsearch.xpack.ml.dataframe.SourceDestValidator;
 import org.elasticsearch.xpack.ml.dataframe.extractor.DataFrameDataExtractorFactory;
 import org.elasticsearch.xpack.ml.dataframe.persistence.DataFrameAnalyticsConfigProvider;
 import org.elasticsearch.xpack.ml.job.JobNodeSelector;
@@ -157,7 +158,10 @@ public void onFailure(Exception e) {
 
         // Validate config
         ActionListener<DataFrameAnalyticsConfig> configListener = ActionListener.wrap(
-            config -> DataFrameDataExtractorFactory.validateConfigAndSourceIndex(client, config, validateListener),
+            config -> {
+                new SourceDestValidator(clusterService.state(), indexNameExpressionResolver).check(config);
+                DataFrameDataExtractorFactory.validateConfigAndSourceIndex(client, config, validateListener);
+            },
             listener::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
new file mode 100644
index 0000000000000..f607387e31774
--- /dev/null
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidator.java
@@ -0,0 +1,65 @@
+/*
+ * 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 = Strings.tokenizeToStringArray(sourceIndex, ",");
+
+        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 [{}]", 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], 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
new file mode 100644
index 0000000000000..fb91673b7a509
--- /dev/null
+++ b/x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/dataframe/SourceDestValidatorTests.java
@@ -0,0 +1,176 @@
+/*
+ * 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("test")
+            .setSource(new DataFrameAnalyticsSource("source-1", null))
+            .setDest(new DataFrameAnalyticsDest("dest", null))
+            .setAnalysis(new OutlierDetection())
+            .build();
+
+        SourceDestValidator validator = new SourceDestValidator(CLUSTER_STATE, new IndexNameExpressionResolver());
+        validator.check(config);
+    }
+
+    public void testCheck_GivenMissingConcreteSourceIndex() {
+        DataFrameAnalyticsConfig config = new DataFrameAnalyticsConfig.Builder("test")
+            .setSource(new DataFrameAnalyticsSource("missing", null))
+            .setDest(new DataFrameAnalyticsDest("dest", null))
+            .setAnalysis(new OutlierDetection())
+            .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("test")
+            .setSource(new DataFrameAnalyticsSource("missing*", null))
+            .setDest(new DataFrameAnalyticsDest("dest", null))
+            .setAnalysis(new OutlierDetection())
+            .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("test")
+            .setSource(new DataFrameAnalyticsSource("source-1", null))
+            .setDest(new DataFrameAnalyticsDest("source-1", null))
+            .setAnalysis(new OutlierDetection())
+            .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("test")
+            .setSource(new DataFrameAnalyticsSource("source-*", null))
+            .setDest(new DataFrameAnalyticsDest(SOURCE_2, null))
+            .setAnalysis(new OutlierDetection())
+            .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("test")
+            .setSource(new DataFrameAnalyticsSource("source-1,source-*", null))
+            .setDest(new DataFrameAnalyticsDest(SOURCE_2, null))
+            .setAnalysis(new OutlierDetection())
+            .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("test")
+            .setSource(new DataFrameAnalyticsSource(SOURCE_1, null))
+            .setDest(new DataFrameAnalyticsDest("dest-alias", null))
+            .setAnalysis(new OutlierDetection())
+            .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("test")
+            .setSource(new DataFrameAnalyticsSource("source-1", null))
+            .setDest(new DataFrameAnalyticsDest("source-1-alias", null))
+            .setAnalysis(new OutlierDetection())
+            .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]"));
+    }
+}
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 ceb8cd04a6f08..f815471ec711f 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
@@ -1,3 +1,9 @@
+---
+setup:
+  - do:
+      indices.create:
+        index: index-source
+
 ---
 "Test get-all and get-all-stats given no analytics exist":
 
@@ -158,6 +164,159 @@
             "analysis": {"outlier_detection":{}}
           }
 
+---
+"Test put config with invalid dest index name":
+
+  - do:
+      catch: /Invalid index name \[<script>Foo\]*/
+      ml.put_data_frame_analytics:
+        id: "config-with-invalid-dest-index"
+        body: >
+          {
+            "source": {
+              "index": "index-source"
+            },
+            "dest": {
+              "index": "<script>Foo"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
+---
+"Test put config with pattern dest index name":
+
+  - do:
+      catch: /Invalid index name \[foo\*\]/
+      ml.put_data_frame_analytics:
+        id: "config-with-invalid-dest-index"
+        body: >
+          {
+            "source": {
+              "index": "index-source"
+            },
+            "dest": {
+              "index": "foo*"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
+---
+"Test put config with missing concrete source index":
+
+  - do:
+      catch: /No index matches source index \[missing]/
+      ml.put_data_frame_analytics:
+        id: "config-with-missing-concrete-source-index"
+        body: >
+          {
+            "source": {
+              "index": "missing"
+            },
+            "dest": {
+              "index": "index-dest"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
+---
+"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: >
+          {
+            "source": {
+              "index": "missing*"
+            },
+            "dest": {
+              "index": "index-dest"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
+---
+"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\]/
+      ml.put_data_frame_analytics:
+        id: "config-with-same-source-dest-index"
+        body: >
+          {
+            "source": {
+              "index": "index-source"
+            },
+            "dest": {
+              "index": "index-source"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
+---
+"Test put config with dest index matching multiple indices":
+
+  - do:
+      indices.create:
+        index: multiple-dest-index-1
+
+  - do:
+      indices.create:
+        index: multiple-dest-index-2
+
+  - do:
+      indices.put_alias:
+        index: multiple-dest-index-1
+        name: multiple-dest-index
+
+  - do:
+      indices.put_alias:
+        index: multiple-dest-index-2
+        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]\]/
+      ml.put_data_frame_analytics:
+        id: "config-with-dest-index-matching-multiple-indices"
+        body: >
+          {
+            "source": {
+              "index": "index-source"
+            },
+            "dest": {
+              "index": "multiple-dest-index"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
+---
+"Test put config with dest index included in source via alias":
+
+  - do:
+      indices.create:
+        index: another-source-index
+
+  - do:
+      indices.put_alias:
+        index: another-source-index
+        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\]/
+      ml.put_data_frame_analytics:
+        id: "config-with-dest-index-included-in-source-via-alias"
+        body: >
+          {
+            "source": {
+              "index": "another-source-index"
+            },
+            "dest": {
+              "index": "dest-alias"
+            },
+            "analysis": {"outlier_detection":{}}
+          }
+
 ---
 "Test put config with unknown top level field":
 
@@ -339,7 +498,7 @@
         body: >
           {
             "source": {
-              "index": "index-foo-1_source"
+              "index": "index-source"
             },
             "dest": {
               "index": "index-foo-1_dest"
@@ -353,7 +512,7 @@
         body: >
           {
             "source": {
-              "index": "index-foo-2_source"
+              "index": "index-source"
             },
             "dest": {
               "index": "index-foo-2_dest"
@@ -368,7 +527,7 @@
         body: >
           {
             "source": {
-              "index": "index-bar_source"
+              "index": "index-source"
             },
             "dest": {
               "index": "index-bar_dest"
@@ -500,7 +659,7 @@
         body: >
           {
             "source": {
-              "index": "index-foo-1_source"
+              "index": "index-source"
             },
             "dest": {
               "index": "index-foo-1_dest"
@@ -514,7 +673,7 @@
         body: >
           {
             "source": {
-              "index": "index-foo-2_source"
+              "index": "index-source"
             },
             "dest": {
               "index": "index-foo-2_dest"
@@ -529,7 +688,7 @@
         body: >
           {
             "source": {
-              "index": "index-bar_source"
+              "index": "index-source"
             },
             "dest": {
               "index": "index-bar_dest"
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 3cc2e91483e18..6a8c6d0e6ed00 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
@@ -8,6 +8,10 @@
 ---
 "Test start given missing source index":
 
+  - do:
+      indices.create:
+        index: missing
+
   - do:
       ml.put_data_frame_analytics:
         id: "missing_index"
@@ -23,7 +27,11 @@
           }
 
   - do:
-      catch: /cannot retrieve data because index \[missing\] does not exist/
+      indices.delete:
+        index: missing
+
+  - do:
+      catch: /No index matches source index \[missing]/
       ml.start_data_frame_analytics:
         id: "missing_index"
 
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/stop_data_frame_analytics.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/stop_data_frame_analytics.yml
index 673910bce9fdc..382fd59d35ad9 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/stop_data_frame_analytics.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/stop_data_frame_analytics.yml
@@ -1,4 +1,8 @@
 setup:
+  - do:
+      indices.create:
+        index: index-source
+
   - do:
       ml.put_data_frame_analytics:
         id: "stop_data_frame_analytics_test_job"