You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2020/11/20 23:44:18 UTC
[airflow] branch v1-10-test updated (4b299e8 -> f421543)
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a change to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git.
from 4b299e8 Fix the default value for VaultBackend's config_path (#12518)
new 415cda4 Create UndefinedJinjaVariablesRule (Resolves #11144) (#11241)
new 0b75a71 Fix connection upgrade rules so they run with SQLite backend: (#12502)
new 27e7981 Add readme for upgrade-check "subpackage". (#12506)
new ce7f0d5 Silence DagBag INFO logs during upgrade check (#12507)
new 81111a3 Add upgrade rule to check for mesos executor and flag to change it. (#11528)
new 9194893 Add upgrade check rule to ensure on "latest" versions (#12514)
new f421543 Add setup.cfg for apache-airflow-upgrade-check (#12517)
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:
airflow/models/dag.py | 4 +-
airflow/upgrade/README.md | 82 +++++++++
airflow/upgrade/rules/__init__.py | 2 +-
airflow/upgrade/rules/aaa_airflow_version_check.py | 87 ++++++++++
airflow/upgrade/rules/conn_id_is_unique.py | 2 +-
airflow/upgrade/rules/conn_type_is_not_nullable.py | 2 +-
...fernet_enabled.py => mesos_executor_removed.py} | 27 ++-
airflow/upgrade/rules/undefined_jinja_varaibles.py | 160 +++++++++++++++++
airflow/upgrade/setup.cfg | 64 +++++++
setup.py | 6 +-
.../rules/test_aaa_airflow_version_check.py | 75 ++++++++
...eprecated.py => test_mesos_executor_removed.py} | 20 ++-
.../rules/test_undefined_jinja_varaibles.py | 192 +++++++++++++++++++++
13 files changed, 693 insertions(+), 30 deletions(-)
create mode 100644 airflow/upgrade/README.md
create mode 100644 airflow/upgrade/rules/aaa_airflow_version_check.py
copy airflow/upgrade/rules/{fernet_enabled.py => mesos_executor_removed.py} (59%)
create mode 100644 airflow/upgrade/rules/undefined_jinja_varaibles.py
create mode 100644 airflow/upgrade/setup.cfg
create mode 100644 tests/upgrade/rules/test_aaa_airflow_version_check.py
copy tests/upgrade/rules/{test_legacy_ui_deprecated.py => test_mesos_executor_removed.py} (66%)
create mode 100644 tests/upgrade/rules/test_undefined_jinja_varaibles.py
[airflow] 07/07: Add setup.cfg for apache-airflow-upgrade-check
(#12517)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f4215436fedd6da64df4b1007fe8c9a46ef527fb
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Nov 20 21:55:22 2020 +0000
Add setup.cfg for apache-airflow-upgrade-check (#12517)
Nothing currently uses this setup.cfg from this folder -- automation for
that will follow shortly.
Now that there is a place list deps for upgrade-check I have moved
`packaging` and `importlib_meta` to test_requires of the main dist.
Build a py2+py3 wheel.
(cherry picked from commit deb7fc0ffe3ddb9bf9aad6f5f9479d20598e2fb5)
---
airflow/upgrade/setup.cfg | 64 +++++++++++++++++++++++++++++++++++++++++++++++
setup.py | 4 ++-
2 files changed, 67 insertions(+), 1 deletion(-)
diff --git a/airflow/upgrade/setup.cfg b/airflow/upgrade/setup.cfg
new file mode 100644
index 0000000..ddd708c
--- /dev/null
+++ b/airflow/upgrade/setup.cfg
@@ -0,0 +1,64 @@
+# 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.
+
+[metadata]
+version=1.0.0
+name = apache-airflow-upgrade-check
+description = Check for compatibility between Airflow versions
+long_description = file: airflow/upgrade/README.md
+long_description_content_type = text/markdown
+url = https://airflow.apache.org
+author = Apache Airflow PMC
+author-email = dev@airflow.apache.org
+license = Apache License 2.0
+license_files =
+ LICENSE
+ NOTICE
+classifiers =
+ Development Status :: 5 - Production/Stable
+ Intended Audience :: Developers
+ License :: OSI Approved :: Apache Software License
+ Programming Language :: Python :: 2.7
+ Programming Language :: Python :: 3
+ Programming Language :: Python :: 3.6
+ Programming Language :: Python :: 3.7
+ Programming Language :: Python :: 3.8
+keywords = airflow, upgrade
+project_urls =
+ Source Code=https://github.com/apache/airflow
+ Bug Tracker=https://github.com/apache/airflow/issues
+ Documentation=https://airflow.apache.org/docs/
+
+[options]
+packages = find:
+install_requires =
+ apache-airflow>=1.10.13,<3
+ importlib-metadata~=2.0; python_version<"3.8"
+ packaging
+python_requires = >=2.7,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*
+setup_requires =
+ setuptools>=40.0
+ wheel
+zip_safe = no
+
+[options.packages.find]
+include =
+ airflow.upgrade
+ airflow.upgrade.*
+
+[bdist_wheel]
+universal=1
diff --git a/setup.py b/setup.py
index 9617ac7..f5f2a53 100644
--- a/setup.py
+++ b/setup.py
@@ -426,12 +426,14 @@ devel = [
'flaky',
'freezegun',
'gitpython',
+ 'importlib-metadata~=2.0; python_version<"3.8"',
'ipdb',
'jira',
'mock;python_version<"3.3"',
'mongomock',
'moto==1.3.14', # TODO - fix Datasync issues to get higher version of moto:
# See: https://github.com/apache/airflow/issues/10985
+ 'packaging',
'parameterized',
'paramiko',
'pre-commit',
@@ -445,7 +447,7 @@ devel = [
'pywinrm',
'qds-sdk>=1.9.6',
'requests_mock',
- 'yamllint'
+ 'yamllint',
]
############################################################################################################
# IMPORTANT NOTE!!!!!!!!!!!!!!!
[airflow] 04/07: Silence DagBag INFO logs during upgrade check
(#12507)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ce7f0d508a3c28858848f66ab3b33a67366d8904
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Nov 20 15:21:15 2020 +0000
Silence DagBag INFO logs during upgrade check (#12507)
By default, the logs would appear in the middle of the status stream,
which makes it slightly harder to parse the output.
Before:
```
============================================= STATUS =============================================
Legacy UI is deprecated by default......................................................SUCCESS
Users must set a kubernetes.pod_template_file value.....................................FAIL
Changes in import paths of hooks, operators, sensors and others.........................FAIL
Remove airflow.AirflowMacroPlugin class.................................................SUCCESS
[2020-11-20 14:26:04,083] {__init__.py:50} INFO - Using executor SequentialExecutor
[2020-11-20 14:26:04,083] {dagbag.py:417} INFO - Filling up the DagBag from /home/ash/airflow/dags
Jinja Template Variables cannot be undefined............................................SUCCESS
```
After:
```
============================================= STATUS =============================================
Legacy UI is deprecated by default......................................................SUCCESS
Users must set a kubernetes.pod_template_file value.....................................FAIL
Changes in import paths of hooks, operators, sensors and others.........................FAIL
Remove airflow.AirflowMacroPlugin class.................................................SUCCESS
Jinja Template Variables cannot be undefined............................................SUCCESS
```
(cherry picked from commit 8e5f7227a4b00149c326637bb51409b8da6caa81)
---
airflow/upgrade/rules/undefined_jinja_varaibles.py | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git a/airflow/upgrade/rules/undefined_jinja_varaibles.py b/airflow/upgrade/rules/undefined_jinja_varaibles.py
index b97cfbc..7e39be4 100644
--- a/airflow/upgrade/rules/undefined_jinja_varaibles.py
+++ b/airflow/upgrade/rules/undefined_jinja_varaibles.py
@@ -17,6 +17,7 @@
from __future__ import absolute_import
+import logging
import re
import jinja2
@@ -131,8 +132,14 @@ The user should do either of the following to fix this -
def check(self, dagbag=None):
if not dagbag:
- dag_folder = conf.get("core", "dags_folder")
- dagbag = DagBag(dag_folder)
+ logger = logging.root
+ old_level = logger.level
+ try:
+ logger.setLevel(logging.ERROR)
+ dag_folder = conf.get("core", "dags_folder")
+ dagbag = DagBag(dag_folder)
+ finally:
+ logger.setLevel(old_level)
dags = dagbag.dags
messages = []
for dag_id, dag in dags.items():
[airflow] 01/07: Create UndefinedJinjaVariablesRule (Resolves
#11144) (#11241)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 415cda438e1d21fb9ca3a64864218f711913b014
Author: Ashmeet Lamba <as...@gmail.com>
AuthorDate: Thu Nov 19 16:33:06 2020 +0530
Create UndefinedJinjaVariablesRule (Resolves #11144) (#11241)
Adding a rule to check for undefined jinja variables when upgrading to Airflow2.0
(cherry picked from commit 18100a0ec96692bb4d7c9e80f206b66a30c65e0d)
---
airflow/models/dag.py | 4 +-
airflow/upgrade/rules/undefined_jinja_varaibles.py | 153 ++++++++++++++++
.../rules/test_undefined_jinja_varaibles.py | 192 +++++++++++++++++++++
3 files changed, 347 insertions(+), 2 deletions(-)
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 348e19d..a1908e3 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -224,7 +224,7 @@ class DAG(BaseDag, LoggingMixin):
end_date=None, # type: Optional[datetime]
full_filepath=None, # type: Optional[str]
template_searchpath=None, # type: Optional[Union[str, Iterable[str]]]
- template_undefined=jinja2.Undefined, # type: Type[jinja2.Undefined]
+ template_undefined=None, # type: Optional[Type[jinja2.Undefined]]
user_defined_macros=None, # type: Optional[Dict]
user_defined_filters=None, # type: Optional[Dict]
default_args=None, # type: Optional[Dict]
@@ -807,7 +807,7 @@ class DAG(BaseDag, LoggingMixin):
# Default values (for backward compatibility)
jinja_env_options = {
'loader': jinja2.FileSystemLoader(searchpath),
- 'undefined': self.template_undefined,
+ 'undefined': self.template_undefined or jinja2.Undefined,
'extensions': ["jinja2.ext.do"],
'cache_size': 0
}
diff --git a/airflow/upgrade/rules/undefined_jinja_varaibles.py b/airflow/upgrade/rules/undefined_jinja_varaibles.py
new file mode 100644
index 0000000..b97cfbc
--- /dev/null
+++ b/airflow/upgrade/rules/undefined_jinja_varaibles.py
@@ -0,0 +1,153 @@
+# 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 __future__ import absolute_import
+
+import re
+
+import jinja2
+import six
+
+from airflow import conf
+from airflow.models import DagBag, TaskInstance
+from airflow.upgrade.rules.base_rule import BaseRule
+from airflow.utils import timezone
+
+
+class UndefinedJinjaVariablesRule(BaseRule):
+
+ title = "Jinja Template Variables cannot be undefined"
+
+ description = """\
+The default behavior for DAG's Jinja templates has changed. Now, more restrictive validation
+of non-existent variables is applied - `jinja2.StrictUndefined`.
+
+The user should do either of the following to fix this -
+1. Fix the Jinja Templates by defining every variable or providing default values
+2. Explicitly declare `template_undefined=jinja2.Undefined` while defining the DAG
+"""
+
+ def _check_rendered_content(self, rendered_content, seen_oids=None):
+ """Replicates the logic in BaseOperator.render_template() to
+ cover all the cases needed to be checked.
+ """
+ if isinstance(rendered_content, six.string_types):
+ return set(re.findall(r"{{(.*?)}}", rendered_content))
+
+ elif isinstance(rendered_content, (int, float, bool)):
+ return set()
+
+ elif isinstance(rendered_content, (tuple, list, set)):
+ debug_error_messages = set()
+ for element in rendered_content:
+ debug_error_messages.update(self._check_rendered_content(element))
+ return debug_error_messages
+
+ elif isinstance(rendered_content, dict):
+ debug_error_messages = set()
+ for key, value in rendered_content.items():
+ debug_error_messages.update(self._check_rendered_content(value))
+ return debug_error_messages
+
+ else:
+ if seen_oids is None:
+ seen_oids = set()
+ return self._nested_check_rendered(rendered_content, seen_oids)
+
+ def _nested_check_rendered(self, rendered_content, seen_oids):
+ debug_error_messages = set()
+ if id(rendered_content) not in seen_oids:
+ seen_oids.add(id(rendered_content))
+ nested_template_fields = rendered_content.template_fields
+ for attr_name in nested_template_fields:
+ nested_rendered_content = getattr(rendered_content, attr_name)
+
+ if nested_rendered_content:
+ errors = list(
+ self._check_rendered_content(nested_rendered_content, seen_oids)
+ )
+ for i in range(len(errors)):
+ errors[i].strip()
+ errors[i] += " NestedTemplateField={}".format(attr_name)
+ debug_error_messages.update(errors)
+ return debug_error_messages
+
+ def _render_task_content(self, task, content, context):
+ completed_rendering = False
+ errors_while_rendering = []
+ while not completed_rendering:
+ # Catch errors such as {{ object.element }} where
+ # object is not defined
+ try:
+ renderend_content = task.render_template(content, context)
+ completed_rendering = True
+ except Exception as e:
+ undefined_variable = re.sub(" is undefined", "", str(e))
+ undefined_variable = re.sub("'", "", undefined_variable)
+ context[undefined_variable] = dict()
+ message = "Could not find the object '{}'".format(undefined_variable)
+ errors_while_rendering.append(message)
+ return renderend_content, errors_while_rendering
+
+ def iterate_over_template_fields(self, task):
+ messages = {}
+ task_instance = TaskInstance(task=task, execution_date=timezone.utcnow())
+ context = task_instance.get_template_context()
+ for attr_name in task.template_fields:
+ content = getattr(task, attr_name)
+ if content:
+ rendered_content, errors_while_rendering = self._render_task_content(
+ task, content, context
+ )
+ debug_error_messages = list(
+ self._check_rendered_content(rendered_content, set())
+ )
+ messages[attr_name] = errors_while_rendering + debug_error_messages
+
+ return messages
+
+ def iterate_over_dag_tasks(self, dag):
+ dag.template_undefined = jinja2.DebugUndefined
+ tasks = dag.tasks
+ messages = {}
+ for task in tasks:
+ error_messages = self.iterate_over_template_fields(task)
+ messages[task.task_id] = error_messages
+ return messages
+
+ def check(self, dagbag=None):
+ if not dagbag:
+ dag_folder = conf.get("core", "dags_folder")
+ dagbag = DagBag(dag_folder)
+ dags = dagbag.dags
+ messages = []
+ for dag_id, dag in dags.items():
+ if dag.template_undefined:
+ continue
+ dag_messages = self.iterate_over_dag_tasks(dag)
+
+ for task_id, task_messages in dag_messages.items():
+ for attr_name, error_messages in task_messages.items():
+ for error_message in error_messages:
+ message = (
+ "Possible UndefinedJinjaVariable -> DAG: {}, Task: {}, "
+ "Attribute: {}, Error: {}".format(
+ dag_id, task_id, attr_name, error_message.strip()
+ )
+ )
+ messages.append(message)
+ return messages
diff --git a/tests/upgrade/rules/test_undefined_jinja_varaibles.py b/tests/upgrade/rules/test_undefined_jinja_varaibles.py
new file mode 100644
index 0000000..83f99a3
--- /dev/null
+++ b/tests/upgrade/rules/test_undefined_jinja_varaibles.py
@@ -0,0 +1,192 @@
+# 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 tempfile import mkdtemp
+from unittest import TestCase
+
+import jinja2
+
+from airflow import DAG
+from airflow.models import DagBag
+from airflow.operators.bash_operator import BashOperator
+from airflow.upgrade.rules.undefined_jinja_varaibles import UndefinedJinjaVariablesRule
+from tests.models import DEFAULT_DATE
+
+
+class ClassWithCustomAttributes:
+ """Class for testing purpose: allows to create objects with custom attributes in one single statement."""
+
+ def __init__(self, **kwargs):
+ for key, value in kwargs.items():
+ setattr(self, key, value)
+
+ def __str__(self):
+ return "{}({})".format(ClassWithCustomAttributes.__name__, str(self.__dict__))
+
+ def __repr__(self):
+ return self.__str__()
+
+ def __eq__(self, other):
+ return self.__dict__ == other.__dict__
+
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
+
+class TestUndefinedJinjaVariablesRule(TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.empty_dir = mkdtemp()
+
+ def setUpValidDag(self):
+ self.valid_dag = DAG(
+ dag_id="test-defined-jinja-variables", start_date=DEFAULT_DATE
+ )
+
+ BashOperator(
+ task_id="templated_string",
+ depends_on_past=False,
+ bash_command="echo",
+ env={
+ "integer": "{{ params.integer }}",
+ "float": "{{ params.float }}",
+ "string": "{{ params.string }}",
+ "boolean": "{{ params.boolean }}",
+ },
+ params={
+ "integer": 1,
+ "float": 1.0,
+ "string": "test_string",
+ "boolean": True,
+ },
+ dag=self.valid_dag,
+ )
+
+ def setUpDagToSkip(self):
+ self.skip_dag = DAG(
+ dag_id="test-defined-jinja-variables",
+ start_date=DEFAULT_DATE,
+ template_undefined=jinja2.Undefined,
+ )
+
+ BashOperator(
+ task_id="templated_string",
+ depends_on_past=False,
+ bash_command="{{ undefined }}",
+ dag=self.skip_dag,
+ )
+
+ def setUpInvalidDag(self):
+ self.invalid_dag = DAG(
+ dag_id="test-undefined-jinja-variables", start_date=DEFAULT_DATE
+ )
+
+ invalid_template_command = """
+ {% for i in range(5) %}
+ echo "{{ params.defined_variable }}"
+ echo "{{ execution_date.today }}"
+ echo "{{ execution_date.invalid_element }}"
+ echo "{{ params.undefined_variable }}"
+ echo "{{ foo }}"
+ {% endfor %}
+ """
+
+ nested_validation = ClassWithCustomAttributes(
+ nested1=ClassWithCustomAttributes(
+ att1="{{ nested.undefined }}", template_fields=["att1"]
+ ),
+ nested2=ClassWithCustomAttributes(
+ att2="{{ bar }}", template_fields=["att2"]
+ ),
+ template_fields=["nested1", "nested2"],
+ )
+
+ BashOperator(
+ task_id="templated_string",
+ depends_on_past=False,
+ bash_command=invalid_template_command,
+ env={
+ "undefined_object": "{{ undefined_object.element }}",
+ "nested_object": nested_validation,
+ },
+ params={"defined_variable": "defined_value"},
+ dag=self.invalid_dag,
+ )
+
+ def setUp(self):
+ self.setUpValidDag()
+ self.setUpDagToSkip()
+ self.setUpInvalidDag()
+
+ def test_description_and_title_is_defined(self):
+ rule = UndefinedJinjaVariablesRule()
+ assert isinstance(rule.description, str)
+ assert isinstance(rule.title, str)
+
+ def test_valid_check(self):
+ dagbag = DagBag(dag_folder=self.empty_dir, include_examples=False)
+ dagbag.dags[self.valid_dag.dag_id] = self.valid_dag
+ rule = UndefinedJinjaVariablesRule()
+
+ messages = rule.check(dagbag)
+
+ assert len(messages) == 0
+
+ def test_skipping_dag_check(self):
+ dagbag = DagBag(dag_folder=self.empty_dir, include_examples=False)
+ dagbag.dags[self.skip_dag.dag_id] = self.skip_dag
+ rule = UndefinedJinjaVariablesRule()
+
+ messages = rule.check(dagbag)
+
+ assert len(messages) == 0
+
+ def test_invalid_check(self):
+ dagbag = DagBag(dag_folder=self.empty_dir, include_examples=False)
+ dagbag.dags[self.invalid_dag.dag_id] = self.invalid_dag
+ rule = UndefinedJinjaVariablesRule()
+
+ messages = rule.check(dagbag)
+
+ expected_messages = [
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: bash_command, Error: no such element: "
+ "dict object['undefined_variable']",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: bash_command, Error: no such element: "
+ "pendulum.pendulum.Pendulum object['invalid_element']",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: bash_command, Error: foo",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: env, Error: Could not find the "
+ "object 'undefined_object",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: env, Error: Could not find the object 'nested'",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: env, Error: bar NestedTemplateField=att2 "
+ "NestedTemplateField=nested2",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: env, Error: no such element: "
+ "dict object['undefined'] NestedTemplateField=att1 NestedTemplateField=nested1",
+ "Possible UndefinedJinjaVariable -> DAG: test-undefined-jinja-variables, "
+ "Task: templated_string, Attribute: env, Error: no such element: dict object['element']",
+ ]
+
+ assert len(messages) == len(expected_messages)
+ assert [m for m in messages if m in expected_messages], len(messages) == len(
+ expected_messages
+ )
[airflow] 06/07: Add upgrade check rule to ensure on "latest"
versions (#12514)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 91948939ebbcc3e302b4d15eac8a41b1dd913cb6
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Nov 20 21:21:13 2020 +0000
Add upgrade check rule to ensure on "latest" versions (#12514)
This checks against PyPI to make sure this is run with the latest
non-preview release of apache-airflow-upgrade-check, and the latest
1.10.x of apache-airflow
(cherry picked from commit d7ace0267c34bd5520d321a495399710c1c49cd1)
---
airflow/upgrade/rules/__init__.py | 2 +-
airflow/upgrade/rules/aaa_airflow_version_check.py | 87 ++++++++++++++++++++++
setup.py | 2 +
.../rules/test_aaa_airflow_version_check.py | 75 +++++++++++++++++++
4 files changed, 165 insertions(+), 1 deletion(-)
diff --git a/airflow/upgrade/rules/__init__.py b/airflow/upgrade/rules/__init__.py
index 4735c7f..97d0160 100644
--- a/airflow/upgrade/rules/__init__.py
+++ b/airflow/upgrade/rules/__init__.py
@@ -21,7 +21,7 @@ def get_rules():
"""Automatically discover all rules"""
rule_classes = []
path = os.path.dirname(os.path.abspath(__file__))
- for file in os.listdir(path):
+ for file in sorted(os.listdir(path)):
if not file.endswith(".py") or file in ("__init__.py", "base_rule.py"):
continue
py_file = file[:-3]
diff --git a/airflow/upgrade/rules/aaa_airflow_version_check.py b/airflow/upgrade/rules/aaa_airflow_version_check.py
new file mode 100644
index 0000000..ad84eb4
--- /dev/null
+++ b/airflow/upgrade/rules/aaa_airflow_version_check.py
@@ -0,0 +1,87 @@
+# 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.
+
+# This module starts with `aaa_` so that it is sorted first alphabetically, but is still a valid python module
+# name (starting with digitis is not valid)
+
+from __future__ import absolute_import
+
+from packaging.version import Version
+import requests
+
+from airflow.upgrade.rules.base_rule import BaseRule
+
+try:
+ import importlib.metadata as importlib_metadata
+except ImportError:
+ import importlib_metadata
+
+
+class VersionCheckRule(BaseRule):
+
+ title = "Check for latest versions of apache-airflow and checker"
+
+ description = """\
+Check that the latest version of apache-airflow-upgrade-check is installed, and
+that you are on the latest 1.10.x release of apache-airflow."""
+
+ def pypi_releases(self, distname):
+ """
+ Get all the non-dev releases of a dist from PyPI
+ """
+
+ resp = requests.get("https://pypi.org/pypi/{}/json".format(distname))
+ resp.raise_for_status()
+
+ for rel_string in resp.json()["releases"].keys():
+ ver = Version(rel_string)
+ if ver.is_devrelease or ver.is_prerelease:
+ continue
+ yield ver
+
+ def check(self):
+
+ current_airflow_version = Version(__import__("airflow").__version__)
+ try:
+ upgrade_check_ver = Version(
+ importlib_metadata.distribution("apache-airflow-upgrade-check").version,
+ )
+ except importlib_metadata.PackageNotFoundError:
+ upgrade_check_ver = Version("0.0.0")
+
+ try:
+ latest_airflow_v1_release = sorted(
+ filter(lambda v: v.major == 1, self.pypi_releases("apache-airflow"))
+ )[-1]
+
+ if current_airflow_version < latest_airflow_v1_release:
+ yield (
+ "There is a more recent version of apache-airflow. Please upgrade to {} and re-run this"
+ " script"
+ ).format(latest_airflow_v1_release)
+
+ latest_upgrade_check_release = sorted(
+ self.pypi_releases("apache-airflow-upgrade-check")
+ )[-1]
+
+ if upgrade_check_ver < latest_upgrade_check_release:
+ yield (
+ "There is a more recent version of apache-airflow-upgrade-check. Please upgrade to {}"
+ " and re-run this script"
+ ).format(latest_upgrade_check_release)
+ except Exception as e:
+ yield "Unable to go ask PyPI.org for latest release information: " + str(e)
diff --git a/setup.py b/setup.py
index 30917d2..9617ac7 100644
--- a/setup.py
+++ b/setup.py
@@ -603,6 +603,7 @@ INSTALL_REQUIREMENTS = [
'future>=0.16.0, <0.19',
'graphviz>=0.12',
'gunicorn>=19.5.0, <21.0',
+ 'importlib-metadata~=2.0; python_version<"3.8"',
'iso8601>=0.1.12',
'jinja2>=2.10.1, <2.12.0',
'json-merge-patch==0.2',
@@ -611,6 +612,7 @@ INSTALL_REQUIREMENTS = [
'markdown>=2.5.2, <3.0',
'marshmallow-sqlalchemy>=0.16.1, <0.24.0;python_version>="3.6"',
'marshmallow-sqlalchemy>=0.16.1, <0.19.0;python_version<"3.6"',
+ 'packaging',
'pandas>=0.17.1, <2.0',
'pendulum==1.4.4',
'pep562~=1.0;python_version<"3.7"',
diff --git a/tests/upgrade/rules/test_aaa_airflow_version_check.py b/tests/upgrade/rules/test_aaa_airflow_version_check.py
new file mode 100644
index 0000000..1c4fce0
--- /dev/null
+++ b/tests/upgrade/rules/test_aaa_airflow_version_check.py
@@ -0,0 +1,75 @@
+# 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 airflow.upgrade.rules.aaa_airflow_version_check import VersionCheckRule
+
+
+class TestVersionCheckRule:
+ def test_new_airflow_version_available(self, requests_mock):
+ requests_mock.get(
+ "https://pypi.org/pypi/apache-airflow/json",
+ json={
+ "releases": {
+ # The values here don't matter
+ "1.10.22b1": None,
+ "1.10.20": None,
+ "1.10.21": None,
+ }
+ },
+ )
+ requests_mock.get(
+ "https://pypi.org/pypi/apache-airflow-upgrade-check/json",
+ json={
+ "releases": {
+ "0.0.0": None,
+ }
+ },
+ )
+
+ results = list(VersionCheckRule().check())
+ assert len(results) == 1, results
+
+ msg = results[0]
+ assert " 1.10.21 " in msg
+ assert "apache-airflow" in msg
+ assert "apache-airflow-upgrade-check" not in msg
+
+ def test_new_airflow_upgrade_check_version_available(self, requests_mock):
+ requests_mock.get(
+ "https://pypi.org/pypi/apache-airflow/json",
+ json={
+ "releases": {
+ # The values here don't matter
+ "1.10.0": None,
+ }
+ },
+ )
+ requests_mock.get(
+ "https://pypi.org/pypi/apache-airflow-upgrade-check/json",
+ json={
+ "releases": {
+ "99.0.0": None,
+ }
+ },
+ )
+
+ results = list(VersionCheckRule().check())
+ assert len(results) == 1, results
+
+ msg = results[0]
+ assert " 99.0.0 " in msg
+ assert "apache-airflow-upgrade-check" in msg
[airflow] 02/07: Fix connection upgrade rules so they run with
SQLite backend: (#12502)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 0b75a713e812aac2f1483d5120a1a4d9d0e48554
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Nov 20 12:24:41 2020 +0000
Fix connection upgrade rules so they run with SQLite backend: (#12502)
When testing out the ugprade check command on 1.10 with the default
SQLite backend I got the following error:
```
[2020-11-20 12:10:28,248] {base.py:1372} ERROR - Error closing cursor
Traceback (most recent call last):
File "/home/ash/.virtualenvs/airflow-1-10/lib/python3.7/site-packages/sqlalchemy/engine/result.py", line 1284, in fetchall
l = self.process_rows(self._fetchall_impl())
File "/home/ash/.virtualenvs/airflow-1-10/lib/python3.7/site-packages/sqlalchemy/engine/result.py", line 1230, in _fetchall_impl
return self.cursor.fetchall()
sqlite3.ProgrammingError: Cannot operate on a closed database.
```
This was caused because the `@provide_session` decorator closed the
connection when the function returned, and since we were using a
generator expression, we hadn't yet fetched all the rows.
This changes it so the rows are fetched before returning.
(cherry picked from commit cb0a2902fc7de73e5defc9ce54f5fd1429ec4fc6)
---
airflow/upgrade/rules/conn_id_is_unique.py | 2 +-
airflow/upgrade/rules/conn_type_is_not_nullable.py | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/airflow/upgrade/rules/conn_id_is_unique.py b/airflow/upgrade/rules/conn_id_is_unique.py
index 8e1e474..edb53c8 100644
--- a/airflow/upgrade/rules/conn_id_is_unique.py
+++ b/airflow/upgrade/rules/conn_id_is_unique.py
@@ -41,5 +41,5 @@ duplicate values in conn_id column.
.having(func.count() > 1)
return (
'Connection.conn_id={} is not unique.'.format(conn_id)
- for conn_id in invalid_connections
+ for conn_id in invalid_connections.all()
)
diff --git a/airflow/upgrade/rules/conn_type_is_not_nullable.py b/airflow/upgrade/rules/conn_type_is_not_nullable.py
index 8f574d9..a411eb0 100644
--- a/airflow/upgrade/rules/conn_type_is_not_nullable.py
+++ b/airflow/upgrade/rules/conn_type_is_not_nullable.py
@@ -42,5 +42,5 @@ If you made any modifications to the table directly, make sure you don't have nu
'Connection<id={}", conn_id={}> have empty conn_type field.'.format(
conn.id, conn.conn_id
)
- for conn in invalid_connections
+ for conn in invalid_connections.all()
)
[airflow] 05/07: Add upgrade rule to check for mesos executor and
flag to change it. (#11528)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 81111a3ccdf0049aaeea4cee7dcb6b96c2db4578
Author: RaviTeja Pothana <ra...@live.com>
AuthorDate: Fri Nov 20 21:07:53 2020 +0530
Add upgrade rule to check for mesos executor and flag to change it. (#11528)
* add upgrade rule to check for mesos config and flag to remove it.
* change from checking the mesos config section to core/executor config
* remove leading new line and indent in desc
(cherry picked from commit 6739b537a016a81f5da495a894a0fe990c8ad25e)
---
airflow/upgrade/rules/mesos_executor_removed.py | 36 ++++++++++++++++
tests/upgrade/rules/test_mesos_executor_removed.py | 48 ++++++++++++++++++++++
2 files changed, 84 insertions(+)
diff --git a/airflow/upgrade/rules/mesos_executor_removed.py b/airflow/upgrade/rules/mesos_executor_removed.py
new file mode 100644
index 0000000..c0e6b52
--- /dev/null
+++ b/airflow/upgrade/rules/mesos_executor_removed.py
@@ -0,0 +1,36 @@
+# 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 airflow.upgrade.rules.base_rule import BaseRule
+from airflow.configuration import conf
+
+
+class MesosExecutorRemovedRule(BaseRule):
+ """
+ MesosExecutorRemovedRule class to ease upgrade to Airflow 2.0
+ """
+ title = "Removal of Mesos Executor"
+ description = "The Mesos Executor has been deprecated as it was not widely used and not maintained."
+
+ def check(self):
+ executor_key = conf.get(section="core", key="executor")
+ if executor_key == "MesosExecutor":
+ return (
+ "The Mesos Executor has been deprecated as it was not widely used and not maintained."
+ "Please migrate to any of the supported executors."
+ "See https://airflow.apache.org/docs/stable/executor/index.html for more details."
+ )
diff --git a/tests/upgrade/rules/test_mesos_executor_removed.py b/tests/upgrade/rules/test_mesos_executor_removed.py
new file mode 100644
index 0000000..2b1e530
--- /dev/null
+++ b/tests/upgrade/rules/test_mesos_executor_removed.py
@@ -0,0 +1,48 @@
+# 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 TestCase
+
+from airflow.upgrade.rules.mesos_executor_removed import MesosExecutorRemovedRule
+from tests.test_utils.config import conf_vars
+
+
+class TestMesosExecutorRemovedRule(TestCase):
+ @conf_vars({("core", "executor"): "MesosExecutor"})
+ def test_invalid_check(self):
+ rule = MesosExecutorRemovedRule()
+
+ assert isinstance(rule.description, str)
+ assert isinstance(rule.title, str)
+
+ msg = (
+ "The Mesos Executor has been deprecated as it was not widely used and not maintained."
+ "Please migrate to any of the supported executors."
+ "See https://airflow.apache.org/docs/stable/executor/index.html for more details."
+ )
+
+ response = rule.check()
+ assert response == msg
+
+ @conf_vars({("core", "executor"): "SequentialExecutor"})
+ def test_check(self):
+ rule = MesosExecutorRemovedRule()
+
+ assert isinstance(rule.description, str)
+ assert isinstance(rule.title, str)
+
+ response = rule.check()
+ assert response is None
[airflow] 03/07: Add readme for upgrade-check "subpackage". (#12506)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 27e79811a7cd62e495711d6d42fc2533f22857b0
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Nov 20 15:21:05 2020 +0000
Add readme for upgrade-check "subpackage". (#12506)
The intent here is this content will be visible on the pypi page for
apache-airflow-upgrade-check (once it is published.)
This is far from perfect, but the _first_ step is the hardest :)
(cherry picked from commit 47fc5c474684c9f59924837a7661e44af106943a)
---
airflow/upgrade/README.md | 82 +++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 82 insertions(+)
diff --git a/airflow/upgrade/README.md b/airflow/upgrade/README.md
new file mode 100644
index 0000000..e5e201d
--- /dev/null
+++ b/airflow/upgrade/README.md
@@ -0,0 +1,82 @@
+<!--
+ 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.
+-->
+
+# Apache Airflow Upgrade Check
+
+[](https://badge.fury.io/py/apache-airflow-upgrade-check)
+[](http://www.apache.org/licenses/LICENSE-2.0.txt)
+[](https://pypi.org/project/apache-airflow-upgrade-check/)
+[](https://pypi.org/project/apache-airflow-upgrade-check/)
+[](https://twitter.com/ApacheAirflow)
+[](https://s.apache.org/airflow-slack)
+
+This package aims to easy the upgrade journey from [Apache Airflow](https://airflow.apache.org/) 1.10 to 2.0.
+
+While we have put a lot of effort in to making this upgrade as painless as possible, with many changes
+providing upgrade path (where the old code continues to work and prints out a deprecation warning) there were
+unfortunately some breaking changes where we couldn't provide a compatibility shim.
+
+The recommended upgrade path to get to Airflow 2.0.0 is to first upgrade to the latest release in the 1.10
+series (at the time of writing: 1.10.13) and to then run this script.
+
+```bash
+pip install apache-airflow-upgrade-check
+airflow upgrade_check
+```
+
+This will then print out a number of action items that you should follow before upgrading to 2.0.0 or above.
+
+The exit code of the command will be 0 (success) if no problems are reported, or 1 otherwise.
+
+For example:
+
+```
+============================================= STATUS =============================================
+
+Legacy UI is deprecated by default......................................................SUCCESS
+Users must set a kubernetes.pod_template_file value.....................................FAIL
+Changes in import paths of hooks, operators, sensors and others.........................FAIL
+Remove airflow.AirflowMacroPlugin class.................................................SUCCESS
+Jinja Template Variables cannot be undefined............................................SUCCESS
+Fernet is enabled by default............................................................FAIL
+Logging configuration has been moved to new section.....................................SUCCESS
+Connection.conn_id is not unique........................................................SUCCESS
+GCP service account key deprecation.....................................................SUCCESS
+Users must delete deprecated configs for KubernetesExecutor.............................FAIL
+Changes in import path of remote task handlers..........................................SUCCESS
+Chain between DAG and operator not allowed..............................................SUCCESS
+SendGrid email uses old airflow.contrib module..........................................SUCCESS
+Connection.conn_type is not nullable....................................................SUCCESS
+Found 16 problems.
+
+======================================== RECOMMENDATIONS =========================================
+
+Users must set a kubernetes.pod_template_file value
+---------------------------------------------------
+In Airflow 2.0, KubernetesExecutor Users need to set a pod_template_file as a base
+value for all pods launched by the KubernetesExecutor
+
+
+Problems:
+
+ 1. Please create a pod_template_file by running `airflow generate_pod_template`.
+This will generate a pod using your aiflow.cfg settings
+
+...
+```