[ML][Inference] stream inflate to parser + throw when byte limit is reached #51644

benwtrent · 2020-01-29T19:01:00Z

Three fixes for when the compressed_definition is utilized on PUT

Update the inflate byte limit to be the minimum of 10% the max heap, or 1GB (what it was previously)
Stream data directly to the JSON parser, so if it is invalid, we don't have to inflate the whole stream to find out
Throw when the maximum bytes are reach indicating that is why the request was rejected

elasticmachine · 2020-01-29T19:01:03Z

Pinging @elastic/ml-core (:ml)

przemekwitek · 2020-01-30T08:35:08Z

...e/src/main/java/org/elasticsearch/xpack/core/ml/inference/InferenceToXContentCompressor.java

@@ -33,7 +34,10 @@
 */
 public final class InferenceToXContentCompressor {
    private static final int BUFFER_SIZE = 4096;
-    private static final long MAX_INFLATED_BYTES = 1_000_000_000; // 1 gb maximum
+    // Either 10% of the configured JVM heap, or 1 GB, which ever is smaller


Why is 10% a good number?

Eventually we could have a dynamic limit that's integrated with the real memory circuit breaker (#31767). Maybe we could reserve a percentage of free memory and use that as the dynamic limit for a given request, then give back that reservation after finding out the actual size required. That's something to investigate for 7.7 or 7.8.

However, I think 10% is an OK first step for 7.6 to reduce the risk of someone accidentally triggering an OOM on a node with a small heap.

przemekwitek · 2020-01-30T08:36:16Z

...e/src/main/java/org/elasticsearch/xpack/core/ml/inference/InferenceToXContentCompressor.java

        byte[] compressedBytes = Base64.getDecoder().decode(compressedString.getBytes(StandardCharsets.UTF_8));
+        if (compressedBytes.length > streamSize) {
+            throw new IOException("compressed stream is longer than maximum allowed bytes [" + streamSize +"]");


Suggested change

throw new IOException("compressed stream is longer than maximum allowed bytes [" + streamSize +"]");

throw new IOException("compressed stream is longer than maximum allowed bytes [" + streamSize + "]");

przemekwitek · 2020-01-30T08:36:50Z

...e/src/main/java/org/elasticsearch/xpack/core/ml/inference/InferenceToXContentCompressor.java

            return parser.mapOrdered();
        }
    }

