[ML] Disable CPS for Dataframes

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidator.java

@@ -69,7 +69,8 @@ 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";
+    public static final String REMOTE_SOURCE_AND_CROSS_PROJECT_INDICES_ARE_NOT_SUPPORTED =
+        "remote source and cross-project indices are not supported";
     public static final String REMOTE_CLUSTERS_TRANSPORT_TOO_OLD =
         "remote clusters are expected to run at least version [{0}] (reason: [{1}])," + " but the following clusters were too old: [{2}]";
     public static final String PIPELINE_MISSING = "Pipeline with id [{0}] could not be found";
@@ -555,7 +556,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_NOT_SUPPORTED);
+                context.addValidationError(REMOTE_SOURCE_AND_CROSS_PROJECT_INDICES_ARE_NOT_SUPPORTED);
             }
             listener.onResponse(context);
         }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/common/validation/SourceDestValidatorTests.java

@@ -67,7 +67,9 @@
 import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_IN_SOURCE_VALIDATION;
 import static org.elasticsearch.xpack.core.common.validation.SourceDestValidator.DESTINATION_PIPELINE_MISSING_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;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -667,6 +669,51 @@ public void testRemoteSourceBasic() throws InterruptedException {
         );
     }
 
+    public void testRemoteSourceNotSupportedValidationWithLocalIndex() throws InterruptedException {
+        Context context = spy(
+            new SourceDestValidator.Context(
+                CLUSTER_STATE,
+                indexNameExpressionResolver,
+                remoteClusterService,
+                remoteClusterLicenseCheckerBasic,
+                ingestService,
+                new String[] { SOURCE_1 },
+                "dest",
+                null,
+                "node_id",
+                "license"
+            )
+        );
+
+        assertValidationWithContext(listener -> REMOTE_SOURCE_NOT_SUPPORTED_VALIDATION.validate(context, listener), c -> {
+            assertNull(c.getValidationException());
+        }, null);
+    }
+
+    public void testRemoteSourceNotSupportedValidationWithRemoteIndex() throws InterruptedException {
+        Context context = spy(
+            new SourceDestValidator.Context(
+                CLUSTER_STATE,
+                indexNameExpressionResolver,
+                remoteClusterService,
+                remoteClusterLicenseCheckerBasic,
+                ingestService,
+                new String[] { REMOTE_BASIC + ":" + SOURCE_1 },
+                "dest",
+                null,
+                "node_id",
+                "license"
+            )
+        );
+
+        assertValidationWithContext(listener -> REMOTE_SOURCE_NOT_SUPPORTED_VALIDATION.validate(context, listener), c -> {
+            assertThat(
+                c.getValidationException().getMessage(),
+                containsString("remote source and cross-project indices are not supported")
+            );
+        }, null);
+    }
+
     public void testRemoteSourcePlatinum() throws InterruptedException {
         final Context context = spy(
             new SourceDestValidator.Context(

x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/DataframeCpsIT.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.ml.integration;
+
+import org.elasticsearch.common.ValidationException;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.plugins.ClusterPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.xpack.core.ml.action.PutDataFrameAnalyticsAction;
+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.BoostedTreeParams;
+import org.elasticsearch.xpack.core.ml.dataframe.analyses.Classification;
+import org.elasticsearch.xpack.core.ml.utils.QueryProvider;
+import org.elasticsearch.xpack.ml.MlSingleNodeTestCase;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.hamcrest.Matchers.containsString;
+
+public class DataframeCpsIT extends MlSingleNodeTestCase {
+    @Override
+    protected Settings nodeSettings() {
+        return Settings.builder().put(super.nodeSettings()).put("serverless.cross_project.enabled", "true").build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> getPlugins() {
+        return Stream.concat(super.getPlugins().stream(), Stream.of(CpsPlugin.class)).toList();
+    }
+
+    public void testCrossProjectFailsForDataFrameAnalytics() throws IOException {
+        var id = "test-cross-project-fails";
+        var sourceIndex = "project1:" + id + "_source_index";
+        var destIndex = id + "_results";
+
+        var config = new DataFrameAnalyticsConfig.Builder().setId(id)
+            .setSource(
+                new DataFrameAnalyticsSource(
+                    new String[] { sourceIndex },
+                    QueryProvider.fromParsedQuery(QueryBuilders.matchAllQuery()),
+                    null,
+                    Collections.emptyMap()
+                )
+            )
+            .setDest(new DataFrameAnalyticsDest(destIndex, null))
+            .setAnalysis(
+                new Classification(
+                    "keyword-field",
+                    BoostedTreeParams.builder().setNumTopFeatureImportanceValues(1).build(),
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null
+                )
+            )
+            .build();
+
+        var request = new PutDataFrameAnalyticsAction.Request(config);
+        var response = client().execute(PutDataFrameAnalyticsAction.INSTANCE, request);
+        var validationException = assertThrows(ValidationException.class, response::actionGet);
+        assertThat(validationException.getMessage(), containsString("remote source and cross-project indices are not supported"));
+    }
+
+    public static class CpsPlugin extends Plugin implements ClusterPlugin {
+        public List<Setting<?>> getSettings() {
+            return List.of(Setting.simpleString("serverless.cross_project.enabled", Setting.Property.NodeScope));
+        }
+    }
+}