Better support for multi cluster for run task (#89442)

This commit introduces a `./gradlew run-ccs` task with the following goals:
* mirror the ease of use of `./gradlew run` for manual cross cluster search development
  * same credentials
  * same well known ports
  * uses ccs specific naming
  * enable debugging across both clusters

This is admittedly kinda hacky. Test clusters have support multi-cluster and are in use for 
 for automated testing. There are some nuances that make that setup (and this setup) 
a bit cumbersome..specifically needing to read one cluster's config to configure another cluster. 

The run task adds a bit more config (well defined ports, etc.) than the tests need to 
so that also complicates this abit more. I found that without the additions here I was unable to get both 
sharing of cluster configuration (like in the [tests](https://github.com/elastic/elasticsearch/blob/main/qa/ccs-common-rest/build.gradle#L55)) 
and the run task's hard coded config to work well together. Hopefully the additions to the run task are not
too hacky as I could not find any other way.

build-tools-internal/src/main/groovy/elasticsearch.run-ccs.gradle

@@ -0,0 +1,60 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import org.elasticsearch.gradle.testclusters.DefaultTestClustersTask
+import org.elasticsearch.gradle.testclusters.RunTask
+
+boolean proxyMode = true;
+
+def fulfillingCluster = testClusters.register('fulfilling-cluster') {
+  setting 'xpack.watcher.enabled', 'false'
+  setting 'xpack.ml.enabled', 'false'
+  setting 'xpack.license.self_generated.type', 'trial'
+
+  user username: 'elastic-admin', password: 'elastic-password', role: '_es_test_root'
+}
+
+def queryingCluster = testClusters.register('querying-cluster') {
+  setting 'xpack.watcher.enabled', 'false'
+  setting 'xpack.ml.enabled', 'false'
+  setting 'xpack.license.self_generated.type', 'trial'
+  if (proxyMode) {
+    setting 'cluster.remote.my_remote_cluster.mode', 'proxy'
+    setting 'cluster.remote.my_remote_cluster.proxy_address', {
+      "\"${fulfillingCluster.get().getAllTransportPortURI().get(0)}\""
+    }
+  } else {
+    setting 'cluster.remote.my_remote_cluster.seeds', {
+      fulfillingCluster.get().getAllTransportPortURI().collect { "\"$it\"" }.toString()
+    }
+  }
+  setting 'cluster.remote.connections_per_cluster', "1"
+
+  user username: 'elastic-admin', password: 'elastic-password', role: '_es_test_root'
+}
+
+// the following task is needed to make sure the fulfilling cluster is fully configured before starting both clusters
+// this allows the quering cluster to use configuration from the fulfilling cluster while honoring the RunTasks configuration (such as use port 9200)
+tasks.register('initfulfillingCluster', RunTask) {
+  useCluster testClusters.named("fulfilling-cluster")
+  initOnly = true //only initialize the testCluster, don't start it
+  portOffset = 1 //when only initializing, instruct to use one above the normal ports to avoid collisions when other cluster also initializes
+  //debug = true //this task doesn't honor the command line options for run-ccs, so need to statically configure debug
+}
+
+tasks.register("run-ccs", RunTask) {
+  dependsOn initfulfillingCluster
+  useCluster testClusters.named("fulfilling-cluster")
+  useCluster testClusters.named("querying-cluster")
+  doFirst {
+    println "** Querying cluster HTTP endpoints are: ${-> queryingCluster.get().allHttpSocketURI.join(",")}"
+    println "** Querying cluster transport endpoints are: ${-> queryingCluster.get().getAllTransportPortURI().join(",")}"
+    println "** Fulfilling cluster HTTP endpoints are: ${-> fulfillingCluster.get().allHttpSocketURI.join(",")}"
+    println "** Fulfilling cluster transport endpoints are: ${-> fulfillingCluster.get().getAllTransportPortURI().join(",")}"
+  }
+}

build-tools/src/main/java/org/elasticsearch/gradle/testclusters/RunTask.java

@@ -35,12 +35,16 @@ public class RunTask extends DefaultTestClustersTask {
 
     private Boolean debug = false;
 
+    private Boolean initOnly = false;
+
     private Boolean preserveData = false;
 
     private Path dataDir = null;
 
     private String keystorePassword = "";
 
+    private Integer offset = 0;
+
     @Option(option = "debug-jvm", description = "Enable debugging configuration, to allow attaching a debugger to elasticsearch.")
     public void setDebug(boolean enabled) {
         this.debug = enabled;
@@ -86,10 +90,36 @@ public class RunTask extends DefaultTestClustersTask {
         return dataDir.toString();
     }
 
+    @Input
+    @Optional
+    Boolean getInitOnly() {
+        return initOnly;
+    }
+
+    /**
+     * Only initialize, but don't actually run. This is useful for multi-cluster run tasks.
+     */
+    public void setInitOnly(Boolean initOnly) {
+        this.initOnly = initOnly;
+    }
+
+    @Input
+    @Optional
+    public Integer getPortOffset() {
+        return offset;
+    }
+
+    /**
+     * Manually increase the port offset. This is useful for multi-cluster run tasks.
+     */
+    public void setPortOffset(Integer offset) {
+        this.offset = offset;
+    }
+
     @Override
     public void beforeStart() {
-        int httpPort = 9200;
-        int transportPort = 9300;
+        int httpPort = 9200 + offset;
+        int transportPort = 9300 + offset;
         Map<String, String> additionalSettings = System.getProperties()
             .entrySet()
             .stream()
@@ -126,12 +156,15 @@ public class RunTask extends DefaultTestClustersTask {
         }
 
         if (debug) {
-            enableDebug();
+            enableDebug(getPortOffset());
         }
     }
 
     @TaskAction
     public void runAndWait() throws IOException {
+        if (initOnly) {
+            return;
+        }
         List<BufferedReader> toRead = new ArrayList<>();
         List<BooleanSupplier> aliveChecks = new ArrayList<>();
 

build-tools/src/main/java/org/elasticsearch/gradle/testclusters/StandaloneRestIntegTestTask.java

@@ -104,7 +104,7 @@ public class StandaloneRestIntegTestTask extends Test implements TestClustersAwa
     @Override
     public void beforeStart() {
         if (debugServer) {
-            enableDebug();
+            enableDebug(0);
         }
     }
 }

build-tools/src/main/java/org/elasticsearch/gradle/testclusters/TestClustersAware.java

@@ -37,8 +37,8 @@ public interface TestClustersAware extends Task {
 
     default void beforeStart() {}
 
-    default void enableDebug() {
-        int debugPort = 5007;
+    default void enableDebug(int portOffset) {
+        int debugPort = 5007 + portOffset;
         for (ElasticsearchCluster cluster : getClusters()) {
             for (ElasticsearchNode node : cluster.getNodes()) {
                 getLogger().lifecycle("Running elasticsearch in debug mode, {} expecting running debug server on port {}", node, debugPort);

build.gradle

@@ -42,6 +42,7 @@ plugins {
   id 'elasticsearch.fips'
   id 'elasticsearch.internal-testclusters'
   id 'elasticsearch.run'
+  id 'elasticsearch.run-ccs'
   id 'elasticsearch.release-tools'
   id 'elasticsearch.versions'
 }