Hey I m pretty new to Prefect just started at a new job on M Prefect Community #ask-community

Hey, I'm pretty new to Prefect (just started at a ...

Scott Melhop

11/16/2023, 3:15 PM

Hey, I'm pretty new to Prefect (just started at a new job on Monday 🙂 ) I noticed some issues with one of our DB servers, which has the Prefect DB hosted on it, and it seems like its the Prefect DB thats hammering it. I looked at the Prefect Server logs and I have a

CancellationCleanup

warning constantly and regular DB timeouts, as shown in the thread. As a newbie I would appreciate any ideas on where to poke 🙏

Scott Melhop

11/16/2023, 3:15 PM

Copy code

21:11:48.947 | ERROR   | prefect.server.services.cancellationcleanup - Unexpected error in: TimeoutError()
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/server/services/loop_service.py", line 78, in start
    await self.run_once()
  File "/usr/local/lib/python3.11/site-packages/prefect/server/database/dependencies.py", line 119, in async_wrapper
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/server/services/cancellation_cleanup.py", line 48, in run_once
    await self.clean_up_cancelled_subflow_runs(db)
  File "/usr/local/lib/python3.11/site-packages/prefect/server/services/cancellation_cleanup.py", line 100, in clean_up_cancelled_subflow_runs
    await self._cancel_subflow(db=db, flow_run=subflow_run)
  File "/usr/local/lib/python3.11/site-packages/prefect/server/services/cancellation_cleanup.py", line 141, in _cancel_subflow
    parent_task_run = await models.task_runs.read_task_run(
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/server/database/dependencies.py", line 119, in async_wrapper
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/server/models/task_runs.py", line 166, in read_task_run
    model = await session.get(db.TaskRun, task_run_id)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/ext/asyncio/session.py", line 604, in get
    result_obj = await greenlet_spawn(
                 ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 190, in greenlet_spawn
    result = context.throw(*sys.exc_info())
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 3571, in get
    return self._get_impl(
           ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 3700, in _get_impl
    return db_load_fn(
           ^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/loading.py", line 666, in load_on_pk_identity
    session.execute(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 2262, in execute
    return self._execute_internal(
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 2144, in _execute_internal
    result: Result[Any] = compile_state_cls.orm_execute_statement(
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/context.py", line 293, in orm_execute_statement
    result = conn.execute(
             ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1412, in execute
    return meth(
           ^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 516, in _execute_on_connection
    return connection._execute_clauseelement(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1635, in _execute_clauseelement
    ret = self._execute_context(
          ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1844, in _execute_context
    return self._exec_single_context(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1984, in _exec_single_context
    self._handle_dbapi_exception(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2342, in _handle_dbapi_exception
    raise exc_info[1].with_traceback(exc_info[2])
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1965, in _exec_single_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 921, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 585, in execute
    self._adapt_connection.await_(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 125, in await_only
    return current.driver.switch(awaitable)  # type: ignore[no-any-return]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 185, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 522, in _prepare_and_execute
    await adapt_connection._start_transaction()
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 855, in _start_transaction
    self._handle_exception(error)
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 804, in _handle_exception
    raise error
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 853, in _start_transaction
    await self._transaction.start()
  File "/usr/local/lib/python3.11/site-packages/asyncpg/transaction.py", line 146, in start
    await self._connection.execute(query)
  File "/usr/local/lib/python3.11/site-packages/asyncpg/connection.py", line 317, in execute
    return await self._protocol.query(query, timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "asyncpg/protocol/protocol.pyx", line 338, in query
TimeoutError
21:11:48.952 | WARNING | prefect.server.services.cancellationcleanup - CancellationCleanup took 252.478202 seconds to run, which is longer than its loop interval of 20.0 seconds.

Bianca Hoch

11/16/2023, 7:53 PM

Hey Scott, thanks for sharing the stack trace. It looks like the CancellationCleanup is causing a timeout to occur:`252.478202 seconds to run, which is longer than its loop interval of 20.0 seconds.`

Bianca Hoch

11/16/2023, 7:55 PM

What version of Prefect are you using for the db? There may be a fix (#9124) in version 2.10.3 that may help you out

Scott Melhop

11/17/2023, 10:09 AM

We are running version 2.13.4

Bianca Hoch

11/21/2023, 5:54 PM

Hmmm..in that case, it's worth a shot updating to the latest version to see if there is any improvements in regard to

CancellationCleanup

Bianca Hoch

11/21/2023, 5:55 PM

You cloud also try raising the value for this environment variable to increase the loop interval for cancellation cleanup: https://docs.prefect.io/2.14.5/api-ref/prefect/settings/#prefect.settings.PREFECT_API_SERVICES_CANCELLATION_CLEANUP_LOOP_SECONDS

Bianca Hoch

11/21/2023, 6:00 PM

If the above doesn't work, opening an issue may be the next best step. Prefect Cloud is also an alternative to self-hosting the orchestration API and DB. You may consider exploring that as an option for running your flows.

4 Views

Open in Slack

Previous Next