QL: retry SQL and EQL requests in a mixed-node (rolling upgrade) cluster

x-pack/plugin/eql/qa/mixed-node/build.gradle

@@ -0,0 +1,66 @@
+apply plugin: 'elasticsearch.testclusters'
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply from : "$rootDir/gradle/bwc-test.gradle"
+apply plugin: 'elasticsearch.rest-test'
+
+import org.elasticsearch.gradle.Version
+import org.elasticsearch.gradle.VersionProperties
+import org.elasticsearch.gradle.info.BuildParams
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask
+
+dependencies {
+  testImplementation project(':x-pack:qa')
+  testImplementation(project(xpackModule('ql:test')))
+  testImplementation project(path: xpackModule('eql'), configuration: 'default')
+}
+
+tasks.named("integTest").configure{ enabled = false}
+
+for (Version bwcVersion : BuildParams.bwcVersions.wireCompatible.findAll { it.onOrAfter('7.10.0') }) {
+  if (bwcVersion == VersionProperties.getElasticsearchVersion()) {
+    // Not really a mixed cluster
+    continue;
+  }
+
+  String baseName = "v${bwcVersion}"
+
+  testClusters {
+    "${baseName}" {
+      versions = [bwcVersion.toString(), project.version]
+      numberOfNodes = 3
+      testDistribution = 'DEFAULT'
+      setting 'xpack.security.enabled', 'false'
+      setting 'xpack.watcher.enabled', 'false'
+      setting 'xpack.ml.enabled', 'false'
+	  setting 'xpack.eql.enabled', 'true'
+      setting 'xpack.license.self_generated.type', 'trial'
+      // for debugging purposes
+      // setting 'logger.org.elasticsearch.xpack.eql.plugin.TransportEqlSearchAction', 'TRACE'
+    }
+  }
+
+  tasks.register("${baseName}#mixedClusterTest", StandaloneRestIntegTestTask) {
+    useCluster testClusters."${baseName}"
+    mustRunAfter("precommit")
+    doFirst {
+      // Getting the endpoints causes a wait for the cluster
+      println "Endpoints are: ${-> testClusters."${baseName}".allHttpSocketURI.join(",")}"
+      println "Upgrading one node to create a mixed cluster"
+      testClusters."${baseName}".nextNodeToNextVersion()
+
+      println "Upgrade complete, endpoints are: ${-> testClusters."${baseName}".allHttpSocketURI.join(",")}"
+      nonInputProperties.systemProperty('tests.rest.cluster', "${-> testClusters."${baseName}".allHttpSocketURI.join(",")}")
+      nonInputProperties.systemProperty('tests.clustername', "${-> testClusters."${baseName}".getName()}")
+    }
+    onlyIf { project.bwc_tests_enabled }
+  }
+
+  tasks.register(bwcTaskName(bwcVersion)) {
+    dependsOn "${baseName}#mixedClusterTest"
+  }
+
+  // run these bwc tests as part of the "check" task
+  tasks.named("check").configure {
+    dependsOn "${baseName}#mixedClusterTest"
+  }
+}

x-pack/plugin/eql/qa/mixed-node/src/test/java/org/elasticsearch/xpack/eql/qa/mixed_node/EqlSearchIT.java

