ESQL: List/get query API (#124832)

This PR adds two new REST endpoints, for listing queries and getting information on a current query.

* Resolves #124827 
* Related to #124828 (initial work)

Changes from the API specified in the above issues:
* The get API is pretty initial, as we don't have a way of fetching the memory used or number of rows processed.

List queries response:
```
GET /_query/queries
// returns for each of the running queries
// query_id, start_time, running_time, query

{ "queries" : {
 "abc": {
  "id": "abc",
  "start_time_millis": 14585858875292,
  "running_time_nanos": 762794,
  "query": "FROM logs* | STATS BY hostname"
  },
 "4321": {
  "id":"4321",
  "start_time_millis": 14585858823573,
  "running_time_nanos": 90231,
  "query": "FROM orders | LOOKUP country_code ON country"
  }
 } 
}
```

Get query response:
```
GET /_query/queries/abc

{
 "id" : "abc",
  "start_time_millis": 14585858875292,
  "running_time_nanos": 762794,
  "query": "FROM logs* | STATS BY hostname"
  "coordinating_node": "oTUltX4IQMOUUVeiohTt8A"
  "data_nodes" : [ "DwrYwfytxthse49X4", "i5msnbUyWlpe86e7"]
}
```

Author: GalLalouche

docs/changelog/124832.yaml

@@ -0,0 +1,6 @@
+pr: 124832
+summary: List/get query API
+area: ES|QL
+type: feature
+issues:
+ - 124827

docs/reference/elasticsearch/security-privileges.md

@@ -194,6 +194,9 @@ This section lists the privileges that you can assign to a role.
 `monitor_enrich`
 :   All read-only operations related to managing and executing enrich policies.
 
+`monitor_esql`
+:   All read-only operations related to ES|QL queries.
+
 `monitor_inference`
 :   All read-only operations related to {{infer}}.
 

rest-api-spec/src/main/resources/rest-api-spec/api/esql.get_query.json

@@ -0,0 +1,32 @@
+{
+  "esql.get_query": {
+    "documentation": {
+      "url": null,
+      "description": "Executes a get ESQL query request"
+    },
+    "stability": "experimental",
+    "visibility": "public",
+    "headers": {
+      "accept": [],
+      "content_type": [
+        "application/json"
+      ]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/_query/queries/{id}",
+          "methods": [
+            "GET"
+          ],
+          "parts": {
+            "id": {
+              "type": "string",
+              "description": "The query ID"
+            }
+          }
+        }
+      ]
+    }
+  }
+}

rest-api-spec/src/main/resources/rest-api-spec/api/esql.list_queries.json

@@ -0,0 +1,26 @@
+{
+  "esql.list_queries": {
+    "documentation": {
+      "url": null,
+      "description": "Executes a list ESQL queries request"
+    },
+    "stability": "experimental",
+    "visibility": "public",
+    "headers": {
+      "accept": [],
+      "content_type": [
+        "application/json"
+      ]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/_query/queries",
+          "methods": [
+            "GET"
+          ]
+        }
+      ]
+    }
+  }
+}

test/framework/src/main/java/org/elasticsearch/test/IntOrLongMatcher.java

