Add x-opaque-id to search slow logs (#31539)

Add x-opaque-id to search slow logs only. Indexing slow log and audit
logs will be handled as separate PRs.

Relates #31521

Author: imotov

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpClient.java

@@ -44,6 +44,7 @@
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.tasks.Task;
 
 import java.io.Closeable;
 import java.net.SocketAddress;
@@ -74,7 +75,7 @@ static Collection<String> returnHttpResponseBodies(Collection<FullHttpResponse>
     static Collection<String> returnOpaqueIds(Collection<FullHttpResponse> responses) {
         List<String> list = new ArrayList<>(responses.size());
         for (HttpResponse response : responses) {
-            list.add(response.headers().get("X-Opaque-Id"));
+            list.add(response.headers().get(Task.X_OPAQUE_ID));
         }
         return list;
     }

plugins/transport-nio/src/test/java/org/elasticsearch/http/nio/Netty4HttpClient.java

@@ -44,6 +44,7 @@
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.tasks.Task;
 
 import java.io.Closeable;
 import java.net.SocketAddress;
@@ -74,7 +75,7 @@ static Collection<String> returnHttpResponseBodies(Collection<FullHttpResponse>
     static Collection<String> returnOpaqueIds(Collection<FullHttpResponse> responses) {
         List<String> list = new ArrayList<>(responses.size());
         for (HttpResponse response : responses) {
-            list.add(response.headers().get("X-Opaque-Id"));
+            list.add(response.headers().get(Task.X_OPAQUE_ID));
         }
         return list;
     }
@@ -90,7 +91,7 @@ public Collection<FullHttpResponse> get(SocketAddress remoteAddress, String... u
         for (int i = 0; i < uris.length; i++) {
             final HttpRequest httpRequest = new DefaultFullHttpRequest(HTTP_1_1, HttpMethod.GET, uris[i]);
             httpRequest.headers().add(HOST, "localhost");
-            httpRequest.headers().add("X-Opaque-ID", String.valueOf(i));
+            httpRequest.headers().add(Task.X_OPAQUE_ID, String.valueOf(i));
             requests.add(httpRequest);
         }
         return sendRequests(remoteAddress, requests);

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -316,6 +316,7 @@
 import org.elasticsearch.rest.action.search.RestMultiSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchScrollAction;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.usage.UsageService;
 
@@ -369,7 +370,7 @@ public ActionModule(boolean transportClient, Settings settings, IndexNameExpress
         destructiveOperations = new DestructiveOperations(settings, clusterSettings);
         Set<String> headers = Stream.concat(
             actionPlugins.stream().flatMap(p -> p.getRestHeaders().stream()),
-            Stream.of("X-Opaque-Id")
+            Stream.of(Task.X_OPAQUE_ID)
         ).collect(Collectors.toSet());
         UnaryOperator<RestHandler> restWrapper = null;
         for (ActionPlugin plugin : actionPlugins) {

server/src/main/java/org/elasticsearch/http/DefaultRestChannel.java

@@ -38,6 +38,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.tasks.Task.X_OPAQUE_ID;
+
 /**
  * The default rest channel for incoming requests. This class implements the basic logic for sending a rest
  * response. It will set necessary headers nad ensure that bytes are released after the response is sent.
@@ -50,7 +52,6 @@ public class DefaultRestChannel extends AbstractRestChannel implements RestChann
     static final String CONTENT_TYPE = "content-type";
     static final String CONTENT_LENGTH = "content-length";
     static final String SET_COOKIE = "set-cookie";
-    static final String X_OPAQUE_ID = "X-Opaque-Id";
 
     private final HttpRequest httpRequest;
     private final BigArrays bigArrays;

server/src/main/java/org/elasticsearch/index/SearchSlowLog.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.index.shard.SearchOperationListener;
 import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.tasks.Task;
 
 import java.util.Collections;
 import java.util.concurrent.TimeUnit;
@@ -174,6 +175,11 @@ public String toString() {
             } else {
                 sb.append("source[], ");
             }
+            if (context.getTask().getHeader(Task.X_OPAQUE_ID) != null) {
+                sb.append("id[").append(context.getTask().getHeader(Task.X_OPAQUE_ID)).append("], ");
+            } else {
+                sb.append("id[], ");
+            }
             return sb.toString();
         }
     }

server/src/main/java/org/elasticsearch/node/Node.java

@@ -136,6 +136,7 @@
 import org.elasticsearch.search.fetch.FetchPhase;
 import org.elasticsearch.snapshots.SnapshotShardsService;
 import org.elasticsearch.snapshots.SnapshotsService;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskResultsService;
 import org.elasticsearch.threadpool.ExecutorBuilder;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -447,7 +448,7 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
             final Transport transport = networkModule.getTransportSupplier().get();
             Set<String> taskHeaders = Stream.concat(
                 pluginsService.filterPlugins(ActionPlugin.class).stream().flatMap(p -> p.getTaskHeaders().stream()),
-                Stream.of("X-Opaque-Id")
+                Stream.of(Task.X_OPAQUE_ID)
             ).collect(Collectors.toSet());
             final TransportService transportService = newTransportService(settings, transport, threadPool,
                 networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings(), taskHeaders);

server/src/main/java/org/elasticsearch/tasks/Task.java

@@ -34,6 +34,11 @@
  */
 public class Task {
 
+    /**
+     * The request header to mark tasks with specific ids
+     */
+    public static final String X_OPAQUE_ID = "X-Opaque-Id";
+
     private final long id;
 
     private final String type;

server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java

@@ -357,7 +357,7 @@ public void testSearchTaskDescriptions() {
             .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
 
         Map<String, String> headers = new HashMap<>();
-        headers.put("X-Opaque-Id", "my_id");
+        headers.put(Task.X_OPAQUE_ID, "my_id");
         headers.put("Foo-Header", "bar");
         headers.put("Custom-Task-Header", "my_value");
         assertSearchResponse(
@@ -404,7 +404,7 @@ public void testSearchTaskHeaderLimit() {
         int maxSize = Math.toIntExact(SETTING_HTTP_MAX_HEADER_SIZE.getDefault(Settings.EMPTY).getBytes() / 2 + 1);
 
         Map<String, String> headers = new HashMap<>();
-        headers.put("X-Opaque-Id", "my_id");
+        headers.put(Task.X_OPAQUE_ID, "my_id");
         headers.put("Custom-Task-Header", randomAlphaOfLengthBetween(maxSize, maxSize + 100));
         IllegalArgumentException ex = expectThrows(
             IllegalArgumentException.class,
@@ -415,7 +415,7 @@ public void testSearchTaskHeaderLimit() {
 
     private void assertTaskHeaders(TaskInfo taskInfo) {
         assertThat(taskInfo.getHeaders().keySet(), hasSize(2));
-        assertEquals("my_id", taskInfo.getHeaders().get("X-Opaque-Id"));
+        assertEquals("my_id", taskInfo.getHeaders().get(Task.X_OPAQUE_ID));
         assertEquals("my_value", taskInfo.getHeaders().get("Custom-Task-Header"));
     }
 

server/src/test/java/org/elasticsearch/http/DefaultRestChannelTests.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -180,7 +181,7 @@ public void testResponse() {
     public void testHeadersSet() {
         Settings settings = Settings.builder().build();
         final TestRequest httpRequest = new TestRequest(HttpRequest.HttpVersion.HTTP_1_1, RestRequest.Method.GET, "/");
-        httpRequest.getHeaders().put(DefaultRestChannel.X_OPAQUE_ID, Collections.singletonList("abc"));
+        httpRequest.getHeaders().put(Task.X_OPAQUE_ID, Collections.singletonList("abc"));
         final RestRequest request = RestRequest.request(xContentRegistry(), httpRequest, httpChannel);
         HttpHandlingSettings handlingSettings = HttpHandlingSettings.fromSettings(settings);
 
@@ -200,15 +201,15 @@ public void testHeadersSet() {
         Map<String, List<String>> headers = httpResponse.headers;
         assertNull(headers.get("non-existent-header"));
         assertEquals(customHeaderValue, headers.get(customHeader).get(0));
-        assertEquals("abc", headers.get(DefaultRestChannel.X_OPAQUE_ID).get(0));
+        assertEquals("abc", headers.get(Task.X_OPAQUE_ID).get(0));
         assertEquals(Integer.toString(resp.content().length()), headers.get(DefaultRestChannel.CONTENT_LENGTH).get(0));
         assertEquals(resp.contentType(), headers.get(DefaultRestChannel.CONTENT_TYPE).get(0));
     }
 
     public void testCookiesSet() {
         Settings settings = Settings.builder().put(HttpTransportSettings.SETTING_HTTP_RESET_COOKIES.getKey(), true).build();
         final TestRequest httpRequest = new TestRequest(HttpRequest.HttpVersion.HTTP_1_1, RestRequest.Method.GET, "/");
-        httpRequest.getHeaders().put(DefaultRestChannel.X_OPAQUE_ID, Collections.singletonList("abc"));
+        httpRequest.getHeaders().put(Task.X_OPAQUE_ID, Collections.singletonList("abc"));
         final RestRequest request = RestRequest.request(xContentRegistry(), httpRequest, httpChannel);
         HttpHandlingSettings handlingSettings = HttpHandlingSettings.fromSettings(settings);
 

server/src/test/java/org/elasticsearch/index/SearchSlowLogTests.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.index;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.action.search.SearchTask;
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -34,12 +35,15 @@
 import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.search.internal.ShardSearchRequest;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESSingleNodeTestCase;
 import org.elasticsearch.test.TestSearchContext;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
+import java.util.Collections;
 
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.hasToString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.not;
@@ -166,10 +170,12 @@ public void testSlowLogSearchContextPrinterToLog() throws IOException {
         SearchContext searchContext = createSearchContext(index);
         SearchSourceBuilder source = SearchSourceBuilder.searchSource().query(QueryBuilders.matchAllQuery());
         searchContext.request().source(source);
+        searchContext.setTask(new SearchTask(0, "n/a", "n/a", "test", null, Collections.singletonMap(Task.X_OPAQUE_ID, "my_id")));
         SearchSlowLog.SlowLogSearchContextPrinter p = new SearchSlowLog.SlowLogSearchContextPrinter(searchContext, 10);
         assertThat(p.toString(), startsWith("[foo][0]"));
         // Makes sure that output doesn't contain any new lines
         assertThat(p.toString(), not(containsString("\n")));
+        assertThat(p.toString(), endsWith("id[my_id], "));
     }
 
     public void testLevelSetting() {