@@ -0,0 +1,239 @@
+/*
+ * 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.eql.qa.mixed_node;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.NotEqualMessageBuilder;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.xpack.ql.TestNode;
+import org.elasticsearch.xpack.ql.TestNodes;
+import org.junit.After;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
+import static java.util.Collections.unmodifiableList;
+import static org.elasticsearch.xpack.ql.TestUtils.buildNodeAndVersions;
+import static org.elasticsearch.xpack.ql.TestUtils.readResource;
+
+/**
+ * Class testing the behavior of events and sequence queries in a mixed cluster scenario (during rolling upgrade).
+ * The test is against a three-node cluster where one node is upgraded, the other two are on the old version.
+ *  
+ */
+public class EqlSearchIT extends ESRestTestCase {
+
+    private static final String index = "test_eql_mixed_versions";
+    private static int numShards;
+    private static int numReplicas = 1;
+    private static int numDocs;
+    private static TestNodes nodes;
+    private static List<TestNode> newNodes;
+    private static List<TestNode> bwcNodes;
+
+    @Before
+    public void createIndex() throws IOException {
+        nodes = buildNodeAndVersions(client());
+        numShards = nodes.size();
+        numDocs = randomIntBetween(numShards, 15);
+        newNodes = new ArrayList<>(nodes.getNewNodes());
+        bwcNodes = new ArrayList<>(nodes.getBWCNodes());
+
+        String mappings = readResource(EqlSearchIT.class.getResourceAsStream("/eql_mapping.json"));
+        createIndex(
+            index,
+            Settings.builder()
+                .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), numShards)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numReplicas)
+                .build(),
+            mappings
+        );
+    }
+
+    @After
+    public void cleanUpIndex() throws IOException {
+        if (indexExists(index)) {
+            deleteIndex(index);
+        }
+    }
+
+    public void testEventsWithRequestToOldNodes() throws Exception {
+        assertEventsQueryOnNodes(bwcNodes);
+    }
+
+    public void testEventsWithRequestToUpgradedNodes() throws Exception {
+        assertEventsQueryOnNodes(newNodes);
+    }
+
+    public void testSequencesWithRequestToOldNodes() throws Exception {
+        assertSequncesQueryOnNodes(bwcNodes);
+    }
+
+    public void testSequencesWithRequestToUpgradedNodes() throws Exception {
+        assertSequncesQueryOnNodes(newNodes);
+    }
+
+    private void assertEventsQueryOnNodes(List<TestNode> nodesList) throws Exception {
+        final String event = randomEvent();
+        Map<String, Object> expectedResponse = prepareEventsTestData(event);
+        try (
+            RestClient client = buildClient(restClientSettings(),
+                nodesList.stream().map(TestNode::getPublishAddress).toArray(HttpHost[]::new))
+        ) {
+            // filter only the relevant bits of the response
+            String filterPath = "filter_path=hits.events._source.@timestamp,hits.events._source.event_type,hits.events._source.sequence";
+
+            Request request = new Request("POST", index + "/_eql/search?" + filterPath);
+            request.setJsonEntity("{\"query\":\"" + event + " where true\"}");
+            assertBusy(() -> { assertResponse(expectedResponse, runEql(client, request)); });
+        }
+    }
+
+    private void assertSequncesQueryOnNodes(List<TestNode> nodesList) throws Exception {
+        Map<String, Object> expectedResponse = prepareSequencesTestData();
+        try (
+            RestClient client = buildClient(restClientSettings(),
+                nodesList.stream().map(TestNode::getPublishAddress).toArray(HttpHost[]::new))
+        ) {
+            String filterPath = "filter_path=hits.sequences.join_keys,hits.sequences.events._id,hits.sequences.events._source";
+            String query = "sequence by `sequence` with maxspan=100ms [success where true] by correlation_success1, correlation_success2 "
+                + "[failure where true] by correlation_failure1, correlation_failure2";
+            String filter = "{\"range\":{\"@timestamp\":{\"gte\":\"1970-05-01\"}}}";
+
+            Request request = new Request("POST", index + "/_eql/search?" + filterPath);
+            request.setJsonEntity("{\"query\":\"" + query + "\",\"filter\":" + filter + "}");
+            assertBusy(() -> { assertResponse(expectedResponse, runEql(client, request)); });
+        }
+    }
+
+    private String randomEvent() {
+        return randomFrom("success", "failure");
+    }
+
+    private Map<String, Object> prepareEventsTestData(String event) throws IOException {
+        List<Map<String, Object>> sourceEvents = new ArrayList<Map<String, Object>>();
+        Map<String, Object> expectedResponse = singletonMap("hits", singletonMap("events", sourceEvents));
+
+        for (int i = 0; i < numDocs; i++) {
+            StringBuilder builder = new StringBuilder();
+            final String randomEvent = randomEvent();
+            builder.append("{");
+            builder.append("\"@timestamp\":" + i + ",");
+            builder.append("\"event_type\":\"" + randomEvent + "\",");
+            builder.append("\"sequence\":" + i);
+            builder.append("}");
+            if (randomEvent.equals(event)) {
+                Map<String, Object> eventSource = new HashMap<>();
+                eventSource.put("@timestamp", i);
+                eventSource.put("event_type", randomEvent);
+                eventSource.put("sequence", i);
+                sourceEvents.add(singletonMap("_source", eventSource));
+            }
+
+            Request request = new Request("PUT", index + "/_doc/" + i);
+            request.setJsonEntity(builder.toString());
+            assertOK(client().performRequest(request));
+        }
+        if (sourceEvents.isEmpty()) {
+            return emptyMap();
+        }
+        return expectedResponse;
+    }
+
+    /*
+     * Output to compare with looks like this:
+     * {
+     *  "hits": {
+     *  "sequences": [
+     *      {
+     *          "join_keys": [
+     *              44,
+     *              "C",
+     *              "D"
+     *          ],
+     *          "events": [
+     *              {
+     *                  "_id": "14",
+     *                  "_source": {
+     *                  ...
+     *                  }
+     *              }
+     *           ]
+     *      }
+     *  }
+     * }
+     *
+     */
+    private Map<String, Object> prepareSequencesTestData() throws IOException {
+        Map<String, Object> event14 = new HashMap<>();
+        Map<String, Object> event14Source = new HashMap<>();
+        event14.put("_id", "14");
+        event14.put("_source", event14Source);
+        event14Source.put("@timestamp", "12345678914");
+        event14Source.put("event_type", "success");
+        event14Source.put("sequence", 44);
+        event14Source.put("correlation_success1", "C");
+        event14Source.put("correlation_success2", "D");
+
+        Map<String, Object> event15 = new HashMap<>();
+        Map<String, Object> event15Source = new HashMap<>();
+        event15.put("_id", "15");
+        event15.put("_source", event15Source);
+        event15Source.put("@timestamp", "12345678999");
+        event15Source.put("event_type", "failure");
+        event15Source.put("sequence", 44);
+        event15Source.put("correlation_failure1", "C");
+        event15Source.put("correlation_failure2", "D");
+
+        Map<String, Object> sequence = new HashMap<>();
+        List<Map<String, Object>> events = unmodifiableList(asList(event14, event15));
+        List<Map<String, Object>> sequences = singletonList(sequence);
+        Map<String, Object> expectedResponse = singletonMap("hits", singletonMap("sequences", sequences));
+
+        sequence.put("join_keys", asList(44, "C", "D"));
+        sequence.put("events", events);
+
+        final String bulkEntries = readResource(EqlSearchIT.class.getResourceAsStream("/eql_data.json"));
+        Request request = new Request("POST", index + "/_bulk?refresh");
+        request.setJsonEntity(bulkEntries);
+        assertOK(client().performRequest(request));
+
+        return expectedResponse;
+    }
+
+    private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
+        if (false == expected.equals(actual)) {
+            NotEqualMessageBuilder message = new NotEqualMessageBuilder();
+            message.compareMaps(actual, expected);
+            fail("Response does not match:\n" + message.toString());
+        }
+    }
+
+    private Map<String, Object> runEql(RestClient client, Request request) throws IOException {
+        Response response = client.performRequest(request);
+        try (InputStream content = response.getEntity().getContent()) {
+            return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
+        }
+    }
+}

