airflow/models/dagrun.py
def update_state(self, session=None):
        """
        Determines the overall state of the DagRun based on the state
        of its TaskInstances.

        :return: State
        """

        dag = self.get_dag()

        tis = self.get_task_instances(session=session)
        self.log.debug("Updating state for %s considering %s task(s)", self, len(tis))

        for ti in list(tis):
            # skip in db?
            if ti.state == State.REMOVED:
                tis.remove(ti)
            else:
                ti.task = dag.get_task(ti.task_id)

        # pre-calculate
        # db is faster
        start_dttm = timezone.utcnow()
        unfinished_tasks = self.get_task_instances(
            state=State.unfinished(),
            session=session
        )
        none_depends_on_past = all(not t.task.depends_on_past for t in unfinished_tasks)
        none_task_concurrency = all(t.task.task_concurrency is None
                                    for t in unfinished_tasks)
        # small speed up
        if unfinished_tasks and none_depends_on_past and none_task_concurrency:
            # todo: this can actually get pretty slow: one task costs between 0.01-015s
            no_dependencies_met = True
            for ut in unfinished_tasks:
                # We need to flag upstream and check for changes because upstream
                # failures/re-schedules can result in deadlock false positives
                old_state = ut.state
                deps_met = ut.are_dependencies_met(
                    dep_context=DepContext(
                        flag_upstream_failed=True,
                        ignore_in_retry_period=True,
                        ignore_in_reschedule_period=True),
                    session=session)
                if deps_met or old_state != ut.current_state(session=session):
                    no_dependencies_met = False
                    break

        duration = (timezone.utcnow() - start_dttm).total_seconds() * 1000
        Stats.timing("dagrun.dependency-check.{}".format(self.dag_id), duration)

        root_ids = [t.task_id for t in dag.roots]
        roots = [t for t in tis if t.task_id in root_ids]

        # if all roots finished and at least one failed, the run failed
        if (not unfinished_tasks and
                any(r.state in (State.FAILED, State.UPSTREAM_FAILED) for r in roots)):
            self.log.info('Marking run %s failed', self)
            self.set_state(State.FAILED)
            dag.handle_callback(self, success=False, reason='task_failure',
                                session=session)

        # if all roots succeeded and no unfinished tasks, the run succeeded
        elif not unfinished_tasks and all(r.state in (State.SUCCESS, State.SKIPPED)
                                          for r in roots):
            self.log.info('Marking run %s successful', self)
            self.set_state(State.SUCCESS)
            dag.handle_callback(self, success=True, reason='success', session=session)

        # if *all tasks* are deadlocked, the run failed
        elif (unfinished_tasks and none_depends_on_past and
              none_task_concurrency and no_dependencies_met):
            self.log.info('Deadlock; marking run %s failed', self)
            self.set_state(State.FAILED)
            dag.handle_callback(self, success=False, reason='all_tasks_deadlocked',
                                session=session)

        # finally, if the roots aren't done, the dag is still running
        else:
            self.set_state(State.RUNNING)

        self._emit_duration_stats_for_finished_state()

        # todo: determine we want to use with_for_update to make sure to lock the run
        session.merge(self)
        session.commit()

        return self.state
