You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ma...@apache.org on 2018/06/05 22:14:50 UTC
incubator-airflow git commit: [AIRFLOW-2558] Clear task/dag is
clearing all executions
Repository: incubator-airflow
Updated Branches:
refs/heads/master 817296a7b -> c2b962ca9
[AIRFLOW-2558] Clear task/dag is clearing all executions
Closes #3465 from feng-tao/airflow_2588_new
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/c2b962ca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/c2b962ca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/c2b962ca
Branch: refs/heads/master
Commit: c2b962ca9831cbf56f6dbb0147c97563b7ebbd8f
Parents: 817296a
Author: Tao feng <tf...@lyft.com>
Authored: Tue Jun 5 15:14:43 2018 -0700
Committer: Maxime Beauchemin <ma...@gmail.com>
Committed: Tue Jun 5 15:14:43 2018 -0700
----------------------------------------------------------------------
airflow/bin/cli.py | 1 -
airflow/models.py | 43 ++++++++++++++++++-------------------------
tests/models.py | 4 ++--
3 files changed, 20 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/c2b962ca/airflow/bin/cli.py
----------------------------------------------------------------------
diff --git a/airflow/bin/cli.py b/airflow/bin/cli.py
index 3bff685..2742df5 100644
--- a/airflow/bin/cli.py
+++ b/airflow/bin/cli.py
@@ -199,7 +199,6 @@ def backfill(args, dag=None):
end_date=args.end_date,
confirm_prompt=True,
include_subdags=False,
- only_backfill_dagruns=True,
)
dag.run(
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/c2b962ca/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index eda4808..c26ec01 100755
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -132,17 +132,15 @@ def clear_task_instances(tis,
session,
activate_dag_runs=True,
dag=None,
- only_backfill_dagruns=False,
):
"""
Clears a set of task instances, but makes sure the running ones
- get killed. Reset backfill dag run state to removed if only_backfill_dagruns is set
+ get killed.
:param tis: a list of task instances
:param session: current session
:param activate_dag_runs: flag to check for active dag run
:param dag: DAG object
- :param only_backfill_dagruns: flag for setting backfill state
"""
job_ids = []
for ti in tis:
@@ -176,13 +174,8 @@ def clear_task_instances(tis,
DagRun.execution_date.in_({ti.execution_date for ti in tis}),
).all()
for dr in drs:
- if only_backfill_dagruns and dr.is_backfill:
- # If the flag is set, we reset backfill dag run for retry.
- # dont reset start date
- dr.state = State.REMOVED
- else:
- dr.state = State.RUNNING
- dr.start_date = timezone.utcnow()
+ dr.state = State.RUNNING
+ dr.start_date = timezone.utcnow()
class DagBag(BaseDagBag, LoggingMixin):
@@ -3697,18 +3690,20 @@ class DAG(BaseDag, LoggingMixin):
self,
state=State.RUNNING,
session=None,
- only_backfill_dagruns=False):
- drs = session.query(DagRun).filter_by(dag_id=self.dag_id).all()
+ start_date=None,
+ end_date=None,
+ ):
+ query = session.query(DagRun).filter_by(dag_id=self.dag_id)
+ if start_date:
+ query = query.filter(DagRun.execution_date >= start_date)
+ if end_date:
+ query = query.filter(DagRun.execution_date <= end_date)
+ drs = query.all()
+
dirty_ids = []
for dr in drs:
- if only_backfill_dagruns:
- if dr.is_backfill:
- dr.state = state
- dirty_ids.append(dr.dag_id)
- else:
- if not dr.is_backfill:
- dr.state = state
- dirty_ids.append(dr.dag_id)
+ dr.state = state
+ dirty_ids.append(dr.dag_id)
DagStat.update(dirty_ids, session=session)
@provide_session
@@ -3721,7 +3716,6 @@ class DAG(BaseDag, LoggingMixin):
reset_dag_runs=True,
dry_run=False,
session=None,
- only_backfill_dagruns=False,
):
"""
Clears a set of task instances associated with the current dag for
@@ -3772,11 +3766,12 @@ class DAG(BaseDag, LoggingMixin):
clear_task_instances(tis.all(),
session,
dag=self,
- only_backfill_dagruns=only_backfill_dagruns,
)
if reset_dag_runs:
self.set_dag_runs_state(session=session,
- only_backfill_dagruns=only_backfill_dagruns)
+ start_date=start_date,
+ end_date=end_date,
+ )
else:
count = 0
print("Bail. Nothing was cleared.")
@@ -3795,7 +3790,6 @@ class DAG(BaseDag, LoggingMixin):
include_subdags=True,
reset_dag_runs=True,
dry_run=False,
- only_backfill_dagruns=False,
):
all_tis = []
for dag in dags:
@@ -3836,7 +3830,6 @@ class DAG(BaseDag, LoggingMixin):
include_subdags=include_subdags,
reset_dag_runs=reset_dag_runs,
dry_run=False,
- only_backfill_dagruns=only_backfill_dagruns,
)
else:
count = 0
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/c2b962ca/tests/models.py
----------------------------------------------------------------------
diff --git a/tests/models.py b/tests/models.py
index 87ba74a..3515516 100644
--- a/tests/models.py
+++ b/tests/models.py
@@ -639,14 +639,14 @@ class DagRunTest(unittest.TestCase):
qry = session.query(TI).filter(
TI.dag_id == dag.dag_id).all()
- clear_task_instances(qry, session, only_backfill_dagruns=True)
+ clear_task_instances(qry, session)
session.commit()
ti0.refresh_from_db()
dr0 = session.query(DagRun).filter(
DagRun.dag_id == dag_id,
DagRun.execution_date == now
).first()
- self.assertEquals(dr0.state, State.REMOVED)
+ self.assertEquals(dr0.state, State.RUNNING)
def test_id_for_date(self):
run_id = models.DagRun.id_for_date(