airflow/jobs.py
def _execute_helper(self):
        """
        The actual scheduler loop. The main steps in the loop are:
            #. Harvest DAG parsing results through DagFileProcessorAgent
            #. Find and queue executable tasks
                #. Change task instance state in DB
                #. Queue tasks in executor
            #. Heartbeat executor
                #. Execute queued tasks in executor asynchronously
                #. Sync on the states of running tasks

        Following is a graphic representation of these steps.

        .. image:: ../docs/img/scheduler_loop.jpg

        :rtype: None
        """
        self.executor.start()

        self.log.info("Resetting orphaned tasks for active dag runs")
        self.reset_state_for_orphaned_tasks()

        # Start after resetting orphaned tasks to avoid stressing out DB.
        self.processor_agent.start()

        execute_start_time = timezone.utcnow()

        # Last time that self.heartbeat() was called.
        last_self_heartbeat_time = timezone.utcnow()

        # For the execute duration, parse and schedule DAGs
        while True:
            self.log.debug("Starting Loop...")
            loop_start_time = time.time()

            if self.using_sqlite:
                self.processor_agent.heartbeat()
                # For the sqlite case w/ 1 thread, wait until the processor
                # is finished to avoid concurrent access to the DB.
                self.log.debug(
                    "Waiting for processors to finish since we're using sqlite")
                self.processor_agent.wait_until_finished()

            self.log.debug("Harvesting DAG parsing results")
            simple_dags = self.processor_agent.harvest_simple_dags()
            self.log.debug("Harvested {} SimpleDAGs".format(len(simple_dags)))

            # Send tasks for execution if available
            simple_dag_bag = SimpleDagBag(simple_dags)
            if len(simple_dags) > 0:
                try:
                    simple_dag_bag = SimpleDagBag(simple_dags)

                    # Handle cases where a DAG run state is set (perhaps manually) to
                    # a non-running state. Handle task instances that belong to
                    # DAG runs in those states

                    # If a task instance is up for retry but the corresponding DAG run
                    # isn't running, mark the task instance as FAILED so we don't try
                    # to re-run it.
                    self._change_state_for_tis_without_dagrun(simple_dag_bag,
                                                              [State.UP_FOR_RETRY],
                                                              State.FAILED)
                    # If a task instance is scheduled or queued or up for reschedule,
                    # but the corresponding DAG run isn't running, set the state to
                    # NONE so we don't try to re-run it.
                    self._change_state_for_tis_without_dagrun(simple_dag_bag,
                                                              [State.QUEUED,
                                                               State.SCHEDULED,
                                                               State.UP_FOR_RESCHEDULE],
                                                              State.NONE)

                    self._execute_task_instances(simple_dag_bag,
                                                 (State.SCHEDULED,))
                except Exception as e:
                    self.log.error("Error queuing tasks")
                    self.log.exception(e)
                    continue

            # Call heartbeats
            self.log.debug("Heartbeating the executor")
            self.executor.heartbeat()

            self._change_state_for_tasks_failed_to_execute()

            # Process events from the executor
            self._process_executor_events(simple_dag_bag)

            # Heartbeat the scheduler periodically
            time_since_last_heartbeat = (timezone.utcnow() -
                                         last_self_heartbeat_time).total_seconds()
            if time_since_last_heartbeat > self.heartrate:
                self.log.debug("Heartbeating the scheduler")
                self.heartbeat()
                last_self_heartbeat_time = timezone.utcnow()

            is_unit_test = conf.getboolean('core', 'unit_test_mode')
            loop_end_time = time.time()
            loop_duration = loop_end_time - loop_start_time
            self.log.debug(
                "Ran scheduling loop in %.2f seconds",
                loop_duration)

            if not is_unit_test:
                self.log.debug("Sleeping for %.2f seconds", self._processor_poll_interval)
                time.sleep(self._processor_poll_interval)

            # Exit early for a test mode, run one additional scheduler loop
            # to reduce the possibility that parsed DAG was put into the queue
            # by the DAG manager but not yet received by DAG agent.
            if self.processor_agent.done:
                self._last_loop = True

            if self._last_loop:
                self.log.info("Exiting scheduler loop as all files"
                              " have been processed {} times".format(self.num_runs))
                break

            if loop_duration < 1 and not is_unit_test:
                sleep_length = 1 - loop_duration
                self.log.debug(
                    "Sleeping for {0:.2f} seconds to prevent excessive logging"
                    .format(sleep_length))
                sleep(sleep_length)

        # Stop any processors
        self.processor_agent.terminate()

        # Verify that all files were processed, and if so, deactivate DAGs that
        # haven't been touched by the scheduler as they likely have been
        # deleted.
        if self.processor_agent.all_files_processed:
            self.log.info(
                "Deactivating DAGs that haven't been touched since %s",
                execute_start_time.isoformat()
            )
            models.DAG.deactivate_stale_dags(execute_start_time)

        self.executor.end()

        settings.Session.remove()
