Too many clients already using dagster-postgres

[Lukashab]

Hello, I have the dagster application with dagster-postgres as a storage deployed using the official helm chart.
Currently, I have about 10 recurring jobs, running each minute. The problem is, once the jobs run all in parallel, there is an obvious concurrence between the tasks regarding connection to the dagster storage (postgres in this case).

Exception :

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at "dagster-postgresql" (*****), port 5432 failed: FATAL:  sorry, too many clients already

(Background on this error at: https://sqlalche.me/e/20/e3q8)
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/api.py", line 764, in job_execution_iterator
    for event in job_context.executor.execute(job_context, execution_plan):
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/executor/in_process.py", line 55, in execute
    yield from iter(
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/api.py", line 875, in __iter__
    yield from self.iterator(
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/executor/in_process.py", line 26, in inprocess_execution_iterator
    yield from inner_plan_execution_iterator(
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_plan.py", line 62, in inner_plan_execution_iterator
    step = active_execution.get_next_step()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/active.py", line 299, in get_next_step
    steps = self.get_steps_to_execute(limit=1)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/active.py", line 361, in get_steps_to_execute
    if not self._instance_concurrency_context.claim(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/instance_concurrency_context.py", line 83, in claim
    if not self._instance.event_log_storage.supports_global_concurrency_limits:
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/storage/event_log/sql_event_log.py", line 2123, in supports_global_concurrency_limits
    return self.has_table(ConcurrencySlotsTable.name)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster_postgres/event_log/event_log.py", line 303, in has_table
    return bool(self._engine.dialect.has_table(self._engine.connect(), table_name))
                                               ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 3269, in connect
    return self._connection_cls(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 147, in __init__
    Connection._handle_dbapi_exception_noconnection(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2431, in _handle_dbapi_exception_noconnection
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 145, in __init__
    self._dbapi_connection = engine.raw_connection()
                             ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 3293, in raw_connection
    return self.pool.connect()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 452, in connect
    return _ConnectionFairy._checkout(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 1269, in _checkout
    fairy = _ConnectionRecord.checkout(pool)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 716, in checkout
    rec = pool._do_get()
          ^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/impl.py", line 284, in _do_get
    return self._create_connection()
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 393, in _create_connection
    return _ConnectionRecord(self)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 678, in __init__
    self.__connect()
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 902, in __connect
    with util.safe_reraise():
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 898, in __connect
    self.dbapi_connection = connection = pool._invoke_creator(self)
                                         ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/create.py", line 645, in connect
    return dialect.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 616, in connect
    return self.loaded_dbapi.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/psycopg2/__init__.py", line 122, in connect
    conn = _connect(dsn, connection_factory=connection_factory, **kwasync)

My question is, can this be avoided by some configuration on the dagster side? For example setting the pool size for the component that is connecting to dagster? Or do I have to either increase the connection limit or setup some pooling between the dagster and my postgres db?

Thank you

[jamiedemaria]

One option would be to limit the number of concurrent runs your deployment allows using concurrency limits https://docs.dagster.io/guides/limiting-concurrency-in-data-pipelines#limiting-overall-runs

[Lukashab]

@jamiedemaria I tried to limit the number of concurrent jobs, but it still sometimes fails. It seems that idle jobs still need to ask postgres for the job running info, so this limit does not affect the number of connections to postgres storage

[NoamMosko]

@Lukashab Hey, was this solved for you?

[gardner]

Looks like this is hard-coded in dagster_postgres without an option to configure it:

dagster/python_modules/libraries/dagster-postgres/dagster_postgres/run_storage/run_storage.py

Line 82 in 90e7048

# Default to not holding any connections open to prevent accumulating connections per DagsterInstance