x-pack/plugin/eql/qa/mixed-node/src/test/resources/eql_data.json

@@ -0,0 +1,30 @@
+{"index":{"_id":1}}
+{"@timestamp":"1234567891","event_type":"success","sequence":1,"correlation_success1":"A","correlation_success2":"B"}
+{"index":{"_id":2}}
+{"@timestamp":"1234567892","event_type":"failure","sequence":2,"correlation_failure1":"A","correlation_failure2":"B"}
+{"index":{"_id":3}}
+{"@timestamp":"1234567893","event_type":"success","sequence":3,"correlation_success1":"A","correlation_success2":"A"}
+{"index":{"_id":4}}
+{"@timestamp":"1234567894","event_type":"success","sequence":4,"correlation_success1":"C","correlation_success2":"C"}
+{"index":{"_id":5}}
+{"@timestamp":"1234567895","event_type":"failure","sequence":5,"correlation_failure1":"B","correlation_failure2":"C"}
+{"index":{"_id":6}}
+{"@timestamp":"1234567896","event_type":"success","sequence":1,"correlation_success1":"A","correlation_success2":"A"}
+{"index":{"_id":7}}
+{"@timestamp":"1234567897","event_type":"failure","sequence":1,"correlation_failure1":"A","correlation_failure2":"A"}
+{"index":{"_id":8}}
+{"@timestamp":"1234567898","event_type":"success","sequence":3,"correlation_success1":"A","correlation_success2":"A"}
+{"index":{"_id":9}}
+{"@timestamp":"1234567899","event_type":"success","sequence":4,"correlation_success1":"C","correlation_success2":"B"}
+{"index":{"_id":10}}
+{"@timestamp":"12345678910","event_type":"failure","sequence":4,"correlation_failure1":"B","correlation_failure2":"B"}
+{"index":{"_id":11}}
+{"@timestamp":"12345678911","event_type":"success","sequence":1,"correlation_success1":"A","correlation_success2":"A"}
+{"index":{"_id":12}}
+{"@timestamp":"12345678912","event_type":"failure","sequence":1,"correlation_failure1":"A","correlation_failure2":"B"}
+{"index":{"_id":13}}
+{"@timestamp":"12345678913","event_type":"success","sequence":3,"correlation_success1":"A","correlation_success2":"A"}
+{"index":{"_id":14}}
+{"@timestamp":"12345678914","event_type":"success","sequence":44,"correlation_success1":"C","correlation_success2":"D"}
+{"index":{"_id":15}}
+{"@timestamp":"12345678999","event_type":"failure","sequence":44,"correlation_failure1":"C","correlation_failure2":"D"}

x-pack/plugin/eql/qa/mixed-node/src/test/resources/eql_mapping.json

@@ -0,0 +1,35 @@
+  "properties": {
+    "@timestamp": {
+      "type": "date"
+    },
+    "event_type": {
+      "type": "keyword"
+    },
+    "sequence": {
+      "type": "long"
+    },
+    "correlation_success1": {
+      "type": "wildcard"
+    },
+    "correlation_failure1": {
+      "type": "wildcard"
+    },
+    "correlation_success2": {
+      "type": "keyword"
+    },
+    "correlation_failure2": {
+      "type": "keyword"
+    },
+    "event": {
+      "properties": {
+        "category": {
+          "type": "alias",
+          "path": "event_type"
+        },
+        "sequence": {
+          "type": "alias",
+          "path": "sequence"
+        }
+      }
+    }
+  }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/BasicQueryClient.java

@@ -57,7 +57,7 @@ public void query(QueryRequest request, ActionListener<SearchResponse> listener)
         // set query timeout
         searchSource.timeout(cfg.requestTimeout());
 
-        SearchRequest search = prepareRequest(client, searchSource, false, indices);
+        SearchRequest search = prepareRequest(searchSource, false, indices);
         search(search, searchLogListener(listener, log));
     }
 
@@ -138,7 +138,7 @@ public void fetchHits(Iterable<List<HitReference>> refs, ActionListener<List<Lis
                 // NB:this is different from mget
                 .size(idQuery.ids().size());
 
-            SearchRequest search = prepareRequest(client, builder, false, entry.getKey());
+            SearchRequest search = prepareRequest(builder, false, entry.getKey());
             multiSearchBuilder.add(search);
         }