airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davy...@apache.org
Subject incubator-airflow git commit: [AIRFLOW-1787] Fix task instance batch clear and set state bugs
Date Tue, 07 Nov 2017 19:23:18 GMT
Repository: incubator-airflow
Updated Branches:
  refs/heads/master 1a7b63eb1 -> 313f5bac4


[AIRFLOW-1787] Fix task instance batch clear and set state bugs

Fixes Batch clear in Task Instances view is not working
for task instances in RUNNING state and all batch
operations in Task instances view cannot work when
manually triggered task instances are selected
because they have a different execution date
format.

Closes #2759 from yrqls21/fix-ti-batch-clear-n
-set-state-bugs


Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/313f5bac
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/313f5bac
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/313f5bac

Branch: refs/heads/master
Commit: 313f5bac4a3f804094bcd583e0e5fbc3b5f405bb
Parents: 1a7b63e
Author: Kevin Yang <kevin.yang@airbnb.com>
Authored: Tue Nov 7 11:22:58 2017 -0800
Committer: Dan Davydov <dan.davydov@airbnb.com>
Committed: Tue Nov 7 11:23:00 2017 -0800

----------------------------------------------------------------------
 airflow/utils/dates.py    | 14 ++++++++++++++
 airflow/www/views.py      | 30 ++++++++++++++++++++----------
 tests/utils/test_dates.py |  9 +++++++++
 3 files changed, 43 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/313f5bac/airflow/utils/dates.py
----------------------------------------------------------------------
diff --git a/airflow/utils/dates.py b/airflow/utils/dates.py
index 43b87f4..81e1c2c 100644
--- a/airflow/utils/dates.py
+++ b/airflow/utils/dates.py
@@ -225,3 +225,17 @@ def days_ago(n, hour=0, minute=0, second=0, microsecond=0):
         second=second,
         microsecond=microsecond)
     return today - timedelta(days=n)
+
+
+def parse_execution_date(execution_date_str):
+    """
+    Parse execution date string to datetime object.
+    """
+    try:
+        # Execution date follows execution date format of scheduled executions,
+        # e.g. '2017-11-02 00:00:00'
+        return datetime.strptime(execution_date_str, '%Y-%m-%d %H:%M:%S')
+    except ValueError:
+        # Execution date follows execution date format of manually triggered executions,
+        # e.g. '2017-11-05 16:18:30..989729'
+        return datetime.strptime(execution_date_str, '%Y-%m-%d %H:%M:%S..%f')

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/313f5bac/airflow/www/views.py
----------------------------------------------------------------------
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 81c44b6..a6d788e 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -76,7 +76,7 @@ from airflow.utils.json import json_ser
 from airflow.utils.state import State
 from airflow.utils.db import create_session, provide_session
 from airflow.utils.helpers import alchemy_to_dict
-from airflow.utils.dates import infer_time_unit, scale_time_units
+from airflow.utils.dates import infer_time_unit, scale_time_units, parse_execution_date
 from airflow.www import utils as wwwutils
 from airflow.www.forms import DateTimeForm, DateTimeWithNumRunsForm
 from airflow.www.validators import GreaterEqualThan
@@ -2502,23 +2502,32 @@ class TaskInstanceModelView(ModelViewOnly):
         try:
             TI = models.TaskInstance
 
+            dag_to_task_details = {}
             dag_to_tis = {}
 
-            for id in ids:
-                task_id, dag_id, execution_date = id.split(',')
+            # Collect dags upfront as dagbag.get_dag() will reset the session
+            for id_str in ids:
+                task_id, dag_id, execution_date = id_str.split(',')
+                dag = dagbag.get_dag(dag_id)
+                task_details = dag_to_task_details.setdefault(dag, [])
+                task_details.append((task_id, execution_date))
 
-                ti = session.query(TI).filter(TI.task_id == task_id,
-                                              TI.dag_id == dag_id,
-                                              TI.execution_date == execution_date).one()
+            for dag, task_details in dag_to_task_details.items():
+                for task_id, execution_date in task_details:
+                    execution_date = parse_execution_date(execution_date)
 
-                dag = dagbag.get_dag(dag_id)
-                tis = dag_to_tis.setdefault(dag, [])
-                tis.append(ti)
+                    ti = session.query(TI).filter(TI.task_id == task_id,
+                                                  TI.dag_id == dag.dag_id,
+                                                  TI.execution_date == execution_date).one()
+
+                    tis = dag_to_tis.setdefault(dag, [])
+                    tis.append(ti)
 
             for dag, tis in dag_to_tis.items():
                 models.clear_task_instances(tis, session, dag=dag)
 
             session.commit()
+
             flash("{0} task instances have been cleared".format(len(ids)))
 
         except Exception as ex:
@@ -2533,7 +2542,8 @@ class TaskInstanceModelView(ModelViewOnly):
             count = len(ids)
             for id in ids:
                 task_id, dag_id, execution_date = id.split(',')
-                execution_date = datetime.strptime(execution_date, '%Y-%m-%d %H:%M:%S')
+                execution_date = parse_execution_date(execution_date)
+
                 ti = session.query(TI).filter(TI.task_id == task_id,
                                               TI.dag_id == dag_id,
                                               TI.execution_date == execution_date).one()

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/313f5bac/tests/utils/test_dates.py
----------------------------------------------------------------------
diff --git a/tests/utils/test_dates.py b/tests/utils/test_dates.py
index 1323034..50e76ba 100644
--- a/tests/utils/test_dates.py
+++ b/tests/utils/test_dates.py
@@ -40,3 +40,12 @@ class Dates(unittest.TestCase):
         self.assertTrue(
             dates.days_ago(0, microsecond=3)
             == today_midnight + timedelta(microseconds=3))
+
+    def test_parse_execution_date(self):
+        execution_date_str_wo_ms = '2017-11-02 00:00:00'
+        execution_date_str_w_ms = '2017-11-05 16:18:30..989729'
+        bad_execution_date_str = '2017-11-06T00:00:00Z'
+
+        self.assertEqual(datetime(2017, 11, 2, 0, 0, 0), dates.parse_execution_date(execution_date_str_wo_ms))
+        self.assertEqual(datetime(2017, 11, 5, 16, 18, 30, 989729), dates.parse_execution_date(execution_date_str_w_ms))
+        self.assertRaises(ValueError, dates.parse_execution_date, bad_execution_date_str)


Mime
View raw message