[AIRFLOW-1424] make the next execution date of DAGs visible

airflow/jobs.py

@@ -772,7 +772,7 @@ def update_import_errors(session, dagbag):
         session.commit()
 
     @provide_session
-    def create_dag_run(self, dag, session=None):
+    def create_dag_run(self, dag, session=None, dry_run=False):
         """
         This method checks whether a new DagRun needs to be created
         for a DAG based on scheduling interval.
@@ -867,7 +867,6 @@ def create_dag_run(self, dag, session=None):
                     "Dag start date: %s. Next run date: %s",
                     dag.start_date, next_run_date
                 )
-
             # don't ever schedule in the future
             if next_run_date > timezone.utcnow():
                 return
@@ -892,6 +891,11 @@ def create_dag_run(self, dag, session=None):
             if next_run_date and min_task_end_date and next_run_date > min_task_end_date:
                 return
 
+            # Don't really schedule the job, we are interested in its next run date
+            # as calculated by the scheduler
+            if dry_run is True:
+                return next_run_date
+
             if next_run_date and period_end and period_end <= timezone.utcnow():
                 next_run = dag.create_dagrun(
                     run_id=DagRun.ID_PREFIX + next_run_date.isoformat(),

airflow/models.py

@@ -3732,6 +3732,32 @@ def latest_execution_date(self, session=None):
         ).scalar()
         return execution_date
 
+    @property
+    @provide_session
+    def schedulable_at(self, session=None):
+        """
+        Returns the earliest time at which the DAG will next be eligible for
+        execution by the scheduler
+        """
+        from airflow.jobs import SchedulerJob
+        scheduler = SchedulerJob()
+        next_run_date = scheduler.create_dag_run(self, dry_run=True)
+        if next_run_date:
+            period_end = self.following_schedule(next_run_date)
+            return period_end
+
+    @property
+    def scheduled_in(self):
+        """
+        Returns a human readable duration before the next schedulable time
+        of the DAG
+        """
+        import pendulum
+        diff = self.schedulable_at - pendulum.now()
+        if diff.in_seconds() <= 0:
+            return "overdue"
+        return diff
+
     @property
     def subdags(self):
         """

airflow/www/templates/airflow/dag.html

@@ -39,6 +39,11 @@ <h3 class="pull-left">
         <span style='color:#AAA;'>ROOT: </span> <span> {{ root }}</span>
       {% endif %}
     </h3>
+    <h4 class="pull-right">
+      <a class="label label-primary" href="/admin/dagrun/?flt2_dag_id_equals={{ dag.dag_id }}">
+        scheduled in: {{ dag.scheduled_in }}
+      </a>
+    </h4>
     <h4 class="pull-right">
       <a class="label label-default" href="/admin/dagrun/?flt2_dag_id_equals={{ dag.dag_id }}">
         schedule: {{ dag.schedule_interval }}

airflow/www/templates/airflow/dags.html

@@ -59,6 +59,9 @@ <h2>DAGs</h2>
                   <span id="statuses_info" class="glyphicon glyphicon-info-sign" aria-hidden="true" title="Status of all previous DAG runs."></span>
                   <img id="loading" width="15" src="{{ url_for("static", filename="loading.gif") }}">
                 </th>
+                <th>Scheduled in
+                  <span id="statuses_info" class="glyphicon glyphicon-info-sign" aria-hidden="true" title="Earliest time at which the DAG will next be scheduled."></span>
+                </th>
                 <th class="text-center">Links</th>
             </tr>
         </thead>
@@ -134,7 +137,16 @@ <h2>DAGs</h2>
                     <svg height="10" width="10" id='dag-run-{{ dag.safe_dag_id }}' style="display: block;"></svg>
                 </td>
 
-                <!-- Column 9: Links -->
+                <!-- Column 9: Dag Scheduled in -->
+                <td>
+                    {% if dag_id in webserver_dags %}
+                    <a class="label label-default" href="/admin/dagrun/?flt2_dag_id_equals={{ dag.dag_id }}" title="Schedulable at: {{ dag.schedulable_at }}">
+                        {{ dag.scheduled_in }}
+                    </a>
+                    {% endif %}
+                </td>
+
+                <!-- Column 10: Links -->
                 <td class="text-center" style="display:flex; flex-direction:row; justify-content:space-around;">
                 {% if dag %}
 

tests/models.py

@@ -454,6 +454,31 @@ def jinja_udf(name):
         result = task.render_template('', "{{ 'world' | hello}}", dict())
         self.assertEqual(result, 'Hello world')
 
+    def test_schedulable_at(self):
+        """
+        [AIRFLOW-1424] test the 'schedulable_after' DAG property.
+        """
+        test_dag_id = 'test_get_num_task_instances_dag'
+        test_task_id = 'task_1'
+
+        test_dag = DAG(dag_id=test_dag_id, start_date=DEFAULT_DATE,
+                       schedule_interval=datetime.timedelta(days=1))
+        DummyOperator(task_id=test_task_id, dag=test_dag)
+        self.assertEqual(test_dag.schedulable_at,
+                         DEFAULT_DATE + datetime.timedelta(days=1))
+
+    def test_scheduled_in_overdue(self):
+        """
+        [AIRFLOW-1424] test the 'scheduled_in' DAG property.
+        """
+        test_dag_id = 'test_get_num_task_instances_dag'
+        test_task_id = 'task_1'
+
+        test_dag = DAG(dag_id=test_dag_id, start_date=DEFAULT_DATE,
+                       schedule_interval=datetime.timedelta(days=1))
+        DummyOperator(task_id=test_task_id, dag=test_dag)
+        self.assertEqual(test_dag.scheduled_in, "overdue")
+
     def test_cycle(self):
         # test empty
         dag = DAG(