Source code for airflow.models.trigger

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
import datetime
from traceback import format_exception
from typing import Any, Dict, Iterable, Optional

from sqlalchemy import Column, Integer, String, func, or_

from airflow.models.base import Base
from airflow.models.taskinstance import TaskInstance
from airflow.triggers.base import BaseTrigger
from airflow.utils import timezone
from airflow.utils.retries import run_with_db_retries
from airflow.utils.session import provide_session
from airflow.utils.sqlalchemy import ExtendedJSON, UtcDateTime
from airflow.utils.state import State


[docs]class Trigger(Base): """ Triggers are a workload that run in an asynchronous event loop shared with other Triggers, and fire off events that will unpause deferred Tasks, start linked DAGs, etc. They are persisted into the database and then re-hydrated into a "triggerer" process, where many are run at once. We model it so that there is a many-to-one relationship between Task and Trigger, for future deduplication logic to use. Rows will be evicted from the database when the triggerer detects no active Tasks/DAGs using them. Events are not stored in the database; when an Event is fired, the triggerer will directly push its data to the appropriate Task/DAG. """
[docs] __tablename__ = "trigger"
[docs] id = Column(Integer, primary_key=True)
[docs] classpath = Column(String(1000), nullable=False)
[docs] kwargs = Column(ExtendedJSON, nullable=False)
[docs] created_date = Column(UtcDateTime, nullable=False)
[docs] triggerer_id = Column(Integer, nullable=True)
def __init__( self, classpath: str, kwargs: Dict[str, Any], created_date: Optional[datetime.datetime] = None ): super().__init__() self.classpath = classpath self.kwargs = kwargs self.created_date = created_date or timezone.utcnow() @classmethod
[docs] def from_object(cls, trigger: BaseTrigger): """ Alternative constructor that creates a trigger row based directly off of a Trigger object. """ classpath, kwargs = trigger.serialize() return cls(classpath=classpath, kwargs=kwargs)
@classmethod @provide_session
[docs] def bulk_fetch(cls, ids: Iterable[int], session=None) -> Dict[int, "Trigger"]: """ Fetches all of the Triggers by ID and returns a dict mapping ID -> Trigger instance """ return {obj.id: obj for obj in session.query(cls).filter(cls.id.in_(ids)).all()}
@classmethod @provide_session
[docs] def clean_unused(cls, session=None): """ Deletes all triggers that have no tasks/DAGs dependent on them (triggers have a one-to-many relationship to both) """ # Update all task instances with trigger IDs that are not DEFERRED to remove them for attempt in run_with_db_retries(): with attempt: session.query(TaskInstance).filter( TaskInstance.state != State.DEFERRED, TaskInstance.trigger_id.isnot(None) ).update({TaskInstance.trigger_id: None}) # Get all triggers that have no task instances depending on them... ids = [ trigger_id for (trigger_id,) in ( session.query(cls.id) .join(TaskInstance, cls.id == TaskInstance.trigger_id, isouter=True) .group_by(cls.id) .having(func.count(TaskInstance.trigger_id) == 0) ) ] # ...and delete them (we can't do this in one query due to MySQL) session.query(Trigger).filter(Trigger.id.in_(ids)).delete(synchronize_session=False)
@classmethod @provide_session
[docs] def submit_event(cls, trigger_id, event, session=None): """ Takes an event from an instance of itself, and triggers all dependent tasks to resume. """ for task_instance in session.query(TaskInstance).filter( TaskInstance.trigger_id == trigger_id, TaskInstance.state == State.DEFERRED ): # Add the event's payload into the kwargs for the task next_kwargs = task_instance.next_kwargs or {} next_kwargs["event"] = event.payload task_instance.next_kwargs = next_kwargs # Remove ourselves as its trigger task_instance.trigger_id = None # Finally, mark it as scheduled so it gets re-queued task_instance.state = State.SCHEDULED
@classmethod @provide_session
[docs] def submit_failure(cls, trigger_id, exc=None, session=None): """ Called when a trigger has failed unexpectedly, and we need to mark everything that depended on it as failed. Notably, we have to actually run the failure code from a worker as it may have linked callbacks, so hilariously we have to re-schedule the task instances to a worker just so they can then fail. We use a special __fail__ value for next_method to achieve this that the runtime code understands as immediate-fail, and pack the error into next_kwargs. TODO: Once we have shifted callback (and email) handling to run on workers as first-class concepts, we can run the failure code here in-process, but we can't do that right now. """ for task_instance in session.query(TaskInstance).filter( TaskInstance.trigger_id == trigger_id, TaskInstance.state == State.DEFERRED ): # Add the error and set the next_method to the fail state traceback = format_exception(type(exc), exc, exc.__traceback__) if exc else None task_instance.next_method = "__fail__" task_instance.next_kwargs = {"error": "Trigger failure", "traceback": traceback} # Remove ourselves as its trigger task_instance.trigger_id = None # Finally, mark it as scheduled so it gets re-queued task_instance.state = State.SCHEDULED
@classmethod @provide_session
[docs] def ids_for_triggerer(cls, triggerer_id, session=None): """Retrieves a list of triggerer_ids.""" return [row[0] for row in session.query(cls.id).filter(cls.triggerer_id == triggerer_id)]
@classmethod @provide_session
[docs] def assign_unassigned(cls, triggerer_id, capacity, session=None): """ Takes a triggerer_id and the capacity for that triggerer and assigns unassigned triggers until that capacity is reached, or there are no more unassigned triggers. """ from airflow.jobs.base_job import BaseJob # To avoid circular import count = session.query(cls.id).filter(cls.triggerer_id == triggerer_id).count() capacity -= count if capacity <= 0: return alive_triggerer_ids = [ row[0] for row in session.query(BaseJob.id).filter( BaseJob.end_date.is_(None), BaseJob.latest_heartbeat > timezone.utcnow() - datetime.timedelta(seconds=30), BaseJob.job_type == "TriggererJob", ) ] # Find triggers who do NOT have an alive triggerer_id, and then assign # up to `capacity` of those to us. trigger_ids_query = ( session.query(cls.id) # notin_ doesn't find NULL rows .filter(or_(cls.triggerer_id.is_(None), cls.triggerer_id.notin_(alive_triggerer_ids))) .limit(capacity) .all() ) session.query(cls).filter(cls.id.in_([i.id for i in trigger_ids_query])).update( {cls.triggerer_id: triggerer_id}, synchronize_session=False, ) session.commit()

Was this entry helpful?