Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Reporting] Fix slow CSV with large max size bytes #120365

Merged
merged 13 commits into from
Dec 7, 2021
Merged

[Reporting] Fix slow CSV with large max size bytes #120365

Changes from 3 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
d82b4ef
use Buffer.alloc + .set API instead of .concat
jloleysens Dec 3, 2021
c3ff734
refactor variable names and actually assign to this.buffer
jloleysens Dec 3, 2021
e6996a2
ok, looks like an array of buffers could work
jloleysens Dec 3, 2021
566296c
added large comment and refactored some variable name
jloleysens Dec 6, 2021
7f3b1ad
Merge branch 'main' into reporting/fix-slow-csv-large-max-size-bytes
kibanamachine Dec 6, 2021
b8c48a6
fix comment
jloleysens Dec 6, 2021
7407b85
refactored logic to deal with an edge case where partial buffers shou…
jloleysens Dec 6, 2021
4edef0d
added new test for detecting when the write stream throws for bad config
jloleysens Dec 6, 2021
df5f5ab
updated logic to not ever call .slice(0), updated the guard for the c…
jloleysens Dec 6, 2021
f82b80e
refactor totalBytesConsumed -> bytesToFlush
jloleysens Dec 6, 2021
a00439a
use the while loop mike wrote
jloleysens Dec 6, 2021
af1dbfb
remove unused variable
jloleysens Dec 6, 2021
938089f
update comment
jloleysens Dec 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 24 additions & 9 deletions x-pack/plugins/reporting/server/lib/content_stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,9 @@ export class ContentStream extends Duplex {
return Math.floor(max / 2);
}

private buffer = Buffer.from('');
private buffers: Buffer[] = [];
private currentByteLength = 0;

jloleysens marked this conversation as resolved.
Show resolved Hide resolved
private bytesRead = 0;
private chunksRead = 0;
private chunksWritten = 0;
Expand Down Expand Up @@ -249,8 +251,20 @@ export class ContentStream extends Duplex {
});
}

private async flush(size = this.buffer.byteLength) {
const chunk = this.buffer.slice(0, size);
private async flush(size = this.currentByteLength) {
const buffersToConsume: Buffer[] = [];
let totalBytesConsumed = 0;
jloleysens marked this conversation as resolved.
Show resolved Hide resolved

for (const buffer of this.buffers) {
if (totalBytesConsumed + buffer.byteLength <= size) {
buffersToConsume.push(buffer);
totalBytesConsumed += buffer.byteLength;
} else {
break;
jloleysens marked this conversation as resolved.
Show resolved Hide resolved
}
}

const chunk = Buffer.concat(buffersToConsume);
jloleysens marked this conversation as resolved.
Show resolved Hide resolved
const content = this.encode(chunk);

if (!this.chunksWritten) {
Expand All @@ -265,22 +279,23 @@ export class ContentStream extends Duplex {
}

this.bytesWritten += chunk.byteLength;
this.buffer = this.buffer.slice(size);

this.buffers = this.buffers.slice(buffersToConsume.length - 1);
this.currentByteLength -= totalBytesConsumed;
jloleysens marked this conversation as resolved.
Show resolved Hide resolved
}

private async flushAllFullChunks() {
const maxChunkSize = await this.getMaxChunkSize();

while (this.buffer.byteLength >= maxChunkSize) {
while (this.currentByteLength >= maxChunkSize) {
await this.flush(maxChunkSize);
}
}

_write(chunk: Buffer | string, encoding: BufferEncoding, callback: Callback) {
this.buffer = Buffer.concat([
this.buffer,
Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding),
]);
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding);
this.currentByteLength += buffer.byteLength;
this.buffers.push(buffer);

this.flushAllFullChunks()
.then(() => callback())
Expand Down