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/15 15:48:54 UTC
[airflow] branch v2-2-test updated (e35588c -> 2e8773b)
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 e35588c BugFix: Null execution date on insert to ``task_fail`` violating NOT NULL (#18979)
new d9af965 CLI: Fail ``backfill`` command before loading DAGs if missing args (#18994)
new f094e6f Fix occassional deadloc on MSSQL test DagMaker cleanup (#18857)
new 2e8773b Don't bake ENV and _cmd into tmp config for non-sudo (#18772)
The 3 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/commands/dag_command.py | 4 +-
airflow/task/task_runner/base_task_runner.py | 4 +-
airflow/utils/configuration.py | 14 ++++++-
tests/cli/commands/test_dag_command.py | 10 +++++
tests/conftest.py | 54 ++++++++++++++++---------
tests/task/task_runner/test_base_task_runner.py | 54 +++++++++++++++++++++++++
6 files changed, 114 insertions(+), 26 deletions(-)
create mode 100644 tests/task/task_runner/test_base_task_runner.py
[airflow] 03/03: Don't bake ENV and _cmd into tmp config for
non-sudo (#18772)
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 2e8773b08e547ff6ca585cc0d5a35fccdaf2b7c5
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Wed Oct 13 15:00:13 2021 +0100
Don't bake ENV and _cmd into tmp config for non-sudo (#18772)
If we are running tasks via sudo then AIRFLOW__ config env vars won't be
visible anymore (without them showing up in `ps`) and we likely might
not have permission to run the _cmd's specified to find the passwords.
But if we are running as the same user then there is no need to "bake"
those options in to the temporary config file -- if the operator decided
they didn't want those values appearing in a config file on disk, then
lets do our best to respect that.
Note: this commit originally appears in 2019 but a critical piece was
missing, meaning that the secrets/envs were still actually appearing.
(cherry picked from commit a90878cf660ffe73973b4e4487c1e691cc212925)
---
airflow/task/task_runner/base_task_runner.py | 4 +-
airflow/utils/configuration.py | 14 ++++++-
tests/conftest.py | 5 +++
tests/task/task_runner/test_base_task_runner.py | 54 +++++++++++++++++++++++++
4 files changed, 73 insertions(+), 4 deletions(-)
diff --git a/airflow/task/task_runner/base_task_runner.py b/airflow/task/task_runner/base_task_runner.py
index f70020d..28bb847 100644
--- a/airflow/task/task_runner/base_task_runner.py
+++ b/airflow/task/task_runner/base_task_runner.py
@@ -66,7 +66,7 @@ class BaseTaskRunner(LoggingMixin):
# want to have to specify them in the sudo call - they would show
# up in `ps` that way! And run commands now, as the other user
# might not be able to run the cmds to get credentials
- cfg_path = tmp_configuration_copy(chmod=0o600)
+ cfg_path = tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True)
# Give ownership of file to user; only they can read and write
subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path], close_fds=True)
@@ -83,7 +83,7 @@ class BaseTaskRunner(LoggingMixin):
# we are running as the same user, and can pass through environment
# variables then we don't need to include those in the config copy
# - the runner can read/execute those values as it needs
- cfg_path = tmp_configuration_copy(chmod=0o600)
+ cfg_path = tmp_configuration_copy(chmod=0o600, include_env=False, include_cmds=False)
self._error_file = NamedTemporaryFile(delete=True)
if self.run_as_user:
diff --git a/airflow/utils/configuration.py b/airflow/utils/configuration.py
index 7057d64..cc9273c 100644
--- a/airflow/utils/configuration.py
+++ b/airflow/utils/configuration.py
@@ -23,13 +23,23 @@ from tempfile import mkstemp
from airflow.configuration import conf
-def tmp_configuration_copy(chmod=0o600):
+def tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):
"""
Returns a path for a temporary file including a full copy of the configuration
settings.
+
+ :param include_env: Should the value of configuration from ``AIRFLOW__``
+ environment variables be included or not
+ :type include_env: bool
+ :param include_cmds: Should the result of calling any *_cmd config be
+ set (True, default), or should the _cmd options be left as the
+ command to run (False)
+ :type include_cmds: bool
:return: a path to a temporary file
"""
- cfg_dict = conf.as_dict(display_sensitive=True, raw=True)
+ cfg_dict = conf.as_dict(
+ display_sensitive=True, raw=True, include_cmds=include_cmds, include_env=include_env
+ )
temp_fd, cfg_path = mkstemp()
with os.fdopen(temp_fd, 'w') as temp_file:
diff --git a/tests/conftest.py b/tests/conftest.py
index 22770c5..94f915f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -476,6 +476,8 @@ def dag_maker(request):
from airflow.utils.log.logging_mixin import LoggingMixin
class DagFactory(LoggingMixin):
+ _own_session = False
+
def __init__(self):
from airflow.models import DagBag
@@ -577,6 +579,7 @@ def dag_maker(request):
from airflow.utils import timezone
if session is None:
+ self._own_session = True
session = settings.Session()
self.kwargs = kwargs
@@ -629,6 +632,8 @@ def dag_maker(request):
synchronize_session=False
)
self.session.commit()
+ if self._own_session:
+ self.session.expunge_all()
factory = DagFactory()
diff --git a/tests/task/task_runner/test_base_task_runner.py b/tests/task/task_runner/test_base_task_runner.py
new file mode 100644
index 0000000..499bba7
--- /dev/null
+++ b/tests/task/task_runner/test_base_task_runner.py
@@ -0,0 +1,54 @@
+#
+# 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.
+from unittest import mock
+
+import pytest
+
+from airflow.jobs.local_task_job import LocalTaskJob
+from airflow.models.baseoperator import BaseOperator
+from airflow.task.task_runner.base_task_runner import BaseTaskRunner
+
+
+@pytest.mark.parametrize(["impersonation"], (("nobody",), (None,)))
+@mock.patch('subprocess.call')
+@mock.patch('os.chown')
+@mock.patch('airflow.task.task_runner.base_task_runner.tmp_configuration_copy')
+def test_config_copy_mode(tmp_configuration_copy, chown, subprocess_call, dag_maker, impersonation):
+ with dag_maker("test"):
+ BaseOperator(task_id="task_1", run_as_user=impersonation)
+
+ dr = dag_maker.create_dagrun()
+
+ ti = dr.task_instances[0]
+ job = LocalTaskJob(ti)
+ runner = BaseTaskRunner(job)
+ # So we don't try to delete it -- cos the file wont exist
+ del runner._cfg_path
+
+ includes = bool(impersonation)
+
+ tmp_configuration_copy.assert_called_with(chmod=0o600, include_env=includes, include_cmds=includes)
+
+ if impersonation:
+ chown.assert_called()
+ subprocess_call.assert_called_with(
+ ['sudo', 'chown', impersonation, tmp_configuration_copy.return_value], close_fds=True
+ )
+ else:
+ chown.assert_not_called()
+ subprocess_call.not_assert_called()
[airflow] 01/03: CLI: Fail ``backfill`` command before loading DAGs
if missing args (#18994)
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 d9af96539694dc85c85a86fc0acf7fea00569b13
Author: Sam Wheating <sa...@shopify.com>
AuthorDate: Fri Oct 15 00:38:23 2021 -0700
CLI: Fail ``backfill`` command before loading DAGs if missing args (#18994)
I was looking through some of the CLI code last week trying to improve the speed of `airflow user` commands and I noticed this small issue.
If neither the `start_date` or `end_date` argument is provided then the command will fail, but it will first parse all of the DAGs which can take up to several minutes in large deployments.
Now the command will fail faster, allowing the user to adjust their command and retry.
(cherry picked from commit b814ab43d62fad83c1083a7bc3a8d009c6103213)
---
airflow/cli/commands/dag_command.py | 4 ++--
tests/cli/commands/test_dag_command.py | 10 ++++++++++
2 files changed, 12 insertions(+), 2 deletions(-)
diff --git a/airflow/cli/commands/dag_command.py b/airflow/cli/commands/dag_command.py
index 30b0148..b94d6cf 100644
--- a/airflow/cli/commands/dag_command.py
+++ b/airflow/cli/commands/dag_command.py
@@ -67,11 +67,11 @@ def dag_backfill(args, dag=None):
if args.ignore_first_depends_on_past is False:
args.ignore_first_depends_on_past = True
- dag = dag or get_dag(args.subdir, args.dag_id)
-
if not args.start_date and not args.end_date:
raise AirflowException("Provide a start_date and/or end_date")
+ dag = dag or get_dag(args.subdir, args.dag_id)
+
# If only one date is passed, using same as start and end
args.end_date = args.end_date or args.start_date
args.start_date = args.start_date or args.end_date
diff --git a/tests/cli/commands/test_dag_command.py b/tests/cli/commands/test_dag_command.py
index 67dcabe..0044761 100644
--- a/tests/cli/commands/test_dag_command.py
+++ b/tests/cli/commands/test_dag_command.py
@@ -158,6 +158,16 @@ class TestCliDags(unittest.TestCase):
)
mock_run.reset_mock()
+ @mock.patch("airflow.cli.commands.dag_command.get_dag")
+ def test_backfill_fails_without_loading_dags(self, mock_get_dag):
+
+ cli_args = self.parser.parse_args(['dags', 'backfill', 'example_bash_operator'])
+
+ with pytest.raises(AirflowException):
+ dag_command.dag_backfill(cli_args)
+
+ mock_get_dag.assert_not_called()
+
def test_show_dag_print(self):
with contextlib.redirect_stdout(io.StringIO()) as temp_stdout:
dag_command.dag_show(self.parser.parse_args(['dags', 'show', 'example_bash_operator']))
[airflow] 02/03: Fix occassional deadloc on MSSQL test DagMaker
cleanup (#18857)
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 f094e6f917447fed4fbd986175432176f448fe93
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Sat Oct 9 22:36:53 2021 +0200
Fix occassional deadloc on MSSQL test DagMaker cleanup (#18857)
Occasionally our tests in CI for MsSQL failed with deadlock on
cleaning SerilizedDag table. After closer inspection, the
deadlock happened in the test dag_maker cleanup() code.
This PR fixes it by attempting to retry the cleaning in case
of deadlock.
(cherry picked from commit fd45f5f3e38b80993d5624480a793be381194f04)
---
tests/conftest.py | 49 +++++++++++++++++++++++++++++--------------------
1 file changed, 29 insertions(+), 20 deletions(-)
diff --git a/tests/conftest.py b/tests/conftest.py
index 36e88b3..22770c5 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -473,7 +473,9 @@ def dag_maker(request):
if serialized_marker:
(want_serialized,) = serialized_marker.args or (True,)
- class DagFactory:
+ from airflow.utils.log.logging_mixin import LoggingMixin
+
+ class DagFactory(LoggingMixin):
def __init__(self):
from airflow.models import DagBag
@@ -601,25 +603,32 @@ def dag_maker(request):
def cleanup(self):
from airflow.models import DagModel, DagRun, TaskInstance, XCom
from airflow.models.serialized_dag import SerializedDagModel
-
- dag_ids = list(self.dagbag.dag_ids)
- if not dag_ids:
- return
- # To isolate problems here with problems from elsewhere on the session object
- self.session.flush()
-
- self.session.query(SerializedDagModel).filter(SerializedDagModel.dag_id.in_(dag_ids)).delete(
- synchronize_session=False
- )
- self.session.query(DagRun).filter(DagRun.dag_id.in_(dag_ids)).delete(synchronize_session=False)
- self.session.query(TaskInstance).filter(TaskInstance.dag_id.in_(dag_ids)).delete(
- synchronize_session=False
- )
- self.session.query(XCom).filter(XCom.dag_id.in_(dag_ids)).delete(synchronize_session=False)
- self.session.query(DagModel).filter(DagModel.dag_id.in_(dag_ids)).delete(
- synchronize_session=False
- )
- self.session.commit()
+ from airflow.utils.retries import run_with_db_retries
+
+ for attempt in run_with_db_retries(logger=self.log):
+ with attempt:
+ dag_ids = list(self.dagbag.dag_ids)
+ if not dag_ids:
+ return
+ # To isolate problems here with problems from elsewhere on the session object
+ self.session.flush()
+
+ self.session.query(SerializedDagModel).filter(
+ SerializedDagModel.dag_id.in_(dag_ids)
+ ).delete(synchronize_session=False)
+ self.session.query(DagRun).filter(DagRun.dag_id.in_(dag_ids)).delete(
+ synchronize_session=False
+ )
+ self.session.query(TaskInstance).filter(TaskInstance.dag_id.in_(dag_ids)).delete(
+ synchronize_session=False
+ )
+ self.session.query(XCom).filter(XCom.dag_id.in_(dag_ids)).delete(
+ synchronize_session=False
+ )
+ self.session.query(DagModel).filter(DagModel.dag_id.in_(dag_ids)).delete(
+ synchronize_session=False
+ )
+ self.session.commit()
factory = DagFactory()