Introduce `ChunkedZipResponse` #109820

DaveCTurner · 2024-06-17T16:01:38Z

Adds a utility for implementing REST APIs which construct a streaming
(i.e. pretty-much-constant-memory) .zip file response as a (pausable)
sequence of ChunkedRestResponseBodyPart instances, where each entry in
the .zip file is itself a (pausable) sequence of
ChunkedRestResponseBodyPart instances.

Relates #104851

Adds a utility for implementing REST APIs which construct a streaming (i.e. pretty-much-constant-memory) `.zip` file response as a (pausable) sequence of `ChunkedRestResponseBodyPart` instances, where each entry in the `.zip` file is itself a (pausable) sequence of `ChunkedRestResponseBodyPart` instances. Relates elastic#104851

elasticsearchmachine · 2024-06-17T16:02:02Z

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

ywangd · 2024-06-19T03:35:31Z

I had a brief look at the main code changes. They make sense to me. But I will need more time to read closely. For the context, do we already return zip response today? Maybe something like ML models? I assume this change does not have user visible impact? If we don't have such usage today, could you please explain a bit on the intended use cases? Thanks!

DaveCTurner · 2024-06-19T05:21:10Z

Yes sorry, some more context. This was also part of my recent on-week project. Today we don't have any APIs that expose zip-format data, at least partly because until #104851 we didn't really have a way to create such a thing in a streaming fashion, and our usual approach of creating the whole response in-memory first would risk making the node go OOM. I have a number of possible use-cases for this in mind, all somewhat to do with supportability:

Diagnostics bundle - today we have to use a separate tool to collect diagnostics bundles, which has the advantage of decoupling the bundle contents definition from the ES version, but also several disadvantages over a built-in API:
- ES doesn't know that all the APIs it's hitting are related to a diagnostics request so it cannot schedule the work very effectively, and that can mean that the act of collecting diagnostics is sometimes itself harmful to the cluster.
- Some customers struggle to install the separate tool.
- It doesn't work especially reliably, e.g. in cases of an unstable cluster or flaky network.
- It apparently involves an awful lot of extra ceremony to run in k8s.
Logs - we don't index all the logs that ES produces (notably omitting TRACE logs and detailed GC logs) so it involves a lot of extra work to get a hold of them if needed. Likewise customers often struggle to share the correct logs in support cases, typically picking a few log files from one node rather than gathering all relevant logs from the whole cluster, but if we had a built-in API to gather a logs bundle then we'd cut down massively on the time we waste in these cases.
Snapshot repository debugging - sometimes we get asked questions about space usage in a snapshot repository, or integrity issues, and it'd be awfully useful to have a dump of all the metadata in the repository for further investigation. Today that's essentially impossible to do, but again if we had a built-in API to expose the info then it'd be easy.

These things all to be discussed separately, but this PR introduces a common prerequisite for them.

DaveCTurner · 2024-08-01T08:54:20Z

Gentle reminder for reviews here if you have time :)

ywangd

Sorry for the extended delay. Honestly, this PR is a bit daunting to review. I just read it again and still trying to build a full mental model for it. I do plan to come back to it soon since otherwise I'll be forgetting about the details again. For the time being, I have left some comments and questions. Thanks!

server/src/internalClusterTest/java/org/elasticsearch/rest/ChunkedZipResponseIT.java

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

DaveCTurner

Thanks @ywangd, quite an interesting exercise coming back to a PR like this with fresh eyes after a few months. Your questions prompted some renaming/commentary/other cleanup that I hope helps make it easier to understand.

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

ywangd

I have a few more comments.

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

ywangd · 2024-08-05T06:56:25Z

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

