Repository navigation
Skip trigger timeout check on occasional db deadlocks - #33172
vchiapaikeo wants to merge 2 commits into
Conversation
vchiapaikeo
changed the title
75ac8b4 to
6317f32
Compare
|
I am not sure - but I believe this is not the right fix. With this query the problem is that it is simply not written in the way to grab the right lock on the DagRun while updating task instance table. I think (unlike some other queries and problems we have with deadlocks - this one could be written in the way that it could grab the locks - possibly with SKIP_LOCKED to make it less contentious with scheduler) and avoid the locks rather than reacting to them. |
|
Just to add some context: Scheduler only operates on task instances tha belong to DAGRuns that it managed to get "For Update" lock on. This means that any query that modifies a bulk of task instances is bound to hit the deadlock, because it might get a lock on one task instance to update, and then wait for dagrun, while scheduler will do that in reverse direction and will attempt to update the two instance in reverse order. |
|
Hmm yes, that makes sense. Would adding the below condition be sufficient to look at TI's queued by the current replica? So then the method would look something like this? @provide_session
def check_trigger_timeouts(self, session: Session = NEW_SESSION) -> None:
"""Mark any "deferred" task as failed if the trigger or execution timeout has passed."""
self.log.debug("Calling SchedulerJob.check_trigger_timeouts method")
try:
num_timed_out_tasks = session.execute(
update(TI)
.where(
TI.state == TaskInstanceState.DEFERRED,
TI.trigger_timeout < timezone.utcnow(),
# Only perform update against the ones that were queued by this scheduler
TI.queued_by_job_id == self.job.id,
)
.values(
state=TaskInstanceState.SCHEDULED,
next_method="__fail__",
next_kwargs={"error": "Trigger/execution timeout"},
trigger_id=None,
)
).rowcount
if num_timed_out_tasks:
self.log.info("Timed out %i deferred tasks without fired triggers", num_timed_out_tasks)
except OperationalError as e:
session.rollback()
self.log.warning(
f"Failed to check trigger timeouts due to {e}. Will reattempt at next scheduled check"
)The problem I see here is that there may be lingering triggers that do not get cleaned up if replicas get dropped. Maybe this overcomplicates the problem... |
Nope. It's different. IMHO you should not limit it by job_id, but you should add dag_run "FOR UPDATE" section with SKIP_LOCKED condition - basically to join the task_instance with dag_run they belong to, and make sure that dag_run is locked "for update". I am not super expert in the sqlalchemy queries to be able to tell exactly how it should be done, my knowledge is more based on the "relational DB knowledge" I have, and learning from @ashb's https://www.youtube.com/watch?v=DYC4-xElccE talk on how scheduler works internally - but there are quite a few such sqlalchemy queries in Airflow code that already do it I believe. Maybe others who know better could chime in as well? |
This comment was marked as off-topic.
This comment was marked as off-topic.
|
So it turns out that this issue actually stemmed from something completely unrelated. Our team will post a discussion / issue about that later. Long story short, we were using Relevant callsites:
Triggerer Model: |
|
That's a very interesting finding. Looks like it needs fixing indeed. |
|
Ping me on an issue when you create it @vchiapaikeo. |
3 participants
closes: #32698
We occasionally hit DB deadlocks during the trigger timeout check process. When this happens, the scheduler crashes. See linked issue for graphs, traceback, etc.. I see two potential options:
At first, I was leaning toward option 1 but after seeing it in code, I worry that the retries will cause even more deadlocks. That is, if a user set AIRFLOW__DATABASE__MAX_DB_RETRIES to a high number (default is 3) and since AIRFLOW__SCHEDULER__TRIGGER_TIMEOUT_CHECK_INTERVAL already defaults to a very low number, 15s, this could increase deadlock likelihood and cause even more stress on the db.
Therefore, going with 2. Please let me know what you think. This should allow us to avoid crashing the scheduler when deadlocks occur on this (not as critical) transaction.
Manual Testing
Started breeze with the following configurations set: