Skip to content
By using AWS re:Post, you agree to the AWS re:Post Terms of Use
AWS re:Postre:Post
About re:Post

Airflow dropping connections to metadatabase

2

Hello!

We have that error every day, sometimes more often, sometimes less. The task gets stuck for 15 minutes in the status queue, then it starts to run and we get an error in the logs. I know that very similar to https://forums.aws.amazon.com/message.jspa?messageID=975621

Have

Class = mw1.medium

Scheduler count = 2

Maximum worker count = 10

Minimum worker count = 2

And set up

celery.sync_parallelism = 1

celery.worker_autoscale = 5,5

[2022-03-29 13:03:35,118: WARNING/ForkPoolWorker-4] Failed operation _store_result. Retrying 2 more times.
Traceback (most recent call last):
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
cursor, statement, parameters, context
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
cursor.execute(statement, parameters)
psycopg2.DatabaseError: could not receive data from server: Connection timed out

The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/celery/backends/database/__init__.py", line 47, in _inner
return fun(*args, **kwargs)
File "/usr/local/lib/python3.7/site-packages/celery/backends/database/__init__.py", line 117, in _store_result
task = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/orm/query.py", line 3535, in __iter__
return self._execute_and_instances(context)
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/orm/query.py", line 3560, in _execute_and_instances
result = conn.execute(querycontext.statement, self._params)
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/base.py", line 1011, in execute
return meth(self, multiparams, params)
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/base.py", line 1130, in _execute_clauseelement
distilled_params,
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/base.py", line 1317, in _execute_context
e, statement, parameters, cursor, context
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/base.py", line 1511, in _handle_dbapi_exception
sqlalchemy_exception, with_traceback=exc_info[2], from_=e
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
raise exception
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
cursor, statement, parameters, context
File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
cursor.execute(statement, parameters)
sqlalchemy.exc.DatabaseError: (psycopg2.DatabaseError) could not receive data from server: Connection timed out

[SQL: SELECT celery_taskmeta.id AS celery_taskmeta_id, celery_taskmeta.task_id AS celery_taskmeta_task_id, celery_taskmeta.status AS celery_taskmeta_status, celery_taskmeta.result AS celery_taskmeta_result, celery_taskmeta.date_done AS celery_taskmeta_date_done, celery_taskmeta.traceback AS celery_taskmeta_traceback
FROM celery_taskmeta
WHERE celery_taskmeta.task_id = %(task_id_1)s]
[parameters: {'task_id_1': '3a54bc75-847d-4538-9ddd-1615ca10ab82'}]
(Background on this error at: http://sqlalche.me/e/13/4xp6)
[2022-03-29 13:03:35,140: INFO/ForkPoolWorker-4] Executing command in Celery: ['airflow', 'tasks', 'run', 'load_mc_data_to_dwh', 'check_mc_files', 'manual__2022-03-29T08:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/load_mc_to_dwh.py']
[2022-03-29 13:03:35,140: INFO/ForkPoolWorker-4] Celery task ID: 3a54bc75-847d-4538-9ddd-1615ca10ab82
[2022-03-29 13:03:35,160: INFO/ForkPoolWorker-4] Filling up the DagBag from /usr/local/airflow/dags/load_mc_to_dwh.py
[2022-03-29 13:03:35,387: WARNING/ForkPoolWorker-4] Running <TaskInstance: load_mc_data_to_dwh.check_mc_files manual__2022-03-29T08:00:00+00:00 [queued]> on host ip-10-0-1-129.eu-west-2.compute.internal
[2022-03-29 13:03:37,115: INFO/ForkPoolWorker-4] Task airflow.executors.celery_executor.execute_command[3a54bc75-847d-4538-9ddd-1615ca10ab82] succeeded in 940.2485898699997s: None

Thanks,

Marko

Topics
Application Integration
Tags
Amazon Managed Workflows for Apache Airflow (MWAA)
Language
English

asked 4 years ago3.7K views

4 Answers
1
Accepted Answer

The key is setting celery.pool equal to solo

Here's our full config:


celery.pool	solo
celery.sync_parallelism	1
celery.worker_autoscale	1,1
core.dag_file_processor_timeout	150
core.dagbag_import_timeout	90
core.killed_task_cleanup_time	604800
core.min_serialized_dag_update_interval	60
scheduler.dag_dir_list_interval	300
scheduler.min_file_process_interval	300
scheduler.parsing_processes	2
scheduler.processor_poll_interval	60
scheduler.schedule_after_task_execution	false

answered 4 years ago

  • AWS-User-5132740
    4 years ago

    hello. Thanks you very much, seems like works. I will write here in few days if everything ok.

0

Hello,

From the error stack trace I see connection timed out issues for the PostGres connection

File "/usr/local/lib64/python3.7/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute cursor.execute(statement, parameters) sqlalchemy.exc.DatabaseError: (psycopg2.DatabaseError) could not receive data from server: Connection timed out

--> But, since you also mentioned that the tasks get stuck for 15 min in status queue and will start running later, I would request for more information from your MWAA environment if it's running multiple DAGs ?

--> I see that the configuration you currently have for celery.worker_autoscale = 5,5, can you please confirm if you are facing the same issue even if the property is set to celery.worker_autoscale = 1,1 ?

This issue needs deeper troubleshooting. I would recommend you to please open a support ticket with MWAA support providing the environment ARN and the time stamp when you faced this issue and we can assist you over screen-share.

REFERENCES:

  1. https://docs.aws.amazon.com/mwaa/latest/userguide/mwaa-autoscaling.html#mwaa-autoscaling-high-volume
AWS
SUPPORT ENGINEER

answered 4 years ago

  • AWS-User-5132740
    4 years ago

    Hi, thank you very match for the replay. Yes with celery.worker_autoscale = 1,1 issue still there. more information from your MWAA environment if it's running multiple DAGs Yes, we run a few DAGs, even though we have a DAG that runs another DAGs. But this issue faced us when we had run only one simple DAG.

0

We have the same scenario as Kayelvo. We set celery.pool to solo, which "fixed" the 15/30 min queuing (task delay) issue with the associated timeout errors connecting to the metadata database. However, as soon as we run a few DAGs with a handful of tasks running in parallel, we start seeing tasks delayed. Thus we would also like the root cause of the connection issues resolved so that we can use the default prefork setting. We cannot current set up a truly functional new environment at this time due to this issue (both version 2.0 and 2.2).

answered 4 years ago

  • AWS-User-5132740
    4 years ago

    we start seeing tasks delayed Hi, looks like its time to up the new worker.

0

Setting celery.pool equal to solo also worked for me. However, my DAGs runs for a longer time when celery.pool is set to solo since it doesn't use multiprocessing. I would like to use the default celery.pool setting of prefork.

I am noticing this issue on both Airflow 2.2.2 and 2.0.2.

Please have this bug resolved for us.

answered 4 years ago

You are not logged in. Log in to post an answer.

A good answer clearly answers the question and provides constructive feedback and encourages professional growth in the question asker.

Relevant content