def _execute_helper(self):
        """
        The actual scheduler loop. The main steps in the loop are:
            #. Harvest DAG parsing results through DagFileProcessorAgent
            #. Find and queue executable tasks
                #. Change task instance state in DB
                #. Queue tasks in executor
            #. Heartbeat executor
                #. Execute queued tasks in executor asynchronously
                #. Sync on the states of running tasks

        Following is a graphic representation of these steps.

        .. image:: ../docs/img/scheduler_loop.jpg

        :rtype: None
        """
        self.executor.start()

        self.log.info("Resetting orphaned tasks for active dag runs")
        self.reset_state_for_orphaned_tasks()

        # Start after resetting orphaned tasks to avoid stressing out DB.
        self.processor_agent.start()

        execute_start_time = timezone.utcnow()

        # Last time that self.heartbeat() was called.
        last_self_heartbeat_time = timezone.utcnow()

        # For the execute duration, parse and schedule DAGs
        while True:
            self.log.debug("Starting Loop...")
            loop_start_time = time.time()

            if self.using_sqlite:
                self.processor_agent.heartbeat()
                # For the sqlite case w/ 1 thread, wait until the processor
                # is finished to avoid concurrent access to the DB.
                self.log.debug(
                    "Waiting for processors to finish since we're using sqlite")
                self.processor_agent.wait_until_finished()

            self.log.debug("Harvesting DAG parsing results")
            simple_dags = self.processor_agent.harvest_simple_dags()
            self.log.debug("Harvested {} SimpleDAGs".format(len(simple_dags)))

            # Send tasks for execution if available
            simple_dag_bag = SimpleDagBag(simple_dags)
            if len(simple_dags) > 0:
                try:
                    simple_dag_bag = SimpleDagBag(simple_dags)

                    # Handle cases where a DAG run state is set (perhaps manually) to
                    # a non-running state. Handle task instances that belong to
                    # DAG runs in those states

                    # If a task instance is up for retry but the corresponding DAG run
                    # isn't running, mark the task instance as FAILED so we don't try
                    # to re-run it.
                    self._change_state_for_tis_without_dagrun(simple_dag_bag,
                                                              [State.UP_FOR_RETRY],
                                                              State.FAILED)
                    # If a task instance is scheduled or queued or up for reschedule,
                    # but the corresponding DAG run isn't running, set the state to
                    # NONE so we don't try to re-run it.
                    self._change_state_for_tis_without_dagrun(simple_dag_bag,
                                                              [State.QUEUED,
                                                               State.SCHEDULED,
                                                               State.UP_FOR_RESCHEDULE],
                                                              State.NONE)

                    self._execute_task_instances(simple_dag_bag,
                                                 (State.SCHEDULED,))
                except Exception as e:
                    self.log.error("Error queuing tasks")
                    self.log.exception(e)
                    continue

            # Call heartbeats
            self.log.debug("Heartbeating the executor")
            self.executor.heartbeat()

            self._change_state_for_tasks_failed_to_execute()

            # Process events from the executor
            self._process_executor_events(simple_dag_bag)

            # Heartbeat the scheduler periodically
            time_since_last_heartbeat = (timezone.utcnow() -
                                         last_self_heartbeat_time).total_seconds()
            if time_since_last_heartbeat > self.heartrate:
                self.log.debug("Heartbeating the scheduler")
                self.heartbeat()
                last_self_heartbeat_time = timezone.utcnow()

            is_unit_test = conf.getboolean('core', 'unit_test_mode')
            loop_end_time = time.time()
            loop_duration = loop_end_time - loop_start_time
            self.log.debug(
                "Ran scheduling loop in %.2f seconds",
                loop_duration)

            if not is_unit_test:
                self.log.debug("Sleeping for %.2f seconds", self._processor_poll_interval)
                time.sleep(self._processor_poll_interval)

            # Exit early for a test mode, run one additional scheduler loop
            # to reduce the possibility that parsed DAG was put into the queue
            # by the DAG manager but not yet received by DAG agent.
            if self.processor_agent.done:
                self._last_loop = True

            if self._last_loop:
                self.log.info("Exiting scheduler loop as all files"
                              " have been processed {} times".format(self.num_runs))
                break

            if loop_duration < 1 and not is_unit_test:
                sleep_length = 1 - loop_duration
                self.log.debug(
                    "Sleeping for {0:.2f} seconds to prevent excessive logging"
                    .format(sleep_length))
                sleep(sleep_length)

        # Stop any processors
        self.processor_agent.terminate()

        # Verify that all files were processed, and if so, deactivate DAGs that
        # haven't been touched by the scheduler as they likely have been
        # deleted.
        if self.processor_agent.all_files_processed:
            self.log.info(
                "Deactivating DAGs that haven't been touched since %s",
                execute_start_time.isoformat()
            )
            models.DAG.deactivate_stale_dags(execute_start_time)

        self.executor.end()

        settings.Session.remove()
