Résoudre les problèmes de crash du consumer hydra lors de l'envoi de beaucoup de messages kakfa en simultanée

[geoffreyaldebert]

No description provided.

[geoffreyaldebert]

Il semblerait que le problème viennent de multiple connexions à postgres. https://github.com/etalab/udata-hydra/blob/aea95a80a25fb5977c1f4699bf6ef40f86e84937/udata_hydra/kafka/consumer.py#L14

Un Traceback de l'erreur :

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 243, in check_url
    await update_check_and_catalog(
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 131, in update_check_and_catalog
    await insert_check(check_data)
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 42, in insert_check
    async with pool.acquire() as connection:
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 986, in __aenter__
    self.connection = await self.pool._acquire(self.timeout)
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 822, in _acquire
    raise exceptions.InterfaceError('pool is closing')
asyncpg.exceptions._base.InterfaceError: pool is closing
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-44' coro=<check_url() done, defined at /usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py:176> exception=InterfaceError('pool is closing')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 221, in check_url
    await update_check_and_catalog(
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 131, in update_check_and_catalog
    await insert_check(check_data)
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 42, in insert_check
    async with pool.acquire() as connection:
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 986, in __aenter__
    self.connection = await self.pool._acquire(self.timeout)
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 822, in _acquire
    raise exceptions.InterfaceError('pool is closing')
asyncpg.exceptions._base.InterfaceError: pool is closing
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-89' coro=<check_url() done, defined at /usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py:176> exception=InterfaceError('pool is closing')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 212, in check_url
    async with _method(
  File "/usr/local/lib/python3.9/site-packages/aiohttp/client.py", line 1138, in __aenter__
    self._resp = await self._coro
  File "/usr/local/lib/python3.9/site-packages/aiohttp/client.py", line 634, in _request
    break
  File "/usr/local/lib/python3.9/site-packages/aiohttp/helpers.py", line 721, in __exit__
    raise asyncio.TimeoutError from None
asyncio.exceptions.TimeoutError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 256, in check_url
    await update_check_and_catalog(
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 131, in update_check_and_catalog
    await insert_check(check_data)
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 42, in insert_check
    async with pool.acquire() as connection:
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 986, in __aenter__
    self.connection = await self.pool._acquire(self.timeout)
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 822, in _acquire
    raise exceptions.InterfaceError('pool is closing')
asyncpg.exceptions._base.InterfaceError: pool is closing
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-87' coro=<check_url() done, defined at /usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py:176> exception=InterfaceError('pool is closing')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 221, in check_url
    await update_check_and_catalog(
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 131, in update_check_and_catalog
    await insert_check(check_data)
  File "/usr/local/lib/python3.9/site-packages/udata_hydra/crawl.py", line 42, in insert_check
    async with pool.acquire() as connection:
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 986, in __aenter__
    self.connection = await self.pool._acquire(self.timeout)
  File "/usr/local/lib/python3.9/site-packages/asyncpg/pool.py", line 822, in _acquire
    raise exceptions.InterfaceError('pool is closing')
asyncpg.exceptions._base.InterfaceError: pool is closing

[geoffreyaldebert]

Avec ce script : https://github.com/geoffreyaldebert/udata-event-orchestration/blob/master/testing/simulatedResourceEvent.py

On peut générer n événements python simulatedResourceEvent.py 1000 # exemple pour 1000 événements

[abulte]

I wonder if the consumer shouldn't be in a separate process. The pattern is very different between the crawler and the consumer and handling a common db connection/pool for both might prove challenging.

[abulte]

On a conceptual level, we might want to consume events without crawling and crawl without consuming events. I think two processes make sense.