Chart: Mount DAGs in triggerer

[jedcunningham]

We need to mount the DAGs in the triggerer, as triggers could come from
the DAGs themselves.

Example DAG to demonstrate this, named custom_trigger.py, in the root of the DAGS dir:

from datetime import datetime

from airflow import DAG
from airflow.models import BaseOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent


class CustomSuccessTrigger(BaseTrigger):
    def serialize(self):
        return ("custom_trigger.CustomSuccessTrigger", {})

    async def run(self):
        yield TriggerEvent(True)


class CustomOperator(BaseOperator):
    def execute(self, context):
        self.defer(trigger=CustomSuccessTrigger(), method_name="next")

    def next(self, context, event=None):
        return None


with DAG(
    "custom_trigger", schedule_interval=None, start_date=datetime(2021, 10, 1)
) as dag:
    CustomOperator(task_id="run")

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.