Memory leak when using AsyncElasticsearch

[teuneboon]

Memory leak when using AsyncElasticsearch

[pquentin]

Thanks @teuneboon, I can reproduce this! 🎉 My observations:

• Please don't do that. :) HTTP clients reuse connections to amortize the cost of TCP and TLS handshakes. Creating one new client per request defeats that. I still would like to fix the memory usage if possible. The slower leak when you create a single instance is probably a totally different issue.
• The required ingredients are 1/ AsyncElasticsearch 2/ SSL. The actual request does not matter (I reproduce with client.info()) and using a single node is also enough to reproduce. AsyncElasticsearch uses aiohttp, which has known memory leaks.
• After about ~30 seconds, the memory usage stabilizes (see figure), which suggests it's maybe not an actual leak, but a reference cycle. However, running gc.collect() in the loop did not help.

The next steps are using memray to understand the peak usage in more detail and trying to reproduce with aiohttp.

[pquentin]

Here's my current attempt with aiohttp:

import asyncio
import aiohttp

async def leaky():
    i = 0
    while i <= 1500:
        async with aiohttp.ClientSession() as session:
            async with session.get(
                "https://localhost:9200/",
                auth=aiohttp.BasicAuth("elastic", "changeme"),
                ssl=False,
            ) as response:
                assert response.status == 200
                await response.text()
        i += 1
        if i % 100 == 0:
            print(i)

if __name__ == "__main__":
    asyncio.run(leaky())

It inexplicably fails after 1000 connections with:

Traceback (most recent call last):
  File "/.../.virtualenvs/elasticsearch-py/lib64/python3.12/site-packages/aiohttp/connector.py", line 1173, in _create_direct_connection
    hosts = await asyncio.shield(host_resolved)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/.../.virtualenvs/elasticsearch-py/lib64/python3.12/site-packages/aiohttp/connector.py", line 884, in _resolve_host
    addrs = await self._resolver.resolve(host, port, family=self._family)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/.../.virtualenvs/elasticsearch-py/lib64/python3.12/site-packages/aiohttp/resolver.py", line 33, in resolve
    infos = await self._loop.getaddrinfo(
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib64/python3.12/asyncio/base_events.py", line 899, in getaddrinfo
    return await self.run_in_executor(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib64/python3.12/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib64/python3.12/socket.py", line 963, in getaddrinfo
    for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
OSError: [Errno 16] Device or resource busy

And only partly reproduces the leak:

[pquentin]

I just remembered that the upcoming release later this month will include HTTPX support, so I tried it too.

import asyncio
from elasticsearch import AsyncElasticsearch

async def leaky():
    i = 0
    while i <= 1500:
        async with AsyncElasticsearch(
            "https://localhost:9200",
            basic_auth=("elastic", "changeme"),
            verify_certs=False,
            node_class="httpxasync",
        ) as es:
            await es.info()
        i += 1
        if i % 100 == 0:
            print(i)

if __name__ == "__main__":
    asyncio.run(leaky())

There's still a leak, maybe? But it's smaller in terms of magnitude and has the same ceiling at some point.