-    static BytesReference inflate(String compressedString, long streamSize) throws IOException {
+    static InputStream inflate(String compressedString, long streamSize) throws IOException {
+        // If the compressed length is already too large, it make sense that the inflated length would be as well


Could you move this line after line 70 (compressedBytes) so that it is clearly visible that it refers to the if check?

przemekwitek · 2020-01-30T08:37:45Z

.../src/main/java/org/elasticsearch/xpack/core/ml/inference/utils/SimpleBoundedInputStream.java

@@ -38,6 +39,9 @@ public SimpleBoundedInputStream(InputStream inputStream, long maxBytes) {
    public int read() throws IOException {
        // We have reached the maximum, signal stream completion.
        if (numBytes >= maxBytes) {
+            if (throwWhenExceeded) {
+                throw new IOException("input stream exceeded maximum bytes of [" + maxBytes +"]");


Suggested change

throw new IOException("input stream exceeded maximum bytes of [" + maxBytes +"]");

throw new IOException("input stream exceeded maximum bytes of [" + maxBytes + "]");

przemekwitek · 2020-01-30T08:38:19Z

.../src/main/java/org/elasticsearch/xpack/core/ml/inference/utils/SimpleBoundedInputStream.java

@@ -38,6 +39,9 @@ public SimpleBoundedInputStream(InputStream inputStream, long maxBytes) {
    public int read() throws IOException {
        // We have reached the maximum, signal stream completion.
        if (numBytes >= maxBytes) {
+            if (throwWhenExceeded) {
+                throw new IOException("input stream exceeded maximum bytes of [" + maxBytes +"]");
+            }
            return -1;


Why do we sometimes throw and sometimes return -1? Would it be possible to have only one exit point?

przemekwitek

LGTM

przemekwitek · 2020-01-30T12:26:26Z

.../test/java/org/elasticsearch/xpack/core/ml/inference/InferenceToXContentCompressorTests.java

-            XContentType.JSON)) {
-            expectThrows(IOException.class, () -> TrainedModelConfig.fromXContent(parser, true));
-        }
+        expectThrows(IOException.class, () -> Streams.readFully(InferenceToXContentCompressor.inflate(firstDeflate, 10L)));


Should we also verify that the IOException has the message containing "input stream exceeded maximum bytes"?

droberts195

LGTM

Adjusting the code for my comment is not essential in the PR, but if you make any other change to the PR before merging then you might as well make my change too.

droberts195 · 2020-01-30T12:48:12Z

...e/src/main/java/org/elasticsearch/xpack/core/ml/inference/InferenceToXContentCompressor.java

        byte[] compressedBytes = Base64.getDecoder().decode(compressedString.getBytes(StandardCharsets.UTF_8));
+        // If the compressed length is already too large, it make sense that the inflated length would be as well


This comment isn't true in general. If you compress a very small string then the compressed size is bigger than the original. For example echo a | gzip -9 | wc -c returns 22.

The assumption is OK with the sort of streamSize values this method is going to be called with given the current code, so it's not essential to change now, but you could make it something like if (compressedBytes.length > Math.max(100L, streamSize)) in case it ever needs to cope with an extreme edge case in the future.

Also, it would be good to adjust the comment to acknowledge the edge case.

…eached (elastic#51644) Three fixes for when the `compressed_definition` is utilized on PUT * Update the inflate byte limit to be the minimum of 10% the max heap, or 1GB (what it was previously) * Stream data directly to the JSON parser, so if it is invalid, we don't have to inflate the whole stream to find out * Throw when the maximum bytes are reach indicating that is why the request was rejected

…eached (#51644) (#51681) Three fixes for when the `compressed_definition` is utilized on PUT * Update the inflate byte limit to be the minimum of 10% the max heap, or 1GB (what it was previously) * Stream data directly to the JSON parser, so if it is invalid, we don't have to inflate the whole stream to find out * Throw when the maximum bytes are reach indicating that is why the request was rejected

…eached (#51644) (#51679) Three fixes for when the `compressed_definition` is utilized on PUT * Update the inflate byte limit to be the minimum of 10% the max heap, or 1GB (what it was previously) * Stream data directly to the JSON parser, so if it is invalid, we don't have to inflate the whole stream to find out * Throw when the maximum bytes are reach indicating that is why the request was rejected

[ML][Inference] indicating when limit is reached, stream to parser

8728411

benwtrent added :ml Machine learning v8.0.0 v7.7.0 v7.6.1 labels Jan 29, 2020

benwtrent changed the title ~~[ML][Inference] indicating when limit is reached, stream to parser~~ [ML][Inference] stream inflate to parser + throw when byte limit is reached Jan 29, 2020

przemekwitek reviewed Jan 30, 2020

View reviewed changes

droberts195 added the >non-issue label Jan 30, 2020

addressing PR commens

1019208

benwtrent requested review from droberts195 and przemekwitek January 30, 2020 12:13

przemekwitek approved these changes Jan 30, 2020

View reviewed changes

droberts195 approved these changes Jan 30, 2020

View reviewed changes

adjusting test and throw conditions

3456259

benwtrent merged commit 8ea9aa2 into elastic:master Jan 30, 2020

benwtrent deleted the feature/ml-inference-adjust-zip-length-check branch January 30, 2020 14:19

benwtrent mentioned this pull request Jan 30, 2020

[7.x] [ML][Inference] stream inflate to parser + throw when byte limit is reached (#51644) #51679

Merged

benwtrent mentioned this pull request Jan 30, 2020

[7.6] [ML][Inference] stream inflate to parser + throw when byte limit is reached (#51644) #51681

Merged

benwtrent added v7.6.0 and removed v7.6.1 labels Jan 30, 2020

jakelandis added v8.0.0-alpha1 and removed v8.0.0 labels Jul 26, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ML][Inference] stream inflate to parser + throw when byte limit is reached #51644

[ML][Inference] stream inflate to parser + throw when byte limit is reached #51644

benwtrent commented Jan 29, 2020

elasticmachine commented Jan 29, 2020

przemekwitek Jan 30, 2020

droberts195 Jan 30, 2020

przemekwitek Jan 30, 2020

przemekwitek Jan 30, 2020

przemekwitek Jan 30, 2020

przemekwitek Jan 30, 2020

przemekwitek left a comment

przemekwitek Jan 30, 2020

droberts195 left a comment

droberts195 Jan 30, 2020

	throw new IOException("compressed stream is longer than maximum allowed bytes [" + streamSize +"]");
	throw new IOException("compressed stream is longer than maximum allowed bytes [" + streamSize + "]");

	throw new IOException("input stream exceeded maximum bytes of [" + maxBytes +"]");
	throw new IOException("input stream exceeded maximum bytes of [" + maxBytes + "]");

		byte[] compressedBytes = Base64.getDecoder().decode(compressedString.getBytes(StandardCharsets.UTF_8));
		// If the compressed length is already too large, it make sense that the inflated length would be as well

[ML][Inference] stream inflate to parser + throw when byte limit is reached #51644

[ML][Inference] stream inflate to parser + throw when byte limit is reached #51644

Conversation

benwtrent commented Jan 29, 2020

elasticmachine commented Jan 29, 2020

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

przemekwitek left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

droberts195 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment