Allow customization of netty channel handles before and during decompression

CHANGELOG.md

@@ -87,6 +87,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Telemetry-Otel] Added support for OtlpGrpcSpanExporter exporter ([#9666](https://github.com/opensearch-project/OpenSearch/pull/9666))
 - Async blob read support for encrypted containers ([#10131](https://github.com/opensearch-project/OpenSearch/pull/10131))
 - Add capability to restrict async durability mode for remote indexes ([#10189](https://github.com/opensearch-project/OpenSearch/pull/10189))
+- Improve compressed request handling ([#10261](https://github.com/opensearch-project/OpenSearch/pull/10261))
 
 ### Dependencies
 - Bump `peter-evans/create-or-update-comment` from 2 to 3 ([#9575](https://github.com/opensearch-project/OpenSearch/pull/9575))

modules/transport-netty4/src/main/java/org/opensearch/http/netty4/AbstractNetty4HttpRequest.java

@@ -0,0 +1,142 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.http.netty4;
+
+import org.opensearch.rest.RestRequest;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+
+public class AbstractNetty4HttpRequest {
+
+    protected HttpHeadersMap headers;
+    protected Exception inboundException;
+
+    protected RestRequest.Method getHttpMethod(HttpRequest request) {
+        HttpMethod httpMethod = request.method();
+        if (httpMethod == HttpMethod.GET) return RestRequest.Method.GET;
+
+        if (httpMethod == HttpMethod.POST) return RestRequest.Method.POST;
+
+        if (httpMethod == HttpMethod.PUT) return RestRequest.Method.PUT;
+
+        if (httpMethod == HttpMethod.DELETE) return RestRequest.Method.DELETE;
+
+        if (httpMethod == HttpMethod.HEAD) {
+            return RestRequest.Method.HEAD;
+        }
+
+        if (httpMethod == HttpMethod.OPTIONS) {
+            return RestRequest.Method.OPTIONS;
+        }
+
+        if (httpMethod == HttpMethod.PATCH) {
+            return RestRequest.Method.PATCH;
+        }
+
+        if (httpMethod == HttpMethod.TRACE) {
+            return RestRequest.Method.TRACE;
+        }
+
+        if (httpMethod == HttpMethod.CONNECT) {
+            return RestRequest.Method.CONNECT;
+        }
+
+        throw new IllegalArgumentException("Unexpected http method: " + httpMethod);
+    }
+
+    /**
+     * A wrapper of {@link HttpHeaders} that implements a map to prevent copying unnecessarily. This class does not support modifications
+     * and due to the underlying implementation, it performs case insensitive lookups of key to values.
+     *
+     * It is important to note that this implementation does have some downsides in that each invocation of the
+     * {@link #values()} and {@link #entrySet()} methods will perform a copy of the values in the HttpHeaders rather than returning a
+     * view of the underlying values.
+     */
+    protected static class HttpHeadersMap implements Map<String, List<String>> {
+
+        private final HttpHeaders httpHeaders;
+
+        HttpHeadersMap(HttpHeaders httpHeaders) {
+            this.httpHeaders = httpHeaders;
+        }
+
+        @Override
+        public int size() {
+            return httpHeaders.size();
+        }
+
+        @Override
+        public boolean isEmpty() {
+            return httpHeaders.isEmpty();
+        }
+
+        @Override
+        public boolean containsKey(Object key) {
+            return key instanceof String && httpHeaders.contains((String) key);
+        }
+
+        @Override
+        public boolean containsValue(Object value) {
+            return value instanceof List && httpHeaders.names().stream().map(httpHeaders::getAll).anyMatch(value::equals);
+        }
+
+        @Override
+        public List<String> get(Object key) {
+            return key instanceof String ? httpHeaders.getAll((String) key) : null;
+        }
+
+        @Override
+        public List<String> put(String key, List<String> value) {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public List<String> remove(Object key) {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public void putAll(Map<? extends String, ? extends List<String>> m) {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public void clear() {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public Set<String> keySet() {
+            return httpHeaders.names();
+        }
+
+        @Override
+        public Collection<List<String>> values() {
+            return httpHeaders.names().stream().map(k -> Collections.unmodifiableList(httpHeaders.getAll(k))).collect(Collectors.toList());
+        }
+
+        @Override
+        public Set<Entry<String, List<String>>> entrySet() {
+            return httpHeaders.names()
+                .stream()
+                .map(k -> new AbstractMap.SimpleImmutableEntry<>(k, httpHeaders.getAll(k)))
+                .collect(Collectors.toSet());
+        }
+    }
+}

modules/transport-netty4/src/main/java/org/opensearch/http/netty4/Netty4DefaultHttpRequest.java

@@ -0,0 +1,114 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.http.netty4;
+
+import org.opensearch.core.common.bytes.BytesArray;
+import org.opensearch.core.common.bytes.BytesReference;
+import org.opensearch.core.rest.RestStatus;
+import org.opensearch.http.HttpRequest;
+import org.opensearch.rest.RestRequest;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import io.netty.handler.codec.http.DefaultHttpRequest;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.cookie.Cookie;
+import io.netty.handler.codec.http.cookie.ServerCookieDecoder;
+import io.netty.handler.codec.http.cookie.ServerCookieEncoder;
+
+public class Netty4DefaultHttpRequest extends AbstractNetty4HttpRequest implements HttpRequest {
+
+    private final DefaultHttpRequest request;
+
+    public Netty4DefaultHttpRequest(DefaultHttpRequest request) {
+        this(request, new HttpHeadersMap(request.headers()), null);
+    }
+
+    private Netty4DefaultHttpRequest(DefaultHttpRequest request, HttpHeadersMap headers, Exception inboundException) {
+        this.request = request;
+        this.headers = headers;
+        this.inboundException = inboundException;
+    }
+
+    @Override
+    public RestRequest.Method method() {
+        return getHttpMethod(request);
+    }
+
+    @Override
+    public String uri() {
+        return request.uri();
+    }
+
+    @Override
+    public BytesReference content() {
+        // throw new RuntimeException("Not implemented");
+        return BytesArray.EMPTY;
+    }
+
+    @Override
+    public final Map<String, List<String>> getHeaders() {
+        return headers;
+    }
+
+    @Override
+    public List<String> strictCookies() {
+        String cookieString = request.headers().get(HttpHeaderNames.COOKIE);
+        if (cookieString != null) {
+            Set<Cookie> cookies = ServerCookieDecoder.STRICT.decode(cookieString);
+            if (!cookies.isEmpty()) {
+                return ServerCookieEncoder.STRICT.encode(cookies);
+            }
+        }
+        return Collections.emptyList();
+    }
+
+    @Override
+    public HttpVersion protocolVersion() {
+        if (request.protocolVersion().equals(io.netty.handler.codec.http.HttpVersion.HTTP_1_0)) {
+            return HttpRequest.HttpVersion.HTTP_1_0;
+        } else if (request.protocolVersion().equals(io.netty.handler.codec.http.HttpVersion.HTTP_1_1)) {
+            return HttpRequest.HttpVersion.HTTP_1_1;
+        } else {
+            throw new IllegalArgumentException("Unexpected http protocol version: " + request.protocolVersion());
+        }
+    }
+
+    @Override
+    public HttpRequest removeHeader(String header) {
+        return null;
+    }
+
+    @Override
+    public Netty4HttpResponse createResponse(RestStatus status, BytesReference content) {
+        return new Netty4HttpResponse(request.headers(), request.protocolVersion(), status, content);
+    }
+
+    @Override
+    public Exception getInboundException() {
+        return inboundException;
+    }
+
+    @Override
+    public void release() {
+        // do nothing
+    }
+
+    @Override
+    public HttpRequest releaseAndCopy() {
+        return this;
+    }
+
+    public DefaultHttpRequest nettyRequest() {
+        return request;
+    }
+}