Improve the esleg bulk response parsing

[alexsapran]

Proposed commit message

Currently, I am raising this PR in order to start a discussion with potential improvements I found while reviewing some Beat profiles.

While doing some synthetic benchmarks I noticed that we are using the package ioutil.ReadAll which is deprecated. Going down that path to replace I took a deeper look into how we do bulk requests using the esleg package.
I introduced a go benchmark to baseline my changes and here is the comparison between main and the proposed changes

goos: darwin
goarch: arm64
pkg: github.com/elastic/beats/v7/libbeat/esleg/eslegclient
                   │ baseline.txt │   initialize_per_connection.txt   │
                   │    sec/op    │   sec/op     vs base              │
ExecHTTPRequest-12    2.995µ ± 2%   2.905µ ± 1%  -3.01% (p=0.000 n=8)

                   │ baseline.txt │   initialize_per_connection.txt    │
                   │     B/op     │     B/op      vs base              │
ExecHTTPRequest-12   6.055Ki ± 0%   5.555Ki ± 0%  -8.26% (p=0.000 n=8)

                   │ baseline.txt │  initialize_per_connection.txt   │
                   │  allocs/op   │ allocs/op   vs base              │
ExecHTTPRequest-12     27.00 ± 0%   26.00 ± 0%  -3.70% (p=0.000 n=8)

We can see this by replacing the ioutil.ReadAll and adding reusable bytes.Buffer for consuming the response we managed to save on memory allocation.

Additionally, I will be pushing another commit, reducing the overall response we get from the _bulk request which is going to even further reduce the amount of memory we need to copy/allocate for the response.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

Author's Checklist

• [ ]

How to test this PR locally

Related issues

• Replace the ioutils.ReadAll usage #36151

Use cases

Screenshots

Logs

Signed-off-by: Alexandros, Sapranidis alexandros@elastic.co

[botelastic]

This pull request doesn't have a Team:<team> label.

[elasticmachine]

💚 Build Succeeded

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

• Start Time: 2023-08-22T14:29:27.262+0000

• Duration: 70 min 6 sec

Test stats 🧪

Test	Results
Failed	0
Passed	28075
Skipped	2015
Total	30090

💚 Flaky test report

Tests succeeded.

🤖 GitHub comments

Expand to view the GitHub comments

To re-run your PR in the CI, just comment with:

• /test : Re-trigger the build.

• /package : Generate the packages and run the E2E tests.

• /beats-tester : Run the installation tests with beats-tester.

• run elasticsearch-ci/docs : Re-trigger the docs validation. (use unformatted text in the comment!)

[ycombinator]

@leehinman Will this improvement automatically benefit the shipper or does the same change need to be made elsewhere for the shipper as well?

[leehinman]

@leehinman Will this improvement automatically benefit the shipper or does the same change need to be made elsewhere for the shipper as well?

should be automatic, when filebeat is acting as an elasticsearch shipper it is using the 'elasticsearch' output.

[alexsapran]

@ycombinator

does the same change need to be made elsewhere for the shipper as well?

A similar change needs to be made for other parts of the codebase that is using the ioutil.ReadAll which uses it to consume the response and allocate more memory than for example the io.copy

[ycombinator]

I think this field needs to be protected by a mutex.

Let's say two goroutines, G1 and G2 are running concurrently, and both end up calling conn.execHTTPRequest at some point in their execution. Now imagine this sequence of events:

1. G1 copies resp.Body into conn.responseBuffer (line 458).
2. G2 calls conn.responseBuffer.Reset() (line 457).
3. G1 returns conn.responseBuffer.Bytes() (line 468).

The result will be that G1 will return an empty slice of bytes.

[alexsapran]

You bring a valid concern here. Before going in and adding a mutex which is potentially defeating what we want to achieve and actually slowing down the process.
I might be wrong, but this is the assumption I had, each go routine would have its own connection. One connection would not be shared between 2 go routines, if 2 goroutines needed a connection to the ES they would create a new connection client thus a new buffer.
If we absolutely need to add a mutex, I would rather change the implementation because if there is a case that 2 go routines shared the same connection then adding a mutex would be similar to not having go routines in the first place.

[ycombinator]

Yes, you're right that as it stands today the only way two goroutines could concurrently call the execHTTPRequest method is if they were sharing the same Connection object. And from what I can tell, that is not happening anywhere in the code.

Besides checking this by eyeballing the code paths that lead to the execHTTPRequest method, I also ran go test -race ./libbeat/... -count 1 | grep -i 'data race' | wc -l from the root folder of the Beats repo, first in the main branch and then in this PR's branch. In both cases I got the same results, which is an quick sanity check that the changes in this PR are most likely not introducing additional data races:

Running in main

$ go test -race ./libbeat/... -count 1 | grep -i 'data race' | wc -l
      22

Running in this PR's branch

$ go test -race ./libbeat/... -count 1 | grep -i 'data race' | wc -l
      22

Side note: for the data races that are being reported on main, I filed #36393.

---

However, in the future, code may be written such that two concurrently-executing goroutines are sharing the same Connection object.

The safest thing to do here would be to add a mutex, but as you noted, this will likely cut into the performance gains we are seeing from the change in this PR.

The next best thing to do, IMO, would be to document clearly in a comment above the execHTTPRequest method AND in comments above each of the constructor functions that return a new Connection — NewConnection, NewClients, and NewConnectedClient — AND in a comment above the Connection struct itself that this method/struct is not threadsafe. You could add a link in your comments to this discussion here on GitHub in case someone wants to know all the details.

[alexsapran]

❤️ for the help here, I agree that race conditions are something to be looked-at at in different work streams.

However, in the future, code may be written such that two concurrently-executing goroutines are sharing the same Connection object.

I think hope that once we move away from the esleg and to a different output client, this part of the code should not be used anymore.

I added some comment about the thread-safety, please take a look at f181725

[ycombinator]

Comments LGTM. Thanks @alexsapran!

[ycombinator]

LGTM. Nice performance improvements!

[rdner]

@alexsapran I think we should not re-use the buffer here because the size is not static. This buffer will never release its allocated memory and the size of this memory will be equal to the size of the biggest ever processed HTTP response. I'd rather prefer to release the memory and re-allocate smaller buffers when needed.

I think we should switch it in favour of elastic/elastic-agent-libs#183 and compare performance. This implementation should be the most optimized one.

[rdner]

For the reference https://cs.opensource.google/go/go/+/refs/tags/go1.22.0:src/bytes/buffer.go;l=103-110

[alexsapran]

Yes, you are correct @rdner to the use case you mention, but this might be just fine in this particular scenario because the output is relational to the batching we do.

So the response size is within a ballpark the same for each bulk we publish.
We can adjust this and make it so that we only allocate a buffer just for the consumption of that response.
Since we are in the process of evaluating other usages of the ioutil.ReadAll it might make sense to adjust this as well.