You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by je...@apache.org on 2021/10/22 23:52:37 UTC
[airflow] branch v2-2-test updated (5b1c76a -> 6ef71d5)
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a change to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git.
from 5b1c76a Add changelog for 2.2.1
new 28f56a0 add detailed information to logging when a dag or a task finishes. (#19097)
new cd87e58 Fix queued dag runs changes catchup=False behaviour (#19130)
new a99d54d Add test for interval timetable catchup=False (#19145)
new be29f5d Crerate TI context with data interval compat layer (#19148)
new fb4deb7 Hide tooltip when next run is none (#19112)
new 6ef71d5 Add TriggererJob to jobs check command (#19179)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
airflow/cli/cli_parser.py | 2 +-
airflow/jobs/scheduler_job.py | 25 ++++++++++
airflow/models/dagrun.py | 25 ++++++++++
airflow/models/taskinstance.py | 5 +-
airflow/timetables/interval.py | 28 ++++++-----
airflow/www/static/js/dag.js | 10 ++--
airflow/www/templates/airflow/dag.html | 3 +-
tests/jobs/test_scheduler_job.py | 48 +++++++++++++++++++
tests/timetables/test_interval_timetable.py | 74 +++++++++++++++++++++++++++++
9 files changed, 201 insertions(+), 19 deletions(-)
create mode 100644 tests/timetables/test_interval_timetable.py
[airflow] 01/06: add detailed information to logging when a dag or
a task finishes. (#19097)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 28f56a007e106d1373c6c231a67b87c45b63697c
Author: Steve Zhang <86...@users.noreply.github.com>
AuthorDate: Fri Oct 22 08:35:26 2021 -0700
add detailed information to logging when a dag or a task finishes. (#19097)
* add detailed information to logging when a dag or a task finishes.
* make logging of start_date/end_date ISO format to be consist
* fix pre-commit
* use only %s in new logging statements to gracefully handle when certain variables are None
* fix precommit
* use self._state instead of get_state(). add computation for dag run duration based on start_date and end_date
* make linter happy with format
* fix typo
* put back missing reference to _state
Co-authored-by: Daniel Imberman <da...@gmail.com>
(cherry picked from commit 324c31c2d7ad756ce3814f74f0b6654d02f19426)
---
airflow/jobs/scheduler_job.py | 25 +++++++++++++++++++++++++
airflow/models/dagrun.py | 25 +++++++++++++++++++++++++
2 files changed, 50 insertions(+)
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index bf2410a..87dc158 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -533,6 +533,31 @@ class SchedulerJob(BaseJob):
self.log.info("Setting external_id for %s to %s", ti, info)
continue
+ msg = (
+ "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s, "
+ "run_start_date=%s, run_end_date=%s, "
+ "run_duration=%s, state=%s, executor_state=%s, try_number=%s, max_tries=%s, job_id=%s, "
+ "pool=%s, queue=%s, priority_weight=%d, operator=%s"
+ )
+ self.log.info(
+ msg,
+ ti.dag_id,
+ ti.task_id,
+ ti.run_id,
+ ti.start_date,
+ ti.end_date,
+ ti.duration,
+ ti.state,
+ state,
+ try_number,
+ ti.max_tries,
+ ti.job_id,
+ ti.pool,
+ ti.queue,
+ ti.priority_weight,
+ ti.operator,
+ )
+
if ti.try_number == buffer_key.try_number and ti.state == State.QUEUED:
Stats.incr('scheduler.tasks.killed_externally')
msg = (
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index 1d53265..e8eb685 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -542,6 +542,31 @@ class DagRun(Base, LoggingMixin):
else:
self.set_state(State.RUNNING)
+ if self._state == State.FAILED or self._state == State.SUCCESS:
+ msg = (
+ "DagRun Finished: dag_id=%s, execution_date=%s, run_id=%s, "
+ "run_start_date=%s, run_end_date=%s, run_duration=%s, "
+ "state=%s, external_trigger=%s, run_type=%s, "
+ "data_interval_start=%s, data_interval_end=%s, dag_hash=%s"
+ )
+ self.log.info(
+ msg,
+ self.dag_id,
+ self.execution_date,
+ self.run_id,
+ self.start_date,
+ self.end_date,
+ (self.end_date - self.start_date).total_seconds()
+ if self.start_date and self.end_date
+ else None,
+ self._state,
+ self.external_trigger,
+ self.run_type,
+ self.data_interval_start,
+ self.data_interval_end,
+ self.dag_hash,
+ )
+
self._emit_true_scheduling_delay_stats_for_finished_state(finished_tasks)
self._emit_duration_stats_for_finished_state()
[airflow] 03/06: Add test for interval timetable catchup=False
(#19145)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit a99d54d27e256c0488d604d312f4a0cab320e09a
Author: Tzu-ping Chung <tp...@astronomer.io>
AuthorDate: Fri Oct 22 15:38:57 2021 +0800
Add test for interval timetable catchup=False (#19145)
(cherry picked from commit 205219c522b77abe9d36d51807c32189c900cfd4)
---
airflow/timetables/interval.py | 24 +++++-----
tests/jobs/test_scheduler_job.py | 48 +++++++++++++++++++
tests/timetables/test_interval_timetable.py | 74 +++++++++++++++++++++++++++++
3 files changed, 134 insertions(+), 12 deletions(-)
diff --git a/airflow/timetables/interval.py b/airflow/timetables/interval.py
index 0dd7ffd..a095565 100644
--- a/airflow/timetables/interval.py
+++ b/airflow/timetables/interval.py
@@ -74,22 +74,22 @@ class _DataIntervalTimetable(Timetable):
earliest = restriction.earliest
if not restriction.catchup:
earliest = self._skip_to_latest(earliest)
+ elif earliest is not None:
+ earliest = self._align(earliest)
if last_automated_data_interval is None:
# First run; schedule the run at the first available time matching
# the schedule, and retrospectively create a data interval for it.
if earliest is None:
return None
- start = self._align(earliest)
- else:
- # There's a previous run.
+ start = earliest
+ else: # There's a previous run.
if earliest is not None:
# Catchup is False or DAG has new start date in the future.
- # Make sure we get the latest start date
+ # Make sure we get the later one.
start = max(last_automated_data_interval.end, earliest)
else:
- # Create a data interval starting from when the end of the previous interval.
+ # Data interval starts from the end of the previous interval.
start = last_automated_data_interval.end
-
if restriction.latest is not None and start > restriction.latest:
return None
end = self._get_next(start)
@@ -189,8 +189,8 @@ class CronDataIntervalTimetable(_DataIntervalTimetable):
def _align(self, current: DateTime) -> DateTime:
"""Get the next scheduled time.
- This is ``current + interval``, unless ``current`` is first interval,
- then ``current`` is returned.
+ This is ``current + interval``, unless ``current`` falls right on the
+ interval boundary, when ``current`` is returned.
"""
next_time = self._get_next(current)
if self._get_prev(next_time) != current:
@@ -205,14 +205,14 @@ class CronDataIntervalTimetable(_DataIntervalTimetable):
This is slightly different from the delta version at terminal values.
If the next schedule should start *right now*, we want the data interval
- that start right now now, not the one that ends now.
+ that start now, not the one that ends now.
"""
current_time = DateTime.utcnow()
- next_start = self._get_next(current_time)
last_start = self._get_prev(current_time)
- if next_start == current_time:
+ next_start = self._get_next(last_start)
+ if next_start == current_time: # Current time is on interval boundary.
new_start = last_start
- elif next_start > current_time:
+ elif next_start > current_time: # Current time is between boundaries.
new_start = self._get_prev(last_start)
else:
raise AssertionError("next schedule shouldn't be earlier")
diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
index 16b2921..4922617 100644
--- a/tests/jobs/test_scheduler_job.py
+++ b/tests/jobs/test_scheduler_job.py
@@ -3578,3 +3578,51 @@ class TestSchedulerJobQueriesCount:
assert start_date is None
assert end_date is None
assert duration is None
+
+ def test_catchup_works_correctly(self, dag_maker):
+ """Test that catchup works correctly"""
+ session = settings.Session()
+ with dag_maker(
+ dag_id='test_catchup_schedule_dag',
+ schedule_interval=timedelta(days=1),
+ start_date=DEFAULT_DATE,
+ catchup=True,
+ max_active_runs=1,
+ session=session,
+ ) as dag:
+ DummyOperator(task_id='dummy')
+
+ self.scheduler_job = SchedulerJob(subdir=os.devnull)
+ self.scheduler_job.executor = MockExecutor()
+ self.scheduler_job.processor_agent = mock.MagicMock(spec=DagFileProcessorAgent)
+
+ self.scheduler_job._create_dag_runs([dag_maker.dag_model], session)
+ self.scheduler_job._start_queued_dagruns(session)
+ # first dagrun execution date is DEFAULT_DATE 2016-01-01T00:00:00+00:00
+ dr = DagRun.find(execution_date=DEFAULT_DATE, session=session)[0]
+ ti = dr.get_task_instance(task_id='dummy')
+ ti.state = State.SUCCESS
+ session.merge(ti)
+ session.flush()
+
+ self.scheduler_job._schedule_dag_run(dr, session)
+ session.flush()
+
+ # Run the second time so _update_dag_next_dagrun will run
+ self.scheduler_job._schedule_dag_run(dr, session)
+ session.flush()
+
+ dag.catchup = False
+ dag.sync_to_db()
+ assert not dag.catchup
+
+ dm = DagModel.get_dagmodel(dag.dag_id)
+ self.scheduler_job._create_dag_runs([dm], session)
+
+ # Check catchup worked correctly by ensuring execution_date is quite new
+ # Our dag is a daily dag
+ assert (
+ session.query(DagRun.execution_date)
+ .filter(DagRun.execution_date != DEFAULT_DATE) # exclude the first run
+ .scalar()
+ ) > (timezone.utcnow() - timedelta(days=2))
diff --git a/tests/timetables/test_interval_timetable.py b/tests/timetables/test_interval_timetable.py
new file mode 100644
index 0000000..53f5aeb
--- /dev/null
+++ b/tests/timetables/test_interval_timetable.py
@@ -0,0 +1,74 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import datetime
+from typing import Optional
+
+import freezegun
+import pendulum
+import pytest
+
+from airflow.settings import TIMEZONE
+from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction, Timetable
+from airflow.timetables.interval import CronDataIntervalTimetable, DeltaDataIntervalTimetable
+
+START_DATE = pendulum.DateTime(2021, 9, 4, tzinfo=TIMEZONE)
+
+PREV_DATA_INTERVAL_START = START_DATE
+PREV_DATA_INTERVAL_END = START_DATE + datetime.timedelta(days=1)
+PREV_DATA_INTERVAL = DataInterval(start=PREV_DATA_INTERVAL_START, end=PREV_DATA_INTERVAL_END)
+
+CURRENT_TIME = pendulum.DateTime(2021, 9, 7, tzinfo=TIMEZONE)
+
+HOURLY_CRON_TIMETABLE = CronDataIntervalTimetable("@hourly", TIMEZONE)
+HOURLY_DELTA_TIMETABLE = DeltaDataIntervalTimetable(datetime.timedelta(hours=1))
+
+
+@pytest.mark.parametrize(
+ "timetable",
+ [pytest.param(HOURLY_CRON_TIMETABLE, id="cron"), pytest.param(HOURLY_DELTA_TIMETABLE, id="delta")],
+)
+@pytest.mark.parametrize(
+ "last_automated_data_interval",
+ [pytest.param(None, id="first-run"), pytest.param(PREV_DATA_INTERVAL, id="subsequent")],
+)
+@freezegun.freeze_time(CURRENT_TIME)
+def test_no_catchup_next_info_starts_at_current_time(
+ timetable: Timetable,
+ last_automated_data_interval: Optional[DataInterval],
+) -> None:
+ """If ``catchup=False``, the next data interval ends at the current time."""
+ next_info = timetable.next_dagrun_info(
+ last_automated_data_interval=last_automated_data_interval,
+ restriction=TimeRestriction(earliest=START_DATE, latest=None, catchup=False),
+ )
+ expected_start = CURRENT_TIME - datetime.timedelta(hours=1)
+ assert next_info == DagRunInfo.interval(start=expected_start, end=CURRENT_TIME)
+
+
+@pytest.mark.parametrize(
+ "timetable",
+ [pytest.param(HOURLY_CRON_TIMETABLE, id="cron"), pytest.param(HOURLY_DELTA_TIMETABLE, id="delta")],
+)
+def test_catchup_next_info_starts_at_previous_interval_end(timetable: Timetable) -> None:
+ """If ``catchup=True``, the next interval starts at the previous's end."""
+ next_info = timetable.next_dagrun_info(
+ last_automated_data_interval=PREV_DATA_INTERVAL,
+ restriction=TimeRestriction(earliest=START_DATE, latest=None, catchup=True),
+ )
+ expected_end = PREV_DATA_INTERVAL_END + datetime.timedelta(hours=1)
+ assert next_info == DagRunInfo.interval(start=PREV_DATA_INTERVAL_END, end=expected_end)
[airflow] 02/06: Fix queued dag runs changes catchup=False
behaviour (#19130)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cd87e5802af08fff0481929db2594955d8a21874
Author: Robin Edwards <ro...@gmail.com>
AuthorDate: Fri Oct 22 02:03:59 2021 +0100
Fix queued dag runs changes catchup=False behaviour (#19130)
(cherry picked from commit 829f90ad337c2ea94db7cd58ccdd71dd680ad419)
---
airflow/timetables/interval.py | 12 +++++++++---
1 file changed, 9 insertions(+), 3 deletions(-)
diff --git a/airflow/timetables/interval.py b/airflow/timetables/interval.py
index 8f5c3f1..0dd7ffd 100644
--- a/airflow/timetables/interval.py
+++ b/airflow/timetables/interval.py
@@ -81,9 +81,15 @@ class _DataIntervalTimetable(Timetable):
return None
start = self._align(earliest)
else:
- # There's a previous run. Create a data interval starting from when
- # the end of the previous interval.
- start = last_automated_data_interval.end
+ # There's a previous run.
+ if earliest is not None:
+ # Catchup is False or DAG has new start date in the future.
+ # Make sure we get the latest start date
+ start = max(last_automated_data_interval.end, earliest)
+ else:
+ # Create a data interval starting from when the end of the previous interval.
+ start = last_automated_data_interval.end
+
if restriction.latest is not None and start > restriction.latest:
return None
end = self._get_next(start)
[airflow] 05/06: Hide tooltip when next run is none (#19112)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit fb4deb74997c791a35c4ebd521f9c1c2868bf90d
Author: Brent Bovenzi <br...@gmail.com>
AuthorDate: Wed Oct 20 15:38:49 2021 -0500
Hide tooltip when next run is none (#19112)
* Hide Next Run when it is None
Check for `is not none` instead of `is defined` when showing the next run tooltip
* check next_dagrun is defined and not none
(cherry picked from commit e6fcf7776a298cff741d8a55f0990ea796fd0540)
---
airflow/www/static/js/dag.js | 10 ++++++----
airflow/www/templates/airflow/dag.html | 3 ++-
2 files changed, 8 insertions(+), 5 deletions(-)
diff --git a/airflow/www/static/js/dag.js b/airflow/www/static/js/dag.js
index d0b329a..6fb7176 100644
--- a/airflow/www/static/js/dag.js
+++ b/airflow/www/static/js/dag.js
@@ -279,10 +279,12 @@ $('#pause_resume').on('change', function onChange() {
$('#next-run').on('mouseover', () => {
$('#next-run').attr('data-original-title', () => {
let newTitle = '';
- newTitle += `<strong>Run After:</strong> ${formatDateTime(nextRun.createAfter)}<br><br>`;
- newTitle += '<strong>Data Interval</strong><br>';
- newTitle += `Start: ${formatDateTime(nextRun.intervalStart)}<br>`;
- newTitle += `End: ${formatDateTime(nextRun.intervalEnd)}`;
+ if (nextRun.createAfter) newTitle += `<strong>Run After:</strong> ${formatDateTime(nextRun.createAfter)}<br><br>`;
+ if (nextRun.intervalStart && nextRun.intervalEnd) {
+ newTitle += '<strong>Data Interval</strong><br>';
+ newTitle += `Start: ${formatDateTime(nextRun.intervalStart)}<br>`;
+ newTitle += `End: ${formatDateTime(nextRun.intervalEnd)}`;
+ }
return newTitle;
});
});
diff --git a/airflow/www/templates/airflow/dag.html b/airflow/www/templates/airflow/dag.html
index 6d4aaee..a88e3d4 100644
--- a/airflow/www/templates/airflow/dag.html
+++ b/airflow/www/templates/airflow/dag.html
@@ -36,7 +36,8 @@
<meta name="extra_links_url" content="{{ url_for('Airflow.extra_links') }}">
<meta name="paused_url" content="{{ url_for('Airflow.paused') }}">
<meta name="tree_data" content="{{ url_for('Airflow.tree_data') }}">
- {% if dag_model is defined and dag_model.next_dagrun_create_after is defined %}
+ {%
+ if dag_model is defined and dag_model.next_dagrun_create_after is defined and dag_model.next_dagrun_create_after is not none %}
<meta name="next_dagrun_create_after" content="{{ dag_model.next_dagrun_create_after }}">
<meta name="next_dagrun_data_interval_start" content="{{ dag_model.next_dagrun_data_interval_start }}">
<meta name="next_dagrun_data_interval_end" content="{{ dag_model.next_dagrun_data_interval_end }}">
[airflow] 04/06: Crerate TI context with data interval compat layer
(#19148)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit be29f5de828580e955cdd87225b1cd8d636ce767
Author: Tzu-ping Chung <tp...@astronomer.io>
AuthorDate: Fri Oct 22 14:50:22 2021 +0800
Crerate TI context with data interval compat layer (#19148)
(cherry picked from commit 1159133040b3513bcc88921823fa001e9773276d)
---
airflow/models/taskinstance.py | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 71ab107..a9f30dc 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1778,6 +1778,7 @@ class TaskInstance(Base, LoggingMixin):
integrate_macros_plugins()
dag_run = self.get_dagrun(session)
+ data_interval = dag.get_run_data_interval(dag_run)
params = ParamsDict(suppress_exception=ignore_param_exceptions)
@@ -1998,8 +1999,8 @@ class TaskInstance(Base, LoggingMixin):
'conf': conf,
'dag': dag,
'dag_run': dag_run,
- 'data_interval_end': timezone.coerce_datetime(dag_run.data_interval_end),
- 'data_interval_start': timezone.coerce_datetime(dag_run.data_interval_start),
+ 'data_interval_end': timezone.coerce_datetime(data_interval.end),
+ 'data_interval_start': timezone.coerce_datetime(data_interval.start),
'ds': ds,
'ds_nodash': ds_nodash,
'execution_date': deprecated_proxy(
[airflow] 06/06: Add TriggererJob to jobs check command (#19179)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6ef71d55f2ce510cabed4fcc3701e924aec2164a
Author: Kamil BreguĊa <mi...@users.noreply.github.com>
AuthorDate: Sat Oct 23 01:32:00 2021 +0200
Add TriggererJob to jobs check command (#19179)
(cherry picked from commit d3ac01052bad07f6ec341ab714faabed913169ce)
---
airflow/cli/cli_parser.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow/cli/cli_parser.py b/airflow/cli/cli_parser.py
index 85687a4..64cc582 100644
--- a/airflow/cli/cli_parser.py
+++ b/airflow/cli/cli_parser.py
@@ -725,7 +725,7 @@ ARG_NAMESPACE = Arg(
# jobs check
ARG_JOB_TYPE_FILTER = Arg(
('--job-type',),
- choices=('BackfillJob', 'LocalTaskJob', 'SchedulerJob'),
+ choices=('BackfillJob', 'LocalTaskJob', 'SchedulerJob', 'TriggererJob'),
action='store',
help='The type of job(s) that will be checked.',
)