Add REST content aggregator #117787

mhl-b · 2024-11-30T01:36:02Z

This PR moves HTTP content aggregation from the Netty pipeline(HttpObjectAggregator) to the REST controller. Immediate benefit is that any RestHandler can consume streamed body by toggling RestHandler#supportContentStream flag and implementing BaseRestHandler.RequestBodyChunkConsumer. Also opens up opportunity for further improvements: circuit-breakers on a stream not full content, fast-fail on headers check, move Authentication to the REST layer and simplify HTTP code, content length limits per API, incremental JSON parsing.

Notable changes:

HttpObjectAggregate is replaced by Netty4HttpContentSizeHandler. Netty's aggregator handled 100-continue and oversized content besides combining content chunks into full content. This PR preserves that behaviour, but removes aggregation.
Add RestContentAggregator to the RestController. See RestController#maybeAggregateAndDispatch.
Coordinated access to Netty's ChannelConfig#setAutoRead with new AutoReadSync class. A more reliable autoRead toggle between Netty4HttpHeaderValidator and Netty4HttpRequestBodyStream. Should be removed in the future when Authentication is moved to the REST layer and HTTP flow control can be simplified.
Add ThreadWatchdog.ActivityTracker to Netty4HttpRequestBodyStream for off-pipeline channel event loop executions.

Closes:

ES-9968
ES-9166
Partly ES-9571
ES-9309

elasticsearchmachine · 2024-11-30T01:46:26Z

Hi @mhl-b, I've created a changelog YAML for you.

…ation

elasticsearchmachine · 2024-12-09T20:15:59Z

Pinging @elastic/es-distributed-coordination (Team:Distributed Coordination)

mhl-b · 2024-12-10T04:06:00Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

                    send();
-                } catch (Exception e) {
-                    channel.pipeline().fireExceptionCaught(e);
+                } catch (Throwable t) {


die-with-dignity test throws OOM exception which is discarded by evenloop executor, need to catch and throw into pipeline

…ation

DaveCTurner

I'm finding this quite hard to get a handle on, there's a lot to review here. Could we pull out the changes to fix the auto-read problems and deal with that first?

I'm also wondering if auto-read is even necessary any more. We're already starting to move to an explicit-read model with stuff like org.elasticsearch.http.HttpBody.Stream#next. Could we go all-in with that model instead and avoid using auto-read at all?

DaveCTurner · 2024-12-30T11:30:47Z

...ternalClusterTest/java/org/elasticsearch/http/netty4/Netty4IncrementalRequestHandlingIT.java

        }

-        int readBytes(int bytes) {
+        int readBytes(int bytes) throws InterruptedException {


Could we make this method interrupt-safe rather than pushing that responsibility up to the caller? I.e. catch InterruptedException, reinstate the thread's interrupt status flag, and throw an AssertionError since we do not expect to be interrupted in these tests.

DaveCTurner · 2024-12-30T11:34:25Z

...ansport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpContentSizeHandler.java

+    );
+
+    static {
+        EXPECTATION_FAILED.headers().set(CONTENT_LENGTH, 0);


nit: I'd rather we finished constructing these values (i.e. setting their headers) before assigning them to their respective fields.

I copied that from the netty's HttpObjectAggregator. Can instantiate those inline.

DaveCTurner · 2024-12-30T11:36:12Z

...ansport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpContentSizeHandler.java

+        if (msg instanceof HttpRequest request) {
+            handleRequest(ctx, request);
+        } else {
+            handleContent(ctx, (HttpContent) msg);


nit: HttpObject has other possibilities (e.g. io.netty.handler.codec.http.DefaultHttpObject), could we assert this is a HttpContent before casting?

HttpObjectDecoder emits HttpRequest and HttpContent only. I dont think DefaultHttpObject is a valid type here.

DaveCTurner · 2024-12-30T11:39:21Z

.../transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

                    }
                })
-                .addLast("aggregator", aggregator);
+                .addLast("content_size", contentSizeHandler);


nit: it'd be more harmonious with the other code here if we inlined contentSizeHandler. Wasn't possible with aggregator because of the setMaxCumulationBufferComponents call, but we can clean that up now.

DaveCTurner · 2024-12-30T11:42:51Z

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequest.java

    private final Map<String, List<String>> headers;
    private final AtomicBoolean released;
    private final Exception inboundException;
    private final boolean pooled;


somewhat unrelated: is this ever false?

It should be true, I did not add these changes here due to already large scope.

DaveCTurner · 2024-12-30T11:55:40Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

    }

