Skip to content

Track completed triggers until deleted from database #19546

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dstandish wants to merge 4 commits into from
Closed

Track completed triggers until deleted from database #19546

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f28b560
Wait for task creation before deleting trigger
dstandish Nov 11, 2021
aef2c70
Revert "Wait for task creation before deleting trigger"
dstandish Nov 13, 2021
73f216d
use completed triggers list instead of managing state in the triggers…
dstandish Nov 13, 2021
4718ae2
don't need this one
dstandish Nov 13, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions airflow/jobs/triggerer_job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ def _run_trigger_loop(self) -> None:
while not self.runner.stop:
# Clean out unused triggers
Trigger.clean_unused()
self.purge_completed_triggers_list()
# Load/delete triggers
self.load_triggers()
# Handle events
Expand All @@ -148,6 +149,25 @@ def load_triggers(self):
ids = Trigger.ids_for_triggerer(self.id)
self.runner.update_triggers(set(ids))

@provide_session
def purge_completed_triggers_list(self, session):

@dstandish dstandish Nov 13, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it would be nice if we didn't have to query the database here but i don't think it can be avoided.

i explored updating clean_unused to return the ids deleted. then we could pass those to purge_completed and purge those rows. but this won't work because clean_unused does not filter w.r.t. triggerer_id, so other triggerer instances could delete the rows belonging to this triggerer, so completed_triggers would grow over time.

on the bright side, we're querying a PK, so it shouldn't be that expensive.

alternatively we could change clean_unused to take a triggerer id and delete only the rows for that triggerer, then the above approach would probably work.

@kaxil kaxil Dec 27, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
def purge_completed_triggers_list(self, session):
def purge_completed_triggers(self, session):

@dstandish dstandish Jan 4, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the reason i put _list there iss because without it, it could signify "purge complleted triggers from the database"

but it's only purging from the instance attribribute completed_triggers. the actual deletion of triggers from the database is done in TriggerJob.clean_unused

what do you think?

"""
The runner will wait to remove a trigger from its ``triggers`` dictionary until it is
purged from the ``completed_triggers`` list.
Once a trigger is deleted from the database, there's no chance it will be created again
and we can remove it.
"""
if self.runner.completed_triggers:
ids = {
trigger_id
for (trigger_id,) in session.query(Trigger.id)
.filter(Trigger.id.in_(self.runner.completed_triggers))
.all()
}
purge_ids = self.runner.completed_triggers.difference(ids)
if purge_ids:
self.runner.completed_triggers.difference_update(purge_ids)

def handle_events(self):
"""
Handles outbound events from triggers - dispatching them into the Trigger
Expand Down Expand Up @@ -213,12 +233,16 @@ class TriggerRunner(threading.Thread, LoggingMixin):
# Outbound queue of failed triggers
failed_triggers: Deque[int]

# Waiting area after triggers have completed but before they've been deleted from the database
completed_triggers: Set[int]

# Should-we-stop flag
stop: bool = False

def __init__(self):
super().__init__()
self.triggers = {}
self.completed_triggers = set()
self.trigger_cache = {}
self.to_create = deque()
self.to_delete = deque()
Expand Down Expand Up @@ -315,6 +339,7 @@ async def cleanup_finished_triggers(self):
details["name"],
)
self.failed_triggers.append(trigger_id)
self.completed_triggers.add(trigger_id)

@andrewgodwin andrewgodwin Nov 16, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It appears that completed_triggers will grow in an unbounded fashion with this code over time - I would suggest a datastructure that is self-limiting. My initial idea is a dict with the IDs as keys and the datetime they were inserted as values, and then just remove all keys whose values are more than 5 minutes old in the first part of purge_completed_triggers.

@dstandish dstandish Nov 17, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how do you envision it growing? the items should only be in completed_triggers for basically one or two loop cycles. it goes submit_event (into completed) -> Trigger.clean_unused (purged from db) -> purge_completed_triggers_list (purged from completed)

but maybe you're saying that in the wild the code may not behave as intended, and triggers will accumulate in the DB, and therefore they'll accumulate in the completed_triggers set? but if they accumulate in the db they'll keep getting recreated and eventually that would be an issue in itself because not only would they exist as ids in a set but they would be running.

using a TTL approach as you have suggested would avoid a db query though.

LMK your thoughts.

@dstandish dstandish Dec 8, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@andrewgodwin gentle nudge here. If you want to go with time-based expiration I have no objection and am happy to rework this. I do recognize that my solution adds a query but if you don't mind explaining I'm having trouble seeingthe unbound growth.

@andrewgodwin andrewgodwin Dec 8, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I just don't quite trust difference_update quite enough, I missed it in my first read-through for "what in here cleans out the datastructure?"

If you're happy there's no edge cases where it can leak entries and write a test to prove so, I think my objection here is void.

del self.triggers[trigger_id]
await asyncio.sleep(0)

Expand Down Expand Up @@ -399,6 +424,9 @@ def update_triggers(self, requested_trigger_ids: Set[int]):
# Either the trigger code or the path to it is bad. Fail the trigger.
self.failed_triggers.append(new_id)
continue
# the trigger recently completed and we should not create it again
if new_id in self.completed_triggers:
continue
self.to_create.append((new_id, trigger_class(**new_triggers[new_id].kwargs)))
# Remove old triggers
for old_id in old_trigger_ids:
Expand Down