def _execute_helper ( self ) : self . executor . start ( ) self . log . info ( "Resetting orphaned tasks for active dag runs" ) self . reset_state_for_orphaned_tasks ( ) # Start after resetting orphaned tasks to avoid stressing out DB. self . processor_agent . start ( ) execute_start_time = timezone . utcnow ( ) # Last time that self.heartbeat() was called. last_self_heartbeat_time = timezone . utcnow ( ) # For the execute duration, parse and schedule DAGs while True : self . log . debug ( "Starting Loop..." ) loop_start_time = time . time ( ) if self . using_sqlite : self . processor_agent . heartbeat ( ) # For the sqlite case w/ 1 thread, wait until the processor # is finished to avoid concurrent access to the DB. self . log . debug ( "Waiting for processors to finish since we're using sqlite" ) self . processor_agent . wait_until_finished ( ) self . log . debug ( "Harvesting DAG parsing results" ) simple_dags = self . processor_agent . harvest_simple_dags ( ) self . log . debug ( "Harvested {} SimpleDAGs" . format ( len ( simple_dags ) ) ) # Send tasks for execution if available simple_dag_bag = SimpleDagBag ( simple_dags ) if len ( simple_dags ) > 0 : try : simple_dag_bag = SimpleDagBag ( simple_dags ) # Handle cases where a DAG run state is set (perhaps manually) to # a non-running state. Handle task instances that belong to # DAG runs in those states # If a task instance is up for retry but the corresponding DAG run # isn't running, mark the task instance as FAILED so we don't try # to re-run it. self . _change_state_for_tis_without_dagrun ( simple_dag_bag , [ State . UP_FOR_RETRY ] , State . FAILED ) # If a task instance is scheduled or queued or up for reschedule, # but the corresponding DAG run isn't running, set the state to # NONE so we don't try to re-run it. self . _change_state_for_tis_without_dagrun ( simple_dag_bag , [ State . QUEUED , State . SCHEDULED , State . UP_FOR_RESCHEDULE ] , State . NONE ) self . _execute_task_instances ( simple_dag_bag , ( State . SCHEDULED , ) ) except Exception as e : self . log . error ( "Error queuing tasks" ) self . log . exception ( e ) continue # Call heartbeats self . log . debug ( "Heartbeating the executor" ) self . executor . heartbeat ( ) self . _change_state_for_tasks_failed_to_execute ( ) # Process events from the executor self . _process_executor_events ( simple_dag_bag ) # Heartbeat the scheduler periodically time_since_last_heartbeat = ( timezone . utcnow ( ) - last_self_heartbeat_time ) . total_seconds ( ) if time_since_last_heartbeat > self . heartrate : self . log . debug ( "Heartbeating the scheduler" ) self . heartbeat ( ) last_self_heartbeat_time = timezone . utcnow ( ) is_unit_test = conf . getboolean ( 'core' , 'unit_test_mode' ) loop_end_time = time . time ( ) loop_duration = loop_end_time - loop_start_time self . log . debug ( "Ran scheduling loop in %.2f seconds" , loop_duration ) if not is_unit_test : self . log . debug ( "Sleeping for %.2f seconds" , self . _processor_poll_interval ) time . sleep ( self . _processor_poll_interval ) # Exit early for a test mode, run one additional scheduler loop # to reduce the possibility that parsed DAG was put into the queue # by the DAG manager but not yet received by DAG agent. if self . processor_agent . done : self . _last_loop = True if self . _last_loop : self . log . info ( "Exiting scheduler loop as all files" " have been processed {} times" . format ( self . num_runs ) ) break if loop_duration < 1 and not is_unit_test : sleep_length = 1 - loop_duration self . log . debug ( "Sleeping for {0:.2f} seconds to prevent excessive logging" . format ( sleep_length ) ) sleep ( sleep_length ) # Stop any processors self . processor_agent . terminate ( ) # Verify that all files were processed, and if so, deactivate DAGs that # haven't been touched by the scheduler as they likely have been # deleted. if self . processor_agent . all_files_processed : self . log . info ( "Deactivating DAGs that haven't been touched since %s" , execute_start_time . isoformat ( ) ) models . DAG . deactivate_stale_dags ( execute_start_time ) self . executor . end ( ) settings . Session . remove ( )
The actual scheduler loop. The main steps in the loop are:
            #. Harvest DAG parsing results through DagFileProcessorAgent
            #. Find and queue executable tasks
                #. Change task instance state in DB
                #. Queue tasks in executor
            #. Heartbeat executor
                #. Execute queued tasks in executor asynchronously
                #. Sync on the states of running tasks

        Following is a graphic representation of these steps.

        .. image:: ../docs/img/scheduler_loop.jpg

        :rtype: None
The actual scheduler loop . The main steps in the loop are : # . Harvest DAG parsing results through DagFileProcessorAgent # . Find and queue executable tasks # . Change task instance state in DB # . Queue tasks in executor # . Heartbeat executor # . Execute queued tasks in executor asynchronously # . Sync on the states of running tasks