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 2022/02/08 00:16:21 UTC
[airflow] branch v2-2-test updated (d4000e8 -> b92cf65)
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 d4000e8 Update stat_name_handler documentation (#21298)
new a04ead2 Fix the incorrect scheduling time for the first run of dag (#21011)
new e11b90c Update error docs to include before_send option (#21275)
new 317dd29 Filter celery stuck task query to exclude completed tasks (#21335)
new 99abbfc Update example DAGs (#21372)
new 8971e0f Fix docs link for smart sensor deprecation (#21394)
new e2f3e94 Avoid deadlock when rescheduling task (#21362)
new b92cf65 Add possibility to create user in the Remote User mode (#19963)
The 7 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:
UPDATING.md | 2 +-
.../example_passing_params_via_test_command.py | 11 +++++-----
airflow/example_dags/tutorial.py | 2 --
airflow/executors/celery_executor.py | 7 +++++-
airflow/models/taskinstance.py | 15 ++++++++++++-
airflow/timetables/interval.py | 2 +-
airflow/www/views.py | 25 ++++++++++++++++++++++
docs/apache-airflow/logging-monitoring/errors.rst | 10 +++++++--
docs/apache-airflow/tutorial.rst | 9 ++------
tests/cli/commands/test_task_command.py | 1 -
tests/executors/test_celery_executor.py | 8 +++++--
tests/timetables/test_interval_timetable.py | 21 ++++++++++++++++++
12 files changed, 90 insertions(+), 23 deletions(-)
[airflow] 02/07: Update error docs to include before_send option (#21275)
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 e11b90cd5800f1071bdee311b8f4cbb5e8d4b611
Author: Abhijeet Prasad <de...@gmail.com>
AuthorDate: Thu Feb 3 18:15:21 2022 -0500
Update error docs to include before_send option (#21275)
https://github.com/apache/airflow/pull/18261 Added support for the `before_send` option when initializing the Sentry SDK in airflow. This patch updates the documentation to reflect this change.
(cherry picked from commit b38391e2f91760e64576723c876341f532a6ee2d)
---
docs/apache-airflow/logging-monitoring/errors.rst | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git a/docs/apache-airflow/logging-monitoring/errors.rst b/docs/apache-airflow/logging-monitoring/errors.rst
index 9f4256a..30bb66d 100644
--- a/docs/apache-airflow/logging-monitoring/errors.rst
+++ b/docs/apache-airflow/logging-monitoring/errors.rst
@@ -42,8 +42,14 @@ Add your ``SENTRY_DSN`` to your configuration file e.g. ``airflow.cfg`` in ``[se
.. note::
If this value is not provided, the SDK will try to read it from the ``SENTRY_DSN`` environment variable.
-You can supply `additional configuration options <https://docs.sentry.io/platforms/python/configuration/options>`__ based on the Python platform via ``[sentry]`` section.
-Unsupported options: ``integrations``, ``in_app_include``, ``in_app_exclude``, ``ignore_errors``, ``before_breadcrumb``, ``before_send``, ``transport``.
+The ``before_send`` option can be used to modify or drop events before they are sent to Sentry. To set this option, provide a dotted path to a before_send function that the sentry SDK should be configured to use.
+
+.. code-block:: ini
+
+ [sentry]
+ before_send = path.to.my.sentry.before_send
+
+You can supply `additional configuration options <https://docs.sentry.io/platforms/python/configuration/options>`__ based on the Python platform via ``[sentry]`` section. Unsupported options: ``integrations``, ``in_app_include``, ``in_app_exclude``, ``ignore_errors``, ``before_breadcrumb``, ``transport``.
Tags
-----
[airflow] 01/07: Fix the incorrect scheduling time for the first run of dag (#21011)
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 a04ead208da620040b718151c0800bb2ff59bbc4
Author: wano <55...@users.noreply.github.com>
AuthorDate: Mon Feb 7 02:02:57 2022 +0800
Fix the incorrect scheduling time for the first run of dag (#21011)
When Catchup_by_default is set to false and start_date in the DAG is the
previous day, the first schedule time for this DAG may be incorrect
Co-authored-by: wanlce <wh...@foxmail.com>
(cherry picked from commit 0bcca55f4881bacc3fbe86f69e71981f5552b398)
---
airflow/timetables/interval.py | 2 +-
tests/timetables/test_interval_timetable.py | 21 +++++++++++++++++++++
2 files changed, 22 insertions(+), 1 deletion(-)
diff --git a/airflow/timetables/interval.py b/airflow/timetables/interval.py
index d669cb6..01fac3a 100644
--- a/airflow/timetables/interval.py
+++ b/airflow/timetables/interval.py
@@ -218,7 +218,7 @@ class CronDataIntervalTimetable(_DataIntervalTimetable):
raise AssertionError("next schedule shouldn't be earlier")
if earliest is None:
return new_start
- return max(new_start, earliest)
+ return max(new_start, self._align(earliest))
def infer_manual_data_interval(self, *, run_after: DateTime) -> DataInterval:
# Get the last complete period before run_after, e.g. if a DAG run is
diff --git a/tests/timetables/test_interval_timetable.py b/tests/timetables/test_interval_timetable.py
index 842cc1f2..fe09e0c 100644
--- a/tests/timetables/test_interval_timetable.py
+++ b/tests/timetables/test_interval_timetable.py
@@ -35,11 +35,32 @@ 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)
+YESTERDAY = CURRENT_TIME - datetime.timedelta(days=1)
HOURLY_CRON_TIMETABLE = CronDataIntervalTimetable("@hourly", TIMEZONE)
HOURLY_TIMEDELTA_TIMETABLE = DeltaDataIntervalTimetable(datetime.timedelta(hours=1))
HOURLY_RELATIVEDELTA_TIMETABLE = DeltaDataIntervalTimetable(dateutil.relativedelta.relativedelta(hours=1))
+CRON_TIMETABLE = CronDataIntervalTimetable("30 16 * * *", TIMEZONE)
+DELTA_FROM_MIDNIGHT = datetime.timedelta(minutes=30, hours=16)
+
+
+@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_first_starts_at_current_time(
+ last_automated_data_interval: Optional[DataInterval],
+) -> None:
+ """If ``catchup=False`` and start_date is a day before"""
+ next_info = CRON_TIMETABLE.next_dagrun_info(
+ last_automated_data_interval=last_automated_data_interval,
+ restriction=TimeRestriction(earliest=YESTERDAY, latest=None, catchup=False),
+ )
+ expected_start = YESTERDAY + DELTA_FROM_MIDNIGHT
+ assert next_info == DagRunInfo.interval(start=expected_start, end=CURRENT_TIME + DELTA_FROM_MIDNIGHT)
+
@pytest.mark.parametrize(
"timetable",
[airflow] 03/07: Filter celery stuck task query to exclude completed tasks (#21335)
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 317dd298e02f54156f362f8d39307e575106b5b5
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Mon Feb 7 10:43:51 2022 +0100
Filter celery stuck task query to exclude completed tasks (#21335)
On testing #19769, it was reported that there was a spike in CPU usage https://github.com/apache/airflow/pull/19769#issuecomment-1029755436
Hopefully, this will fix it
(cherry picked from commit a49224fa7ce45e9765c0d752edc30430e0d3ce14)
---
airflow/executors/celery_executor.py | 7 ++++++-
tests/executors/test_celery_executor.py | 8 ++++++--
2 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/airflow/executors/celery_executor.py b/airflow/executors/celery_executor.py
index 8daced6..9185f1b 100644
--- a/airflow/executors/celery_executor.py
+++ b/airflow/executors/celery_executor.py
@@ -404,7 +404,12 @@ class CeleryExecutor(BaseExecutor):
session_ = app.backend.ResultSession()
task_cls = getattr(app.backend, "task_cls", TaskDb)
with session_cleanup(session_):
- celery_task_ids = [t.task_id for t in session_.query(task_cls.task_id).all()]
+ celery_task_ids = [
+ t.task_id
+ for t in session_.query(task_cls.task_id)
+ .filter(~task_cls.status.in_([celery_states.SUCCESS, celery_states.FAILURE]))
+ .all()
+ ]
self.log.debug("Checking for stuck queued tasks")
max_allowed_time = utcnow() - self.task_adoption_timeout
diff --git a/tests/executors/test_celery_executor.py b/tests/executors/test_celery_executor.py
index 5632f7d..f057440 100644
--- a/tests/executors/test_celery_executor.py
+++ b/tests/executors/test_celery_executor.py
@@ -42,7 +42,7 @@ from parameterized import parameterized
from airflow.configuration import conf
from airflow.exceptions import AirflowException, AirflowTaskTimeout
from airflow.executors import celery_executor
-from airflow.executors.celery_executor import BulkStateFetcher
+from airflow.executors.celery_executor import BulkStateFetcher, CeleryExecutor
from airflow.models.baseoperator import BaseOperator
from airflow.models.dag import DAG
from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance, TaskInstanceKey
@@ -532,6 +532,7 @@ class TestCeleryExecutor:
assert ti.state == state
@mock.patch("celery.backends.database.DatabaseBackend.ResultSession")
+ @mock.patch.object(CeleryExecutor, "update_all_task_states")
@pytest.mark.backend("mysql", "postgres")
@freeze_time("2020-01-01")
@pytest.mark.parametrize(
@@ -543,6 +544,7 @@ class TestCeleryExecutor:
)
def test_the_check_interval_to_clear_stuck_queued_task_is_correct_for_db_query(
self,
+ mock_update_all_task_states,
mock_result_session,
task_id,
state,
@@ -557,7 +559,9 @@ class TestCeleryExecutor:
mock_backend = DatabaseBackend(app=celery_executor.app, url="sqlite3://")
with mock.patch('airflow.executors.celery_executor.Celery.backend', mock_backend):
mock_session = mock_backend.ResultSession.return_value
- mock_session.query.return_value.all.return_value = [result_obj("SUCCESS", task_id)]
+ mock_session.query.return_value.filter.return_value.all.return_value = [
+ result_obj("SUCCESS", task_id)
+ ]
last_check_time = time.time() - 302 # should clear ti state
[airflow] 05/07: Fix docs link for smart sensor deprecation (#21394)
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 8971e0ff4283febfdb8a292e978ee1247f583f0d
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Mon Feb 7 08:55:20 2022 -0700
Fix docs link for smart sensor deprecation (#21394)
We are releasing the deprecation in version 2.2.4, not 2.3.0 like
originally planned.
(cherry picked from commit 3a780380d8f5d50ffc876c326e70ee0eee033c0d)
---
UPDATING.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/UPDATING.md b/UPDATING.md
index a75b2d6..2ed4aac 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -87,7 +87,7 @@ https://developers.google.com/style/inclusive-documentation
Smart sensors, an "early access" feature added in Airflow 2, are now deprecated and will be removed in Airflow 2.4.0. They have been superseded by Deferable Operators, added in Airflow 2.2.0.
-See [Migrating to Deferrable Operators](https://airflow.apache.org/docs/apache-airflow/2.3.0/concepts/smart-sensors.html#migrating-to-deferrable-operators) for details on how to migrate.
+See [Migrating to Deferrable Operators](https://airflow.apache.org/docs/apache-airflow/2.2.4/concepts/smart-sensors.html#migrating-to-deferrable-operators) for details on how to migrate.
## Airflow 2.2.3
[airflow] 04/07: Update example DAGs (#21372)
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 99abbfc873d3df070fff5926d69c31bfad462359
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Mon Feb 7 11:24:31 2022 -0700
Update example DAGs (#21372)
(cherry picked from commit 7a38ec2ad3b3bd6fda5e1ee9fe9e644ccb8b4c12)
---
.../example_dags/example_passing_params_via_test_command.py | 11 ++++++-----
airflow/example_dags/tutorial.py | 2 --
docs/apache-airflow/tutorial.rst | 9 ++-------
tests/cli/commands/test_task_command.py | 1 -
4 files changed, 8 insertions(+), 15 deletions(-)
diff --git a/airflow/example_dags/example_passing_params_via_test_command.py b/airflow/example_dags/example_passing_params_via_test_command.py
index e3f04c4..d4781af 100644
--- a/airflow/example_dags/example_passing_params_via_test_command.py
+++ b/airflow/example_dags/example_passing_params_via_test_command.py
@@ -68,17 +68,18 @@ with DAG(
) as dag:
run_this = my_py_command(params={"miff": "agg"})
- my_templated_command = dedent(
+ my_command = dedent(
+ """
+ echo "'foo' was passed in via Airflow CLI Test command with value '$FOO'"
+ echo "'miff' was passed in via BashOperator with value '$MIFF'"
"""
- echo " 'foo was passed in via Airflow CLI Test command with value {{ params.foo }} "
- echo " 'miff was passed in via BashOperator with value {{ params.miff }} "
- """
)
also_run_this = BashOperator(
task_id='also_run_this',
- bash_command=my_templated_command,
+ bash_command=my_command,
params={"miff": "agg"},
+ env={"FOO": "{{ params.foo }}", "MIFF": "{{ params.miff }}"},
)
env_var_test_task = print_env_vars()
diff --git a/airflow/example_dags/tutorial.py b/airflow/example_dags/tutorial.py
index 1049772..ff2bd2f 100644
--- a/airflow/example_dags/tutorial.py
+++ b/airflow/example_dags/tutorial.py
@@ -109,7 +109,6 @@ with DAG(
{% for i in range(5) %}
echo "{{ ds }}"
echo "{{ macros.ds_add(ds, 7)}}"
- echo "{{ params.my_param }}"
{% endfor %}
"""
)
@@ -118,7 +117,6 @@ with DAG(
task_id='templated',
depends_on_past=False,
bash_command=templated_command,
- params={'my_param': 'Parameter I passed in'},
)
# [END jinja_template]
diff --git a/docs/apache-airflow/tutorial.rst b/docs/apache-airflow/tutorial.rst
index 1c32e78..0034b2c 100644
--- a/docs/apache-airflow/tutorial.rst
+++ b/docs/apache-airflow/tutorial.rst
@@ -151,13 +151,8 @@ stamp").
:end-before: [END jinja_template]
Notice that the ``templated_command`` contains code logic in ``{% %}`` blocks,
-references parameters like ``{{ ds }}``, calls a function as in
-``{{ macros.ds_add(ds, 7)}}``, and references a user-defined parameter
-in ``{{ params.my_param }}``.
-
-The ``params`` hook in ``BaseOperator`` allows you to pass a dictionary of
-parameters and/or objects to your templates. Please take the time
-to understand how the parameter ``my_param`` makes it through to the template.
+references parameters like ``{{ ds }}``, and calls a function as in
+``{{ macros.ds_add(ds, 7)}}``.
Files can also be passed to the ``bash_command`` argument, like
``bash_command='templated_command.sh'``, where the file location is relative to
diff --git a/tests/cli/commands/test_task_command.py b/tests/cli/commands/test_task_command.py
index 201af16..76c6cdb 100644
--- a/tests/cli/commands/test_task_command.py
+++ b/tests/cli/commands/test_task_command.py
@@ -263,7 +263,6 @@ class TestCliTasks(unittest.TestCase):
assert 'echo "2016-01-01"' in output
assert 'echo "2016-01-08"' in output
- assert 'echo "Parameter I passed in"' in output
def test_cli_run_when_pickle_and_dag_cli_method_selected(self):
"""
[airflow] 06/07: Avoid deadlock when rescheduling task (#21362)
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 e2f3e94f054bc7217378aa743d185e5e508468ea
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Mon Feb 7 20:12:05 2022 +0100
Avoid deadlock when rescheduling task (#21362)
The scheduler job performs scheduling after locking the "scheduled"
DagRun row for writing. This should prevent from modifying DagRun
and related task instances by another scheduler or "mini-scheduler"
run after task is completed.
However there is apparently one more case where the DagRun is being
locked by "Task" processes - namely when task throws
AirflowRescheduleException. In this case a new "TaskReschedule"
entity is inserted into the database and it also performs lock
on the DagRun (because TaskReschedule has "DagRun" relationship.
This PR modifies handling the AirflowRescheduleException to obtain the
very same DagRun lock before it attempts to insert TaskReschedule
entity.
Seems that TaskReschedule is the only one that has this relationship
so likely all the misterious SchedulerJob deadlock cases we
experienced might be explained (and fixed) by this one.
It is likely that this one:
* Fixes: #16982
* Fixes: #19957
(cherry picked from commit 6d110b565a505505351d1ff19592626fb24e4516)
---
airflow/models/taskinstance.py | 15 ++++++++++++++-
1 file changed, 14 insertions(+), 1 deletion(-)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index ec34156..2dcc923 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -93,7 +93,7 @@ from airflow.utils.operator_helpers import context_to_airflow_vars
from airflow.utils.platform import getuser
from airflow.utils.retries import run_with_db_retries
from airflow.utils.session import create_session, provide_session
-from airflow.utils.sqlalchemy import ExtendedJSON, UtcDateTime
+from airflow.utils.sqlalchemy import ExtendedJSON, UtcDateTime, with_row_locks
from airflow.utils.state import DagRunState, State
from airflow.utils.timeout import timeout
@@ -1657,11 +1657,24 @@ class TaskInstance(Base, LoggingMixin):
# Don't record reschedule request in test mode
if test_mode:
return
+
+ from airflow.models.dagrun import DagRun # Avoid circular import
+
self.refresh_from_db(session)
self.end_date = timezone.utcnow()
self.set_duration()
+ # Lock DAG run to be sure not to get into a deadlock situation when trying to insert
+ # TaskReschedule which apparently also creates lock on corresponding DagRun entity
+ with_row_locks(
+ session.query(DagRun).filter_by(
+ dag_id=self.dag_id,
+ run_id=self.run_id,
+ ),
+ session=session,
+ ).one()
+
# Log reschedule request
session.add(
TaskReschedule(
[airflow] 07/07: Add possibility to create user in the Remote User mode (#19963)
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 b92cf65f0300bbac305f19980384110283eba4d5
Author: Łukasz Wyszomirski <wy...@google.com>
AuthorDate: Fri Jan 28 06:18:05 2022 +0100
Add possibility to create user in the Remote User mode (#19963)
(cherry picked from commit cdd9ea66208e3d70d1cf2a34530ba69bc3c58a50)
---
airflow/www/views.py | 25 +++++++++++++++++++++++++
1 file changed, 25 insertions(+)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index f2642a7..2ed2a67 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -4731,3 +4731,28 @@ class CustomUserOIDModelView(MultiResourceUserMixin, UserOIDModelView):
class CustomUserRemoteUserModelView(MultiResourceUserMixin, UserRemoteUserModelView):
"""Customize permission names for FAB's builtin UserRemoteUserModelView."""
+
+ _class_permission_name = permissions.RESOURCE_USER
+
+ class_permission_name_mapping = {
+ 'userinfoedit': permissions.RESOURCE_MY_PROFILE,
+ 'userinfo': permissions.RESOURCE_MY_PROFILE,
+ }
+
+ method_permission_name = {
+ 'add': 'create',
+ 'userinfo': 'read',
+ 'download': 'read',
+ 'show': 'read',
+ 'list': 'read',
+ 'edit': 'edit',
+ 'userinfoedit': 'edit',
+ 'delete': 'delete',
+ }
+
+ base_permissions = [
+ permissions.ACTION_CAN_CREATE,
+ permissions.ACTION_CAN_READ,
+ permissions.ACTION_CAN_EDIT,
+ permissions.ACTION_CAN_DELETE,
+ ]