-    public void handleNettyContent(HttpContent httpContent) {
+    public void handleNettyContent(HttpContent httpContent) throws Exception {


I'm uncomfortable with relaxing all these signatures to permit any checked exception. AFAICT this is because we can now end up calling dispatchRequest which calls various methods that claim they might throw an IOException, although in practice they shouldn't do so. Could we handle this better in dispatchRequest instead?

DaveCTurner · 2024-12-30T11:59:41Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

        } else {
-            channel.eventLoop().submit(this::doClose);
+            if (channel.eventLoop().isShutdown() == false) {
+                channel.eventLoop().submit(this::doClose);


I suspect this might not be threadsafe: what if the event loop shuts down after checking isShutdown but before submitting the task? Should we do something more like org.elasticsearch.transport.netty4.Netty4Utils#safeWriteAndFlush?

That said, don't we clean up all in-flight requests and close all HTTP channels etc. before shutting down the event loop anyway? If so, isShutdown() is always false here right?

Yup, not threadsafe. I think I should catch RejectedException here from eventLoop().submit(). I dont remember all details, but shutdown integ tests invoke this code and isShutdown() is true.

DaveCTurner · 2024-12-30T12:06:49Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

        this.threadContext = threadContext;
+        this.activityTracker = activityTracker;
        this.requestContext = threadContext.newStoredContext();
+        this.autoRead = AutoReadSync.getHandle(channel);


I'm uncomfortable with dynamically acquiring a new auto-read flag for each body stream in the channel. We're only reading one of these at once, can we do something more obviously bounded here?

We can read fairly large chunk from network that can contain multiple HTTP requests. We will dispatch each request headers to RestHandlers while holding content for each of them in the Netty4HttpRequestBodyStream until it's processed. So it's possible that multiple stream handlers can exists in the channel that toggle off auto-read at the same time. In particular when stream.next() is not invoked immediately by ChunkHandler.

The ultimate and clean solution that I see here is moving Netty4HttpHeaderValidator to the rest layer. Place FlowControlHandler in front of HttpContentDecompressor that will hold all incoming http parts, and toggle auto-read only in the Netty4HttpRequestBodyStream. The bitset toggle I introduced here only for the time being until we move auth stuff.

mhl-b · 2025-06-19T16:06:44Z

closed in favour of #129302

elasticsearchmachine added the v9.0.0 label Nov 30, 2024

mhl-b added >enhancement :Distributed Coordination/Network Http and internode communication implementations Team:Distributed Coordination Meta label for Distributed Coordination team labels Nov 30, 2024

mhl-b force-pushed the rest-content-aggregation branch from cae8263 to c7d3e4d Compare December 5, 2024 05:55

wip

650738e

mhl-b force-pushed the rest-content-aggregation branch from 37ba19b to 650738e Compare December 6, 2024 01:09

mhl-b added 7 commits December 5, 2024 18:33

remove old http continue test

983b877

remove HttpObjectAggregator

cf907fa

Merge remote-tracking branch 'upstream/main' into rest-content-aggreg…

81e4811

…ation

cleanup

eecc278

Merge remote-tracking branch 'upstream/main' into rest-content-aggreg…

df42681

…ation

Merge remote-tracking branch 'upstream/main' into rest-content-aggreg…

a749650

…ation

100-continue test fix

3c92567

mhl-b changed the title ~~[WIP] Add rest content aggregator~~ [WIP] Add REST content aggregator Dec 9, 2024

mhl-b requested review from DaveCTurner and Tim-Brooks December 9, 2024 20:15

mhl-b marked this pull request as ready for review December 9, 2024 20:15

mhl-b requested a review from a team December 9, 2024 20:15

mhl-b changed the title ~~[WIP] Add REST content aggregator~~ Add REST content aggregator Dec 9, 2024

mhl-b commented Dec 10, 2024

View reviewed changes

nielsbauman mentioned this pull request Dec 23, 2024

Unmute IngestCommonClientYamlTestSuiteIT #119188

Merged

Merge remote-tracking branch 'upstream/main' into rest-content-aggreg…

c0b6db4

…ation

DaveCTurner reviewed Dec 30, 2024

View reviewed changes

mhl-b mentioned this pull request Jan 5, 2025

Http stream activity tracker and exceptions handling #119564

Merged

This was referenced Jan 16, 2025

Add http stream content size handler #120246

Merged

Backport http stream content size #120246 #120417

Merged

mhl-b mentioned this pull request Jan 28, 2025

Indexing a single doc to a data stream bulk endpoint trips assertion and returns class cast exception #120746

Closed

mhl-b mentioned this pull request Jan 28, 2025

Add http stream content size handler (fixed #120246) #121095

Merged

elasticsearchmachine added v9.1.0 and removed v9.0.0 labels Jan 30, 2025

mhl-b closed this Jun 19, 2025

Add REST content aggregator #117787

Add REST content aggregator #117787

Uh oh!

Conversation

mhl-b commented Nov 30, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

elasticsearchmachine commented Nov 30, 2024

Uh oh!

elasticsearchmachine commented Dec 9, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DaveCTurner left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Dec 30, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b commented Jun 19, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mhl-b commented Nov 30, 2024 •

edited

Loading

mhl-b Dec 30, 2024 •

edited

Loading