Add http-only headers to ElasticsearchException (#130348)

When return an error from Elasticsearch exceptions may contain values
written as http response headers. ElasticsearchException contains a map
of headers that are added to the response. But these values are also
written to a special "header" section of the response body.

This commit renames the existing "headers" in ElasticsearchException to
"body headers", which are both http headers and written to the response
body. A new "http headers" is added for headers that should only be
written as response headers.

Author: rjernst

modules/reindex/src/test/java/org/elasticsearch/reindex/ReindexFromRemoteWithAuthTests.java

@@ -209,7 +209,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void app
             String auth = context.getHeader(AUTHORIZATION_HEADER);
             if (auth == null) {
                 ElasticsearchSecurityException e = new ElasticsearchSecurityException("Authentication required", RestStatus.UNAUTHORIZED);
-                e.addHeader("WWW-Authenticate", "Basic realm=auth-realm");
+                e.addBodyHeader("WWW-Authenticate", "Basic realm=auth-realm");
                 throw e;
             }
             if (false == REQUIRED_AUTH.equals(auth)) {

server/src/internalClusterTest/java/org/elasticsearch/ingest/IngestClientIT.java

@@ -322,8 +322,8 @@ public void testPipelineOriginHeader() throws Exception {
             client().index(indexRequest).get();
         });
         IngestProcessorException ingestException = (IngestProcessorException) e.getCause();
-        assertThat(ingestException.getHeader("processor_type"), equalTo(List.of("fail")));
-        assertThat(ingestException.getHeader("pipeline_origin"), equalTo(List.of("3", "2", "1")));
+        assertThat(ingestException.getBodyHeader("processor_type"), equalTo(List.of("fail")));
+        assertThat(ingestException.getBodyHeader("pipeline_origin"), equalTo(List.of("3", "2", "1")));
     }
 
     public void testPipelineProcessorOnFailure() throws Exception {

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -122,7 +122,8 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     private static final Map<Integer, CheckedFunction<StreamInput, ? extends ElasticsearchException, IOException>> ID_TO_SUPPLIER;
     private static final Map<Class<? extends ElasticsearchException>, ElasticsearchExceptionHandle> CLASS_TO_ELASTICSEARCH_EXCEPTION_HANDLE;
     private final Map<String, List<String>> metadata = new HashMap<>();
-    private final Map<String, List<String>> headers = new HashMap<>();
+    private final Map<String, List<String>> bodyHeaders = new HashMap<>();
+    private final Map<String, List<String>> httpHeaders = new HashMap<>();
 
     /**
      * Construct a <code>ElasticsearchException</code> with the specified cause exception.
@@ -169,14 +170,14 @@ public ElasticsearchException(String msg, Throwable cause, Object... args) {
     public ElasticsearchException(StreamInput in) throws IOException {
         super(in.readOptionalString(), in.readException());
         readStackTrace(this, in);
-        headers.putAll(in.readMapOfLists(StreamInput::readString));
+        bodyHeaders.putAll(in.readMapOfLists(StreamInput::readString));
         metadata.putAll(in.readMapOfLists(StreamInput::readString));
     }
 
     private void maybePutTimeoutHeader() {
         if (isTimeout()) {
             // see https://www.rfc-editor.org/rfc/rfc8941.html#section-4.1.9 for booleans in structured headers
-            headers.put(TIMED_OUT_HEADER, List.of("?1"));
+            bodyHeaders.put(TIMED_OUT_HEADER, List.of("?1"));
         }
     }
 
@@ -220,42 +221,77 @@ protected Map<String, List<String>> getMetadata() {
     }
 
     /**
-     * Adds a new header with the given key.
+     * Adds a new header with the given key that is part of the response body and http headers.
      * This method will replace existing header if a header with the same key already exists
      */
-    public void addHeader(String key, List<String> value) {
+    public void addBodyHeader(String key, List<String> value) {
         // we need to enforce this otherwise bw comp doesn't work properly, as "es." was the previous criteria to split headers in two sets
         if (key.startsWith("es.")) {
             throw new IllegalArgumentException("exception headers must not start with [es.], found [" + key + "] instead");
         }
-        this.headers.put(key, value);
+        this.bodyHeaders.put(key, value);
     }
 
     /**
-     * Adds a new header with the given key.
+     * Adds a new header with the given key that is part of the response body and http headers.
      * This method will replace existing header if a header with the same key already exists
      */
-    public void addHeader(String key, String... value) {
-        addHeader(key, Arrays.asList(value));
+    public void addBodyHeader(String key, String... value) {
+        addBodyHeader(key, Arrays.asList(value));
     }
 
     /**
-     * Returns a set of all header keys on this exception
+     * Returns a set of all body header keys on this exception
      */
-    public Set<String> getHeaderKeys() {
-        return headers.keySet();
+    public Set<String> getBodyHeaderKeys() {
+        return bodyHeaders.keySet();
     }
 
     /**
-     * Returns the list of header values for the given key or {@code null} if no header for the
+     * Returns the list of body header values for the given key or {@code null} if no header for the
      * given key exists.
      */
-    public List<String> getHeader(String key) {
-        return headers.get(key);
+    public List<String> getBodyHeader(String key) {
+        return bodyHeaders.get(key);
     }
 
-    protected Map<String, List<String>> getHeaders() {
-        return headers;
+    protected Map<String, List<String>> getBodyHeaders() {
+        return bodyHeaders;
+    }
+
+    /**
+     * Adds a new http header with the given key.
+     * This method will replace existing http header if a header with the same key already exists
+     */
+    public void addHttpHeader(String key, List<String> value) {
+        this.httpHeaders.put(key, value);
+    }
+
+    /**
+     * Adds a new http header with the given key.
+     * This method will replace existing http header if a header with the same key already exists
+     */
+    public void addHttpHeader(String key, String... value) {
+        this.httpHeaders.put(key, List.of(value));
+    }
+
+    /**
+     * Returns a set of all body header keys on this exception
+     */
+    public Set<String> getHttpHeaderKeys() {
+        return httpHeaders.keySet();
+    }
+
+    /**
+     * Returns the list of http header values for the given key or {@code null} if no header for the
+     * given key exists.
+     */
+    public List<String> getHttpHeader(String key) {
+        return httpHeaders.get(key);
+    }
+
+    protected Map<String, List<String>> getHttpHeaders() {
+        return httpHeaders;
     }
 
     /**
@@ -335,7 +371,7 @@ private static Writer<Throwable> createNestingFunction(int thisLevel, Runnable n
     protected void writeTo(StreamOutput out, Writer<Throwable> nestedExceptionsWriter) throws IOException {
         out.writeOptionalString(this.getMessage());
         nestedExceptionsWriter.write(out, this);
-        out.writeMap(headers, StreamOutput::writeStringCollection);
+        out.writeMap(bodyHeaders, StreamOutput::writeStringCollection);
         out.writeMap(metadata, StreamOutput::writeStringCollection);
     }
 
@@ -384,7 +420,7 @@ protected XContentBuilder toXContent(XContentBuilder builder, Params params, int
         if (ex != this) {
             generateThrowableXContent(builder, params, this, nestedLevel);
         } else {
-            innerToXContent(builder, params, this, headers, metadata, getCause(), nestedLevel);
+            innerToXContent(builder, params, this, bodyHeaders, metadata, getCause(), nestedLevel);
         }
         return builder;
     }
@@ -581,7 +617,7 @@ public static ElasticsearchException innerFromXContent(XContentParser parser, bo
             e.addMetadata("es." + entry.getKey(), entry.getValue());
         }
         for (Map.Entry<String, List<String>> header : headers.entrySet()) {
-            e.addHeader(header.getKey(), header.getValue());
+            e.addBodyHeader(header.getKey(), header.getValue());
         }
 
         // Adds root causes as suppressed exception. This way they are not lost

server/src/main/java/org/elasticsearch/action/bulk/FailureStoreDocumentConverter.java

@@ -115,22 +115,23 @@ private static XContentBuilder createSource(IndexRequest source, Exception excep
                 // we can't instantiate it in tests, so we'll have to check for the headers directly.
                 var ingestException = ExceptionsHelper.<ElasticsearchException>unwrapCausesAndSuppressed(
                     exception,
-                    t -> t instanceof ElasticsearchException e && Sets.haveNonEmptyIntersection(e.getHeaderKeys(), INGEST_EXCEPTION_HEADERS)
+                    t -> t instanceof ElasticsearchException e
+                        && Sets.haveNonEmptyIntersection(e.getBodyHeaderKeys(), INGEST_EXCEPTION_HEADERS)
                 ).orElse(null);
                 if (ingestException != null) {
-                    if (ingestException.getHeaderKeys().contains(PIPELINE_ORIGIN_EXCEPTION_HEADER)) {
-                        List<String> pipelineOrigin = ingestException.getHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER);
+                    if (ingestException.getBodyHeaderKeys().contains(PIPELINE_ORIGIN_EXCEPTION_HEADER)) {
+                        List<String> pipelineOrigin = ingestException.getBodyHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER);
                         Collections.reverse(pipelineOrigin);
                         if (pipelineOrigin.isEmpty() == false) {
                             builder.field("pipeline_trace", pipelineOrigin);
                             builder.field("pipeline", pipelineOrigin.get(pipelineOrigin.size() - 1));
                         }
                     }
-                    if (ingestException.getHeaderKeys().contains(PROCESSOR_TAG_EXCEPTION_HEADER)) {
-                        builder.field("processor_tag", ingestException.getHeader(PROCESSOR_TAG_EXCEPTION_HEADER).get(0));
+                    if (ingestException.getBodyHeaderKeys().contains(PROCESSOR_TAG_EXCEPTION_HEADER)) {
+                        builder.field("processor_tag", ingestException.getBodyHeader(PROCESSOR_TAG_EXCEPTION_HEADER).get(0));
                     }
-                    if (ingestException.getHeaderKeys().contains(PROCESSOR_TYPE_EXCEPTION_HEADER)) {
-                        builder.field("processor_type", ingestException.getHeader(PROCESSOR_TYPE_EXCEPTION_HEADER).get(0));
+                    if (ingestException.getBodyHeaderKeys().contains(PROCESSOR_TYPE_EXCEPTION_HEADER)) {
+                        builder.field("processor_type", ingestException.getBodyHeader(PROCESSOR_TYPE_EXCEPTION_HEADER).get(0));
                     }
                 }
             }

server/src/main/java/org/elasticsearch/action/search/SearchPhaseExecutionException.java

@@ -128,7 +128,7 @@ protected XContentBuilder toXContent(XContentBuilder builder, Params params, int
             // We don't have a cause when all shards failed, but we do have shards failures so we can "guess" a cause
             // (see {@link #getCause()}). Here, we use super.getCause() because we don't want the guessed exception to
             // be rendered twice (one in the "cause" field, one in "failed_shards")
-            innerToXContent(builder, params, this, getHeaders(), getMetadata(), super.getCause(), nestedLevel);
+            innerToXContent(builder, params, this, getBodyHeaders(), getMetadata(), super.getCause(), nestedLevel);
         }
         return builder;
     }

server/src/main/java/org/elasticsearch/common/io/stream/NotSerializableExceptionWrapper.java

@@ -36,8 +36,8 @@ public NotSerializableExceptionWrapper(Throwable other) {
             addSuppressed(otherSuppressed);
         }
         if (other instanceof ElasticsearchException ex) {
-            for (String key : ex.getHeaderKeys()) {
-                this.addHeader(key, ex.getHeader(key));
+            for (String key : ex.getBodyHeaderKeys()) {
+                this.addBodyHeader(key, ex.getBodyHeader(key));
             }
             for (String key : ex.getMetadataKeys()) {
                 this.addMetadata(key, ex.getMetadata(key));

server/src/main/java/org/elasticsearch/ingest/CompoundProcessor.java

@@ -292,9 +292,9 @@ void executeOnFailure(
     }
 
     private static void putFailureMetadata(IngestDocument ingestDocument, ElasticsearchException cause) {
-        List<String> processorTypeHeader = cause.getHeader(PROCESSOR_TYPE_EXCEPTION_HEADER);
-        List<String> processorTagHeader = cause.getHeader(PROCESSOR_TAG_EXCEPTION_HEADER);
-        List<String> processorOriginHeader = cause.getHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER);
+        List<String> processorTypeHeader = cause.getBodyHeader(PROCESSOR_TYPE_EXCEPTION_HEADER);
+        List<String> processorTagHeader = cause.getBodyHeader(PROCESSOR_TAG_EXCEPTION_HEADER);
+        List<String> processorOriginHeader = cause.getBodyHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER);
         String failedProcessorType = (processorTypeHeader != null) ? processorTypeHeader.get(0) : null;
         String failedProcessorTag = (processorTagHeader != null) ? processorTagHeader.get(0) : null;
         String failedPipelineId = (processorOriginHeader != null) ? processorOriginHeader.get(0) : null;
@@ -316,24 +316,24 @@ private static void removeFailureMetadata(IngestDocument ingestDocument) {
     }
 
     static IngestProcessorException newCompoundProcessorException(Exception e, Processor processor, IngestDocument document) {
-        if (e instanceof IngestProcessorException ipe && ipe.getHeader(PROCESSOR_TYPE_EXCEPTION_HEADER) != null) {
+        if (e instanceof IngestProcessorException ipe && ipe.getBodyHeader(PROCESSOR_TYPE_EXCEPTION_HEADER) != null) {
             return ipe;
         }
 
         IngestProcessorException exception = new IngestProcessorException(e);
 
         String processorType = processor.getType();
         if (processorType != null) {
-            exception.addHeader(PROCESSOR_TYPE_EXCEPTION_HEADER, processorType);
+            exception.addBodyHeader(PROCESSOR_TYPE_EXCEPTION_HEADER, processorType);
         }
         String processorTag = processor.getTag();
         if (processorTag != null) {
-            exception.addHeader(PROCESSOR_TAG_EXCEPTION_HEADER, processorTag);
+            exception.addBodyHeader(PROCESSOR_TAG_EXCEPTION_HEADER, processorTag);
         }
         if (document != null) {
             List<String> pipelineStack = document.getPipelineStack();
             if (pipelineStack.isEmpty() == false) {
-                exception.addHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER, pipelineStack);
+                exception.addBodyHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER, pipelineStack);
             }
         }
 

server/src/main/java/org/elasticsearch/ingest/IngestPipelineException.java

@@ -28,7 +28,7 @@ public class IngestPipelineException extends ElasticsearchException implements E
 
     IngestPipelineException(final String pipeline, final Exception cause) {
         super(cause);
-        this.addHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER, List.of(pipeline));
+        this.addBodyHeader(PIPELINE_ORIGIN_EXCEPTION_HEADER, List.of(pipeline));
     }
 
     public IngestPipelineException(final StreamInput in) throws IOException {

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -1604,8 +1604,8 @@ public synchronized void reloadPipeline(ProjectId projectId, String id) throws E
     }
 
     private static Pipeline substitutePipeline(String id, ElasticsearchParseException e) {
-        String tag = e.getHeaderKeys().contains("processor_tag") ? e.getHeader("processor_tag").get(0) : null;
-        String type = e.getHeaderKeys().contains("processor_type") ? e.getHeader("processor_type").get(0) : "unknown";
+        String tag = e.getBodyHeaderKeys().contains("processor_tag") ? e.getBodyHeader("processor_tag").get(0) : null;
+        String type = e.getBodyHeaderKeys().contains("processor_type") ? e.getBodyHeader("processor_type").get(0) : "unknown";
         String errorMessage = "pipeline with id [" + id + "] could not be loaded, caused by [" + e.getDetailedMessage() + "]";
         Processor failureProcessor = new AbstractProcessor(tag, "this is a placeholder processor") {
             @Override

server/src/main/java/org/elasticsearch/rest/RestResponse.java

@@ -212,12 +212,16 @@ static RestResponse createSimpleErrorResponse(RestChannel channel, RestStatus st
     }
 
     public void copyHeaders(ElasticsearchException ex) {
-        Set<String> headerKeySet = ex.getHeaderKeys();
+        Set<String> bodyHeaderKeySet = ex.getBodyHeaderKeys();
+        Set<String> httpHeaderKeySet = ex.getHttpHeaderKeys();
         if (customHeaders == null) {
-            customHeaders = Maps.newMapWithExpectedSize(headerKeySet.size());
+            customHeaders = Maps.newMapWithExpectedSize(bodyHeaderKeySet.size() + httpHeaderKeySet.size());
         }
-        for (String key : headerKeySet) {
-            customHeaders.computeIfAbsent(key, k -> new ArrayList<>()).addAll(ex.getHeader(key));
+        for (String key : bodyHeaderKeySet) {
+            customHeaders.computeIfAbsent(key, k -> new ArrayList<>()).addAll(ex.getBodyHeader(key));
+        }
+        for (String key : httpHeaderKeySet) {
+            customHeaders.computeIfAbsent(key, k -> new ArrayList<>()).addAll(ex.getHttpHeader(key));
         }
     }