def update_state(self, session=None):
        """
        Determines the overall state of the DagRun based on the state
        of its TaskInstances.

        :return: State
        """

        dag = self.get_dag()

        tis = self.get_task_instances(session=session)
        self.log.debug("Updating state for %s considering %s task(s)", self, len(tis))

        for ti in list(tis):
            # skip in db?
            if ti.state == State.REMOVED:
                tis.remove(ti)
            else:
                ti.task = dag.get_task(ti.task_id)

        # pre-calculate
        # db is faster
        start_dttm = timezone.utcnow()
        unfinished_tasks = self.get_task_instances(
            state=State.unfinished(),
            session=session
        )
        none_depends_on_past = all(not t.task.depends_on_past for t in unfinished_tasks)
        none_task_concurrency = all(t.task.task_concurrency is None
                                    for t in unfinished_tasks)
        # small speed up
        if unfinished_tasks and none_depends_on_past and none_task_concurrency:
            # todo: this can actually get pretty slow: one task costs between 0.01-015s
            no_dependencies_met = True
            for ut in unfinished_tasks:
                # We need to flag upstream and check for changes because upstream
                # failures/re-schedules can result in deadlock false positives
                old_state = ut.state
                deps_met = ut.are_dependencies_met(
                    dep_context=DepContext(
                        flag_upstream_failed=True,
                        ignore_in_retry_period=True,
                        ignore_in_reschedule_period=True),
                    session=session)
                if deps_met or old_state != ut.current_state(session=session):
                    no_dependencies_met = False
                    break

        duration = (timezone.utcnow() - start_dttm).total_seconds() * 1000
        Stats.timing("dagrun.dependency-check.{}".format(self.dag_id), duration)

        root_ids = [t.task_id for t in dag.roots]
        roots = [t for t in tis if t.task_id in root_ids]

        # if all roots finished and at least one failed, the run failed
        if (not unfinished_tasks and
                any(r.state in (State.FAILED, State.UPSTREAM_FAILED) for r in roots)):
            self.log.info('Marking run %s failed', self)
            self.set_state(State.FAILED)
            dag.handle_callback(self, success=False, reason='task_failure',
                                session=session)

        # if all roots succeeded and no unfinished tasks, the run succeeded
        elif not unfinished_tasks and all(r.state in (State.SUCCESS, State.SKIPPED)
                                          for r in roots):
            self.log.info('Marking run %s successful', self)
            self.set_state(State.SUCCESS)
            dag.handle_callback(self, success=True, reason='success', session=session)

        # if *all tasks* are deadlocked, the run failed
        elif (unfinished_tasks and none_depends_on_past and
              none_task_concurrency and no_dependencies_met):
            self.log.info('Deadlock; marking run %s failed', self)
            self.set_state(State.FAILED)
            dag.handle_callback(self, success=False, reason='all_tasks_deadlocked',
                                session=session)

        # finally, if the roots aren't done, the dag is still running
        else:
            self.set_state(State.RUNNING)

        self._emit_duration_stats_for_finished_state()

        # todo: determine we want to use with_for_update to make sure to lock the run
        session.merge(self)
        session.commit()

        return self.state
def update_state ( self , session = None ) : dag = self . get_dag ( ) tis = self . get_task_instances ( session = session ) self . log . debug ( "Updating state for %s considering %s task(s)" , self , len ( tis ) ) for ti in list ( tis ) : # skip in db? if ti . state == State . REMOVED : tis . remove ( ti ) else : ti . task = dag . get_task ( ti . task_id ) # pre-calculate # db is faster start_dttm = timezone . utcnow ( ) unfinished_tasks = self . get_task_instances ( state = State . unfinished ( ) , session = session ) none_depends_on_past = all ( not t . task . depends_on_past for t in unfinished_tasks ) none_task_concurrency = all ( t . task . task_concurrency is None for t in unfinished_tasks ) # small speed up if unfinished_tasks and none_depends_on_past and none_task_concurrency : # todo: this can actually get pretty slow: one task costs between 0.01-015s no_dependencies_met = True for ut in unfinished_tasks : # We need to flag upstream and check for changes because upstream # failures/re-schedules can result in deadlock false positives old_state = ut . state deps_met = ut . are_dependencies_met ( dep_context = DepContext ( flag_upstream_failed = True , ignore_in_retry_period = True , ignore_in_reschedule_period = True ) , session = session ) if deps_met or old_state != ut . current_state ( session = session ) : no_dependencies_met = False break duration = ( timezone . utcnow ( ) - start_dttm ) . total_seconds ( ) * 1000 Stats . timing ( "dagrun.dependency-check.{}" . format ( self . dag_id ) , duration ) root_ids = [ t . task_id for t in dag . roots ] roots = [ t for t in tis if t . task_id in root_ids ] # if all roots finished and at least one failed, the run failed if ( not unfinished_tasks and any ( r . state in ( State . FAILED , State . UPSTREAM_FAILED ) for r in roots ) ) : self . log . info ( 'Marking run %s failed' , self ) self . set_state ( State . FAILED ) dag . handle_callback ( self , success = False , reason = 'task_failure' , session = session ) # if all roots succeeded and no unfinished tasks, the run succeeded elif not unfinished_tasks and all ( r . state in ( State . SUCCESS , State . SKIPPED ) for r in roots ) : self . log . info ( 'Marking run %s successful' , self ) self . set_state ( State . SUCCESS ) dag . handle_callback ( self , success = True , reason = 'success' , session = session ) # if *all tasks* are deadlocked, the run failed elif ( unfinished_tasks and none_depends_on_past and none_task_concurrency and no_dependencies_met ) : self . log . info ( 'Deadlock; marking run %s failed' , self ) self . set_state ( State . FAILED ) dag . handle_callback ( self , success = False , reason = 'all_tasks_deadlocked' , session = session ) # finally, if the roots aren't done, the dag is still running else : self . set_state ( State . RUNNING ) self . _emit_duration_stats_for_finished_state ( ) # todo: determine we want to use with_for_update to make sure to lock the run session . merge ( self ) session . commit ( ) return self . state
Determines the overall state of the DagRun based on the state
        of its TaskInstances.

        :return: State
Determines the overall state of the DagRun based on the state of its TaskInstances .