AIP-40: Deferrable Tasks

[andrewgodwin]

This is the implementation of AIP-40, Deferrable "Async" Tasks (https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=177050929).

The main changes are:

• A new concept of a Trigger is introduced, as a small piece of asyncio code that can fire off events
  • There is a BaseTrigger and some time-related triggers under a new airflow.triggers package
  • There is a new Trigger database model and associated trigger table
  • Async versions of the various date/time sensors have been added which defer rather than poke.
• There is a new persistent process (Job) called triggerer
  • It only runs on Python 3.7+
  • It handles polling the database for which triggers need running, running them, and re-scheduling task instances whose triggers have fired events
  • If a trigger throws an exception or exits without firing an event, it logs why and marks dependent task instances as failed
  • It monitors the asyncio event loop with a watchdog task and alerts the user if anything is overrunning (i.e. not using await) and blocking the loop.
  • It is designed to run in parallel with itself in a highly-available manner, and also has built-in consistent-hash based partitioning (sharding) support
• Task Instances have a new deferred state which indicates they are waiting on a trigger to run
  • The trigger they are waiting for is stored in a new trigger_id column, and a failure timeout is in a trigger_timeout column
  • The scheduler takes care of timing out task instances into the failed state
  • Deferral is triggered by raising the TaskDeferred exception, or calling self.defer on the TaskInstance which does the same thing.
  • A next_method and next_kwargs column are added to specify what a task instance/operator's execution entry point should be if it's not the default of execute(). They are currently only used by deferral, but have been written to be independent in case they are useful elsewhere.
• Two new dependencies are added
  • jump-consistent-hash is a small MIT licensed library that implements a fast, consistent hash algorithm
  • pytest-asyncio is an Apache 2 licensed library that enables async tests to be written easily

Changes that are deliberately not in here and will be in a future PR for them specifically:

• UI warning when the triggerer is not running and you have deferred task instances
• Updating Breeze to include triggerer in what it runs
• Updating the Docker Compose files to include triggerer
• Updating the Helm Chart to include triggerer
• Some way of detecting/preventing DB access within triggers

Remaining fixes:

• SQLite does not seem to like deferral
• Deferred tasks should fail if their trigger fails to load

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. Building image for the PR has been cancelled

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. Building image for the PR has been cancelled

[jedcunningham]

Suggested change

	self.capacity = conf.getint('triggerer', 'default_capacity', fallback=1000)
	self.capacity = conf.getint('triggerer', 'default_capacity')

(Eventually I'll start my crusade against fallback, but this will mean 1 less when that time comes)

Stale review - we can followup with more changes

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[kaxil]

Looks good to me -- I think we can follow up if there are more changes