Add query to esql task (ESQL-1511)

This adds the `query` to the main ESQL task so you can see long running
queries. And adds some docs about it including an example of cancelling
a query.

---------

Co-authored-by: Abdon Pijpelink <abdon.pijpelink@elastic.co>

docs/reference/esql/index.asciidoc

@@ -130,5 +130,7 @@ include::aggregation-functions.asciidoc[]
 
 include::multivalued-fields.asciidoc[]
 
+include::task-management.asciidoc[]
+
 :esql-tests!:
 :esql-specs!:

docs/reference/esql/task-management.asciidoc

@@ -0,0 +1,35 @@
+[[esql-task-management]]
+== {esql} task management
+
+++++
+<titleabbrev>Task management</titleabbrev>
+++++
+
+You can get running {esql} queries with the <<tasks,task management API>>:
+
+[source,console,id=esql-task-management-get-all]
+----
+GET /_tasks?pretty&detailed&group_by=parents&human&actions=*data/read/esql
+----
+
+Which returns a list of statuses like this:
+
+[source,js]
+----
+include::{esql-specs}/query_task.json[]
+----
+// NOTCONSOLE
+// Tested in a unit test
+
+<1> The user submitted query.
+<2> Time the query has been running.
+
+You can use this to find long running queries and, if you need to, cancel them
+with the <<task-cancellation, task cancellation API>>:
+
+[source,console,id=esql-task-management-cancelEsqlQueryRequestTests]
+----
+POST _tasks/2j8UKw1bRO283PMwDugNNg:5326/_cancel
+----
+
+It may take a few seconds for the query to be stopped.

x-pack/plugin/esql/qa/testFixtures/src/main/resources/query_task.json

@@ -0,0 +1,14 @@
+{
+  "node" : "2j8UKw1bRO283PMwDugNNg",
+  "id" : 5326,
+  "type" : "transport",
+  "action" : "indices:data/read/esql",
+  "description" : "FROM test | STATS MAX(d) by a, b",  <1>
+  "start_time" : "2023-07-31T15:46:32.328Z",
+  "start_time_in_millis" : 1690818392328,
+  "running_time" : "41.7ms",                           <2>
+  "running_time_in_nanos" : 41770830,
+  "cancellable" : true,
+  "cancelled" : false,
+  "headers" : { }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlQueryRequest.java

@@ -241,7 +241,8 @@ public class EsqlQueryRequest extends ActionRequest implements CompositeIndicesR
 
     @Override
     public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
-        return new CancellableTask(id, type, action, "", parentTaskId, headers);
+        // Pass the query as the description
+        return new CancellableTask(id, type, action, query, parentTaskId, headers);
     }
 
     protected static void validateParams(List<TypedParamValue> params) {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/action/EsqlQueryRequestTests.java

@@ -7,11 +7,17 @@
 
 package org.elasticsearch.xpack.esql.action;
 
+import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.RangeQueryBuilder;
 import org.elasticsearch.index.query.TermQueryBuilder;
 import org.elasticsearch.search.SearchModule;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParser;
@@ -23,8 +29,10 @@ import java.time.ZoneId;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 
 public class EsqlQueryRequestTests extends ESTestCase {
 
@@ -97,7 +105,33 @@ public class EsqlQueryRequestTests extends ESTestCase {
             }""");
         assertNotNull(request.validate());
         assertThat(request.validate().getMessage(), containsString("[query] is required"));
+    }
+
+    public void testTask() throws IOException {
+        String query = randomAlphaOfLength(10);
+        int id = randomInt();
 
+        EsqlQueryRequest request = parseEsqlQueryRequest("""
+            {
+                "query": "QUERY"
+            }""".replace("QUERY", query));
+        Task task = request.createTask(id, "transport", EsqlQueryAction.NAME, TaskId.EMPTY_TASK_ID, Map.of());
+        assertThat(task.getDescription(), equalTo(query));
+
+        String localNode = randomAlphaOfLength(2);
+        TaskInfo taskInfo = task.taskInfo(localNode, true);
+        String json = taskInfo.toString();
+        String expected = Streams.readFully(getClass().getClassLoader().getResourceAsStream("query_task.json")).utf8ToString();
+        expected = expected.replaceAll("\s*<\\d+>", "")
+            .replaceAll("FROM test \\| STATS MAX\\(d\\) by a, b", query)
+            .replaceAll("5326", Integer.toString(id))
+            .replaceAll("2j8UKw1bRO283PMwDugNNg", localNode)
+            .replaceAll("2023-07-31T15:46:32\\.328Z", DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.formatMillis(taskInfo.startTime()))
+            .replaceAll("1690818392328", Long.toString(taskInfo.startTime()))
+            .replaceAll("41.7ms", TimeValue.timeValueNanos(taskInfo.runningTimeNanos()).toString())
+            .replaceAll("41770830", Long.toString(taskInfo.runningTimeNanos()))
+            .trim();
+        assertThat(json, equalTo(expected));
     }
 
     private static void assertParserErrorMessage(String json, String message) {