+        /**
+         * Transfer {@link #currentEntryReleasable} into the supplied collection (i.e. add it to {@code releasables} and then clear
+         * {@link #currentEntryReleasable}). Called when the last chunk of the last part of the current entry is serialized, so that we can
+         * start serializing chunks of the next entry straight away whilst delaying the release of the current entry's resources until the
+         * transmission of the chunk that is currently under construction.
+         */
+        private void transferCurrentEntryReleasable(ArrayList<Releasable> releasables) {


This transfer of entry releasables is somewhat complext to follow for me. Conceptually, an entry releasable is released when the entry is completed or cancelled. That seems most intuitive. But we accumulate them for efficiency to prioritize ongoing bytes writing? Could it also be a concern that we are not releasing them in time? I wonder whether it might be simpler to release them once per entry?

We're delaying the release of the releasable until the bytes are actually sent. That's by design: the entry is still consuming some memory in the network layer until we hand those bytes off to the OS, so we shouldn't consider it as completed earlier.

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

nicktindall

LGTM (once @ywangd's concerns are addressed also)

clearing the approved flag until Yang has taken a look too

ywangd

A few more comments. I don't expect to find anything of significance. But do plan to take yet another look. Thanks!

ywangd · 2024-08-06T05:12:29Z

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

+                    // request aborted, nothing more to send (queue is being cleared by queueRefs#closeInternal)
+                    isPartComplete = true;
+                    isLastPart = true;
+                    return new ReleasableBytesReference(BytesArray.EMPTY, () -> {});


Why don't we throw AlreadyClosedException similar to what enqueueEntry does? The underlying channel should be closed at this point. So no need to be gentle here?

Fair point, no real need for an ACE in enqueueEntry either - see 8d73a28.

The change of using Releasable as parameter for newEntryListener looks nice to me. In most cases, I think the caller indeed does not care about getting notified when the resource is released other than it gets released at some point. 👍

ywangd · 2024-08-06T05:22:54Z

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

+
+        private void finishCurrentPart(ArrayList<Releasable> releasables) throws IOException {
+            if (bodyPart.isLastPart()) {
+                zipOutputStream.closeEntry();


Can we add a comment to say that we don't set isLastPart = true here because we bridge the last part of this entry to the first part of the next entry so that the caller sees only continuous parts instead of having to be aware of entries. The fact that the networking code is unware of entries and they are for producing side only is something that I didn't immediately realize. In hindsight, it would be helpful to see this first to build the mental model.

Sure, see 9d5c060.

ywangd · 2024-08-06T05:42:57Z

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

+        /**
+         * A cache for an empty list to be used to collect the {@code Releasable} instances to be released when the next chunk has been
+         * fully transmitted. It's a list because a call to {@link #encodeChunk} may yield a chunk that completes several entries, each of
+         * which has its own resources to release. We cache this value across chunks because most chunks won't release anything, so we can
+         * keep the empty list around for later to save on allocations.
+         */
+        private ArrayList<Releasable> nextReleasablesCache = new ArrayList<>();


Based on my understanding of the other comment, it is necessary to accumulate the currentEntryReleasable for an entry because the do loop for writeNextBytes may write more than one entry. Until these written bytes are actually sent later by the networking layer, we must retain releaseables for all the entries. The releaseables are released each time data is sent out, i.e. each time the response is paused or finished at the end. If this sounds correct, can we please somehow incorporate it into the comments, maybe here or somewhere else more suitable?

We cache this value across chunks because most chunks won't release anything, so we can keep the empty list around for later to save on allocations.

I am not sure whether this helps the understanding. Unless my above understanding is wrong, I think the need for a list instead of a single entry is best covered by the fact that a loop of writeNextBytes can span multiple entries.

The releaseables are released each time data is sent out, i.e. each time the response is paused or finished at the end.

No, that's not right, the releasables are released when the chunk that completes the entries has been sent, but we do not wait all the way until the end (pause or finish) of the part that contains those entries.

Not sure what else to add to the comments to help here. Although the sentence about caching may not be what you're looking for, the preceding sentence explains why it's a list.

Sorry my comment was imprecise. I was meant to say

The releaseables are packaged to be released each time data is sent out, i.e. each time the response is paused or finished at the end.

Essentially it means how we pass the current list of entry releaseables each time after the do/while loop.

Not sure what else to add to the comments to help here

I think it could help to add a comment right before the do/while loop to say that writeNextBytes can work through multiple entries and accumulate entry releaseables which are then released in a single batch once the processed bytes are fully sent out. Feels like a good compliment to the comment here about "a call ... completes serveral entries".

server/src/internalClusterTest/java/org/elasticsearch/rest/ChunkedZipResponseIT.java

…nkedZipResponseIT.java Co-authored-by: Yang Wang <[email protected]>

ywangd · 2024-08-07T01:40:36Z

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

+    private SubscribableListener<ChunkedRestResponseBodyPart> nextAvailableChunksListener;
+
+    /**
+     * A resource to be released when the transmission of the current entry is complete.


Nit

Suggested change

* A resource to be released when the transmission of the current entry is complete.

* A resource to be released when the transmission of the current entry is complete.

* Multiple of them maybe released in a single batch if their associated entries are transmitted together.

?

Expanded in 5aa4c08.

ywangd · 2024-08-07T01:51:40Z

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

+
+                @Override
+                public void onFailure(Exception e) {
+                    Releasables.closeExpectNoException(releasable);


Might worth a logging message here since I don't think it should happen normally and it could lead to an unusable zip file if happens?

No this is fine, it's covered by the docs and by ChunkedZipResponseIT#testRandomZipResponse (see the comment about NPEs in handleZipRestRequest). It just means no entry is sent.

ywangd · 2024-08-07T01:55:33Z

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

+     * @param releasable A resource which is released when the entry has been completely processed: either fully sent, or else the request
+     *                   was cancelled and the response will not be used any further.


Nit: It's also released when the entry is "skipped" (for lack of a better word). But maybe it's obvious from the code

Suggested change

* @param releasable A resource which is released when the entry has been completely processed: either fully sent, or else the request

* was cancelled and the response will not be used any further.

* @param releasable A resource which is released when the entry has been skipped or completely processed: either fully sent, or else the request

* was cancelled and the response will not be used any further.

Expanded in 5aa4c08.

ywangd · 2024-08-07T02:02:57Z

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

+                    // request aborted, nothing more to send (queue is being cleared by queueRefs#closeInternal)
+                    isPartComplete = true;
+                    isLastPart = true;
+                    return new ReleasableBytesReference(BytesArray.EMPTY, () -> {});


The change of using Releasable as parameter for newEntryListener looks nice to me. In most cases, I think the caller indeed does not care about getting notified when the resource is released other than it gets released at some point. 👍

ywangd

LGTM

Thanks for the iterations and the opportunity to review this work 👍

server/src/internalClusterTest/java/org/elasticsearch/rest/ChunkedZipResponseIT.java

DaveCTurner · 2024-08-07T06:22:01Z

Thanks both for the reviews!

* upstream/main: (132 commits) Fix compile after several merges Update docs with new behavior on skip conditions (elastic#111640) Skip on any instance of node or version features being present (elastic#111268) Skip on any node capability being present (elastic#111585) [DOCS] Publishes Anthropic inference service docs. (elastic#111619) Introduce `ChunkedZipResponse` (elastic#109820) [Gradle] fix esql compile cacheability (elastic#111651) Mute org.elasticsearch.datastreams.logsdb.qa.StandardVersusLogsIndexModeChallengeRestIT testTermsQuery elastic#111666 Mute org.elasticsearch.datastreams.logsdb.qa.StandardVersusLogsIndexModeChallengeRestIT testMatchAllQuery elastic#111664 Mute org.elasticsearch.xpack.esql.analysis.VerifierTests testMatchCommand elastic#111661 Mute org.elasticsearch.xpack.esql.optimizer.LocalPhysicalPlanOptimizerTests testMatchCommandWithMultipleMatches {default} elastic#111660 Mute org.elasticsearch.xpack.esql.optimizer.LocalPhysicalPlanOptimizerTests testMatchCommand {default} elastic#111659 Mute org.elasticsearch.xpack.esql.optimizer.LocalPhysicalPlanOptimizerTests testMatchCommandWithWhereClause {default} elastic#111658 LogsDB qa tests - add specific matcher for source (elastic#111568) ESQL: Move `randomLiteral` (elastic#111647) [ESQL] Clean up UNSUPPORTED type blocks (elastic#111648) ESQL: Remove the `NESTED` DataType (elastic#111495) ESQL: Move more out of esql-core (elastic#111604) Improve MvPSeriesWeightedSum edge case and add more tests (elastic#111552) Add link to flood-stage watermark exception message (elastic#111315) ... # Conflicts: # server/src/main/java/org/elasticsearch/TransportVersions.java

Adds a utility for implementing REST APIs which construct a streaming (i.e. pretty-much-constant-memory) `.zip` file response as a (pausable) sequence of `ChunkedRestResponseBodyPart` instances, where each entry in the `.zip` file is itself a (pausable) sequence of `ChunkedRestResponseBodyPart` instances. Relates elastic#104851

Adds a utility for implementing REST APIs which construct a streaming (i.e. pretty-much-constant-memory) `.zip` file response as a (pausable) sequence of `ChunkedRestResponseBodyPart` instances, where each entry in the `.zip` file is itself a (pausable) sequence of `ChunkedRestResponseBodyPart` instances. Relates #104851

Similar to `ChunkedZipResponse` (elastic#109820) this utility allows Elasticsearch to send an `XContent`-based response constructed out of a sequence of `ChunkedToXContent` fragments, provided in a streaming and asynchronous fashion. This will enable elastic#93735 to proceed without needing to create a temporary index to hold the intermediate results.

Similar to `ChunkedZipResponse` (#109820) this utility allows Elasticsearch to send an `XContent`-based response constructed out of a sequence of `ChunkedToXContent` fragments, provided in a streaming and asynchronous fashion. This will enable #93735 to proceed without needing to create a temporary index to hold the intermediate results.

Adds a utility for implementing REST APIs which construct a streaming (i.e. pretty-much-constant-memory) `.zip` file response as a (pausable) sequence of `ChunkedRestResponseBodyPart` instances, where each entry in the `.zip` file is itself a (pausable) sequence of `ChunkedRestResponseBodyPart` instances. Relates elastic#104851

Similar to `ChunkedZipResponse` (elastic#109820) this utility allows Elasticsearch to send an `XContent`-based response constructed out of a sequence of `ChunkedToXContent` fragments, provided in a streaming and asynchronous fashion. This will enable elastic#93735 to proceed without needing to create a temporary index to hold the intermediate results.

DaveCTurner added >non-issue :Distributed Coordination/Network Http and internode communication implementations v8.15.0 labels Jun 17, 2024

DaveCTurner requested a review from ywangd June 17, 2024 16:01

elasticsearchmachine added the Team:Distributed Meta label for distributed team (obsolete) label Jun 17, 2024

elasticsearchmachine added v8.16.0 and removed v8.15.0 labels Jul 4, 2024

Merge branch 'main' into 2024/06/17/ChunkedZipResponse

0ec41bc

DaveCTurner requested a review from nicktindall July 18, 2024 05:42

Merge branch 'main' into 2024/06/17/ChunkedZipResponse

7261f2b

ywangd reviewed Aug 2, 2024

View reviewed changes

DaveCTurner added 2 commits August 4, 2024 10:14

Merge branch 'main' into 2024/06/17/ChunkedZipResponse

4336c9c

Review feedback

45468e4

DaveCTurner commented Aug 4, 2024

View reviewed changes

DaveCTurner requested a review from ywangd August 4, 2024 09:24

DaveCTurner added 2 commits August 5, 2024 06:39

Comment typos

4adf929

Merge branch 'main' into 2024/06/17/ChunkedZipResponse

b6aa400

ywangd reviewed Aug 5, 2024

View reviewed changes

server/src/main/java/org/elasticsearch/rest/ChunkedZipResponse.java Outdated Show resolved Hide resolved

server/src/main/java/org/elasticsearch/rest/ChunkedZipResponse.java Outdated Show resolved Hide resolved

Tighter assert

b67a6d1

nicktindall reviewed Aug 5, 2024

View reviewed changes

server/src/main/java/org/elasticsearch/rest/ChunkedZipResponse.java Outdated Show resolved Hide resolved

ywangd reviewed Aug 5, 2024

View reviewed changes

DaveCTurner added 3 commits August 5, 2024 08:05

nextReleasables -> nextReleasablesCache

88f2ff3

More comments

7db200b

NO_MORE_ENTRIES instead of null

0b01e76

DaveCTurner requested a review from ywangd August 5, 2024 08:58

DaveCTurner requested a review from nicktindall August 5, 2024 08:58

nicktindall previously approved these changes Aug 5, 2024

View reviewed changes

ywangd reviewed Aug 6, 2024

View reviewed changes

DaveCTurner and others added 4 commits August 6, 2024 07:46

Update server/src/internalClusterTest/java/org/elasticsearch/rest/Chu…

16911a0

…nkedZipResponseIT.java Co-authored-by: Yang Wang <[email protected]>

Merge branch 'main' into 2024/06/17/ChunkedZipResponse

d31a3b1

Remove ACE in enqueueEntry

8d73a28

Clarify flags

9d5c060

DaveCTurner requested a review from ywangd August 6, 2024 07:16

DaveCTurner added 2 commits August 6, 2024 09:17

Spotless

56f18b6

Add test for failures in getNextPart()

c2812c0

ywangd reviewed Aug 7, 2024

View reviewed changes

ywangd approved these changes Aug 7, 2024

View reviewed changes

server/src/internalClusterTest/java/org/elasticsearch/rest/ChunkedZipResponseIT.java Show resolved Hide resolved

DaveCTurner added 2 commits August 7, 2024 06:57

Merge branch 'main' into 2024/06/17/ChunkedZipResponse

cffd1eb

Comments

5aa4c08

DaveCTurner added the auto-merge-without-approval Automatically merge pull request when CI checks pass (NB doesn't wait for reviews!) label Aug 7, 2024

Limit entry count before failure in testGetNextPartFailure

59c716b

elasticsearchmachine merged commit 4272164 into elastic:main Aug 7, 2024
15 checks passed

DaveCTurner deleted the 2024/06/17/ChunkedZipResponse branch August 7, 2024 07:20

DaveCTurner mentioned this pull request Aug 15, 2024

Introduce StreamingXContentResponse #111933

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce `ChunkedZipResponse` #109820

Introduce `ChunkedZipResponse` #109820

DaveCTurner commented Jun 17, 2024

elasticsearchmachine commented Jun 17, 2024

ywangd commented Jun 19, 2024

DaveCTurner commented Jun 19, 2024

DaveCTurner commented Aug 1, 2024

ywangd left a comment

DaveCTurner left a comment

ywangd left a comment

ywangd Aug 5, 2024

DaveCTurner Aug 5, 2024

nicktindall left a comment •

edited

Loading

ywangd left a comment

ywangd Aug 6, 2024

DaveCTurner Aug 6, 2024 •

edited

Loading

ywangd Aug 7, 2024

ywangd Aug 6, 2024

DaveCTurner Aug 6, 2024 •

edited

Loading

ywangd Aug 6, 2024

DaveCTurner Aug 6, 2024

ywangd Aug 7, 2024

ywangd Aug 7, 2024

DaveCTurner Aug 7, 2024

ywangd Aug 7, 2024

DaveCTurner Aug 7, 2024

ywangd Aug 7, 2024

DaveCTurner Aug 7, 2024

ywangd Aug 7, 2024

ywangd left a comment

DaveCTurner commented Aug 7, 2024

	* A resource to be released when the transmission of the current entry is complete.
	* A resource to be released when the transmission of the current entry is complete.
	* Multiple of them maybe released in a single batch if their associated entries are transmitted together.

		* @param releasable A resource which is released when the entry has been completely processed: either fully sent, or else the request
		* was cancelled and the response will not be used any further.

Introduce ChunkedZipResponse #109820

Introduce ChunkedZipResponse #109820

Conversation

DaveCTurner commented Jun 17, 2024

elasticsearchmachine commented Jun 17, 2024

ywangd commented Jun 19, 2024

DaveCTurner commented Jun 19, 2024

DaveCTurner commented Aug 1, 2024

ywangd left a comment

Choose a reason for hiding this comment

DaveCTurner left a comment

Choose a reason for hiding this comment

ywangd left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nicktindall left a comment • edited Loading

Choose a reason for hiding this comment

ywangd left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

DaveCTurner Aug 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

DaveCTurner Aug 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ywangd left a comment

Choose a reason for hiding this comment

DaveCTurner commented Aug 7, 2024

Introduce `ChunkedZipResponse` #109820

Introduce `ChunkedZipResponse` #109820

nicktindall left a comment •

edited

Loading

DaveCTurner Aug 6, 2024 •

edited

Loading

DaveCTurner Aug 6, 2024 •

edited

Loading