@@ -0,0 +1,79 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.test;
+
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.isA;
+
+/**
+ * A type-agnostic way of comparing integer values, not caring if it's a long or an integer.
+ */
+public abstract sealed class IntOrLongMatcher<T> extends BaseMatcher<T> {
+    public static IntOrLongMatcher<Integer> matches(int expected) {
+        return new IntMatcher(expected);
+    }
+
+    public static IntOrLongMatcher<Long> matches(long expected) {
+        return new LongMatcher(expected);
+    }
+
+    private static final class IntMatcher extends IntOrLongMatcher<Integer> {
+        private final int expected;
+
+        private IntMatcher(int expected) {
+            this.expected = expected;
+        }
+
+        @Override
+        public boolean matches(Object o) {
+            return switch (o) {
+                case Integer i -> expected == i;
+                case Long l -> expected == l;
+                default -> false;
+            };
+        }
+
+        @Override
+        public void describeTo(Description description) {
+            equalTo(expected).describeTo(description);
+        }
+    }
+
+    private static final class LongMatcher extends IntOrLongMatcher<Long> {
+        private final long expected;
+
+        LongMatcher(long expected) {
+            this.expected = expected;
+        }
+
+        @Override
+        public boolean matches(Object o) {
+            return switch (o) {
+                case Integer i -> expected == i;
+                case Long l -> expected == l;
+                default -> false;
+            };
+        }
+
+        @Override
+        public void describeTo(Description description) {
+            equalTo(expected).describeTo(description);
+        }
+    }
+
+    public static Matcher<Object> isIntOrLong() {
+        return anyOf(isA(Integer.class), isA(Long.class));
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ClientHelper.java

@@ -196,6 +196,7 @@ private static String maybeRewriteSingleAuthenticationHeaderForVersion(
     public static final String APM_ORIGIN = "apm";
     public static final String OTEL_ORIGIN = "otel";
     public static final String REINDEX_DATA_STREAM_ORIGIN = "reindex_data_stream";
+    public static final String ESQL_ORIGIN = "esql";
 
     private ClientHelper() {}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolver.java

@@ -110,6 +110,7 @@ public class ClusterPrivilegeResolver {
     private static final Set<String> MONITOR_WATCHER_PATTERN = Set.of("cluster:monitor/xpack/watcher/*");
     private static final Set<String> MONITOR_ROLLUP_PATTERN = Set.of("cluster:monitor/xpack/rollup/*");
     private static final Set<String> MONITOR_ENRICH_PATTERN = Set.of("cluster:monitor/xpack/enrich/*", "cluster:admin/xpack/enrich/get");
+    private static final Set<String> MONITOR_ESQL_PATTERN = Set.of("cluster:monitor/xpack/esql/*");
     // intentionally cluster:monitor/stats* to match cluster:monitor/stats, cluster:monitor/stats[n] and cluster:monitor/stats/remote
     private static final Set<String> MONITOR_STATS_PATTERN = Set.of("cluster:monitor/stats*");
 
@@ -249,6 +250,7 @@ public class ClusterPrivilegeResolver {
     public static final NamedClusterPrivilege MONITOR_WATCHER = new ActionClusterPrivilege("monitor_watcher", MONITOR_WATCHER_PATTERN);
     public static final NamedClusterPrivilege MONITOR_ROLLUP = new ActionClusterPrivilege("monitor_rollup", MONITOR_ROLLUP_PATTERN);
     public static final NamedClusterPrivilege MONITOR_ENRICH = new ActionClusterPrivilege("monitor_enrich", MONITOR_ENRICH_PATTERN);
+    public static final NamedClusterPrivilege MONITOR_ESQL = new ActionClusterPrivilege("monitor_esql", MONITOR_ESQL_PATTERN);
     public static final NamedClusterPrivilege MONITOR_STATS = new ActionClusterPrivilege("monitor_stats", MONITOR_STATS_PATTERN);
     public static final NamedClusterPrivilege MANAGE = new ActionClusterPrivilege("manage", ALL_CLUSTER_PATTERN, ALL_SECURITY_PATTERN);
     public static final NamedClusterPrivilege MANAGE_INFERENCE = new ActionClusterPrivilege("manage_inference", MANAGE_INFERENCE_PATTERN);
@@ -431,6 +433,7 @@ public class ClusterPrivilegeResolver {
             MONITOR_WATCHER,
             MONITOR_ROLLUP,
             MONITOR_ENRICH,
+            MONITOR_ESQL,
             MONITOR_STATS,
             MANAGE,
             MANAGE_CONNECTOR,

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/async/AsyncTaskManagementService.java

@@ -172,6 +172,8 @@ public AsyncTaskManagementService(
         this.threadPool = threadPool;
     }
 
+    public static String ASYNC_ACTION_SUFFIX = "[a]";
+
     public void asyncExecute(
         Request request,
         TimeValue waitForCompletionTimeout,
@@ -182,7 +184,7 @@ public void asyncExecute(
         String nodeId = clusterService.localNode().getId();
         try (var ignored = threadPool.getThreadContext().newTraceContext()) {
             @SuppressWarnings("unchecked")
-            T searchTask = (T) taskManager.register("transport", action + "[a]", new AsyncRequestWrapper(request, nodeId));
+            T searchTask = (T) taskManager.register("transport", action + ASYNC_ACTION_SUFFIX, new AsyncRequestWrapper(request, nodeId));
             boolean operationStarted = false;
             try {
                 operation.execute(

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlSecurityIT.java

@@ -42,6 +42,7 @@
 import static org.elasticsearch.test.MapMatcher.matchesMap;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
 
 public class EsqlSecurityIT extends ESRestTestCase {
     @ClassRule
@@ -69,6 +70,8 @@ public class EsqlSecurityIT extends ESRestTestCase {
         .user("logs_foo_after_2021", "x-pack-test-password", "logs_foo_after_2021", false)
         .user("logs_foo_after_2021_pattern", "x-pack-test-password", "logs_foo_after_2021_pattern", false)
         .user("logs_foo_after_2021_alias", "x-pack-test-password", "logs_foo_after_2021_alias", false)
+        .user("user_without_monitor_privileges", "x-pack-test-password", "user_without_monitor_privileges", false)
+        .user("user_with_monitor_privileges", "x-pack-test-password", "user_with_monitor_privileges", false)
         .build();
 
     @Override
@@ -309,7 +312,7 @@ public void testIndexPatternErrorMessageComparison_ESQL_SearchDSL() throws Excep
         json.endObject();
         Request searchRequest = new Request("GET", "/index-user1,index-user2/_search");
         searchRequest.setJsonEntity(Strings.toString(json));
-        searchRequest.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("es-security-runas-user", "metadata1_read2"));
+        setUser(searchRequest, "metadata1_read2");
 
         // ES|QL query on the same index pattern
         var esqlResp = expectThrows(ResponseException.class, () -> runESQLCommand("metadata1_read2", "FROM index-user1,index-user2"));
@@ -429,13 +432,13 @@ public void testFieldLevelSecurityAllow() throws Exception {
 
     public void testFieldLevelSecurityAllowPartial() throws Exception {
         Request request = new Request("GET", "/index*/_field_caps");
-        request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("es-security-runas-user", "fls_user"));
+        setUser(request, "fls_user");
         request.addParameter("error_trace", "true");
         request.addParameter("pretty", "true");
         request.addParameter("fields", "*");
 
         request = new Request("GET", "/index*/_search");
-        request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("es-security-runas-user", "fls_user"));
+        setUser(request, "fls_user");
         request.addParameter("error_trace", "true");
         request.addParameter("pretty", "true");
 
@@ -761,6 +764,36 @@ public void testFromLookupIndexForbidden() throws Exception {
         assertThat(resp.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
     }
 
+    public void testListQueryAllowed() throws Exception {
+        Request request = new Request("GET", "_query/queries");
+        setUser(request, "user_with_monitor_privileges");
+        var resp = client().performRequest(request);
+        assertOK(resp);
+    }
+
+    public void testListQueryForbidden() throws Exception {
+        Request request = new Request("GET", "_query/queries");
+        setUser(request, "user_without_monitor_privileges");
+        var resp = expectThrows(ResponseException.class, () -> client().performRequest(request));
+        assertThat(resp.getResponse().getStatusLine().getStatusCode(), equalTo(403));
+        assertThat(resp.getMessage(), containsString("this action is granted by the cluster privileges [monitor_esql,monitor,manage,all]"));
+    }
+
+    public void testGetQueryAllowed() throws Exception {
+        // This is a bit tricky, since there is no such running query. We just make sure it didn't fail on forbidden privileges.
+        Request request = new Request("GET", "_query/queries/foo:1234");
+        var resp = expectThrows(ResponseException.class, () -> client().performRequest(request));
+        assertThat(resp.getResponse().getStatusLine().getStatusCode(), not(equalTo(404)));
+    }
+
+    public void testGetQueryForbidden() throws Exception {
+        Request request = new Request("GET", "_query/queries/foo:1234");
+        setUser(request, "user_without_monitor_privileges");
+        var resp = expectThrows(ResponseException.class, () -> client().performRequest(request));
+        assertThat(resp.getResponse().getStatusLine().getStatusCode(), equalTo(403));
+        assertThat(resp.getMessage(), containsString("this action is granted by the cluster privileges [monitor_esql,monitor,manage,all]"));
+    }
+
     private void createEnrichPolicy() throws Exception {
         createIndex("songs", Settings.EMPTY, """
             "properties":{"song_id": {"type": "keyword"}, "title": {"type": "keyword"}, "artist": {"type": "keyword"} }
@@ -837,11 +870,16 @@ protected Response runESQLCommand(String user, String command) throws IOExceptio
         json.endObject();
         Request request = new Request("POST", "_query");
         request.setJsonEntity(Strings.toString(json));
-        request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("es-security-runas-user", user));
+        setUser(request, user);
         request.addParameter("error_trace", "true");
         return client().performRequest(request);
     }
 
+    private static void setUser(Request request, String user) {
+        request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("es-security-runas-user", user));
+
+    }
+
     static void addRandomPragmas(XContentBuilder builder) throws IOException {
         if (Build.current().isSnapshot()) {
             Settings pragmas = randomPragmas();
@@ -853,7 +891,7 @@ static void addRandomPragmas(XContentBuilder builder) throws IOException {
         }
     }
 
-    static Settings randomPragmas() {
+    private static Settings randomPragmas() {
         Settings.Builder settings = Settings.builder();
         if (randomBoolean()) {
             settings.put("page_size", between(1, 5));

x-pack/plugin/esql/qa/security/src/javaRestTest/resources/roles.yml

@@ -193,3 +193,10 @@ logs_foo_after_2021_alias:
             "@timestamp": {"gte": "2021-01-01T00:00:00"}
           }
         }
+
+user_without_monitor_privileges:
+  cluster: []
+
+user_with_monitor_privileges:
+  cluster:
+    - monitor_esql