Bugfix for BufferedTokenizer to completely consume lines in case of lines bigger then sizeLimit #16482
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…er instead of a JRuby array, added more tests to cover some 'buffer full' conditions
andsel
changed the title
jsvd
reviewed
Oct 14, 2024
There was a problem hiding this comment.
A few questions:
- there's quite a bit going on here and this is feature that is in the critical path, any clue on the performance impact of this PR?
- Since the size checks are done between the buffered data and the first message, it's possible for the codec to output messages larger than the setting if the incoming data never touches the buffer:
jruby-9.4.8.0 :019 > buffer = FileWatch::BufferedTokenizer.new("\n", 5)
=> #<FileWatch::BufferedTokenizer:0xff2f2ae>
jruby-9.4.8.0 :020 > buffer.extract("1234").each { |line| puts line; puts line.length }
=> []
jruby-9.4.8.0 :021 > buffer.extract("5\n0123456789\n").each { |line| puts line; puts line.length }
12345
5
0123456789
10
=> ["12345", "0123456789"]
This is OK if codecs and plugins clearly describe that the setting exposed to the user controls only the "leftover buffer size" and not maximum message size. This is hard for the user to understand, which will likely expect it to be message size, but they could set the limit to 5MB but still get a 50MB message.
An alternative would be to perform the size check much earlier in extract , maybe right after data.convertToString (this is half-baked thought), what do you think?
| throw new IllegalStateException("input buffer full"); | ||
| } | ||
| this.inputSize = inputSize + entitiesSize; | ||
| } | ||
| input.append(entities.shift(context)); | ||
| if (entities.isEmpty()) { | ||
| headToken.append(input.shift(context)); // remove head |
There was a problem hiding this comment.
Maybe we can avoid some overhead when we know there's no leftover content at the end of the incoming data:
if (input.getLength() < 2) {
headToken.append(input.shift(context)); // remove head
return RubyUtil.RUBY.newArray();
} else {
if (headToken.length() > 0) {
headToken.append(input.shift(context)); // append buffer to first element and
input.unshift(RubyUtil.toRubyObject(headToken.toString())); // reinsert it into the array
headToken = new StringBuilder();
}
headToken.append(input.pop(context)); // put the leftovers in headToken for later
inputSize = headToken.length();
return input;
}There was a problem hiding this comment.
We can do that but complicates even more the flow. Before applying that maybe a measure of effective performance gain should be done.
There was a problem hiding this comment.
About question 1, I haven't done any performance check on this. However, we should define the use cases to check. The buffer full condition in the existing implementation at certain point didn't provide correct results, so if we use the original implementation as baseline, we aren't testing apples-to-apples. The only thing we can test is aggregation process, while any token size doesn't fail the size limit.
I could think to:
- tokens that are smaller then a full segment
- tokens that span multiple segments
- tokens that are always spread between the end of a preceding segment and the subsequent one.
On point number 2, this is a bug. The expectation for the client of this tokenizer is that lines (or tokens) bigger than the predefined limit generates an exception and that the normal flow is able to resume on the next extract call immediately after the token delimiter that marks the end of the offending token. So if the token that should trigger the error is not the first of a segment, is must still raise the error.
There was a problem hiding this comment.
@andsel given my proposal improves performance should we add it to this PR?
There was a problem hiding this comment.
yes, I missed to add it
BenchmarkSummaryAs suggested by the flamegraphs that @jsvd measured, this implementation has a performance loss, and the big suspect is the copying of splitted data into the input accumulator array: https://github.com/elastic/logstash/pull/16482/files#diff-5c7f8990e98f54782395d29b4b1b5b68cf6f782b34af5eb8f1b5a77331e0172eR84. With this PR the total consumption of A variation < 5% could be considered negligible, in particular with the usage of the tokenizer inside an input codec, where the latencies of the network or the locks to access the in memory queue, could make a bigger impact in the overall performance. So we could assume that this PR doesn't significantly influence the overall performance. DetailsResult with performance metering of the
Raw baseline results
Bugfix results
Flamegraphs5 feeding pipelines feeding one pipeline with employed the With Logstash 8.15.1 With this PR The loader LS can be configured with these in - pipeline.id: test_1
pipeline.workers: 2
pipeline.batch.size: 200
config.string: "input { java_generator {} } filter { ruby { code => 'event.set(\"data\", \"*\"*rand(400))' } } output { tcp { port => 3333 host => localhost codec => line }}"
- pipeline.id: test_2
pipeline.workers: 2
pipeline.batch.size: 200
config.string: "input { java_generator {} } filter { ruby { code => 'event.set(\"data\", \"*\"*rand(400))' } } output { tcp { port => 3333 host => localhost codec => line }}"
- pipeline.id: test_3
pipeline.workers: 2
pipeline.batch.size: 200
config.string: "input { java_generator {} } filter { ruby { code => 'event.set(\"data\", \"*\"*rand(400))' } } output { tcp { port => 3333 host => localhost codec => line }}"
- pipeline.id: test_4
pipeline.workers: 2
pipeline.batch.size: 200
config.string: "input { java_generator {} } filter { ruby { code => 'event.set(\"data\", \"*\"*rand(400))' } } output { tcp { port => 3333 host => localhost codec => line }}"
- pipeline.id: test_5
pipeline.workers: 2
pipeline.batch.size: 200
config.string: "input { java_generator {} } filter { ruby { code => 'event.set(\"data\", \"*\"*rand(400))' } } output { tcp { port => 3333 host => localhost codec => line }}"While the system under test can be executed with bin/logstash -e "input { tcp { port => 3333 codec => line } } output { null {} }"Used the asynch porofiler from https://github.com/async-profiler/async-profiler running with bin/asprof -d 20 -i 10ms -f /tmp/flamegraph.html <LS PID> |
|
💛 Build succeeded, but was flaky
Failed CI StepsHistory
cc @andsel |
|
@logstashmachine backport 8.x |
github-actions bot
pushed a commit
that referenced
this pull request
Oct 16, 2024
…ines bigger then sizeLimit (#16482) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests. (cherry picked from commit 85493ce)
4 tasks
andsel
added a commit
that referenced
this pull request
Oct 16, 2024
… consume lines in case of lines bigger then sizeLimit (#16569) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests. (cherry picked from commit 85493ce) Co-authored-by: Andrea Selva <selva.andre@gmail.com>
|
@logstashmachine backport 7.17 |
1 similar comment
|
@logstashmachine backport 7.17 |
andsel
added a commit
to andsel/logstash
that referenced
this pull request
Oct 17, 2024
…ines bigger then sizeLimit (elastic#16482) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests.
5 tasks
|
@logstashmachine backport 8.15 |
github-actions bot
pushed a commit
that referenced
this pull request
Oct 17, 2024
…ines bigger then sizeLimit (#16482) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests. (cherry picked from commit 85493ce)
4 tasks
|
@logstashmachine backport 8.16 |
github-actions bot
pushed a commit
that referenced
this pull request
Oct 17, 2024
…ines bigger then sizeLimit (#16482) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests. (cherry picked from commit 85493ce)
4 tasks
edmocosta
pushed a commit
that referenced
this pull request
Oct 18, 2024
…ines bigger then sizeLimit (#16482) (#16579) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests. (cherry picked from commit 85493ce) Co-authored-by: Andrea Selva <selva.andre@gmail.com>
edmocosta
pushed a commit
that referenced
this pull request
Oct 22, 2024
…onsume lines in case of lines bigger then sizeLimit (#16577) * Bugfix for BufferedTokenizer to completely consume lines in case of lines bigger then sizeLimit (#16482) Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met. Updates BufferedTokenizerExt so that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter. Updated the accumulation variable from RubyArray containing strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in the input array. Furthermore it translates the `buftok_spec` tests into JUnit tests. * Fixed compilation error due to libraries incompatibilities - usage of `data.convertToString().split(context, delimiter, MINUS_ONE);` instead of `data.convertToString().split(delimiter, -1);` - avoid to extend BuffererdTokenir test cases from `org.logstash.RubyTestBase` which was introduced in #13159 - JDK 8 compatibilities: - `Arrays.asList` vs `List.of` - `assertThrows` method from JUnit5 not available in JUnit4 so reimplemented in the test
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Release notes
[rn:skip]
What does this PR do?
Updates
BufferedTokenizerExtso that can accumulate token fragments coming from different data segments. When a "buffer full" condition is matched, it record this state in a local field so that on next data segment it can consume all the token fragments till the next token delimiter.Updated the accumulation variable from
RubyArraycontaining strings to a StringBuilder which contains the head token, plus the remaining token fragments are stored in theinputarray.Port the tests present at
logstash/logstash-core/spec/logstash/util/buftok_spec.rb
Line 20 in f35e10d
Why is it important/What is the impact to the user?
Fixes the behaviour of the tokenizer to be able to work properly when buffer full conditions are met.
Checklist
[ ] I have made corresponding changes to the documentation[ ] I have made corresponding change to the default configuration files (and/or docker env variables)Author's Checklist
How to test this PR locally
Follow the instructions in #16483
Related issues
Use cases
Screenshots
Logs