Run more search rest tests in a CCS setup

Currently we only test a small subset of cross cluster search in rest tests in
the 'multi-cluster-search' qa module. In order to increase test coverage for
basic CCS setups, this change adds a new qa modula that uses a subset of
existing search rest tests tests and runs them in a CCS scenario.

Relates to #84481

qa/ccs-common-rest/build.gradle

@@ -0,0 +1,98 @@
+/*
+ * 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.Version
+import org.elasticsearch.gradle.internal.test.RestIntegTestTask
+import org.elasticsearch.gradle.testclusters.DefaultTestClustersTask
+
+
+apply plugin: 'elasticsearch.internal-testclusters'
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply plugin: 'elasticsearch.rest-resources'
+
+restResources {
+  restApi {
+    include '_common', 'bulk', 'count', 'cluster', 'index', 'indices', 'field_caps', 'msearch',
+      'search', 'async_search', 'graph', '*_point_in_time', "info"
+  }
+  restTests {
+    includeCore 'field_caps', 'search', 'suggest'
+  }
+}
+
+def writeCluster = testClusters.register("ccs-write") {
+  numberOfNodes = 2
+  setting 'node.roles', '[data,ingest,master]'
+}
+
+def searchCluster = testClusters.register("ccs-search") {
+  setting 'node.roles', '[data,ingest,master,remote_cluster_client]'
+  setting 'cluster.remote.connections_per_cluster', '1'
+  setting 'cluster.remote.remote_cluster.seeds',
+    { "\"${writeCluster.get().getAllTransportPortURI().get(0)}\"" }
+}
+
+testClusters.configureEach {
+  setting 'xpack.security.enabled', 'false'
+  requiresFeature 'es.index_mode_feature_flag_registered', Version.fromString("8.0.0")
+}
+
+tasks.register('startWriteCluster', DefaultTestClustersTask) {
+  useCluster writeCluster
+  doLast {
+    clusters.each { c ->
+      print "Writer cluster transport uri for ccs configuration is: "
+      println c.getAllTransportPortURI().get(0)
+    }
+  }
+}
+
+tasks.register("ccs-write", RestIntegTestTask) {
+  mustRunAfter("precommit")
+  dependsOn startWriteCluster
+
+  useCluster writeCluster
+  useCluster searchCluster
+
+  systemProperty 'tests.rest.suite',
+    [
+      'field_caps',
+      'search',
+      'search.aggregation',
+      'search.highlight',
+      'search.inner_hits',
+      'suggest',
+    ].join(',')
+
+  systemProperty 'tests.rest.blacklist',
+    [
+      // TODO look into fixing these
+      'search/70_response_filtering/Search with response filtering', // makes assertions on "_clusters" section
+      'search/150_rewrite_on_coordinator/Ensure that we fetch the document only once', // terms lookup query with index
+      'search/170_terms_query/Terms Query with No.of terms exceeding index.max_terms_count should FAIL', // terms lookup query with index
+      'search/350_point_in_time/basic', // [indices] cannot be used with point in time
+      'search/350_point_in_time/point-in-time with slicing', // [indices] cannot be used with point in time
+      'search/350_point_in_time/wildcard', // [indices] cannot be used with point in time
+      'search.aggregation/220_filters_bucket/cache busting', // node_selector?
+      'search.aggregation/220_filters_bucket/cache hits', // node_selector?
+      'search.aggregation/50_filter/Standard queries get cached',
+      'search.aggregation/50_filter/Terms lookup gets cached', // terms lookup by "index" doesn't seem to work correctly
+      'search.aggregation/70_adjacency_matrix/Terms lookup' // terms lookup by "index" doesn't seem to work correctly
+    ].join(',')
+
+
+  doFirst {
+    // Getting the endpoints causes a wait for the cluster
+    println "Writer cluster endpoints are: ${-> writeCluster.get().allHttpSocketURI.join(",")}"
+    println "Search cluster endpoints are: ${-> searchCluster.get().allHttpSocketURI.join(",")}"
+    nonInputProperties.systemProperty('tests.rest.cluster', writeCluster.map(c -> c.allHttpSocketURI.join(",")))
+    nonInputProperties.systemProperty('tests.rest.search_cluster', searchCluster.map(c -> c.allHttpSocketURI.join(",")))
+  }
+}
+

qa/ccs-common-rest/src/test/java/org/elasticsearch/test/rest/yaml/CcsCommonYamlTestSuiteIT.java

@@ -0,0 +1,188 @@
+/*
+ * 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.
+ */
+
+package org.elasticsearch.test.rest.yaml;
+
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+import com.carrotsearch.randomizedtesting.annotations.TimeoutSuite;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.lucene.tests.util.TimeUnits;
+import org.elasticsearch.Version;
+import org.elasticsearch.client.NodeSelector;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.core.Tuple;
+import org.elasticsearch.test.rest.ObjectPath;
+import org.elasticsearch.test.rest.yaml.section.ClientYamlTestSection;
+import org.elasticsearch.test.rest.yaml.section.ExecutableSection;
+import org.elasticsearch.test.rest.yaml.section.MatchAssertion;
+import org.junit.AfterClass;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.unmodifiableList;
+
+@TimeoutSuite(millis = 15 * TimeUnits.MINUTE) // to account for slow as hell VMs
+public class CcsCommonYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
+
+    private static final Logger logger = LogManager.getLogger(CcsCommonYamlTestSuiteIT.class);
+    private static RestClient searchClient;
+    private static RestClient adminSearchClient;
+    private static List<HttpHost> clusterHosts;
+    private static ClientYamlTestClient searchYamlTestClient;
+
+    private static final String REMOTE_CLUSTER_NAME = "remote_cluster";
+
+    @Before
+    public void initSearchClient() throws IOException {
+        if (searchClient == null) {
+            assert adminSearchClient == null;
+            assert clusterHosts == null;
+
+            final String cluster = System.getProperty("tests.rest.search_cluster");
+            assertNotNull("[tests.rest.search_cluster] is not configured", cluster);
+            String[] stringUrls = cluster.split(",");
+            List<HttpHost> hosts = new ArrayList<>(stringUrls.length);
+            for (String stringUrl : stringUrls) {
+                int portSeparator = stringUrl.lastIndexOf(':');
+                if (portSeparator < 0) {
+                    throw new IllegalArgumentException("Illegal cluster url [" + stringUrl + "]");
+                }
+                String host = stringUrl.substring(0, portSeparator);
+                int port = Integer.parseInt(stringUrl.substring(portSeparator + 1));
+                hosts.add(buildHttpHost(host, port));
+            }
+            clusterHosts = unmodifiableList(hosts);
+            logger.info("initializing REST search clients against {}", clusterHosts);
+            searchClient = buildClient(restClientSettings(), clusterHosts.toArray(new HttpHost[clusterHosts.size()]));
+            adminSearchClient = buildClient(restAdminSettings(), clusterHosts.toArray(new HttpHost[clusterHosts.size()]));
+
+            Tuple<Version, Version> versionVersionTuple = readVersionsFromCatNodes(adminSearchClient);
+            final Version esVersion = versionVersionTuple.v1();
+            final Version masterVersion = versionVersionTuple.v2();
+            final String os = readOsFromNodesInfo(adminSearchClient);
+
+            searchYamlTestClient = new ClientYamlTestClient(
+                getRestSpec(),
+                searchClient,
+                hosts,
+                esVersion,
+                masterVersion,
+                os,
+                this::getClientBuilderWithSniffedHosts
+            ) {
+                public ClientYamlTestResponse callApi(
+                    String apiName,
+                    Map<String, String> params,
+                    HttpEntity entity,
+                    Map<String, String> headers,
+                    NodeSelector nodeSelector
+                ) throws IOException {
+                    // on request, we need to replace index specifications by prefixing the remote cluster
+                    String originalIndices = params.get("index");
+                    String expandedIndices = REMOTE_CLUSTER_NAME + ":*";
+                    if (originalIndices != null && (originalIndices.isEmpty() == false)) {
+                        String[] indices = originalIndices.split(",");
+                        List<String> newIndices = new ArrayList<>();
+                        for (String indexName : indices) {
+                            newIndices.add(REMOTE_CLUSTER_NAME + ":" + indexName);
+                        }
+                        expandedIndices = String.join(",", newIndices);
+                    }
+                    params.put("index", String.join(",", expandedIndices));
+                    ClientYamlTestResponse clientYamlTestResponse = super.callApi(apiName, params, entity, headers, nodeSelector);
+                    return clientYamlTestResponse;
+                };
+            };
+
+            // check that we have an established CCS connection
+            Request request = new Request("GET", "_remote/info");
+            Response response = adminSearchClient.performRequest(request);
+            assertOK(response);
+            ObjectPath responseObject = ObjectPath.createFromResponse(response);
+            assertNotNull(responseObject.evaluate(REMOTE_CLUSTER_NAME));
+            logger.info("Established connection to remote cluster [" + REMOTE_CLUSTER_NAME + "]");
+        }
+
+        assert searchClient != null;
+        assert adminSearchClient != null;
+        assert clusterHosts != null;
+    }
+
+    public CcsCommonYamlTestSuiteIT(ClientYamlTestCandidate testCandidate) throws IOException {
+        super(rewrite(testCandidate));
+    }
+
+    private static ClientYamlTestCandidate rewrite(ClientYamlTestCandidate clientYamlTestCandidate) {
+        ClientYamlTestSection testSection = clientYamlTestCandidate.getTestSection();
+        List<ExecutableSection> executableSections = testSection.getExecutableSections();
+        List<ExecutableSection> modifiedExecutableSections = new ArrayList<>();
+        for (ExecutableSection section : executableSections) {
+            if (section instanceof MatchAssertion) {
+                MatchAssertion matchSection = (MatchAssertion) section;
+                if (matchSection.getField().endsWith("_index")) {
+                    String modifiedExpectedValue = REMOTE_CLUSTER_NAME + ":" + matchSection.getExpectedValue();
+                    modifiedExecutableSections.add(
+                        new MatchAssertion(matchSection.getLocation(), matchSection.getField(), modifiedExpectedValue)
+                    );
+                }
+            } else {
+                modifiedExecutableSections.add(section);
+            }
+        }
+        return new ClientYamlTestCandidate(
+            clientYamlTestCandidate.getRestTestSuite(),
+            new ClientYamlTestSection(
+                testSection.getLocation(),
+                testSection.getName(),
+                testSection.getSkipSection(),
+                modifiedExecutableSections
+            )
+        );
+    };
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() throws Exception {
+        return createParameters();
+    }
+
+    @Override
+    protected ClientYamlTestExecutionContext createRestTestExecutionContext(
+        ClientYamlTestCandidate clientYamlTestCandidate,
+        ClientYamlTestClient clientYamlTestClient
+    ) {
+        return new ClientYamlTestExecutionContext(clientYamlTestCandidate, clientYamlTestClient, randomizeContentType()) {
+            protected ClientYamlTestClient clientYamlTestClient(String apiName) {
+                if (apiName.equals("search") || apiName.equals("field_caps")) {
+                    return searchYamlTestClient;
+                } else {
+                    return super.clientYamlTestClient(apiName);
+                }
+            }
+        };
+    }
+
+    @AfterClass
+    public static void closeSearchClients() throws IOException {
+        try {
+            IOUtils.close(searchClient, adminSearchClient);
+        } finally {
+            clusterHosts = null;
+        }
+    }
+}

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/ClientYamlTestCandidate.java

@@ -45,6 +45,10 @@ public SetupSection getSetupSection() {
         return restTestSuite.getSetupSection();
     }
 
+    public ClientYamlTestSuite getRestTestSuite() {
+        return restTestSuite;
+    }
+
     public TeardownSection getTeardownSection() {
         return restTestSuite.getTeardownSection();
     }

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/ClientYamlTestExecutionContext.java

@@ -183,7 +183,11 @@ ClientYamlTestResponse callApiInternal(
         Map<String, String> headers,
         NodeSelector nodeSelector
     ) throws IOException {
-        return clientYamlTestClient.callApi(apiName, params, entity, headers, nodeSelector);
+        return clientYamlTestClient(apiName).callApi(apiName, params, entity, headers, nodeSelector);
+    }
+
+    protected ClientYamlTestClient clientYamlTestClient(String apiName) {
+        return clientYamlTestClient;
     }
 
     /**

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/ESClientYamlSuiteTestCase.java

@@ -108,6 +108,8 @@ public abstract class ESClientYamlSuiteTestCase extends ESRestTestCase {
 
     private final ClientYamlTestCandidate testCandidate;
 
+    private static ClientYamlSuiteRestSpec restSpecification;
+
     protected ESClientYamlSuiteTestCase(ClientYamlTestCandidate testCandidate) {
         this.testCandidate = testCandidate;
     }
@@ -126,6 +128,7 @@ public void initAndResetContext() throws Exception {
             assert blacklistPathMatchers == null;
             final ClientYamlSuiteRestSpec restSpec = ClientYamlSuiteRestSpec.load(SPEC_PATH);
             validateSpec(restSpec);
+            restSpecification = restSpec;
             final List<HttpHost> hosts = getClusterHosts();
             Tuple<Version, Version> versionVersionTuple = readVersionsFromCatNodes(adminClient());
             final Version esVersion = versionVersionTuple.v1();
@@ -317,6 +320,10 @@ protected ClientYamlTestExecutionContext getAdminExecutionContext() {
         return adminExecutionContext;
     }
 
+    static ClientYamlSuiteRestSpec getRestSpec() {
+        return restSpecification;
+    }
+
     private static void validateSpec(ClientYamlSuiteRestSpec restSpec) {
         boolean validateSpec = RandomizedTest.systemPropertyAsBoolean(REST_TESTS_VALIDATE_SPEC, true);
         if (validateSpec) {
@@ -341,7 +348,7 @@ private static void validateSpec(ClientYamlSuiteRestSpec restSpec) {
         }
     }
 
-    private Tuple<Version, Version> readVersionsFromCatNodes(RestClient restClient) throws IOException {
+    Tuple<Version, Version> readVersionsFromCatNodes(RestClient restClient) throws IOException {
         // we simply go to the _cat/nodes API and parse all versions in the cluster
         final Request request = new Request("GET", "/_cat/nodes");
         request.addParameter("h", "version,master");
@@ -370,7 +377,7 @@ private Tuple<Version, Version> readVersionsFromCatNodes(RestClient restClient)
         return new Tuple<>(version, masterVersion);
     }
 
-    private String readOsFromNodesInfo(RestClient restClient) throws IOException {
+    String readOsFromNodesInfo(RestClient restClient) throws IOException {
         final Request request = new Request("GET", "/_nodes/os");
         Response response = restClient.performRequest(request);
         ClientYamlTestResponse restTestResponse = new ClientYamlTestResponse(response);