You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/12/15 04:14:30 UTC
[GitHub] [airflow] Greetlist opened a new issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Greetlist opened a new issue #20308:
URL: https://github.com/apache/airflow/issues/20308
### Apache Airflow version
2.2.1
### What happened
I find some Deferrable Operators are marked failed unexpectedly. When I clear these tasks, some of them will success, but some of them will fail as before.Even I clear the same task, sometime it is marked success, sometime it is marked failed.
I find this log in scheduler log
```
ERROR[0m - Executor reports task instance <TaskInstance: x [queued]> finished (success) although the task says its queued. (Info: None) Was the task killed externally?[0m
[[34m2021-12-14 10:22:20,208[0m] {[34mtaskinstance.py:[0m1706} ERROR[0m - Executor reports task instance <TaskInstance: x [queued]> finished (success) although the task says its queued. (Info: None) Was the task killed externally?[0m
[[34m2021-12-14 10:22:20,215[0m] {[34mtaskinstance.py:[0m1281} INFO[0m - Marking task as FAILED. dag_id=x, task_id=x, execution_date=20211212T170000, start_date=20211214T102218, end_date=20211214T102220[0m
```
### What you expected to happen
When I clear a Deferrable Operator task, task final state must be the same instead of sometime it is marked success, sometime it is marked failed.
### How to reproduce
Create a Deferrable Operator and this operator's run function **will yield immediately**.Like this:
```
class ExampleTrigger(BaseTrigger):
async def run(self):
while datetime.now() > datetime.strptime("1900-01-01", "%Y-%m-%d"):
await asyncio.sleep(1)
yield TriggerEvent('success')
```
In my case, I just query a task's state from another airflow database, and the execution_date(old version airflow) is prev_day [*would yield immediately*]
This task will be mark failed by scheduler randomly.
### Operating System
Ubuntu 16.04
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other
### Deployment details
1. One Triggerer process.
2. Two Schedulers.
3. One WebServer.
### Anything else
I check the code of:
1. scheduler
2. trigger
3. taskinstance
4. timestamp of log
and figure out what happend in this condition:
Assume TA is a deferrable task.
1. Scheduler loop 1 change TA task state: SCHEDULED -> QUEUED -> RUNNING -> DEFERRED, but this scheduler loop do not quit.
2. Meanwhile Trigger change TA task state: DEFERRED -> SCHEDULED
3. Scheduler loop 2 start, find the TA task is SCHEDULED and then change TA task state: SCHEDULED -> QUEUED
4. Scheduler loop 1 start to run the *_process_executor_events* function to check Worker-Process return status, and fetch a **WRONG state QUEUED** from database, finally run into this code, and mark the task as failed.
5. Scheduler loop 2 enqueue the *airflow task run xxx* command and then worker also fetch a failed task to execute.
In my opinion, I think there are two reasons cause this problem:
1. the query in *_executable_task_instances_to_queued* function:
can not guarantee one task is executed only in one scheduler loop.
2. The Trigger-Process can modify this task state to SCHEDULED asynchronously.
**How to solve this problem**
1. Add a new column in task_instance table named is_still_processing default is False
2. While query SCHEDULED task, add a filter like ```TI.is_still_processing == False``` at
3. Before enqueue task instance to executor queue, change is_still_processing to True in database.
4. At the end of *_executable_task_instances_to_queued* loop:
change this task's is_still_processing to False in database.
If I can guarantee one Deferrable Operator task is executed only in one scheduler loop and quit successfully,I can solve this problem.
I think only Deferrable Operator task which can yield immediately have this problem.
or
Just change task instance state to SUCCESS instead of SCHEDULED, but this Deferrable task's next_method has no chance to be executed.
Any Suggestions? I need to solve this problem ASAP.
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Greetlist commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
Greetlist commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057740624
I doubt if this code would release the lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
potiuk commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1055554824
@andrewgodwin - maybe there is an extra commit somewhere that releases the lock. I know @ashb implemented mechanisms to discover that that has been used in scheduler row-level locking - maybe we should use it here (unless it is done already ?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Greetlist commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
Greetlist commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1005435631
At this moment, I solve this problem by this:
Pass *custom_skip* args to method_name when invoking the **defer** function.
Tricky solution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk edited a comment on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
potiuk edited a comment on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057869795
Actually It might, in case Lock is kept as an object in the session. I am not sure what kind of "objects" are kept in the session. Funny thing that statement about in the documentation of SQLAlchemy is a bit ambiguous. I am not a native speaker but IMHO it's not obvious at all oif "and releases any transactional/connection resources" refers to "close()" or "expunge_all()" method.
At a first glance, it should be about "close()", but if Locks are represented as objects in the session, expunge_all() might also release locks. WDYT @andrewgodwin @ashb ?
https://docs.sqlalchemy.org/en/14/orm/session_basics.html#closing
> The Session.close() method issues a Session.expunge_all() which removes all ORM-mapped objects from the session, and releases any transactional/connection resources from the Engine object(s) to which it is bound. When connections are returned to the connection pool, transactional state is rolled back as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] akaa07 commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
akaa07 commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1046405909
I am having the same problem.
Multiple delayable operators randomly terminate abnormally.
Since I am using MWAA, there is no workaround solution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057052689
https://github.com/apache/airflow/blob/aa2cb5545f09d694b9143b323efcd4f6b6c66e60/airflow/utils/sqlalchemy.py#L284-L303 if it is useful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
potiuk commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057949483
> The lock is not an object -- it is a database side thing tied to the open transaction. The only way of releasing the lock is to COMMIT or ROLLBACK the DB session.
Yeah. It seems that even you can't explicitly release such lock even if you want, so it would be impossible to implement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
potiuk commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057869795
Actually It might in case Lock is kept as an object in the session. I am not sure what kind of "objects" are kept in the session. Funny thing that statement about in the documentation of SQLAlchemy is a bit ambiguous. I am not a native speaker but IMHO it's not obvious at all oif "and releases any transactional/connection resources" refers to "close()" or "expunge_all()" method.
At a first glance, it should be about "close()", but if Locks are represented as objects in the session, expunge_all() might also release locks. WDYT @andrewgodwin @ashb ?
https://docs.sqlalchemy.org/en/14/orm/session_basics.html#closing
> The [Session.close()](https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.close) method issues a [Session.expunge_all()](https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session.expunge_all) which removes all ORM-mapped objects from the session, and [releases](https://docs.sqlalchemy.org/en/14/glossary.html#term-releases) any transactional/connection resources from the [Engine](https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.Engine) object(s) to which it is bound. When connections are returned to the connection pool, transactional state is rolled back as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] andrewgodwin commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
andrewgodwin commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1054700837
Hmm, I would have thought the scheduler row locking would have prevented this. What database are you using to run against?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057960984
(I haven't been following the context, so this suggestion may make no sense).
One option might be to use savepoints --
```sql
SAVEPOINT before_lock;
SELECT * from trigger FOR UPDATE;
ROLLBACK TO before_lock;
-- Lock would be released now.
```
That rollback would only work if there have been no changes flushed.
(SQLA has the means to do this as "nested" transactions so we don't need to resort to raw SQL)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
potiuk commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1052679061
cc: @dstandish @andrewgodwin - seems like a likely case to take a look at.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Greetlist commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
Greetlist commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1054980898
The DataBase is in K8S, and Image is: ```Percona Server 8.0.17```
I make a test on test DataBase whose version is ```(MySQL Community Server - GPL 8.0.26)```.
``` for update ``` do lock these rows when I try to modify one of these rows at another session.
As I comment before, I fix it with a tricky way.Prevent the deferrable task is executed twice, this problem is never show again.
It seem that the trigger process do modify task_instance record even the record locked by ``` begin; for update; commit ```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057881165
The lock is not an object -- it is a database side thing tied to the open transaction. The only way of releasing the lock is to COMMIT or ROLLBACK the DB session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk edited a comment on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
potiuk edited a comment on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1057869795
Actually It might, in case Lock is kept as an object in the session. I am not sure what kind of "objects" are kept in the session. Funny thing that statement about in the documentation of SQLAlchemy is a bit ambiguous. I am not a native speaker but IMHO it's not obvious at all if "and releases any transactional/connection resources" refers to "close()" or "expunge_all()" method.
At a first glance, it should be about "close()", but if Locks are represented as objects in the session, expunge_all() might also release locks. WDYT @andrewgodwin @ashb ?
https://docs.sqlalchemy.org/en/14/orm/session_basics.html#closing
> The Session.close() method issues a Session.expunge_all() which removes all ORM-mapped objects from the session, and releases any transactional/connection resources from the Engine object(s) to which it is bound. When connections are returned to the connection pool, transactional state is rolled back as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Greetlist commented on issue #20308: Multi Schedulers don't synchronize task instance state when using Triggerer
Posted by GitBox <gi...@apache.org>.
Greetlist commented on issue #20308:
URL: https://github.com/apache/airflow/issues/20308#issuecomment-1058771937
I re-scan the **scheduler_job.py, models/trigger.py ** again.
``` _do_scheduling ``` function has RowLock, ``` _process_executor_events ``` function also has RowLock and two functions use the same session.
https://github.com/apache/airflow/blob/17d3e78e1b4011267e81846b5d496769934a5bcc/airflow/models/trigger.py#L109 has no RowLock, it will select some records which already selected by scheduler.But it will update these records, will wait the RowLock, wait until one scheduler loop is done.
The process above is make sense. But it do has inconsistent problem.I have no idea where the key is.
Confusing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org