You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ep...@apache.org on 2023/07/06 22:27:13 UTC
[airflow] branch v2-6-test updated (b012eae95d -> ccee9c5ec1)
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a change to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
omit b012eae95d Update RELEASE_NOTES.rst
omit 7a0043e201 Update version to 2.6.3
omit 75934b1ddb Fix failing main due to new release of jsonschema (#32405)
omit 3cb9d1cfd0 Fix triggerers alive check and add a new conf for triggerer heartbeat rate (#32123)
omit 6396916301 Less precise check for error message in trigger initialization (#32035)
omit 31f14d34a2 Catch the exception that triggerer initialization failed (#31999)
omit f638d9df4e Add information on ARM MySQL support in Docker images (#32401)
omit 0f3b9793af Fix ts/eslint warnings (#32321)
omit 8016e686e0 Hide sensitive values from extra in connection edit form (#32309)
omit 55b0032e7f Sanitize `DagRun.run_id` and allow flexibility (#32293)
omit 0d79275794 feat(jobs/triggerer_job_runner): add triggerer canceled log (#31757)
omit b5520fbb64 Update pipeline.rst (#32369)
omit aefdff6449 Fix try_number shown in the `task?task_id=&dag_id=&execution_date=` (#32361)
omit 0e4ee0092b Retry transactions on occasional deadlocks for rendered fields (#32341)
omit 81d1060342 Update index.rst doc to correct grammar (#32315)
omit 72668572bc Update python.rst with multi-Scheduler LocalExecutor info (#32310)
omit 8053bddeef Clarify Listener API behavior (#32269)
omit f6a5ca2233 Add information for users who ask for requirements (#32262)
omit f41c102e6d Fix behaviour of LazyDictWithCache wheni import fails (#32248)
omit e8163fd639 Add links to DAGRun / DAG / Task in templates-ref.rst (#32245)
omit 3fa9b6468b Add comment to warn off a potential wrong fix (#32230)
omit bbf671a642 Remove executor_class from Job - fixing backfil for custom executors (#32219)
omit aa4f83930c fix bugged singleton implementation (#32218)
omit 04e93f5351 Add libeos_dev on ARM runtime PROD image (#32200)
omit 0577d001f6 Documentation Changes (#32185)
omit 0815954685 Use mapIndex to display extra links per mapped task. (#32154)
omit d032a16274 Update callbacks.rst (#32148)
omit 22dcdfe80b docs(authoring-and-scheduling/deferring): add a note that we'll need to restart triggerer to reflect any trigger change (#32140)
omit d8bc96ccf8 Adding missing hyperlink to the tutorial documentation (#32105)
omit 2600dc4c44 Ensure that main triggerer thread exits if the async thread fails (#32092)
omit 8a5419f7ea Update comment for editable installs wrt setuptools (#32090)
omit 8da9f281e6 Use re2 for matching untrusted regex (#32060)
omit 9010dc67a8 Render list items in rendered fields view (#32042)
omit e257bbde75 Fix hashing of dag_dependencies in serialized dag (#32037)
omit 11f0a9c060 Return None if an XComArg fails to resolve in a multiple_outputs Task (#32027)
omit 52ba6abef0 Check for DAG ID in query param from url as well as kwargs (#32014)
omit 059b017a36 Flash an error msg instead of failure in `rendered-templates` when map index is not found (#32011)
omit 3815a48b97 Fix ExternalTaskSensor when there is not task group TIs for the current execution date (#32009)
omit 62c5ef11e1 Fix number param html type in trigger template (#31980)
omit 9174653ad7 Skip python version check in some pre-commit when loading setup.py (#31965)
omit 4382e99e20 Fix masking nested variable fields (#31964)
omit d9eb020f88 Fixed failing main on too fast merged change (#31967)
new 497509e9b7 Fix masking nested variable fields (#31964)
new 352cfa8df1 Skip python version check in some pre-commit when loading setup.py (#31965)
new e991f6689d Fix number param html type in trigger template (#31980)
new c000015083 Fix ExternalTaskSensor when there is not task group TIs for the current execution date (#32009)
new ed2d3acedd Flash an error msg instead of failure in `rendered-templates` when map index is not found (#32011)
new c78e16588e Check for DAG ID in query param from url as well as kwargs (#32014)
new 23d117049a Return None if an XComArg fails to resolve in a multiple_outputs Task (#32027)
new 2813c41c1b Fix hashing of dag_dependencies in serialized dag (#32037)
new e0e8d71b0e Render list items in rendered fields view (#32042)
new fd7f63f196 Use re2 for matching untrusted regex (#32060)
new 3e8a2174cd Update comment for editable installs wrt setuptools (#32090)
new d6cc9e4bb1 Ensure that main triggerer thread exits if the async thread fails (#32092)
new a7a676eb08 Adding missing hyperlink to the tutorial documentation (#32105)
new 1f64d3c426 docs(authoring-and-scheduling/deferring): add a note that we'll need to restart triggerer to reflect any trigger change (#32140)
new 050b3856dc Update callbacks.rst (#32148)
new 028af0cafc Use mapIndex to display extra links per mapped task. (#32154)
new 3ed537c934 Documentation Changes (#32185)
new 600a9254b4 Add libeos_dev on ARM runtime PROD image (#32200)
new 7d0d67198e fix bugged singleton implementation (#32218)
new 14480751dd Remove executor_class from Job - fixing backfil for custom executors (#32219)
new 1a24e08d1f Add comment to warn off a potential wrong fix (#32230)
new 9c7e2897b7 Add links to DAGRun / DAG / Task in templates-ref.rst (#32245)
new 38df98bac8 Fix behaviour of LazyDictWithCache wheni import fails (#32248)
new 555d73c0a1 Add information for users who ask for requirements (#32262)
new 3e4c5e840a Clarify Listener API behavior (#32269)
new 288fb12e83 Update python.rst with multi-Scheduler LocalExecutor info (#32310)
new 9d7a7e7558 Update index.rst doc to correct grammar (#32315)
new bc7516b565 Retry transactions on occasional deadlocks for rendered fields (#32341)
new fb30cbdb94 Fix try_number shown in the `task?task_id=&dag_id=&execution_date=` (#32361)
new cd210181ee Update pipeline.rst (#32369)
new 3b51035b5b feat(jobs/triggerer_job_runner): add triggerer canceled log (#31757)
new 8ff7dfbd9e Sanitize `DagRun.run_id` and allow flexibility (#32293)
new 191daab7a4 Hide sensitive values from extra in connection edit form (#32309)
new 04a3a0ab27 Fix ts/eslint warnings (#32321)
new 43720fb18a Add information on ARM MySQL support in Docker images (#32401)
new f10204b59f Catch the exception that triggerer initialization failed (#31999)
new c92574f471 Less precise check for error message in trigger initialization (#32035)
new cf31c6f136 Fix triggerers alive check and add a new conf for triggerer heartbeat rate (#32123)
new fc61875ca3 Fix failing main due to new release of jsonschema (#32405)
new 9ae7f0a0fe Update version to 2.6.3
new 9cddb22b64 Update RELEASE_NOTES.rst
new ccee9c5ec1 Update trigger url to fix test
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b012eae95d)
\
N -- N -- N refs/heads/v2-6-test (ccee9c5ec1)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 42 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:
tests/jobs/test_scheduler_job.py | 2 +-
tests/www/views/test_views_trigger_dag.py | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
[airflow] 14/42: docs(authoring-and-scheduling/deferring): add a note that we'll need to restart triggerer to reflect any trigger change (#32140)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1f64d3c42644efa649a606aa99daf39be1e3c2c1
Author: Wei Lee <we...@gmail.com>
AuthorDate: Mon Jun 26 22:19:57 2023 +0800
docs(authoring-and-scheduling/deferring): add a note that we'll need to restart triggerer to reflect any trigger change (#32140)
(cherry picked from commit 51dbbaf40b90aa676cc706fd8c40c12cac66b9cc)
---
docs/apache-airflow/authoring-and-scheduling/deferring.rst | 1 +
1 file changed, 1 insertion(+)
diff --git a/docs/apache-airflow/authoring-and-scheduling/deferring.rst b/docs/apache-airflow/authoring-and-scheduling/deferring.rst
index 4755566670..b1e6c6be98 100644
--- a/docs/apache-airflow/authoring-and-scheduling/deferring.rst
+++ b/docs/apache-airflow/authoring-and-scheduling/deferring.rst
@@ -55,6 +55,7 @@ Writing a deferrable operator takes a bit more work. There are some main points
* Your Operator will be stopped and removed from its worker while deferred, and no state will persist automatically. You can persist state by asking Airflow to resume you at a certain method or pass certain kwargs, but that's it.
* You can defer multiple times, and you can defer before/after your Operator does significant work, or only defer if certain conditions are met (e.g. a system does not have an immediate answer). Deferral is entirely under your control.
* Any Operator can defer; no special marking on its class is needed, and it's not limited to Sensors.
+* In order for any changes to a Trigger to be reflected, the *triggerer* needs to be restarted whenever the Trigger is modified.
Triggering Deferral
[airflow] 02/42: Skip python version check in some pre-commit when loading setup.py (#31965)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 352cfa8df10f3eac519946af61ddd40f9a636121
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Fri Jun 16 19:24:52 2023 +0200
Skip python version check in some pre-commit when loading setup.py (#31965)
Some providers are excluded for some versions of Python (for example
the `apache.hive` provider is excluded on Python 3.11. However
this might make some pre-commits confused when running on Python 3.11
because the list of generated extras will be different for them
than those on other Python versions.
We do not want to set python used in pre-commits to, say, python3.8,
because that would mean that would require our users to have python3.8
as binary available on their system, which is not always setup. Python3
on the other hand is going to be universally available.
In order to mitigate the problem we will run the precommits with
_SKIP_PYTHON_VERSION_CHECK variable set before import of setup is done,
so that the python version check is skipped and all python versions
will produce the same set of extras.
(cherry picked from commit 578cbb24fbd7e4bbeef2e61078c5d333aeda0448)
---
scripts/ci/pre_commit/pre_commit_check_setup_extra_packages_ref.py | 2 ++
scripts/ci/pre_commit/pre_commit_insert_extras.py | 3 +++
setup.py | 6 +++++-
3 files changed, 10 insertions(+), 1 deletion(-)
diff --git a/scripts/ci/pre_commit/pre_commit_check_setup_extra_packages_ref.py b/scripts/ci/pre_commit/pre_commit_check_setup_extra_packages_ref.py
index a384c958de..a523ac62c0 100755
--- a/scripts/ci/pre_commit/pre_commit_check_setup_extra_packages_ref.py
+++ b/scripts/ci/pre_commit/pre_commit_check_setup_extra_packages_ref.py
@@ -37,6 +37,8 @@ PY_IDENTIFIER = r"[a-zA-Z_][a-zA-Z0-9_\.]*"
sys.path.insert(0, AIRFLOW_SOURCES_DIR)
+os.environ["_SKIP_PYTHON_VERSION_CHECK"] = "true"
+
from setup import ( # noqa # isort:skip
add_all_provider_packages,
EXTRAS_DEPRECATED_ALIASES,
diff --git a/scripts/ci/pre_commit/pre_commit_insert_extras.py b/scripts/ci/pre_commit/pre_commit_insert_extras.py
index fac926f611..2750e9dbae 100755
--- a/scripts/ci/pre_commit/pre_commit_insert_extras.py
+++ b/scripts/ci/pre_commit/pre_commit_insert_extras.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import os
import sys
from pathlib import Path
from textwrap import wrap
@@ -27,6 +28,8 @@ sys.path.insert(0, str(Path(__file__).parent.resolve())) # make sure common_pre
sys.path.insert(0, str(AIRFLOW_SOURCES_DIR)) # make sure setup is imported from Airflow
# flake8: noqa: F401
+os.environ["_SKIP_PYTHON_VERSION_CHECK"] = "true"
+
from common_precommit_utils import insert_documentation # isort: skip
from setup import EXTRAS_DEPENDENCIES # isort:skip
diff --git a/setup.py b/setup.py
index 77506af9e3..cc8ef0a14f 100644
--- a/setup.py
+++ b/setup.py
@@ -66,13 +66,17 @@ CURRENT_PYTHON_VERSION = f"{sys.version_info.major}.{sys.version_info.minor}"
# corresponding provider.yaml file.
#
def fill_provider_dependencies() -> dict[str, dict[str, list[str]]]:
+ # in case we are loading setup from pre-commits, we want to skip the check for python version
+ # because if someone uses a version of Python where providers are excluded, the setup will fail
+ # to see the extras for those providers
+ skip_python_version_check = os.environ.get("_SKIP_PYTHON_VERSION_CHECK")
try:
with AIRFLOW_SOURCES_ROOT.joinpath("generated", "provider_dependencies.json").open() as f:
dependencies = json.load(f)
return {
key: value
for key, value in dependencies.items()
- if CURRENT_PYTHON_VERSION not in value["excluded-python-versions"]
+ if CURRENT_PYTHON_VERSION not in value["excluded-python-versions"] or skip_python_version_check
}
except Exception as e:
print(f"Exception while loading provider dependencies {e}")
[airflow] 31/42: feat(jobs/triggerer_job_runner): add triggerer canceled log (#31757)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3b51035b5b4a7823da043637e9db3362e6529658
Author: Wei Lee <we...@gmail.com>
AuthorDate: Fri Jun 16 16:31:51 2023 +0800
feat(jobs/triggerer_job_runner): add triggerer canceled log (#31757)
Emit log message when trigger is cancelled
Co-authored-by: Daniel Standish <15...@users.noreply.github.com>
Co-authored-by: Jed Cunningham <66...@users.noreply.github.com>
(cherry picked from commit a60429eadfffb5fb0f867c220a6cecf628692dcf)
---
airflow/jobs/triggerer_job_runner.py | 9 +++++++++
tests/jobs/test_triggerer_job.py | 29 ++++++++++++++++++++++++++++-
2 files changed, 37 insertions(+), 1 deletion(-)
diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py
index 7051dec5a0..7ca3fc2e94 100644
--- a/airflow/jobs/triggerer_job_runner.py
+++ b/airflow/jobs/triggerer_job_runner.py
@@ -40,6 +40,7 @@ from airflow.serialization.pydantic.job import JobPydantic
from airflow.stats import Stats
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.typing_compat import TypedDict
+from airflow.utils import timezone
from airflow.utils.log.file_task_handler import FileTaskHandler
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.log.trigger_handler import (
@@ -615,6 +616,14 @@ class TriggerRunner(threading.Thread, LoggingMixin):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ timeout = trigger.task_instance.trigger_timeout
+ if timeout:
+ timeout = timeout.replace(tzinfo=timezone.utc) if not timeout.tzinfo else timeout
+ if timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
+ self.log.error("Trigger cancelled; message=%s", err)
+ raise
finally:
# CancelledError will get injected when we're stopped - which is
# fine, the cleanup process will understand that, but we want to
diff --git a/tests/jobs/test_triggerer_job.py b/tests/jobs/test_triggerer_job.py
index dd35dd4cc8..d0ce887a57 100644
--- a/tests/jobs/test_triggerer_job.py
+++ b/tests/jobs/test_triggerer_job.py
@@ -22,7 +22,7 @@ import datetime
import importlib
import time
from threading import Thread
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
import pendulum
import pytest
@@ -255,6 +255,33 @@ def test_trigger_lifecycle(session):
job_runner.trigger_runner.stop = True
+class TestTriggerRunner:
+ @pytest.mark.asyncio
+ @patch("airflow.jobs.triggerer_job_runner.TriggerRunner.set_individual_trigger_logging")
+ async def test_run_trigger_canceled(self, session) -> None:
+ trigger_runner = TriggerRunner()
+ trigger_runner.triggers = {1: {"task": MagicMock(), "name": "mock_name", "events": 0}}
+ mock_trigger = MagicMock()
+ mock_trigger.task_instance.trigger_timeout = None
+ mock_trigger.run.side_effect = asyncio.CancelledError()
+
+ with pytest.raises(asyncio.CancelledError):
+ await trigger_runner.run_trigger(1, mock_trigger)
+
+ @pytest.mark.asyncio
+ @patch("airflow.jobs.triggerer_job_runner.TriggerRunner.set_individual_trigger_logging")
+ async def test_run_trigger_timeout(self, session, caplog) -> None:
+ trigger_runner = TriggerRunner()
+ trigger_runner.triggers = {1: {"task": MagicMock(), "name": "mock_name", "events": 0}}
+ mock_trigger = MagicMock()
+ mock_trigger.task_instance.trigger_timeout = timezone.utcnow() - datetime.timedelta(hours=1)
+ mock_trigger.run.side_effect = asyncio.CancelledError()
+
+ with pytest.raises(asyncio.CancelledError):
+ await trigger_runner.run_trigger(1, mock_trigger)
+ assert "Trigger cancelled due to timeout" in caplog.text
+
+
def test_trigger_create_race_condition_18392(session, tmp_path):
"""
This verifies the resolution of race condition documented in github issue #18392.
[airflow] 21/42: Add comment to warn off a potential wrong fix (#32230)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1a24e08d1fbd0d66b1683654f8295b8b132694ab
Author: Tzu-ping Chung <ur...@gmail.com>
AuthorDate: Thu Jun 29 06:09:26 2023 +0800
Add comment to warn off a potential wrong fix (#32230)
(cherry picked from commit 73dd48083d1102b81b76ff6f5668cc41261c362e)
---
airflow/models/dag.py | 10 +++++++++-
1 file changed, 9 insertions(+), 1 deletion(-)
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 331feae28e..479c70524b 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -869,7 +869,9 @@ class DAG(LoggingMixin):
this method only considers ``schedule_interval`` values valid prior to
Airflow 2.2.
- DO NOT use this method is there is a known data interval.
+ DO NOT call this method if there is a known data interval.
+
+ :meta private:
"""
timetable_type = type(self.timetable)
if issubclass(timetable_type, (NullTimetable, OnceTimetable, DatasetTriggeredTimetable)):
@@ -879,6 +881,12 @@ class DAG(LoggingMixin):
end = cast(CronDataIntervalTimetable, self.timetable)._get_next(start)
elif issubclass(timetable_type, DeltaDataIntervalTimetable):
end = cast(DeltaDataIntervalTimetable, self.timetable)._get_next(start)
+ # Contributors: When the exception below is raised, you might want to
+ # add an 'elif' block here to handle custom timetables. Stop! The bug
+ # you're looking for is instead at when the DAG run (represented by
+ # logical_date) was created. See GH-31969 for an example:
+ # * Wrong fix: GH-32074 (modifies this function).
+ # * Correct fix: GH-32118 (modifies the DAG run creation code).
else:
raise ValueError(f"Not a valid timetable: {self.timetable!r}")
return DataInterval(start, end)
[airflow] 28/42: Retry transactions on occasional deadlocks for rendered fields (#32341)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit bc7516b565531a1b333e6b3984cfc0e527083da2
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Wed Jul 5 13:28:16 2023 +0200
Retry transactions on occasional deadlocks for rendered fields (#32341)
This is a follow-up on #18616 where we introduced retries on the
occassional deadlocks when rendered task fields have been deleted
by parallel threads (this is not a real deadlock, it's because
MySQL locks too many things when queries are executed and will
deadlock when one of those queries wait too much).
Adding retry - while not perfect - should allow to handle the
problem and significantly decrease the likelihood of such
deadlocks.
We can probably think about different approach for rendered fields,
but for now retrying is - I think - acceptable short-term fix.
Fixes: #32294
Fixes: #29687
(cherry picked from commit c8a3c112a7bae345d37bb8b90d68c8d6ff2ef8fc)
---
airflow/models/renderedtifields.py | 1 +
1 file changed, 1 insertion(+)
diff --git a/airflow/models/renderedtifields.py b/airflow/models/renderedtifields.py
index 7fd2c29edf..4620b38f5c 100644
--- a/airflow/models/renderedtifields.py
+++ b/airflow/models/renderedtifields.py
@@ -175,6 +175,7 @@ class RenderedTaskInstanceFields(Base):
return result.k8s_pod_yaml if result else None
@provide_session
+ @retry_db_transaction
def write(self, session: Session = None):
"""Write instance to database
[airflow] 39/42: Fix failing main due to new release of jsonschema (#32405)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit fc61875ca3e2708997767be55324b2b63f2875cb
Author: eladkal <45...@users.noreply.github.com>
AuthorDate: Thu Jul 6 17:36:58 2023 +0300
Fix failing main due to new release of jsonschema (#32405)
(cherry picked from commit 89d5e8b456c3f8e2cbe2691a4e1fa52ef2c1b6c5)
---
scripts/in_container/verify_providers.py | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/scripts/in_container/verify_providers.py b/scripts/in_container/verify_providers.py
index bc56a5fa93..78cd4c8118 100755
--- a/scripts/in_container/verify_providers.py
+++ b/scripts/in_container/verify_providers.py
@@ -164,6 +164,10 @@ KNOWN_COMMON_DEPRECATED_MESSAGES: set[str] = {
"urllib3 (1.26.9) or chardet (5.1.0)/charset_normalizer (2.0.12) doesn't match a supported version!",
"urllib3 (1.26.12) or chardet (5.1.0)/charset_normalizer (2.0.12) doesn't match a supported version!",
"Jupyter is migrating its paths to use standard platformdirs given by the platformdirs library.",
+ (
+ "Importing ErrorTree directly from the jsonschema package is deprecated and will become"
+ " an ImportError. Import it from jsonschema.exceptions instead."
+ ),
}
# The set of warning messages generated by direct importing of some deprecated modules. We should only
[airflow] 01/42: Fix masking nested variable fields (#31964)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 497509e9b722c5089db4c3cd118b8ada79f1d240
Author: Hussein Awala <hu...@awala.fr>
AuthorDate: Fri Jun 16 21:05:01 2023 +0200
Fix masking nested variable fields (#31964)
* Fix masking nested variable fields
Signed-off-by: Hussein Awala <hu...@awala.fr>
* add a unit test
Signed-off-by: Hussein Awala <hu...@awala.fr>
---------
Signed-off-by: Hussein Awala <hu...@awala.fr>
(cherry picked from commit e22ce9baed19ddf771db59b7da1d25e240430625)
---
airflow/models/variable.py | 2 +-
tests/models/test_variable.py | 35 +++++++++++++++++++++++++++++++++++
2 files changed, 36 insertions(+), 1 deletion(-)
diff --git a/airflow/models/variable.py b/airflow/models/variable.py
index 5e5a42887c..9b69e28f1e 100644
--- a/airflow/models/variable.py
+++ b/airflow/models/variable.py
@@ -143,7 +143,7 @@ class Variable(Base, LoggingMixin):
else:
if deserialize_json:
obj = json.loads(var_val)
- mask_secret(var_val, key)
+ mask_secret(obj, key)
return obj
else:
mask_secret(var_val, key)
diff --git a/tests/models/test_variable.py b/tests/models/test_variable.py
index b67027a905..67e267b8ca 100644
--- a/tests/models/test_variable.py
+++ b/tests/models/test_variable.py
@@ -258,3 +258,38 @@ class TestVariable:
]
finally:
session.rollback()
+
+
+@pytest.mark.parametrize(
+ "variable_value, deserialize_json, expected_masked_values",
+ [
+ ("s3cr3t", False, ["s3cr3t"]),
+ ('{"api_key": "s3cr3t"}', True, ["s3cr3t"]),
+ ('{"api_key": "s3cr3t", "normal_key": "normal_value"}', True, ["s3cr3t"]),
+ ('{"api_key": "s3cr3t", "another_secret": "123456"}', True, ["s3cr3t", "123456"]),
+ ],
+)
+def test_masking_only_secret_values(variable_value, deserialize_json, expected_masked_values):
+ from airflow.utils.log.secrets_masker import _secrets_masker
+
+ session = settings.Session()
+
+ try:
+ var = Variable(
+ key=f"password-{os.getpid()}",
+ val=variable_value,
+ )
+ session.add(var)
+ session.flush()
+
+ # Make sure we re-load it, not just get the cached object back
+ session.expunge(var)
+ _secrets_masker().patterns = set()
+
+ Variable.get(var.key, deserialize_json=deserialize_json)
+
+ for expected_masked_value in expected_masked_values:
+ assert expected_masked_value in _secrets_masker().patterns
+ finally:
+ session.rollback()
+ db.clear_db_variables()
[airflow] 03/42: Fix number param html type in trigger template (#31980)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e991f6689d9af2baf0da0b8884b622b5a86ae092
Author: Hussein Awala <hu...@awala.fr>
AuthorDate: Mon Jun 19 11:31:52 2023 +0200
Fix number param html type in trigger template (#31980)
* Keep valuetype="number" for both param number and integer types and change only type value
Signed-off-by: Hussein Awala <hu...@awala.fr>
* Update airflow/www/templates/airflow/trigger.html
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
---------
Signed-off-by: Hussein Awala <hu...@awala.fr>
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
(cherry picked from commit 232771869030d708c57f840aea735b18bd4bffb2)
---
airflow/www/templates/airflow/trigger.html | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/airflow/www/templates/airflow/trigger.html b/airflow/www/templates/airflow/trigger.html
index 4c03e243f4..08f17176dc 100644
--- a/airflow/www/templates/airflow/trigger.html
+++ b/airflow/www/templates/airflow/trigger.html
@@ -72,7 +72,7 @@
{% elif "enum" in form_details.schema and form_details.schema.enum %}
<select class="my_select2 form-control" name="element_{{ form_key }}" id="element_{{ form_key }}" data-placeholder="Select Value"
onchange="updateJSONconf();"
- {%- if "integer" in form_details.schema.type %} valuetype="number" {% elif "number" in form_details.schema.type %} valuetype="decimal" {%- endif %}
+ {%- if "integer" in form_details.schema.type or "number" in form_details.schema.type %} valuetype="number"{% endif %}
{%- if not "null" in form_details.schema.type %} required=""{% endif %}>
{% for option in form_details.schema.enum -%}
<option{% if form_details.value == option %} selected="true"{% endif %}>{{ option }}</option>
@@ -94,7 +94,7 @@
</textarea>
{% elif form_details.schema and ("integer" in form_details.schema.type or "number" in form_details.schema.type) %}
<input class="form-control" name="element_{{ form_key }}" id="element_{{ form_key }}"
- {% if "integer" in form_details.schema.type %} valuetype="number" type="number" {%else%} valuetype="decimal" type="decimal" {% endif %}
+ valuetype="number" {% if "integer" in form_details.schema.type %} type="number" {% else %} type="decimal" {% endif %}
value="{% if form_details.value %}{{ form_details.value }}{% endif %}"
{%- if form_details.schema.minimum %} min="{{ form_details.schema.minimum }}"{% endif %}
{%- if form_details.schema.maximum %} max="{{ form_details.schema.maximum }}"{% endif %}
[airflow] 07/42: Return None if an XComArg fails to resolve in a multiple_outputs Task (#32027)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 23d117049a1da3e19f4ff21775702fbedd0a331c
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Jun 21 10:55:57 2023 +0100
Return None if an XComArg fails to resolve in a multiple_outputs Task (#32027)
* Return None if an XComArg fails to resolve in a multiple_outputs Task
Tasks with multiple_outputs set to True returns XComs with different keys
which are not known to Airflow. Because they have multiple_outputs set, we
should return None if we can't find the XCom, just like we return None when
the key is equal to XCOM_RETURN_KEY known to Airflow.
Closes: https://github.com/apache/airflow/issues/29199
* Apply suggestions from code review
(cherry picked from commit 79eac7687cf7c6bcaa4df2b8735efaad79a7fee2)
---
airflow/models/xcom_arg.py | 7 ++++++
tests/decorators/test_python.py | 47 +++++++++++++++++++++++++++++++++++++++--
2 files changed, 52 insertions(+), 2 deletions(-)
diff --git a/airflow/models/xcom_arg.py b/airflow/models/xcom_arg.py
index d8b42ba819..ab9a46641b 100644
--- a/airflow/models/xcom_arg.py
+++ b/airflow/models/xcom_arg.py
@@ -370,6 +370,13 @@ class PlainXComArg(XComArg):
return result
if self.key == XCOM_RETURN_KEY:
return None
+ if getattr(self.operator, "multiple_outputs", False):
+ # If the operator is set to have multiple outputs and it was not executed,
+ # we should return "None" instead of showing an error. This is because when
+ # multiple outputs XComs are created, the XCom keys associated with them will have
+ # different names than the predefined "XCOM_RETURN_KEY" and won't be found.
+ # Therefore, it's better to return "None" like we did above where self.key==XCOM_RETURN_KEY.
+ return None
raise XComNotFound(ti.dag_id, task_id, self.key)
diff --git a/tests/decorators/test_python.py b/tests/decorators/test_python.py
index 435c56de2e..7eebe1cf5f 100644
--- a/tests/decorators/test_python.py
+++ b/tests/decorators/test_python.py
@@ -18,14 +18,14 @@
import sys
from collections import namedtuple
from datetime import date, timedelta
-from typing import TYPE_CHECKING, Dict, Tuple
+from typing import TYPE_CHECKING, Dict, Tuple, Union
import pytest
from airflow import PY38, PY311
from airflow.decorators import task as task_decorator
from airflow.decorators.base import DecoratedMappedOperator
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, XComNotFound
from airflow.models import DAG
from airflow.models.baseoperator import BaseOperator
from airflow.models.expandinput import DictOfListsExpandInput
@@ -803,6 +803,49 @@ def test_upstream_exception_produces_none_xcom(dag_maker, session):
assert result == "'example' None"
+@pytest.mark.parametrize("multiple_outputs", [True, False])
+def test_multiple_outputs_produces_none_xcom_when_task_is_skipped(dag_maker, session, multiple_outputs):
+ from airflow.exceptions import AirflowSkipException
+ from airflow.utils.trigger_rule import TriggerRule
+
+ result = None
+
+ with dag_maker(session=session) as dag:
+
+ @dag.task()
+ def up1() -> str:
+ return "example"
+
+ @dag.task(multiple_outputs=multiple_outputs)
+ def up2(x) -> Union[dict, None]:
+ if x == 2:
+ return {"x": "example"}
+ raise AirflowSkipException()
+
+ @dag.task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
+ def down(a, b):
+ nonlocal result
+ result = f"{a!r} {b!r}"
+
+ down(up1(), up2(1)["x"])
+
+ dr = dag_maker.create_dagrun()
+
+ decision = dr.task_instance_scheduling_decisions(session=session)
+ assert len(decision.schedulable_tis) == 2 # "up1" and "up2"
+ for ti in decision.schedulable_tis:
+ ti.run(session=session)
+
+ decision = dr.task_instance_scheduling_decisions(session=session)
+ assert len(decision.schedulable_tis) == 1 # "down"
+ if multiple_outputs:
+ decision.schedulable_tis[0].run(session=session)
+ assert result == "'example' None"
+ else:
+ with pytest.raises(XComNotFound):
+ decision.schedulable_tis[0].run(session=session)
+
+
@pytest.mark.filterwarnings("error")
def test_no_warnings(reset_logging_config, caplog):
@task_decorator
[airflow] 27/42: Update index.rst doc to correct grammar (#32315)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9d7a7e75585fc93102843265e4ca6322b2602b24
Author: Pankaj Koti <pa...@gmail.com>
AuthorDate: Wed Jul 5 13:10:32 2023 +0530
Update index.rst doc to correct grammar (#32315)
* Update index.rst doc to correct grammar
* Address @uranusjr's comments
(cherry picked from commit 43f3e57bf162293b92154f16a8ce33e6922fbf4e)
---
docs/apache-airflow/index.rst | 16 ++++++++++++----
1 file changed, 12 insertions(+), 4 deletions(-)
diff --git a/docs/apache-airflow/index.rst b/docs/apache-airflow/index.rst
index feaf2d388b..adc5f47eff 100644
--- a/docs/apache-airflow/index.rst
+++ b/docs/apache-airflow/index.rst
@@ -98,14 +98,22 @@ Rich scheduling and execution semantics enable you to easily define complex pipe
intervals. Backfilling allows you to (re-)run pipelines on historical data after making changes to your logic.
And the ability to rerun partial pipelines after resolving an error helps maximize efficiency.
-Airflow's user interface provides both in-depth views of pipelines and individual tasks, and an overview of
-pipelines over time. From the interface, you can inspect logs and manage tasks, for example retrying a task in
+Airflow's user interface provides:
+
+ 1. In-depth views of two things:
+
+ i. Pipelines
+ ii. Tasks
+
+ 2. Overview of your pipelines over time
+
+From the interface, you can inspect logs and manage tasks, for example retrying a task in
case of failure.
The open-source nature of Airflow ensures you work on components developed, tested, and used by many other
`companies <https://github.com/apache/airflow/blob/main/INTHEWILD.md>`_ around the world. In the active
`community <https://airflow.apache.org/community>`_ you can find plenty of helpful resources in the form of
-blogs posts, articles, conferences, books, and more. You can connect with other peers via several channels
+blog posts, articles, conferences, books, and more. You can connect with other peers via several channels
such as `Slack <https://s.apache.org/airflow-slack>`_ and mailing lists.
Airflow as a Platform is highly customizable. By utilizing :doc:`public-airflow-interface` you can extend
@@ -115,7 +123,7 @@ Why not Airflow™?
=================
Airflow™ was built for finite batch workflows. While the CLI and REST API do allow triggering workflows,
-Airflow was not built for infinitely-running event-based workflows. Airflow is not a streaming solution.
+Airflow was not built for infinitely running event-based workflows. Airflow is not a streaming solution.
However, a streaming system such as Apache Kafka is often seen working together with Apache Airflow. Kafka can
be used for ingestion and processing in real-time, event data is written to a storage location, and Airflow
periodically starts a workflow processing a batch of data.
[airflow] 18/42: Add libeos_dev on ARM runtime PROD image (#32200)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 600a9254b403dded433eff6034f51eee088efd01
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Tue Jun 27 22:11:46 2023 +0200
Add libeos_dev on ARM runtime PROD image (#32200)
This is a follow-up after #31437. Seems that for ARM image, the
BigQuery libraries require libgeos-dev library installed on
system level not only to be built but also to be imported.
(cherry picked from commit 771362af4784f3d913d6c3d3b44c78269280a96e)
---
Dockerfile | 2 +-
Dockerfile.ci | 2 +-
docs/docker-stack/changelog.rst | 7 ++++++-
scripts/docker/install_os_dependencies.sh | 2 +-
4 files changed, 9 insertions(+), 4 deletions(-)
diff --git a/Dockerfile b/Dockerfile
index 77ab5e70de..600da4a587 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -105,7 +105,7 @@ software-properties-common sqlite3 sudo unixodbc unixodbc-dev"
function get_runtime_apt_deps() {
if [[ "${RUNTIME_APT_DEPS=}" == "" ]]; then
RUNTIME_APT_DEPS="apt-transport-https apt-utils ca-certificates \
-curl dumb-init freetds-bin gosu krb5-user \
+curl dumb-init freetds-bin gosu krb5-user libgeos-dev \
ldap-utils libffi7 libldap-2.4-2 libsasl2-2 libsasl2-modules libssl1.1 locales \
lsb-release netcat openssh-client python3-selinux rsync sasl2-bin sqlite3 sudo unixodbc"
export RUNTIME_APT_DEPS
diff --git a/Dockerfile.ci b/Dockerfile.ci
index 985b677b43..d89da4655b 100644
--- a/Dockerfile.ci
+++ b/Dockerfile.ci
@@ -65,7 +65,7 @@ software-properties-common sqlite3 sudo unixodbc unixodbc-dev"
function get_runtime_apt_deps() {
if [[ "${RUNTIME_APT_DEPS=}" == "" ]]; then
RUNTIME_APT_DEPS="apt-transport-https apt-utils ca-certificates \
-curl dumb-init freetds-bin gosu krb5-user \
+curl dumb-init freetds-bin gosu krb5-user libgeos-dev \
ldap-utils libffi7 libldap-2.4-2 libsasl2-2 libsasl2-modules libssl1.1 locales \
lsb-release netcat openssh-client python3-selinux rsync sasl2-bin sqlite3 sudo unixodbc"
export RUNTIME_APT_DEPS
diff --git a/docs/docker-stack/changelog.rst b/docs/docker-stack/changelog.rst
index 063748d250..b40c0ae68f 100644
--- a/docs/docker-stack/changelog.rst
+++ b/docs/docker-stack/changelog.rst
@@ -58,7 +58,12 @@ here so that users affected can find the reason for the changes.
Airflow 2.6
~~~~~~~~~~~~~
-* 2.6
+* 2.6.3
+
+ * Add ``libgeos-dev`` library to runtime PROD image as it is required by BigQuery library on ARM image
+
+
+* 2.6.0
* Snowflake provider installed by default
diff --git a/scripts/docker/install_os_dependencies.sh b/scripts/docker/install_os_dependencies.sh
index 73bb43c6cc..6aaa0ba561 100644
--- a/scripts/docker/install_os_dependencies.sh
+++ b/scripts/docker/install_os_dependencies.sh
@@ -47,7 +47,7 @@ software-properties-common sqlite3 sudo unixodbc unixodbc-dev"
function get_runtime_apt_deps() {
if [[ "${RUNTIME_APT_DEPS=}" == "" ]]; then
RUNTIME_APT_DEPS="apt-transport-https apt-utils ca-certificates \
-curl dumb-init freetds-bin gosu krb5-user \
+curl dumb-init freetds-bin gosu krb5-user libgeos-dev \
ldap-utils libffi7 libldap-2.4-2 libsasl2-2 libsasl2-modules libssl1.1 locales \
lsb-release netcat openssh-client python3-selinux rsync sasl2-bin sqlite3 sudo unixodbc"
export RUNTIME_APT_DEPS
[airflow] 10/42: Use re2 for matching untrusted regex (#32060)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit fd7f63f1963b1b5f64d6e44a052f7a62d716e667
Author: Pierre Jeambrun <pi...@gmail.com>
AuthorDate: Fri Jun 23 09:55:19 2023 +0200
Use re2 for matching untrusted regex (#32060)
(cherry picked from commit 116e607ddcb32480e57c342f48226545ac6fc315)
---
airflow/models/dag.py | 7 ++++---
setup.cfg | 1 +
2 files changed, 5 insertions(+), 3 deletions(-)
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 97c973f785..331feae28e 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -26,7 +26,6 @@ import logging
import os
import pathlib
import pickle
-import re
import sys
import traceback
import warnings
@@ -43,6 +42,7 @@ from typing import (
Iterable,
Iterator,
List,
+ Pattern,
Sequence,
Union,
cast,
@@ -52,6 +52,7 @@ from urllib.parse import urlsplit
import jinja2
import pendulum
+import re2 as re
from dateutil.relativedelta import relativedelta
from pendulum.tz.timezone import Timezone
from sqlalchemy import Boolean, Column, ForeignKey, Index, Integer, String, Text, and_, case, func, not_, or_
@@ -2169,7 +2170,7 @@ class DAG(LoggingMixin):
def partial_subset(
self,
- task_ids_or_regex: str | re.Pattern | Iterable[str],
+ task_ids_or_regex: str | Pattern | Iterable[str],
include_downstream=False,
include_upstream=True,
include_direct_upstream=False,
@@ -2196,7 +2197,7 @@ class DAG(LoggingMixin):
memo = {id(self.task_dict): None, id(self._task_group): None}
dag = copy.deepcopy(self, memo) # type: ignore
- if isinstance(task_ids_or_regex, (str, re.Pattern)):
+ if isinstance(task_ids_or_regex, (str, Pattern)):
matched_tasks = [t for t in self.tasks if re.findall(task_ids_or_regex, t.task_id)]
else:
matched_tasks = [t for t in self.tasks if t.task_id in task_ids_or_regex]
diff --git a/setup.cfg b/setup.cfg
index 4f73a4aa4a..6e09495c9b 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -100,6 +100,7 @@ install_requires =
flask-login>=0.6.2
flask-session>=0.4.0
flask-wtf>=0.15
+ google-re2>=1.0
graphviz>=0.12
gunicorn>=20.1.0
httpx
[airflow] 11/42: Update comment for editable installs wrt setuptools (#32090)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3e8a2174cdf9aab8d91830ff9c8c16830df634d4
Author: Pankaj Koti <pa...@gmail.com>
AuthorDate: Fri Jun 23 15:42:45 2023 +0530
Update comment for editable installs wrt setuptools (#32090)
(cherry picked from commit ca13c7b77ea0e7d37bfe893871bab565d26884d0)
---
pyproject.toml | 12 +++++++-----
1 file changed, 7 insertions(+), 5 deletions(-)
diff --git a/pyproject.toml b/pyproject.toml
index 0568ed62f9..5643c246ae 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -17,11 +17,13 @@
[tool.black]
line-length = 110
target-version = ['py37', 'py38', 'py39', 'py310']
-# The build system section is needed in order to workaround the side-effect introduced by recent
-# setup tools version. The recent setuptools version update (64.0.0) broke paths of editable installations
-# and we have to pin it to 63.4.3 version
-# The problem is tracked (and this limitation might be removed if it is solved) in:
-# https://github.com/pypa/setuptools/issues/3548
+
+# Editable installs are currently broken using setuptools 64.0.0 and above. The problem is tracked in
+# https://github.com/pypa/setuptools/issues/3548. We're also discussing how we could potentially fix
+# this problem on our end in issue https://github.com/apache/airflow/issues/30764. Until then we need
+# to use one of the following workarounds locally for editable installs:
+# 1) Pin setuptools <= 63.4.3 below in the [build-system] section.
+# 2) Include your airflow source code directory in PYTHONPATH.
[build-system]
requires = ['setuptools==67.2.0']
build-backend = "setuptools.build_meta"
[airflow] 32/42: Sanitize `DagRun.run_id` and allow flexibility (#32293)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 8ff7dfbd9e76aa40b04adeb231df3820606f5ba3
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Sun Jul 2 12:31:13 2023 +0100
Sanitize `DagRun.run_id` and allow flexibility (#32293)
This commit sanitizes the DagRun.run_id parameter by introducing a configurable option.
Users now have the ability to select a specific run_id pattern for their runs,
ensuring stricter control over the values used. This update does not impact the default run_id
generation performed by the scheduler for scheduled DAG runs or for Dag runs triggered without
modifying the run_id parameter in the run configuration page.
The configuration flexibility empowers users to align the run_id pattern with their specific requirements.
(cherry picked from commit 05bd90f563649f2e9c8f0c85cf5838315a665a02)
---
airflow/config_templates/config.yml | 9 +++++
airflow/config_templates/default_airflow.cfg | 5 +++
airflow/models/dag.py | 55 +++++++++++++++-------------
airflow/models/dagrun.py | 16 +++++++-
airflow/www/views.py | 32 +++++++++++-----
tests/models/test_dagrun.py | 32 ++++++++++++++++
tests/www/views/test_views_trigger_dag.py | 30 +++++++++++++++
7 files changed, 143 insertions(+), 36 deletions(-)
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index 68c2001e00..4653a82416 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -2440,6 +2440,15 @@ scheduler:
type: float
example: ~
default: "120.0"
+ allowed_run_id_pattern:
+ description: |
+ The run_id pattern used to verify the validity of user input to the run_id parameter when
+ triggering a DAG. This pattern cannot change the pattern used by scheduler to generate run_id
+ for scheduled DAG runs or DAG runs triggered without changing the run_id parameter.
+ version_added: 2.6.3
+ type: string
+ example: ~
+ default: "^[A-Za-z0-9_.~:+-]+$"
triggerer:
description: ~
options:
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index 04971050c4..c765f17ead 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -1245,6 +1245,11 @@ task_queued_timeout = 600.0
# longer than `[scheduler] task_queued_timeout`.
task_queued_timeout_check_interval = 120.0
+# The run_id pattern used to verify the validity of user input to the run_id parameter when
+# triggering a DAG. This pattern cannot change the pattern used by scheduler to generate run_id
+# for scheduled DAG runs or DAG runs triggered without changing the run_id parameter.
+allowed_run_id_pattern = ^[A-Za-z0-9_.~:+-]+$
+
[triggerer]
# How many triggers a single Triggerer will run at once, by default.
default_capacity = 1000
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 479c70524b..97b35e5a78 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -66,7 +66,7 @@ import airflow.templates
from airflow import settings, utils
from airflow.api_internal.internal_api_call import internal_api_call
from airflow.compat.functools import cached_property
-from airflow.configuration import conf, secrets_backend_list
+from airflow.configuration import conf as airflow_conf, secrets_backend_list
from airflow.exceptions import (
AirflowDagInconsistent,
AirflowException,
@@ -80,7 +80,7 @@ from airflow.models.abstractoperator import AbstractOperator
from airflow.models.base import Base, StringID
from airflow.models.dagcode import DagCode
from airflow.models.dagpickle import DagPickle
-from airflow.models.dagrun import DagRun
+from airflow.models.dagrun import RUN_ID_REGEX, DagRun
from airflow.models.operator import Operator
from airflow.models.param import DagParam, ParamsDict
from airflow.models.taskinstance import Context, TaskInstance, TaskInstanceKey, clear_task_instances
@@ -402,13 +402,13 @@ class DAG(LoggingMixin):
user_defined_filters: dict | None = None,
default_args: dict | None = None,
concurrency: int | None = None,
- max_active_tasks: int = conf.getint("core", "max_active_tasks_per_dag"),
- max_active_runs: int = conf.getint("core", "max_active_runs_per_dag"),
+ max_active_tasks: int = airflow_conf.getint("core", "max_active_tasks_per_dag"),
+ max_active_runs: int = airflow_conf.getint("core", "max_active_runs_per_dag"),
dagrun_timeout: timedelta | None = None,
sla_miss_callback: None | SLAMissCallback | list[SLAMissCallback] = None,
- default_view: str = conf.get_mandatory_value("webserver", "dag_default_view").lower(),
- orientation: str = conf.get_mandatory_value("webserver", "dag_orientation"),
- catchup: bool = conf.getboolean("scheduler", "catchup_by_default"),
+ default_view: str = airflow_conf.get_mandatory_value("webserver", "dag_default_view").lower(),
+ orientation: str = airflow_conf.get_mandatory_value("webserver", "dag_orientation"),
+ catchup: bool = airflow_conf.getboolean("scheduler", "catchup_by_default"),
on_success_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None,
on_failure_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None,
doc_md: str | None = None,
@@ -2429,7 +2429,7 @@ class DAG(LoggingMixin):
mark_success=False,
local=False,
executor=None,
- donot_pickle=conf.getboolean("core", "donot_pickle"),
+ donot_pickle=airflow_conf.getboolean("core", "donot_pickle"),
ignore_task_deps=False,
ignore_first_depends_on_past=True,
pool=None,
@@ -2666,13 +2666,14 @@ class DAG(LoggingMixin):
"Creating DagRun needs either `run_id` or both `run_type` and `execution_date`"
)
- if run_id and "/" in run_id:
- warnings.warn(
- "Using forward slash ('/') in a DAG run ID is deprecated. Note that this character "
- "also makes the run impossible to retrieve via Airflow's REST API.",
- RemovedInAirflow3Warning,
- stacklevel=3,
- )
+ regex = airflow_conf.get("scheduler", "allowed_run_id_pattern")
+
+ if run_id and not re.match(RUN_ID_REGEX, run_id):
+ if not regex.strip() or not re.match(regex.strip(), run_id):
+ raise AirflowException(
+ f"The provided run ID '{run_id}' is invalid. It does not match either "
+ f"the configured pattern: '{regex}' or the built-in pattern: '{RUN_ID_REGEX}'"
+ )
# create a copy of params before validating
copied_params = copy.deepcopy(self.params)
@@ -2960,7 +2961,7 @@ class DAG(LoggingMixin):
def get_default_view(self):
"""This is only there for backward compatible jinja2 templates"""
if self.default_view is None:
- return conf.get("webserver", "dag_default_view").lower()
+ return airflow_conf.get("webserver", "dag_default_view").lower()
else:
return self.default_view
@@ -3177,7 +3178,7 @@ class DagModel(Base):
root_dag_id = Column(StringID())
# A DAG can be paused from the UI / DB
# Set this default value of is_paused based on a configuration value!
- is_paused_at_creation = conf.getboolean("core", "dags_are_paused_at_creation")
+ is_paused_at_creation = airflow_conf.getboolean("core", "dags_are_paused_at_creation")
is_paused = Column(Boolean, default=is_paused_at_creation)
# Whether the DAG is a subdag
is_subdag = Column(Boolean, default=False)
@@ -3251,7 +3252,9 @@ class DagModel(Base):
"TaskOutletDatasetReference",
cascade="all, delete, delete-orphan",
)
- NUM_DAGS_PER_DAGRUN_QUERY = conf.getint("scheduler", "max_dagruns_to_create_per_loop", fallback=10)
+ NUM_DAGS_PER_DAGRUN_QUERY = airflow_conf.getint(
+ "scheduler", "max_dagruns_to_create_per_loop", fallback=10
+ )
def __init__(self, concurrency=None, **kwargs):
super().__init__(**kwargs)
@@ -3264,10 +3267,10 @@ class DagModel(Base):
)
self.max_active_tasks = concurrency
else:
- self.max_active_tasks = conf.getint("core", "max_active_tasks_per_dag")
+ self.max_active_tasks = airflow_conf.getint("core", "max_active_tasks_per_dag")
if self.max_active_runs is None:
- self.max_active_runs = conf.getint("core", "max_active_runs_per_dag")
+ self.max_active_runs = airflow_conf.getint("core", "max_active_runs_per_dag")
if self.has_task_concurrency_limits is None:
# Be safe -- this will be updated later once the DAG is parsed
@@ -3346,7 +3349,7 @@ class DagModel(Base):
have a value
"""
# This is for backwards-compatibility with old dags that don't have None as default_view
- return self.default_view or conf.get_mandatory_value("webserver", "dag_default_view").lower()
+ return self.default_view or airflow_conf.get_mandatory_value("webserver", "dag_default_view").lower()
@property
def safe_dag_id(self):
@@ -3529,13 +3532,13 @@ def dag(
user_defined_filters: dict | None = None,
default_args: dict | None = None,
concurrency: int | None = None,
- max_active_tasks: int = conf.getint("core", "max_active_tasks_per_dag"),
- max_active_runs: int = conf.getint("core", "max_active_runs_per_dag"),
+ max_active_tasks: int = airflow_conf.getint("core", "max_active_tasks_per_dag"),
+ max_active_runs: int = airflow_conf.getint("core", "max_active_runs_per_dag"),
dagrun_timeout: timedelta | None = None,
sla_miss_callback: None | SLAMissCallback | list[SLAMissCallback] = None,
- default_view: str = conf.get_mandatory_value("webserver", "dag_default_view").lower(),
- orientation: str = conf.get_mandatory_value("webserver", "dag_orientation"),
- catchup: bool = conf.getboolean("scheduler", "catchup_by_default"),
+ default_view: str = airflow_conf.get_mandatory_value("webserver", "dag_default_view").lower(),
+ orientation: str = airflow_conf.get_mandatory_value("webserver", "dag_orientation"),
+ catchup: bool = airflow_conf.getboolean("scheduler", "catchup_by_default"),
on_success_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None,
on_failure_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None,
doc_md: str | None = None,
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index ba0fd9fda3..028a81bb75 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -24,6 +24,7 @@ from collections import defaultdict
from datetime import datetime
from typing import TYPE_CHECKING, Any, Callable, Iterable, Iterator, NamedTuple, Sequence, TypeVar, overload
+import re2 as re
from sqlalchemy import (
Boolean,
Column,
@@ -43,7 +44,7 @@ from sqlalchemy import (
)
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.associationproxy import association_proxy
-from sqlalchemy.orm import Session, declared_attr, joinedload, relationship, synonym
+from sqlalchemy.orm import Session, declared_attr, joinedload, relationship, synonym, validates
from sqlalchemy.sql.expression import false, select, true
from airflow import settings
@@ -75,6 +76,8 @@ if TYPE_CHECKING:
CreatedTasks = TypeVar("CreatedTasks", Iterator["dict[str, Any]"], Iterator[TI])
TaskCreator = Callable[[Operator, Iterable[int]], CreatedTasks]
+RUN_ID_REGEX = r"^(?:manual|scheduled|dataset_triggered)__(?:\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00)$"
+
class TISchedulingDecision(NamedTuple):
"""Type of return for DagRun.task_instance_scheduling_decisions"""
@@ -238,6 +241,17 @@ class DagRun(Base, LoggingMixin):
external_trigger=self.external_trigger,
)
+ @validates("run_id")
+ def validate_run_id(self, key: str, run_id: str) -> str | None:
+ if not run_id:
+ return None
+ regex = airflow_conf.get("scheduler", "allowed_run_id_pattern")
+ if not re.match(regex, run_id) and not re.match(RUN_ID_REGEX, run_id):
+ raise ValueError(
+ f"The run_id provided '{run_id}' does not match the pattern '{regex}' or '{RUN_ID_REGEX}'"
+ )
+ return run_id
+
@property
def logical_date(self) -> datetime:
return self.execution_date
diff --git a/airflow/www/views.py b/airflow/www/views.py
index c29d952583..18188d3de5 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -96,7 +96,7 @@ from airflow.models import Connection, DagModel, DagTag, Log, SlaMiss, TaskFail,
from airflow.models.abstractoperator import AbstractOperator
from airflow.models.dag import DAG, get_dataset_triggered_next_run_info
from airflow.models.dagcode import DagCode
-from airflow.models.dagrun import DagRun, DagRunType
+from airflow.models.dagrun import RUN_ID_REGEX, DagRun, DagRunType
from airflow.models.dataset import DagScheduleDatasetReference, DatasetDagRunQueue, DatasetEvent, DatasetModel
from airflow.models.mappedoperator import MappedOperator
from airflow.models.operator import Operator
@@ -1896,7 +1896,7 @@ class Airflow(AirflowBaseView):
def trigger(self, session: Session = NEW_SESSION):
"""Triggers DAG Run."""
dag_id = request.values["dag_id"]
- run_id = request.values.get("run_id", "")
+ run_id = request.values.get("run_id", "").replace(" ", "+")
origin = get_safe_url(request.values.get("origin"))
unpause = request.values.get("unpause")
request_conf = request.values.get("conf")
@@ -2016,13 +2016,27 @@ class Airflow(AirflowBaseView):
flash(message, "error")
return redirect(origin)
- # Flash a warning when slash is used, but still allow it to continue on.
- if run_id and "/" in run_id:
- flash(
- "Using forward slash ('/') in a DAG run ID is deprecated. Note that this character "
- "also makes the run impossible to retrieve via Airflow's REST API.",
- "warning",
- )
+ regex = conf.get("scheduler", "allowed_run_id_pattern")
+ if run_id and not re.match(RUN_ID_REGEX, run_id):
+ if not regex.strip() or not re.match(regex.strip(), run_id):
+ flash(
+ f"The provided run ID '{run_id}' is invalid. It does not match either "
+ f"the configured pattern: '{regex}' or the built-in pattern: '{RUN_ID_REGEX}'",
+ "error",
+ )
+
+ form = DateTimeForm(data={"execution_date": execution_date})
+ return self.render_template(
+ "airflow/trigger.html",
+ form_fields=form_fields,
+ dag=dag,
+ dag_id=dag_id,
+ origin=origin,
+ conf=request_conf,
+ form=form,
+ is_dag_run_conf_overrides_params=is_dag_run_conf_overrides_params,
+ recent_confs=recent_confs,
+ )
run_conf = {}
if request_conf:
diff --git a/tests/models/test_dagrun.py b/tests/models/test_dagrun.py
index 38f882c687..5cb2c61e7b 100644
--- a/tests/models/test_dagrun.py
+++ b/tests/models/test_dagrun.py
@@ -29,6 +29,7 @@ from sqlalchemy.orm.session import Session
from airflow import settings
from airflow.callbacks.callback_requests import DagCallbackRequest
from airflow.decorators import task, task_group
+from airflow.exceptions import AirflowException
from airflow.models import (
DAG,
DagBag,
@@ -53,6 +54,7 @@ from airflow.utils.trigger_rule import TriggerRule
from airflow.utils.types import DagRunType
from tests.models import DEFAULT_DATE as _DEFAULT_DATE
from tests.test_utils import db
+from tests.test_utils.config import conf_vars
from tests.test_utils.mock_operators import MockOperator
DEFAULT_DATE = pendulum.instance(_DEFAULT_DATE)
@@ -2328,3 +2330,33 @@ def test_dagrun_with_note(dag_maker, session):
assert session.query(DagRun).filter(DagRun.id == dr.id).one_or_none() is None
assert session.query(DagRunNote).filter(DagRunNote.dag_run_id == dr.id).one_or_none() is None
+
+
+@pytest.mark.parametrize(
+ "pattern, run_id, result",
+ [
+ ["^[A-Z]", "ABC", True],
+ ["^[A-Z]", "abc", False],
+ ["^[0-9]", "123", True],
+ # The below params tests that user configuration does not affect internally generated
+ # run_ids
+ ["", "scheduled__2023-01-01T00:00:00+00:00", True],
+ ["", "manual__2023-01-01T00:00:00+00:00", True],
+ ["", "dataset_triggered__2023-01-01T00:00:00+00:00", True],
+ ["", "scheduled_2023-01-01T00", False],
+ ["", "manual_2023-01-01T00", False],
+ ["", "dataset_triggered_2023-01-01T00", False],
+ ["^[0-9]", "scheduled__2023-01-01T00:00:00+00:00", True],
+ ["^[0-9]", "manual__2023-01-01T00:00:00+00:00", True],
+ ["^[a-z]", "dataset_triggered__2023-01-01T00:00:00+00:00", True],
+ ],
+)
+def test_dag_run_id_config(session, dag_maker, pattern, run_id, result):
+ with conf_vars({("scheduler", "allowed_run_id_pattern"): pattern}):
+ with dag_maker():
+ ...
+ if result:
+ dag_maker.create_dagrun(run_id=run_id)
+ else:
+ with pytest.raises(AirflowException):
+ dag_maker.create_dagrun(run_id=run_id)
diff --git a/tests/www/views/test_views_trigger_dag.py b/tests/www/views/test_views_trigger_dag.py
index 1c7cc5827e..b6fcd20d3e 100644
--- a/tests/www/views/test_views_trigger_dag.py
+++ b/tests/www/views/test_views_trigger_dag.py
@@ -30,6 +30,7 @@ from airflow.utils import timezone
from airflow.utils.session import create_session
from airflow.utils.types import DagRunType
from tests.test_utils.api_connexion_utils import create_test_client
+from tests.test_utils.config import conf_vars
from tests.test_utils.www import check_content_in_response
@@ -286,3 +287,32 @@ def test_trigger_dag_params_array_value_none_render(admin_client, dag_maker, ses
f'<textarea style="display: none;" id="json_start" name="json_start">{expected_dag_conf}</textarea>',
resp,
)
+
+
+@pytest.mark.parametrize(
+ "pattern, run_id, result",
+ [
+ ["^[A-Z]", "ABC", True],
+ ["^[A-Z]", "abc", False],
+ ["^[0-9]", "123", True],
+ # The below params tests that user configuration does not affect internally generated
+ # run_ids. We use manual__ as a prefix for manually triggered DAGs due to a restriction
+ # in manually triggered DAGs that the run_id must not start with scheduled__.
+ ["", "manual__2023-01-01T00:00:00+00:00", True],
+ ["", "scheduled_2023-01-01T00", False],
+ ["", "manual_2023-01-01T00", False],
+ ["", "dataset_triggered_2023-01-01T00", False],
+ ["^[0-9]", "manual__2023-01-01T00:00:00+00:00", True],
+ ["^[a-z]", "manual__2023-01-01T00:00:00+00:00", True],
+ ],
+)
+def test_dag_run_id_pattern(session, admin_client, pattern, run_id, result):
+ with conf_vars({("scheduler", "allowed_run_id_pattern"): pattern}):
+ test_dag_id = "example_bash_operator"
+ admin_client.post(f"dags/{test_dag_id}/trigger?&run_id={run_id}")
+ run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
+ if result:
+ assert run is not None
+ assert run.run_type == DagRunType.MANUAL
+ else:
+ assert run is None
[airflow] 29/42: Fix try_number shown in the `task?task_id=&dag_id=&execution_date=` (#32361)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit fb30cbdb941b4cfe11148e7a4de7a825457659fa
Author: Akash Sharma <35...@users.noreply.github.com>
AuthorDate: Wed Jul 5 14:00:51 2023 +0530
Fix try_number shown in the `task?task_id=&dag_id=&execution_date=` (#32361)
* try_number fix
* change test name
(cherry picked from commit a8e4b8aee602e8c672ab879b7392a65b5c2bb34e)
---
airflow/www/views.py | 3 ++-
tests/www/views/test_views_tasks.py | 9 +++++++++
2 files changed, 11 insertions(+), 1 deletion(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 37b756d332..c29d952583 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1721,7 +1721,8 @@ class Airflow(AirflowBaseView):
with warnings.catch_warnings():
warnings.simplefilter("ignore", RemovedInAirflow3Warning)
all_ti_attrs = (
- (name, getattr(ti, name))
+ # fetching the value of _try_number to be shown under name try_number in UI
+ (name, getattr(ti, "_try_number" if name == "try_number" else name))
for name in dir(ti)
if not name.startswith("_") and name not in ti_attrs_to_skip
)
diff --git a/tests/www/views/test_views_tasks.py b/tests/www/views/test_views_tasks.py
index 8ed25f5614..c63ec5d23e 100644
--- a/tests/www/views/test_views_tasks.py
+++ b/tests/www/views/test_views_tasks.py
@@ -323,6 +323,15 @@ def test_views_get(admin_client, url, contents):
check_content_in_response(content, resp)
+def test_rendered_task_view(admin_client):
+ url = f"task?task_id=runme_0&dag_id=example_bash_operator&execution_date={DEFAULT_VAL}"
+ resp = admin_client.get(url, follow_redirects=True)
+ resp_html = resp.data.decode("utf-8")
+ assert resp.status_code == 200
+ assert "<td>_try_number</td>" not in resp_html
+ assert "<td>try_number</td>" in resp_html
+
+
def test_rendered_k8s(admin_client):
url = f"rendered-k8s?task_id=runme_0&dag_id=example_bash_operator&execution_date={DEFAULT_VAL}"
with unittest.mock.patch.object(settings, "IS_K8S_OR_K8SCELERY_EXECUTOR", True):
[airflow] 35/42: Add information on ARM MySQL support in Docker images (#32401)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 43720fb18a239b650c120ed3639093057ef36c3c
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Thu Jul 6 17:28:29 2023 +0200
Add information on ARM MySQL support in Docker images (#32401)
The change #29519 added experimental MariaDB client support for
MySQL for ARM images but we missed that in the changelog.
(cherry picked from commit ef75a3a6757a033586c933f7b62ab86f846af754)
---
docs/docker-stack/changelog.rst | 1 +
docs/docker-stack/index.rst | 25 +++++++++++++++++--------
2 files changed, 18 insertions(+), 8 deletions(-)
diff --git a/docs/docker-stack/changelog.rst b/docs/docker-stack/changelog.rst
index b40c0ae68f..619dfd9e74 100644
--- a/docs/docker-stack/changelog.rst
+++ b/docs/docker-stack/changelog.rst
@@ -107,6 +107,7 @@ Airflow 2.3
This can be done by setting ``DOCKER_BUILDKIT=1`` as an environment variable
or by installing `the buildx plugin <https://docs.docker.com/buildx/working-with-buildx/>`_
and running ``docker buildx build`` command.
+ * MySQL is experimentally supported on ARM through MariaDB client library
* Add Python 3.10 support
* Switch to ``Debian Bullseye`` based images. ``Debian Buster`` is deprecated and only available for
customized image building.
diff --git a/docs/docker-stack/index.rst b/docs/docker-stack/index.rst
index d54c374b70..5866f444e8 100644
--- a/docs/docker-stack/index.rst
+++ b/docs/docker-stack/index.rst
@@ -124,14 +124,23 @@ Support
The reference Docker Image supports the following platforms and database:
-* Intel platform (x86_64)
- * Postgres Client
- * MySQL Client
- * MSSQL Client
-
-* ARM platform (aarch64) - experimental support, might change any time
- * Postgres Client
- * MSSQL Client
+Intel platform (x86_64)
+-----------------------
+
+* Postgres Client
+* MySQL Client
+* MSSQL Client
+
+ARM platform (aarch64)
+----------------------
+
+ARM support is experimental, might change any time.
+
+* Postgres Client
+* MySQL Client (MySQL 8)
+* MSSQL Client
+
+Note that MySQL on arm has experimental support through MariaDB client library.
Usage
=====
[airflow] 05/42: Flash an error msg instead of failure in `rendered-templates` when map index is not found (#32011)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ed2d3acedd02de3ac22b98d9641c8bb4ecb0b07b
Author: Hussein Awala <hu...@awala.fr>
AuthorDate: Sun Jun 25 10:06:34 2023 +0200
Flash an error msg instead of failure in `rendered-templates` when map index is not found (#32011)
---------
Signed-off-by: Hussein Awala <hu...@awala.fr>
(cherry picked from commit 62a534dbc7fa8ddb4c249ade85c558b64d1630dd)
---
airflow/www/views.py | 21 ++++++++++++++++++---
1 file changed, 18 insertions(+), 3 deletions(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 4907891d1b..8eaac3eec5 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1340,6 +1340,9 @@ class Airflow(AirflowBaseView):
dag_run = dag.get_dagrun(execution_date=dttm, session=session)
raw_task = dag.get_task(task_id).prepare_for_execution()
+ title = "Rendered Template"
+ html_dict = {}
+
ti: TaskInstance
if dag_run is None:
# No DAG run matching given logical date. This usually means this
@@ -1351,7 +1354,21 @@ class Airflow(AirflowBaseView):
ti.dag_run = DagRun(dag_id=dag_id, execution_date=dttm)
else:
ti = dag_run.get_task_instance(task_id=task_id, map_index=map_index, session=session)
- ti.refresh_from_task(raw_task)
+ if ti:
+ ti.refresh_from_task(raw_task)
+ else:
+ flash(f"there is no task instance with the provided map_index {map_index}", "error")
+ return self.render_template(
+ "airflow/ti_code.html",
+ html_dict=html_dict,
+ dag=dag,
+ task_id=task_id,
+ execution_date=execution_date,
+ map_index=map_index,
+ form=form,
+ root=root,
+ title=title,
+ )
try:
ti.get_rendered_template_fields(session=session)
@@ -1370,8 +1387,6 @@ class Airflow(AirflowBaseView):
# but we'll display some quasi-meaingful field names.
task = ti.task.unmap(None)
- title = "Rendered Template"
- html_dict = {}
renderers = wwwutils.get_attr_renderer()
for template_field in task.template_fields:
[airflow] 13/42: Adding missing hyperlink to the tutorial documentation (#32105)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit a7a676eb082ea97d810e5f19ac58280bb54f3eeb
Author: Vijayasarathi Balasubramanian <vi...@gmail.com>
AuthorDate: Sat Jun 24 02:14:15 2023 -0400
Adding missing hyperlink to the tutorial documentation (#32105)
* Adding missing hyperlink to the tutorial documentation
(cherry picked from commit 4a6efe520f6b357002072bf5955eb731d5bfd716)
---
docs/apache-airflow/tutorial/fundamentals.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/tutorial/fundamentals.rst b/docs/apache-airflow/tutorial/fundamentals.rst
index 069de57790..695d748c00 100644
--- a/docs/apache-airflow/tutorial/fundamentals.rst
+++ b/docs/apache-airflow/tutorial/fundamentals.rst
@@ -82,7 +82,7 @@ of default parameters that we can use when creating tasks.
:end-before: [END default_args]
For more information about the BaseOperator's parameters and what they do,
-refer to the :py:class:`airflow.models.BaseOperator` documentation.
+refer to the :py:class:`airflow.models.baseoperator.BaseOperator` documentation.
Also, note that you could easily define different sets of arguments that
would serve different purposes. An example of that would be to have
[airflow] 15/42: Update callbacks.rst (#32148)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 050b3856dc85298d8e1afebc90fb854233dc7317
Author: Frank Cash <fc...@astronomer.io>
AuthorDate: Mon Jun 26 16:38:02 2023 -0400
Update callbacks.rst (#32148)
(cherry picked from commit 45bd9c993ee87789e01aae55f8abb34a0ff1de48)
---
.../administration-and-deployment/logging-monitoring/callbacks.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/administration-and-deployment/logging-monitoring/callbacks.rst b/docs/apache-airflow/administration-and-deployment/logging-monitoring/callbacks.rst
index 88e6dd7a0b..ccf54f2147 100644
--- a/docs/apache-airflow/administration-and-deployment/logging-monitoring/callbacks.rst
+++ b/docs/apache-airflow/administration-and-deployment/logging-monitoring/callbacks.rst
@@ -83,7 +83,7 @@ In the following example, failures in any task call the ``task_failure_alert`` f
on_success_callback=None,
on_failure_callback=task_failure_alert,
tags=["example"],
- ) as dag:
+ ):
task1 = EmptyOperator(task_id="task1")
task2 = EmptyOperator(task_id="task2")
[airflow] 41/42: Update RELEASE_NOTES.rst
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9cddb22b646b88e8bf5509a5e328745e44b0fde2
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu Jul 6 21:21:20 2023 +0100
Update RELEASE_NOTES.rst
---
RELEASE_NOTES.rst | 67 +++++++++++++++++++++++++++++++++++++++++++++-
docs/spelling_wordlist.txt | 1 +
2 files changed, 67 insertions(+), 1 deletion(-)
diff --git a/RELEASE_NOTES.rst b/RELEASE_NOTES.rst
index 0a30d047f2..95de23ba25 100644
--- a/RELEASE_NOTES.rst
+++ b/RELEASE_NOTES.rst
@@ -21,7 +21,72 @@
.. towncrier release notes start
-Airflow 2.6.2 (2023-06-17)
+Airflow 2.6.3 (2023-07-17)
+--------------------------
+
+Significant Changes
+^^^^^^^^^^^^^^^^^^^
+
+No significant changes.
+
+Bug Fixes
+"""""""""
+- Fix triggerers alive check and add a new conf for triggerer heartbeat rate (#32123)
+- Catch the exception that triggerer initialization failed (#31999)
+- Hide sensitive values from extra in connection edit form (#32309)
+- Sanitize ``DagRun.run_id`` and allow flexibility (#32293)
+- Add triggerer canceled log (#31757)
+- Fix try_number shown in the ``task?task_id=&dag_id=&execution_date=`` (#32361)
+- Retry transactions on occasional deadlocks for rendered fields (#32341)
+- Fix behaviour of LazyDictWithCache when import fails (#32248)
+- Remove ``executor_class`` from Job - fixing backfill for custom executors (#32219)
+- Fix bugged singleton implementation (#32218)
+- Use ``mapIndex`` to display extra links per mapped task. (#32154)
+- Ensure that main triggerer thread exits if the async thread fails (#32092)
+- Use ``re2`` for matching untrusted regex (#32060)
+- Render list items in rendered fields view (#32042)
+- Fix hashing of ``dag_dependencies`` in serialized dag (#32037)
+- Return ``None`` if an XComArg fails to resolve in a multiple_outputs Task (#32027)
+- Check for DAG ID in query param from url as well as kwargs (#32014)
+- Flash an error message instead of failure in ``rendered-templates`` when map index is not found (#32011)
+- Fix ``ExternalTaskSensor`` when there is not task group TIs for the current execution date (#32009)
+- Fix number param html type in trigger template (#31980, #31946)
+- Fix masking nested variable fields (#31964)
+- Fix ``operator_extra_links`` property serialization in mapped tasks (#31904)
+- Decode old-style nested Xcom value (#31866)
+- Add a check for trailing slash in webserver base_url (#31833)
+- Fix connection uri parsing when the host includes a scheme (#31465)
+- This fixes an issue were a database session got stuck (#31128)
+- Fix DAG's ``on_failure_callback`` is not invoked when task failed during testing dag. (#30965)
+- Fix airflow module version check when using ``ExternalPythonOperator`` and debug logging level (#30367)
+
+Misc/Internal
+"""""""""""""
+- Fix ``task.sensor`` annotation in type stub (#31954)
+- Limit ``Pydantic`` to ``< 2.0.0`` until we solve ``2.0.0`` incompatibilities (#32312)
+- Fix ``Pydantic`` 2 pickiness about model definition (#32307)
+
+Doc only changes
+""""""""""""""""
+- Minor updates to docs(#32369,#32315,#32310)
+- Clarify Listener API behavior (#32269)
+- Add information for users who ask for requirements (#32262)
+- Add links to DAGRun / DAG / Task in templates-ref.rst (#32245)
+- Add comment to warn off a potential wrong fix (#32230)
+- Add a note that we'll need to restart triggerer to reflect any trigger change (#32140)
+- Adding missing hyperlink to the tutorial documentation (#32105)
+- Added difference between Deferrable and Non-Deferrable Operators (#31840)
+- Add comments explaining need for special "trigger end" log message (#31812)
+- Increase readability for prerequisites.rst (#31794)
+- Documentation update on Plugin updates. (#31781)
+- Fix SemVer link in security documentation (#32320)
+- Update security model of Airflow (#32098)
+- Update references to restructured documentation from airflow core (#32282)
+- Separate out advanced logging configuration (#32131)
+- Add ``™`` to Airflow in prominent places (#31977)
+
+
+Airflow 2.6.2 (2023-06-12)
--------------------------
Significant Changes
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index c046d7c53c..192be0fa76 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -1574,6 +1574,7 @@ unprocessable
unqueued
unterminated
untransformed
+untrusted
unutilized
updateMask
updateonly
[airflow] 22/42: Add links to DAGRun / DAG / Task in templates-ref.rst (#32245)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9c7e2897b717e61aaf1db68e7c9e8e7b86cb57f3
Author: fritz-astronomer <80...@users.noreply.github.com>
AuthorDate: Wed Jun 28 18:35:02 2023 -0400
Add links to DAGRun / DAG / Task in templates-ref.rst (#32245)
* Update templates-ref.rst
Add links to DAGRun / DAG / Task
* add TaskInstance link
(cherry picked from commit b6477d0c3d498a8a1962fa67e0a92326ee6393fb)
---
docs/apache-airflow/templates-ref.rst | 24 ++++++++++++++++--------
1 file changed, 16 insertions(+), 8 deletions(-)
diff --git a/docs/apache-airflow/templates-ref.rst b/docs/apache-airflow/templates-ref.rst
index 34ba7d2ece..7f59091405 100644
--- a/docs/apache-airflow/templates-ref.rst
+++ b/docs/apache-airflow/templates-ref.rst
@@ -47,16 +47,16 @@ Variable Type Description
| Example: ``20180101T000000+0000``.
``{{ ts_nodash }}`` str | Same as ``{{ dag_run.logical_date | ts_nodash }}``.
| Example: ``20180101T000000``.
-``{{ prev_data_interval_start_success }}`` `pendulum.DateTime`_ | Start of the data interval of the prior successful DAG run.
+``{{ prev_data_interval_start_success }}`` `pendulum.DateTime`_ | Start of the data interval of the prior successful `DAG Run`_.
| ``None`` | Added in version 2.2.
-``{{ prev_data_interval_end_success }}`` `pendulum.DateTime`_ | End of the data interval of the prior successful DAG run.
+``{{ prev_data_interval_end_success }}`` `pendulum.DateTime`_ | End of the data interval of the prior successful `DAG Run`_.
| ``None`` | Added in version 2.2.
-``{{ prev_start_date_success }}`` `pendulum.DateTime`_ Start date from prior successful DAG run (if available).
+``{{ prev_start_date_success }}`` `pendulum.DateTime`_ Start date from prior successful `DAG Run`_ (if available).
| ``None``
-``{{ dag }}`` DAG The currently running DAG.
-``{{ task }}`` BaseOperator | The currently running task.
+``{{ dag }}`` DAG The currently running `DAG`_. You can read more about DAGs in :doc:`DAGs <core-concepts/dags>`.
+``{{ task }}`` BaseOperator | The currently running `Task`_. You can read more about Tasks in :doc:`core-concepts/operators`
``{{ macros }}`` | A reference to the macros package. See Macros_ below.
-``{{ task_instance }}`` TaskInstance The currently running task instance.
+``{{ task_instance }}`` TaskInstance The currently running `Task Instance`_.
``{{ ti }}`` TaskInstance Same as ``{{ task_instance }}``.
``{{ params }}`` dict[str, Any] | The user-defined params. This can be overridden by the mapping
| passed to ``trigger_dag -c`` if ``dag_run_conf_overrides_params``
@@ -68,8 +68,8 @@ Variable Type Description
| ``{dag_id}__{task_id}__{ds_nodash}``.
``{{ conf }}`` AirflowConfigParser | The full configuration object representing the content of your
| ``airflow.cfg``. See :mod:`airflow.configuration.conf`.
-``{{ run_id }}`` str The currently running DAG run's run ID.
-``{{ dag_run }}`` DagRun The currently running DAG run.
+``{{ run_id }}`` str The currently running `DAG Run`_'s run ID.
+``{{ dag_run }}`` DagRun The currently running `DAG Run`_.
``{{ test_mode }}`` bool Whether the task instance was run by the ``airflow test`` CLI.
``{{ expanded_ti_count }}`` int | ``None`` | Number of task instances that a mapped task was expanded into. If
| the current task is not mapped, this should be ``None``.
@@ -179,3 +179,11 @@ Some airflow specific macros are also defined:
:members:
.. _pendulum.DateTime: https://pendulum.eustace.io/docs/#introduction
+
+.. _DAG Run: :class:`~airflow.models.dagrun.DagRun`
+
+.. _DAG: :class:`~airflow.models.dag.DAG`
+
+.. _Task: :class:`~airflow.models.baseoperator.BaseOperator`
+
+.. _Task Instance: :class:`~airflow.models.taskinstance.TaskInstance`
[airflow] 30/42: Update pipeline.rst (#32369)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cd210181ee19ff02052a409eaf9b4e5098bbc143
Author: Mats Holm <32...@users.noreply.github.com>
AuthorDate: Wed Jul 5 18:39:37 2023 +0200
Update pipeline.rst (#32369)
Typo in final dag, merge_data.
(cherry picked from commit 383efbcbe00a0ce25251ba3a06a625f0ae40d806)
---
docs/apache-airflow/tutorial/pipeline.rst | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/docs/apache-airflow/tutorial/pipeline.rst b/docs/apache-airflow/tutorial/pipeline.rst
index c9bfcfa601..fca2418a70 100644
--- a/docs/apache-airflow/tutorial/pipeline.rst
+++ b/docs/apache-airflow/tutorial/pipeline.rst
@@ -288,7 +288,10 @@ Putting all of the pieces together, we have our completed DAG.
FROM employees_temp
) t
ON CONFLICT ("Serial Number") DO UPDATE
- SET "Serial Number" = excluded."Serial Number";
+ SET
+ "Employee Markme" = excluded."Employee Markme",
+ "Description" = excluded."Description",
+ "Leave" = excluded."Leave";
"""
try:
postgres_hook = PostgresHook(postgres_conn_id="tutorial_pg_conn")
[airflow] 24/42: Add information for users who ask for requirements (#32262)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 555d73c0a179965cb6e2155e7014228a7d84d0ed
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Sat Jul 1 23:56:03 2023 +0200
Add information for users who ask for requirements (#32262)
* Add information for users who ask for requirements
This change is based on a number of discussions with the users asking
what are the minimum requirements for Airflow to run.
While we cannot give precise answer, we should also make the users
aware that simple answers are not possible, and that when they are
deciding to install airflow and manage it on their own, they also
take the responsibility to monitor and adjust the resources they
need based on the monitoring they have to run.
* Apply suggestions from code review
Co-authored-by: Pankaj Koti <pa...@gmail.com>
* Update docs/apache-airflow/installation/index.rst
---------
Co-authored-by: Pankaj Koti <pa...@gmail.com>
(cherry picked from commit f6db66e16374e504665972feba0831d4148c6d50)
---
.../administration-and-deployment/scheduler.rst | 1 +
docs/apache-airflow/installation/index.rst | 72 +++++++++++++++++++++-
2 files changed, 72 insertions(+), 1 deletion(-)
diff --git a/docs/apache-airflow/administration-and-deployment/scheduler.rst b/docs/apache-airflow/administration-and-deployment/scheduler.rst
index dfc97e6120..1a2a41b136 100644
--- a/docs/apache-airflow/administration-and-deployment/scheduler.rst
+++ b/docs/apache-airflow/administration-and-deployment/scheduler.rst
@@ -154,6 +154,7 @@ The following databases are fully supported and provide an "optimal" experience:
Microsoft SQLServer has not been tested with HA.
+.. _fine-tuning-scheduler:
Fine-tuning your Scheduler performance
--------------------------------------
diff --git a/docs/apache-airflow/installation/index.rst b/docs/apache-airflow/installation/index.rst
index 1ddbf5d66b..8f37ca208d 100644
--- a/docs/apache-airflow/installation/index.rst
+++ b/docs/apache-airflow/installation/index.rst
@@ -77,6 +77,9 @@ More details: :doc:`installing-from-sources`
* You should develop and handle the deployment for all components of Airflow.
* You are responsible for setting up database, creating and managing database schema with ``airflow db`` commands,
automated startup and recovery, maintenance, cleanup and upgrades of Airflow and the Airflow Providers.
+* You need to setup monitoring of your system allowing you to observe resources and react to problems.
+* You are expected to configure and manage appropriate resources for the installation (memory, CPU, etc) based
+ on the monitoring of your installation and feedback loop. See the notes about requirements.
**What Apache Airflow Community provides for that method**
@@ -123,6 +126,9 @@ More details: :doc:`/installation/installing-from-pypi`
* You should develop and handle the deployment for all components of Airflow.
* You are responsible for setting up database, creating and managing database schema with ``airflow db`` commands,
automated startup and recovery, maintenance, cleanup and upgrades of Airflow and Airflow Providers.
+* You need to setup monitoring of your system allowing you to observe resources and react to problems.
+* You are expected to configure and manage appropriate resources for the installation (memory, CPU, etc) based
+ on the monitoring of your installation and feedback loop.
**What Apache Airflow Community provides for that method**
@@ -181,6 +187,9 @@ and official constraint files- same that are used for installing Airflow from Py
deployments of containers. You can use your own custom mechanism, custom Kubernetes deployments,
custom Docker Compose, custom Helm charts etc., and you should choose it based on your experience
and expectations.
+* You need to setup monitoring of your system allowing you to observe resources and react to problems.
+* You are expected to configure and manage appropriate resources for the installation (memory, CPU, etc) based
+ on the monitoring of your installation and feedback loop.
**What Apache Airflow Community provides for that method**
@@ -238,6 +247,9 @@ More details: :doc:`helm-chart:index`
those changes when released by upgrading the base image. However, you are responsible in creating a
pipeline of building your own custom images with your own added dependencies and Providers and need to
repeat the customization step and building your own image when new version of Airflow image is released.
+* You need to setup monitoring of your system allowing you to observe resources and react to problems.
+* You are expected to configure and manage appropriate resources for the installation (memory, CPU, etc) based
+ on the monitoring of your installation and feedback loop.
**What Apache Airflow Community provides for that method**
@@ -256,7 +268,6 @@ More details: :doc:`helm-chart:index`
* If you can provide description of a reproducible problem with Airflow software, you can open
issue at `GitHub issues <https://github.com/apache/airflow/issues>`__
-
Using Managed Airflow Services
''''''''''''''''''''''''''''''
@@ -316,3 +327,62 @@ Follow the `Ecosystem <https://airflow.apache.org/ecosystem/>`__ page to find a
**Where to ask for help**
* Depends on what the 3rd-party provides. Look at the documentation of the 3rd-party deployment you use.
+
+
+Notes about minimum requirements
+''''''''''''''''''''''''''''''''
+
+There are often questions about minimum requirements for Airflow for production systems, but it is
+not possible to give a simple answer to that question.
+
+The requirements that Airflow might need depend on many factors, including (but not limited to):
+ * The deployment your Airflow is installed with (see above ways of installing Airflow)
+ * The requirements of the deployment environment (for example Kubernetes, Docker, Helm, etc.) that
+ are completely independent from Airflow (for example DNS resources, sharing the nodes/resources)
+ with more (or less) pods and containers that are needed that might depend on particular choice of
+ the technology/cloud/integration of monitoring etc.
+ * Technical details of database, hardware, network, etc. that your deployment is running on
+ * The complexity of the code you add to your DAGS, configuration, plugins, settings etc. (note, that
+ Airflow runs the code that DAG author and Deployment Manager provide)
+ * The number and choice of providers you install and use (Airflow has more than 80 providers) that can
+ be installed by choice of the Deployment Manager and using them might require more resources.
+ * The choice of parameters that you use when tuning Airflow. Airflow has many configuration parameters
+ that can fine-tuned to your needs
+ * The number of DagRuns and tasks instances you run with parallel instances of each in consideration
+ * How complex are the tasks you run
+
+The above "DAG" characteristics will change over time and even will change depending on the time of the day
+or week, so you have to be prepared to continuously monitor the system and adjust the parameters to make
+it works smoothly.
+
+While we can provide some specific minimum requirements for some development "quick start" - such as
+in case of our :ref:`running-airflow-in-docker` quick-start guide, it is not possible to provide any minimum
+requirements for production systems.
+
+The best way to think of resource allocation for Airflow instance is to think of it in terms of process
+control theory - where there are two types of systems:
+
+1. Fully predictable, with few knobs and variables, where you can reliably set the values for the
+ knobs and have an easy way to determine the behaviour of the system
+
+2. Complex systems with multiple variables, that are hard to predict and where you need to monitor
+ the system and adjust the knobs continuously to make sure the system is running smoothly.
+
+Airflow (and generally any modern system running usually on cloud services, with multiple layers responsible
+for resources as well multiple parameters to control their behaviour) is a complex system and they fall
+much more in the second category. If you decide to run Airflow in production on your own, you should be
+prepared for the monitor/observe/adjust feedback loop to make sure the system is running smoothly.
+
+Having a good monitoring system that will allow you to monitor the system and adjust the parameters
+is a must to put that in practice.
+
+There are few guidelines that you can use for optimizing your resource usage as well. The
+:ref:`fine-tuning-scheduler` is a good starting point to fine-tune your scheduler, you can also follow
+the :ref:`best_practice` guide to make sure you are using Airflow in the most efficient way.
+
+Also, one of the important things that Managed Services for Airflow provide is that they make a lot
+of opinionated choices and fine-tune the system for you, so you don't have to worry about it too much.
+With such managed services, there are usually far less number of knobs to turn and choices to make and one
+of the things you pay for is that the Managed Service provider manages the system for you and provides
+paid support and allows you to scale the system as needed and allocate the right resources - following the
+choices made there when it comes to the kinds of deployment you might have.
[airflow] 06/42: Check for DAG ID in query param from url as well as kwargs (#32014)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit c78e16588ee399f6eaf60425eb1ad7fa6d3fe352
Author: Niko Oliveira <on...@amazon.com>
AuthorDate: Tue Jun 20 10:04:45 2023 -0700
Check for DAG ID in query param from url as well as kwargs (#32014)
Previously the dag id was only being checked in request args and form
but not kwargs, so it was possible for the id when passed as kwargs
to be None. This can allow auth for a user who does not have the
permissions to view a particular DAG.
(cherry picked from commit ac65b82eeeeaa670e09a83c7da65cbac7e89f8db)
---
airflow/www/auth.py | 3 ++-
tests/www/views/test_views_tasks.py | 47 +++++++++++++++++++++++++++++++++++++
2 files changed, 49 insertions(+), 1 deletion(-)
diff --git a/airflow/www/auth.py b/airflow/www/auth.py
index 8fdafa1f63..0056af55fd 100644
--- a/airflow/www/auth.py
+++ b/airflow/www/auth.py
@@ -38,7 +38,8 @@ def has_access(permissions: Sequence[tuple[str, str]] | None = None) -> Callable
appbuilder = current_app.appbuilder
dag_id = (
- request.args.get("dag_id")
+ kwargs.get("dag_id")
+ or request.args.get("dag_id")
or request.form.get("dag_id")
or (request.is_json and request.json.get("dag_id"))
or None
diff --git a/tests/www/views/test_views_tasks.py b/tests/www/views/test_views_tasks.py
index 35538e5bf9..8ed25f5614 100644
--- a/tests/www/views/test_views_tasks.py
+++ b/tests/www/views/test_views_tasks.py
@@ -369,6 +369,20 @@ def test_graph_trigger_origin_graph_view(app, admin_client):
check_content_in_response(href, resp)
+def test_graph_view_without_dag_permission(app, one_dag_perm_user_client):
+ url = "/dags/example_bash_operator/graph"
+ resp = one_dag_perm_user_client.get(url, follow_redirects=True)
+ assert resp.status_code == 200
+ assert resp.request.url == "http://localhost/dags/example_bash_operator/graph"
+ check_content_in_response("example_bash_operator", resp)
+
+ url = "/dags/example_xcom/graph"
+ resp = one_dag_perm_user_client.get(url, follow_redirects=True)
+ assert resp.status_code == 200
+ assert resp.request.url == "http://localhost/home"
+ check_content_in_response("Access is Denied", resp)
+
+
def test_dag_details_trigger_origin_dag_details_view(app, admin_client):
app.dag_bag.get_dag("test_graph_view").create_dagrun(
run_type=DagRunType.SCHEDULED,
@@ -581,6 +595,39 @@ def per_dag_perm_user_client(app, new_dag_to_delete):
delete_roles(app)
+@pytest.fixture()
+def one_dag_perm_user_client(app):
+ username = "test_user_one_dag_perm"
+ dag_id = "example_bash_operator"
+ sm = app.appbuilder.sm
+ perm = f"{permissions.RESOURCE_DAG_PREFIX}{dag_id}"
+
+ sm.create_permission(permissions.ACTION_CAN_READ, perm)
+
+ create_user(
+ app,
+ username=username,
+ role_name="User with permission to access only one dag",
+ permissions=[
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_LOG),
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+ (permissions.ACTION_CAN_READ, perm),
+ ],
+ )
+
+ sm.find_user(username=username)
+
+ yield client_with_login(
+ app,
+ username=username,
+ password=username,
+ )
+
+ delete_user(app, username=username) # type: ignore
+ delete_roles(app)
+
+
def test_delete_just_dag_per_dag_permissions(new_dag_to_delete, per_dag_perm_user_client):
resp = per_dag_perm_user_client.post(
f"delete?dag_id={new_dag_to_delete.dag_id}&next=/home", follow_redirects=True
[airflow] 04/42: Fix ExternalTaskSensor when there is not task group TIs for the current execution date (#32009)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit c00001508315999f1b0f9e8077daafc270ff0f4c
Author: Hussein Awala <hu...@awala.fr>
AuthorDate: Wed Jun 21 11:55:45 2023 +0200
Fix ExternalTaskSensor when there is not task group TIs for the current execution date (#32009)
* Add a check on none TIs for the current execution date
Signed-off-by: Hussein Awala <hu...@awala.fr>
* replace inline if-else by old one
Signed-off-by: Hussein Awala <hu...@awala.fr>
---------
Signed-off-by: Hussein Awala <hu...@awala.fr>
(cherry picked from commit 14eb1d3116ecef15be7be9a8f9d08757e74f981c)
---
airflow/sensors/external_task.py | 13 ++++++++-----
tests/sensors/test_external_task_sensor.py | 20 ++++++++++++++++++++
2 files changed, 28 insertions(+), 5 deletions(-)
diff --git a/airflow/sensors/external_task.py b/airflow/sensors/external_task.py
index 959ebe5131..69ba41ef09 100644
--- a/airflow/sensors/external_task.py
+++ b/airflow/sensors/external_task.py
@@ -366,11 +366,14 @@ class ExternalTaskSensor(BaseSensorOperator):
) / len(self.external_task_ids)
elif self.external_task_group_id:
external_task_group_task_ids = self.get_external_task_group_task_ids(session, dttm_filter)
- count = (
- self._count_query(TI, session, states, dttm_filter)
- .filter(tuple_in_condition((TI.task_id, TI.map_index), external_task_group_task_ids))
- .scalar()
- ) / len(external_task_group_task_ids)
+ if not external_task_group_task_ids:
+ count = 0
+ else:
+ count = (
+ self._count_query(TI, session, states, dttm_filter)
+ .filter(tuple_in_condition((TI.task_id, TI.map_index), external_task_group_task_ids))
+ .scalar()
+ ) / len(external_task_group_task_ids)
else:
count = self._count_query(DR, session, states, dttm_filter).scalar()
return count
diff --git a/tests/sensors/test_external_task_sensor.py b/tests/sensors/test_external_task_sensor.py
index 966c3a2139..7c62357e9e 100644
--- a/tests/sensors/test_external_task_sensor.py
+++ b/tests/sensors/test_external_task_sensor.py
@@ -808,6 +808,26 @@ exit 0
):
op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
+ def test_external_task_group_when_there_is_no_TIs(self):
+ """Test that the sensor does not fail when there are no TIs to check."""
+ self.add_time_sensor()
+ self.add_dummy_task_group_with_dynamic_tasks(State.FAILED)
+ op = ExternalTaskSensor(
+ task_id="test_external_task_sensor_check",
+ external_dag_id=TEST_DAG_ID,
+ external_task_group_id=TEST_TASK_GROUP_ID,
+ failed_states=[State.FAILED],
+ dag=self.dag,
+ poke_interval=1,
+ timeout=3,
+ )
+ with pytest.raises(AirflowSensorTimeout):
+ op.run(
+ start_date=DEFAULT_DATE + timedelta(hours=1),
+ end_date=DEFAULT_DATE + timedelta(hours=1),
+ ignore_ti_state=True,
+ )
+
def test_external_task_sensor_check_zipped_dag_existence(dag_zip_maker):
with dag_zip_maker("test_external_task_sensor_check_existense.py") as dagbag:
[airflow] 08/42: Fix hashing of dag_dependencies in serialized dag (#32037)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 2813c41c1b93ea020d5788d4bb08c775abb6b90f
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Tue Jun 20 20:41:58 2023 -0700
Fix hashing of dag_dependencies in serialized dag (#32037)
(cherry picked from commit c76d57a67f1091e49cc9a8255054970e05ab1de5)
---
airflow/serialization/serialized_objects.py | 4 +--
tests/models/test_serialized_dag.py | 46 ++++++++++++++++++++++++++++-
2 files changed, 47 insertions(+), 3 deletions(-)
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index 04c427309c..058da502a8 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -1240,7 +1240,7 @@ class SerializedDAG(DAG, BaseSerialization):
for dep in SerializedBaseOperator.detect_dependencies(task)
}
dag_deps.update(DependencyDetector.detect_dag_dependencies(dag))
- serialized_dag["dag_dependencies"] = [x.__dict__ for x in dag_deps]
+ serialized_dag["dag_dependencies"] = [x.__dict__ for x in sorted(dag_deps)]
serialized_dag["_task_group"] = TaskGroupSerialization.serialize_task_group(dag.task_group)
# Edge info in the JSON exactly matches our internal structure
@@ -1446,7 +1446,7 @@ class TaskGroupSerialization(BaseSerialization):
return group
-@dataclass(frozen=True)
+@dataclass(frozen=True, order=True)
class DagDependency:
"""Dataclass for representing dependencies between DAGs.
These are calculated during serialization and attached to serialized DAGs.
diff --git a/tests/models/test_serialized_dag.py b/tests/models/test_serialized_dag.py
index b425cd8f65..61be12676f 100644
--- a/tests/models/test_serialized_dag.py
+++ b/tests/models/test_serialized_dag.py
@@ -20,13 +20,17 @@ from __future__ import annotations
from unittest import mock
+import pendulum
import pytest
-from airflow import DAG, example_dags as example_dags_module
+from airflow import DAG, Dataset, example_dags as example_dags_module
from airflow.models import DagBag
from airflow.models.dagcode import DagCode
from airflow.models.serialized_dag import SerializedDagModel as SDM
+from airflow.operators.bash import BashOperator
from airflow.serialization.serialized_objects import SerializedDAG
+from airflow.settings import json
+from airflow.utils.hashlib_wrapper import md5
from airflow.utils.session import create_session
from tests.test_utils import db
from tests.test_utils.asserts import assert_queries_count
@@ -196,3 +200,43 @@ class TestSerializedDagModel:
expected_dependencies = {dag_id: [] for dag_id in example_dags}
assert SDM.get_dag_dependencies() == expected_dependencies
+
+ def test_order_of_deps_is_consistent(self):
+ """
+ Previously the 'dag_dependencies' node in serialized dag was converted to list from set.
+ This caused the order, and thus the hash value, to be unreliable, which could produce
+ excessive dag parsing.
+ """
+ first_dag_hash = None
+ for r in range(10):
+ with DAG(
+ dag_id="example",
+ start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
+ schedule=[
+ Dataset("1"),
+ Dataset("2"),
+ Dataset("3"),
+ Dataset("4"),
+ Dataset("5"),
+ ],
+ ) as dag6:
+ BashOperator(
+ task_id="any",
+ outlets=[Dataset("0*"), Dataset("6*")],
+ bash_command="sleep 5",
+ )
+ deps_order = [x["dependency_id"] for x in SerializedDAG.serialize_dag(dag6)["dag_dependencies"]]
+ # in below assert, 0 and 6 both come at end because "source" is different for them and source
+ # is the first field in DagDependency class
+ assert deps_order == ["1", "2", "3", "4", "5", "0*", "6*"]
+
+ # for good measure, let's check that the dag hash is consistent
+ dag_json = json.dumps(SerializedDAG.to_dict(dag6), sort_keys=True).encode("utf-8")
+ this_dag_hash = md5(dag_json).hexdigest()
+
+ # set first dag hash on first pass
+ if first_dag_hash is None:
+ first_dag_hash = this_dag_hash
+
+ # dag hash should not change without change in structure (we're in a loop)
+ assert this_dag_hash == first_dag_hash
[airflow] 17/42: Documentation Changes (#32185)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3ed537c934534773dee356e48a108fe4e7bb5bb3
Author: Vijayasarathi Balasubramanian <vi...@gmail.com>
AuthorDate: Tue Jun 27 14:09:55 2023 -0400
Documentation Changes (#32185)
Readability change in the kubernetes.rst tutorial file
(cherry picked from commit 23b8e839a35e84d57c5cf38b0b21171ac3bd1ecc)
---
docs/apache-airflow/administration-and-deployment/kubernetes.rst | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/apache-airflow/administration-and-deployment/kubernetes.rst b/docs/apache-airflow/administration-and-deployment/kubernetes.rst
index 961a93328b..160694fa11 100644
--- a/docs/apache-airflow/administration-and-deployment/kubernetes.rst
+++ b/docs/apache-airflow/administration-and-deployment/kubernetes.rst
@@ -27,7 +27,7 @@ and autoscaling options that Kubernetes provides.
Helm Chart for Kubernetes
=========================
-We maintain :doc:`official Helm chart <helm-chart:index>` for Airflow that helps you define, install, and upgrade deployment. The Helm Chart uses :doc:`official Docker image and Dockerfile <docker-stack:index>` that is also maintained and released by the community.
+We maintain an :doc:`official Helm chart <helm-chart:index>` for Airflow that helps you define, install, and upgrade deployment. The Helm Chart uses :doc:`official Docker image and Dockerfile <docker-stack:index>` that is also maintained and released by the community.
Kubernetes Executor
^^^^^^^^^^^^^^^^^^^
@@ -47,7 +47,7 @@ Pod Mutation Hook
The Airflow local settings file (``airflow_local_settings.py``) can define a ``pod_mutation_hook`` function
that has the ability to mutate pod objects before sending them to the Kubernetes client
for scheduling. It receives a single argument as a reference to pod objects, and
-is expected to alter its attributes.
+are expected to alter its attributes.
This could be used, for instance, to add sidecar or init containers
to every worker pod launched by KubernetesExecutor or KubernetesPodOperator.
[airflow] 26/42: Update python.rst with multi-Scheduler LocalExecutor info (#32310)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 288fb12e836785f87f7bb4e6bc6ea24945aa3b37
Author: Sai Pragna Etikyala <pr...@gmail.com>
AuthorDate: Wed Jul 5 02:22:30 2023 -0700
Update python.rst with multi-Scheduler LocalExecutor info (#32310)
Co-authored-by: Pankaj Koti <pa...@gmail.com>
(cherry picked from commit e53320d62030a53c6ffe896434bcf0fc85803f31)
---
docs/apache-airflow/core-concepts/executor/local.rst | 8 ++++++++
1 file changed, 8 insertions(+)
diff --git a/docs/apache-airflow/core-concepts/executor/local.rst b/docs/apache-airflow/core-concepts/executor/local.rst
index 256c612553..94a3f51211 100644
--- a/docs/apache-airflow/core-concepts/executor/local.rst
+++ b/docs/apache-airflow/core-concepts/executor/local.rst
@@ -47,3 +47,11 @@ Arguably, :class:`~airflow.executors.sequential_executor.SequentialExecutor` cou
parallelism of just 1 worker, i.e. ``self.parallelism = 1``.
This option could lead to the unification of the executor implementations, running
locally, into just one :class:`~airflow.executors.local_executor.LocalExecutor` with multiple modes.
+
+.. note::
+
+ When multiple Schedulers are configured with ``executor = LocalExecutor`` in the ``[core]`` section of your ``airflow.cfg``, each Scheduler will run a LocalExecutor. This means tasks would be processed in a distributed fashion across the machines running the Schedulers.
+
+ One consideration should be taken into account:
+
+ - Restarting a Scheduler: If a Scheduler is restarted, it may take some time for other Schedulers to recognize the orphaned tasks and restart or fail them.
[airflow] 37/42: Less precise check for error message in trigger initialization (#32035)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit c92574f47112316267f87cea5b0fb481906f7482
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Tue Jun 20 23:10:15 2023 +0200
Less precise check for error message in trigger initialization (#32035)
The test for error message (introduced in #31999) in trigger
initialization expected precise error message, however Python 3.10
changed the details printed when TypeError was converted to string
(including the information about class being created not only method).
This caused the tests to fail on Python 3.10 and 3.11.
Making two asserts with two parts of the message expected solves
the problem for all python versions.
(cherry picked from commit 12c8099846a55ba43bf7d48bab3d01ab697f97a4)
---
tests/jobs/test_triggerer_job.py | 6 ++----
1 file changed, 2 insertions(+), 4 deletions(-)
diff --git a/tests/jobs/test_triggerer_job.py b/tests/jobs/test_triggerer_job.py
index 3e43d8f520..9115372777 100644
--- a/tests/jobs/test_triggerer_job.py
+++ b/tests/jobs/test_triggerer_job.py
@@ -296,10 +296,8 @@ class TestTriggerRunner:
trigger_runner.update_triggers({1})
- assert (
- "Trigger failed; message=__init__() got an unexpected keyword argument 'not_exists_arg'"
- in caplog.text
- )
+ assert "Trigger failed" in caplog.text
+ assert "got an unexpected keyword argument 'not_exists_arg'" in caplog.text
def test_trigger_create_race_condition_18392(session, tmp_path):
[airflow] 20/42: Remove executor_class from Job - fixing backfil for custom executors (#32219)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 14480751dd4ba07a0b8dc37bbdc1548193e962b2
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Wed Jun 28 07:32:36 2023 +0200
Remove executor_class from Job - fixing backfil for custom executors (#32219)
The executor_class in job was used in the past to pass information
about executor being used but it has been replaced by executor
passed via JobRunner when needed. The executor_class has only
been used in backfill job, but it has been used wrongly - because
it had no fully qualified path there.
The right way of using it is to check the executor passed via
JobRunner, and since executor_class is not used any more, we
can safely remove it.
This fixes a bug when custom exceutors could not be used
during backfill.
Co-authored-by: Andrew Halpern <57...@users.noreply.github.com>
(cherry picked from commit df4026a47aa6141142e186ec853ba286c2be0e95)
---
airflow/jobs/backfill_job_runner.py | 5 +----
airflow/jobs/job.py | 3 ---
tests/jobs/test_base_job.py | 1 -
3 files changed, 1 insertion(+), 8 deletions(-)
diff --git a/airflow/jobs/backfill_job_runner.py b/airflow/jobs/backfill_job_runner.py
index a44ebfa9de..e3e87c1a81 100644
--- a/airflow/jobs/backfill_job_runner.py
+++ b/airflow/jobs/backfill_job_runner.py
@@ -536,10 +536,7 @@ class BackfillJobRunner(BaseJobRunner[Job], LoggingMixin):
cfg_path = None
- executor_class, _ = ExecutorLoader.import_executor_cls(
- self.job.executor_class,
- )
- if executor_class.is_local:
+ if executor.is_local:
cfg_path = tmp_configuration_copy()
executor.queue_task_instance(
diff --git a/airflow/jobs/job.py b/airflow/jobs/job.py
index 399fa39f25..48dd94d58c 100644
--- a/airflow/jobs/job.py
+++ b/airflow/jobs/job.py
@@ -104,9 +104,6 @@ class Job(Base, LoggingMixin):
self.hostname = get_hostname()
if executor:
self.executor = executor
- self.executor_class = executor.__class__.__name__
- else:
- self.executor_class = conf.get("core", "EXECUTOR")
self.start_date = timezone.utcnow()
self.latest_heartbeat = timezone.utcnow()
if heartrate is not None:
diff --git a/tests/jobs/test_base_job.py b/tests/jobs/test_base_job.py
index c646979237..0e36bd4571 100644
--- a/tests/jobs/test_base_job.py
+++ b/tests/jobs/test_base_job.py
@@ -204,7 +204,6 @@ class TestJob:
test_job = Job(heartrate=10, dag_id="example_dag", state=State.RUNNING)
MockJobRunner(job=test_job)
- assert test_job.executor_class == "SequentialExecutor"
assert test_job.heartrate == 10
assert test_job.dag_id == "example_dag"
assert test_job.hostname == "test_hostname"
[airflow] 23/42: Fix behaviour of LazyDictWithCache wheni import fails (#32248)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 38df98bac8101ac61d1a27bcaa07b82a24b1487e
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Thu Jun 29 09:09:19 2023 +0200
Fix behaviour of LazyDictWithCache wheni import fails (#32248)
When #15330 added docker.task, it also optimized replacement of the
callable with it's result in LazyDictWithCache. LazyDictWithCache
is used by Provider's Manager to optimize access to hooks - basically
hook is only actually imported, when it is accessed. This helps with
speeding up importing of connection information
The optimization added result of running callable to _resolved
set, but it missed the case when None was returned. Previously,
when None was returned, the callable was not replaced and it
was called again. After the change - the _resolved set was
updated with the key and None was returned. But since the key
has not been replaced, next time when the same key was retrieved,
the original "callable" was returned, not the None value. So if
callable returned None, and the same key was retrieved twice, the second
time, instead of None, the dictionary returned Callable.
This PR fixes it by setting the value to dictionary even if it
was None.
(cherry picked from commit 462df0d05abbf023ea367ad2591cc889a49c815a)
---
airflow/providers_manager.py | 3 +--
tests/always/test_providers_manager.py | 31 ++++++++++++++++++++++++++++++-
2 files changed, 31 insertions(+), 3 deletions(-)
diff --git a/airflow/providers_manager.py b/airflow/providers_manager.py
index 84fd957a0e..dd214c755c 100644
--- a/airflow/providers_manager.py
+++ b/airflow/providers_manager.py
@@ -110,8 +110,7 @@ class LazyDictWithCache(MutableMapping):
# callable itself
value = value()
self._resolved.add(key)
- if value:
- self._raw_dict.__setitem__(key, value)
+ self._raw_dict.__setitem__(key, value)
return value
def __delitem__(self, key):
diff --git a/tests/always/test_providers_manager.py b/tests/always/test_providers_manager.py
index 31a6f70823..ab834ec5a9 100644
--- a/tests/always/test_providers_manager.py
+++ b/tests/always/test_providers_manager.py
@@ -28,7 +28,7 @@ from flask_babel import lazy_gettext
from wtforms import BooleanField, Field, StringField
from airflow.exceptions import AirflowOptionalProviderFeatureException
-from airflow.providers_manager import HookClassProvider, ProviderInfo, ProvidersManager
+from airflow.providers_manager import HookClassProvider, LazyDictWithCache, ProviderInfo, ProvidersManager
class TestProviderManager:
@@ -372,3 +372,32 @@ class TestProviderManager:
assert [
"Optional provider feature disabled when importing 'HookClass' from 'test_package' package"
] == self._caplog.messages
+
+
+@pytest.mark.parametrize(
+ "value, expected_outputs,",
+ [
+ ("a", "a"),
+ (1, 1),
+ (None, None),
+ (lambda: 0, 0),
+ (lambda: None, None),
+ (lambda: "z", "z"),
+ ],
+)
+def test_lazy_cache_dict_resolving(value, expected_outputs):
+ lazy_cache_dict = LazyDictWithCache()
+ lazy_cache_dict["key"] = value
+ assert lazy_cache_dict["key"] == expected_outputs
+ # Retrieve it again to see if it is correctly returned again
+ assert lazy_cache_dict["key"] == expected_outputs
+
+
+def test_lazy_cache_dict_raises_error():
+ def raise_method():
+ raise Exception("test")
+
+ lazy_cache_dict = LazyDictWithCache()
+ lazy_cache_dict["key"] = raise_method
+ with pytest.raises(Exception, match="test"):
+ _ = lazy_cache_dict["key"]
[airflow] 25/42: Clarify Listener API behavior (#32269)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3e4c5e840af6f7b9047f67f66e54b39523824897
Author: Vijayasarathi Balasubramanian <vi...@gmail.com>
AuthorDate: Tue Jul 4 16:57:49 2023 -0400
Clarify Listener API behavior (#32269)
* Fixes: #31180 - Plugin for listeners - on_dag_run_running hook ignored
Documentation update for Plugin for listeners - on_dag_run_running hook ignored
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
(cherry picked from commit ab2c861dd8a96f22b0fda692368ce9b103175322)
---
docs/apache-airflow/administration-and-deployment/listeners.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/administration-and-deployment/listeners.rst b/docs/apache-airflow/administration-and-deployment/listeners.rst
index c34689c53f..2847eb514d 100644
--- a/docs/apache-airflow/administration-and-deployment/listeners.rst
+++ b/docs/apache-airflow/administration-and-deployment/listeners.rst
@@ -54,7 +54,7 @@ Airflow defines the specification as `hookspec <https://github.com/apache/airflo
To include the listener in your Airflow installation, include it as a part of an :doc:`Airflow Plugin </authoring-and-scheduling/plugins>`
-Listener API is meant to be called across all DAGs and all operators. You can't listen to events generated by specific DAGs. For that behavior, try methods like ``on_success_callback`` and ``pre_execute``. These provide callbacks for particular DAG authors or operator creators.
+Listener API is meant to be called across all DAGs and all operators. You can't listen to events generated by specific DAGs. For that behavior, try methods like ``on_success_callback`` and ``pre_execute``. These provide callbacks for particular DAG authors or operator creators. The logs and ``print()`` calls will be handled as part of the listeners.
|experimental|
[airflow] 40/42: Update version to 2.6.3
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9ae7f0a0fe5de904b881b6ab349febd319234529
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu Jul 6 20:39:11 2023 +0100
Update version to 2.6.3
---
README.md | 30 +++++++++++-----------
airflow/__init__.py | 2 +-
airflow/api_connexion/openapi/v1.yaml | 2 +-
airflow/utils/db.py | 1 +
.../installation/supported-versions.rst | 2 +-
docs/docker-stack/README.md | 10 ++++----
.../extending/add-airflow-configuration/Dockerfile | 2 +-
.../extending/add-apt-packages/Dockerfile | 2 +-
.../add-build-essential-extend/Dockerfile | 2 +-
.../extending/add-providers/Dockerfile | 2 +-
.../extending/add-pypi-packages/Dockerfile | 2 +-
.../extending/add-requirement-packages/Dockerfile | 2 +-
.../extending/custom-providers/Dockerfile | 2 +-
.../extending/embedding-dags/Dockerfile | 2 +-
.../extending/writable-directory/Dockerfile | 2 +-
docs/docker-stack/entrypoint.rst | 14 +++++-----
.../ci/pre_commit/pre_commit_supported_versions.py | 2 +-
17 files changed, 41 insertions(+), 40 deletions(-)
diff --git a/README.md b/README.md
index 733f2fbb8b..40ef40a41a 100644
--- a/README.md
+++ b/README.md
@@ -86,15 +86,15 @@ Airflow is not a streaming solution, but it is often used to process real-time d
Apache Airflow is tested with:
-| | Main version (dev) | Stable version (2.6.2) |
-|------------|------------------------------|---------------------------|
-| Python | 3.7, 3.8, 3.9, 3.10, 3.11 | 3.7, 3.8, 3.9, 3.10, 3.11 |
-| Platform | AMD64/ARM64(\*) | AMD64/ARM64(\*) |
-| Kubernetes | 1.23, 1.24, 1.25, 1.26, 1.27 | 1.23, 1.24, 1.25, 1.26 |
-| PostgreSQL | 11, 12, 13, 14, 15 | 11, 12, 13, 14, 15 |
-| MySQL | 5.7, 8 | 5.7, 8 |
-| SQLite | 3.15.0+ | 3.15.0+ |
-| MSSQL | 2017(\*), 2019(\*) | 2017(\*), 2019(\*) |
+| | Main version (dev) | Stable version (2.6.3) |
+|------------|------------------------------|------------------------------|
+| Python | 3.7, 3.8, 3.9, 3.10, 3.11 | 3.7, 3.8, 3.9, 3.10, 3.11 |
+| Platform | AMD64/ARM64(\*) | AMD64/ARM64(\*) |
+| Kubernetes | 1.23, 1.24, 1.25, 1.26, 1.27 | 1.23, 1.24, 1.25, 1.26, 1.27 |
+| PostgreSQL | 11, 12, 13, 14, 15 | 11, 12, 13, 14, 15 |
+| MySQL | 5.7, 8 | 5.7, 8 |
+| SQLite | 3.15.0+ | 3.15.0+ |
+| MSSQL | 2017(\*), 2019(\*) | 2017(\*), 2019(\*) |
\* Experimental
@@ -158,15 +158,15 @@ them to the appropriate format and workflow that your tool requires.
```bash
-pip install 'apache-airflow==2.6.2' \
- --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.6.2/constraints-3.7.txt"
+pip install 'apache-airflow==2.6.3' \
+ --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.6.3/constraints-3.7.txt"
```
2. Installing with extras (i.e., postgres, google)
```bash
-pip install 'apache-airflow[postgres,google]==2.6.2' \
- --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.6.2/constraints-3.7.txt"
+pip install 'apache-airflow[postgres,google]==2.6.3' \
+ --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.6.3/constraints-3.7.txt"
```
For information on installing provider packages, check
@@ -275,7 +275,7 @@ Apache Airflow version life cycle:
| Version | Current Patch/Minor | State | First Release | Limited Support | EOL/Terminated |
|-----------|-----------------------|-----------|-----------------|-------------------|------------------|
-| 2 | 2.6.2 | Supported | Dec 17, 2020 | TBD | TBD |
+| 2 | 2.6.3 | Supported | Dec 17, 2020 | TBD | TBD |
| 1.10 | 1.10.15 | EOL | Aug 27, 2018 | Dec 17, 2020 | June 17, 2021 |
| 1.9 | 1.9.0 | EOL | Jan 03, 2018 | Aug 27, 2018 | Aug 27, 2018 |
| 1.8 | 1.8.2 | EOL | Mar 19, 2017 | Jan 03, 2018 | Jan 03, 2018 |
@@ -305,7 +305,7 @@ They are based on the official release schedule of Python and Kubernetes, nicely
2. The "oldest" supported version of Python/Kubernetes is the default one until we decide to switch to
later version. "Default" is only meaningful in terms of "smoke tests" in CI PRs, which are run using this
default version and the default reference image available. Currently `apache/airflow:latest`
- and `apache/airflow:2.6.2` images are Python 3.7 images. This means that default reference image will
+ and `apache/airflow:2.6.3` images are Python 3.7 images. This means that default reference image will
become the default at the time when we start preparing for dropping 3.7 support which is few months
before the end of life for Python 3.7.
diff --git a/airflow/__init__.py b/airflow/__init__.py
index 733e906573..3f91e9f327 100644
--- a/airflow/__init__.py
+++ b/airflow/__init__.py
@@ -26,7 +26,7 @@ isort:skip_file
"""
from __future__ import annotations
-__version__ = "2.6.2"
+__version__ = "2.6.3"
# flake8: noqa: F401
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
index 6ad9e84056..a91a6beee7 100644
--- a/airflow/api_connexion/openapi/v1.yaml
+++ b/airflow/api_connexion/openapi/v1.yaml
@@ -231,7 +231,7 @@ info:
This means that the server encountered an unexpected condition that prevented it from
fulfilling the request.
- version: '2.6.2'
+ version: '2.6.3'
license:
name: Apache 2.0
url: http://www.apache.org/licenses/LICENSE-2.0.html
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index ad505f33ad..ab79d51056 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -82,6 +82,7 @@ REVISION_HEADS_MAP = {
"2.6.0": "98ae134e6fff",
"2.6.1": "98ae134e6fff",
"2.6.2": "c804e5c76e3e",
+ "2.6.3": "c804e5c76e3e",
}
diff --git a/docs/apache-airflow/installation/supported-versions.rst b/docs/apache-airflow/installation/supported-versions.rst
index e4fcb7c603..9ff4b68292 100644
--- a/docs/apache-airflow/installation/supported-versions.rst
+++ b/docs/apache-airflow/installation/supported-versions.rst
@@ -29,7 +29,7 @@ Apache Airflow™ version life cycle:
========= ===================== ========= =============== ================= ================
Version Current Patch/Minor State First Release Limited Support EOL/Terminated
========= ===================== ========= =============== ================= ================
-2 2.6.2 Supported Dec 17, 2020 TBD TBD
+2 2.6.3 Supported Dec 17, 2020 TBD TBD
1.10 1.10.15 EOL Aug 27, 2018 Dec 17, 2020 June 17, 2021
1.9 1.9.0 EOL Jan 03, 2018 Aug 27, 2018 Aug 27, 2018
1.8 1.8.2 EOL Mar 19, 2017 Jan 03, 2018 Jan 03, 2018
diff --git a/docs/docker-stack/README.md b/docs/docker-stack/README.md
index 151c420144..b90d0ced99 100644
--- a/docs/docker-stack/README.md
+++ b/docs/docker-stack/README.md
@@ -31,12 +31,12 @@ Every time a new version of Airflow is released, the images are prepared in the
[apache/airflow DockerHub](https://hub.docker.com/r/apache/airflow)
for all the supported Python versions.
-You can find the following images there (Assuming Airflow version `2.6.2`):
+You can find the following images there (Assuming Airflow version `2.6.3`):
* `apache/airflow:latest` - the latest released Airflow image with default Python version (3.7 currently)
* `apache/airflow:latest-pythonX.Y` - the latest released Airflow image with specific Python version
-* `apache/airflow:2.6.2` - the versioned Airflow image with default Python version (3.7 currently)
-* `apache/airflow:2.6.2-pythonX.Y` - the versioned Airflow image with specific Python version
+* `apache/airflow:2.6.3` - the versioned Airflow image with default Python version (3.7 currently)
+* `apache/airflow:2.6.3-pythonX.Y` - the versioned Airflow image with specific Python version
Those are "reference" regular images. They contain the most common set of extras, dependencies and providers that are
often used by the users and they are good to "try-things-out" when you want to just take Airflow for a spin,
@@ -47,8 +47,8 @@ via [Building the image](https://airflow.apache.org/docs/docker-stack/build.html
* `apache/airflow:slim-latest` - the latest released Airflow image with default Python version (3.7 currently)
* `apache/airflow:slim-latest-pythonX.Y` - the latest released Airflow image with specific Python version
-* `apache/airflow:slim-2.6.2` - the versioned Airflow image with default Python version (3.7 currently)
-* `apache/airflow:slim-2.6.2-pythonX.Y` - the versioned Airflow image with specific Python version
+* `apache/airflow:slim-2.6.3` - the versioned Airflow image with default Python version (3.7 currently)
+* `apache/airflow:slim-2.6.3-pythonX.Y` - the versioned Airflow image with specific Python version
The Apache Airflow image provided as convenience package is optimized for size, and
it provides just a bare minimal set of the extras and dependencies installed and in most cases
diff --git a/docs/docker-stack/docker-examples/extending/add-airflow-configuration/Dockerfile b/docs/docker-stack/docker-examples/extending/add-airflow-configuration/Dockerfile
index 65a4e308ce..5f04cbeb6c 100644
--- a/docs/docker-stack/docker-examples/extending/add-airflow-configuration/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/add-airflow-configuration/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.7.0.dev0
+FROM apache/airflow:2.6.3
ENV AIRFLOW__CORE__LOAD_EXAMPLES=True
ENV AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=my_conn_string
# [END Dockerfile]
diff --git a/docs/docker-stack/docker-examples/extending/add-apt-packages/Dockerfile b/docs/docker-stack/docker-examples/extending/add-apt-packages/Dockerfile
index 5167360207..ba712c5b69 100644
--- a/docs/docker-stack/docker-examples/extending/add-apt-packages/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/add-apt-packages/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
USER root
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
diff --git a/docs/docker-stack/docker-examples/extending/add-build-essential-extend/Dockerfile b/docs/docker-stack/docker-examples/extending/add-build-essential-extend/Dockerfile
index 4d685bdb03..3d82143574 100644
--- a/docs/docker-stack/docker-examples/extending/add-build-essential-extend/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/add-build-essential-extend/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
USER root
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
diff --git a/docs/docker-stack/docker-examples/extending/add-providers/Dockerfile b/docs/docker-stack/docker-examples/extending/add-providers/Dockerfile
index 16571468d9..7cc7f4eee7 100644
--- a/docs/docker-stack/docker-examples/extending/add-providers/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/add-providers/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
USER root
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
diff --git a/docs/docker-stack/docker-examples/extending/add-pypi-packages/Dockerfile b/docs/docker-stack/docker-examples/extending/add-pypi-packages/Dockerfile
index 2c9ea8df80..6aea5d23b7 100644
--- a/docs/docker-stack/docker-examples/extending/add-pypi-packages/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/add-pypi-packages/Dockerfile
@@ -15,6 +15,6 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" lxml
# [END Dockerfile]
diff --git a/docs/docker-stack/docker-examples/extending/add-requirement-packages/Dockerfile b/docs/docker-stack/docker-examples/extending/add-requirement-packages/Dockerfile
index 8500fdaf72..6e096bf7bd 100644
--- a/docs/docker-stack/docker-examples/extending/add-requirement-packages/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/add-requirement-packages/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
COPY requirements.txt /
RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" -r /requirements.txt
# [END Dockerfile]
diff --git a/docs/docker-stack/docker-examples/extending/custom-providers/Dockerfile b/docs/docker-stack/docker-examples/extending/custom-providers/Dockerfile
index 49f77e441d..85fdd71521 100644
--- a/docs/docker-stack/docker-examples/extending/custom-providers/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/custom-providers/Dockerfile
@@ -15,6 +15,6 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
RUN pip install "apache-airflow==${AIRFLOW_VERSION}" --no-cache-dir apache-airflow-providers-docker==2.5.1
# [END Dockerfile]
diff --git a/docs/docker-stack/docker-examples/extending/embedding-dags/Dockerfile b/docs/docker-stack/docker-examples/extending/embedding-dags/Dockerfile
index c463c0eae3..266c9271f0 100644
--- a/docs/docker-stack/docker-examples/extending/embedding-dags/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/embedding-dags/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
COPY --chown=airflow:root test_dag.py /opt/airflow/dags
diff --git a/docs/docker-stack/docker-examples/extending/writable-directory/Dockerfile b/docs/docker-stack/docker-examples/extending/writable-directory/Dockerfile
index 96e87ecbe2..dd190761ea 100644
--- a/docs/docker-stack/docker-examples/extending/writable-directory/Dockerfile
+++ b/docs/docker-stack/docker-examples/extending/writable-directory/Dockerfile
@@ -15,7 +15,7 @@
# This is an example Dockerfile. It is not intended for PRODUCTION use
# [START Dockerfile]
-FROM apache/airflow:2.6.2
+FROM apache/airflow:2.6.3
RUN umask 0002; \
mkdir -p ~/writeable-directory
# [END Dockerfile]
diff --git a/docs/docker-stack/entrypoint.rst b/docs/docker-stack/entrypoint.rst
index 6395dfb62a..9c56d85b43 100644
--- a/docs/docker-stack/entrypoint.rst
+++ b/docs/docker-stack/entrypoint.rst
@@ -132,7 +132,7 @@ if you specify extra arguments. For example:
.. code-block:: bash
- docker run -it apache/airflow:2.6.2-python3.7 bash -c "ls -la"
+ docker run -it apache/airflow:2.6.3-python3.7 bash -c "ls -la"
total 16
drwxr-xr-x 4 airflow root 4096 Jun 5 18:12 .
drwxr-xr-x 1 root root 4096 Jun 5 18:12 ..
@@ -144,7 +144,7 @@ you pass extra parameters. For example:
.. code-block:: bash
- > docker run -it apache/airflow:2.6.2-python3.7 python -c "print('test')"
+ > docker run -it apache/airflow:2.6.3-python3.7 python -c "print('test')"
test
If first argument equals to "airflow" - the rest of the arguments is treated as an airflow command
@@ -152,13 +152,13 @@ to execute. Example:
.. code-block:: bash
- docker run -it apache/airflow:2.6.2-python3.7 airflow webserver
+ docker run -it apache/airflow:2.6.3-python3.7 airflow webserver
If there are any other arguments - they are simply passed to the "airflow" command
.. code-block:: bash
- > docker run -it apache/airflow:2.6.2-python3.7 help
+ > docker run -it apache/airflow:2.6.3-python3.7 help
usage: airflow [-h] GROUP_OR_COMMAND ...
positional arguments:
@@ -363,7 +363,7 @@ database and creating an ``admin/admin`` Admin user with the following command:
--env "_AIRFLOW_DB_UPGRADE=true" \
--env "_AIRFLOW_WWW_USER_CREATE=true" \
--env "_AIRFLOW_WWW_USER_PASSWORD=admin" \
- apache/airflow:2.6.2-python3.8 webserver
+ apache/airflow:2.6.3-python3.8 webserver
.. code-block:: bash
@@ -372,7 +372,7 @@ database and creating an ``admin/admin`` Admin user with the following command:
--env "_AIRFLOW_DB_UPGRADE=true" \
--env "_AIRFLOW_WWW_USER_CREATE=true" \
--env "_AIRFLOW_WWW_USER_PASSWORD_CMD=echo admin" \
- apache/airflow:2.6.2-python3.8 webserver
+ apache/airflow:2.6.3-python3.8 webserver
The commands above perform initialization of the SQLite database, create admin user with admin password
and Admin role. They also forward local port ``8080`` to the webserver port and finally start the webserver.
@@ -412,6 +412,6 @@ Example:
--env "_AIRFLOW_DB_UPGRADE=true" \
--env "_AIRFLOW_WWW_USER_CREATE=true" \
--env "_AIRFLOW_WWW_USER_PASSWORD_CMD=echo admin" \
- apache/airflow:2.6.2-python3.8 webserver
+ apache/airflow:2.6.3-python3.8 webserver
This method is only available starting from Docker image of Airflow 2.1.1 and above.
diff --git a/scripts/ci/pre_commit/pre_commit_supported_versions.py b/scripts/ci/pre_commit/pre_commit_supported_versions.py
index b4d38098bb..c31f1222a2 100755
--- a/scripts/ci/pre_commit/pre_commit_supported_versions.py
+++ b/scripts/ci/pre_commit/pre_commit_supported_versions.py
@@ -27,7 +27,7 @@ AIRFLOW_SOURCES = Path(__file__).resolve().parent.parent.parent.parent
HEADERS = ("Version", "Current Patch/Minor", "State", "First Release", "Limited Support", "EOL/Terminated")
SUPPORTED_VERSIONS = (
- ("2", "2.6.2", "Supported", "Dec 17, 2020", "TBD", "TBD"),
+ ("2", "2.6.3", "Supported", "Dec 17, 2020", "TBD", "TBD"),
("1.10", "1.10.15", "EOL", "Aug 27, 2018", "Dec 17, 2020", "June 17, 2021"),
("1.9", "1.9.0", "EOL", "Jan 03, 2018", "Aug 27, 2018", "Aug 27, 2018"),
("1.8", "1.8.2", "EOL", "Mar 19, 2017", "Jan 03, 2018", "Jan 03, 2018"),
[airflow] 12/42: Ensure that main triggerer thread exits if the async thread fails (#32092)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d6cc9e4bb1efe9713eccd8e62e46f11bad294a36
Author: tomrutter <60...@users.noreply.github.com>
AuthorDate: Tue Jun 27 23:46:16 2023 +0200
Ensure that main triggerer thread exits if the async thread fails (#32092)
Co-authored-by: tom.rutter <to...@zurich.com>
(cherry picked from commit e585b588fc49b1b1c73a8952e9b257d7a9e13314)
---
airflow/jobs/triggerer_job_runner.py | 29 +++++++++++++++---------
tests/jobs/test_triggerer_job.py | 43 ++++++++++++++++++++++++++++++++++++
2 files changed, 61 insertions(+), 11 deletions(-)
diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py
index 0d56244709..7051dec5a0 100644
--- a/airflow/jobs/triggerer_job_runner.py
+++ b/airflow/jobs/triggerer_job_runner.py
@@ -350,6 +350,9 @@ class TriggererJobRunner(BaseJobRunner["Job | JobPydantic"], LoggingMixin):
This runs synchronously and handles all database reads/writes.
"""
while not self.trigger_runner.stop:
+ if not self.trigger_runner.is_alive():
+ self.log.error("Trigger runner thread has died! Exiting.")
+ break
# Clean out unused triggers
Trigger.clean_unused()
# Load/delete triggers
@@ -466,17 +469,21 @@ class TriggerRunner(threading.Thread, LoggingMixin):
watchdog = asyncio.create_task(self.block_watchdog())
last_status = time.time()
while not self.stop:
- # Run core logic
- await self.create_triggers()
- await self.cancel_triggers()
- await self.cleanup_finished_triggers()
- # Sleep for a bit
- await asyncio.sleep(1)
- # Every minute, log status
- if time.time() - last_status >= 60:
- count = len(self.triggers)
- self.log.info("%i triggers currently running", count)
- last_status = time.time()
+ try:
+ # Run core logic
+ await self.create_triggers()
+ await self.cancel_triggers()
+ await self.cleanup_finished_triggers()
+ # Sleep for a bit
+ await asyncio.sleep(1)
+ # Every minute, log status
+ if time.time() - last_status >= 60:
+ count = len(self.triggers)
+ self.log.info("%i triggers currently running", count)
+ last_status = time.time()
+ except Exception:
+ self.stop = True
+ raise
# Wait for watchdog to complete
await watchdog
diff --git a/tests/jobs/test_triggerer_job.py b/tests/jobs/test_triggerer_job.py
index 0e19927f17..dd35dd4cc8 100644
--- a/tests/jobs/test_triggerer_job.py
+++ b/tests/jobs/test_triggerer_job.py
@@ -412,6 +412,49 @@ def test_trigger_from_expired_triggerer(session):
assert [x for x, y in job_runner.trigger_runner.to_create] == [1]
+def test_trigger_runner_exception_stops_triggerer(session):
+ """
+ Checks that if an exception occurs when creating triggers, that the triggerer
+ process stops
+ """
+
+ class MockTriggerException(Exception):
+ pass
+
+ class TriggerRunner_(TriggerRunner):
+ async def create_triggers(self):
+ raise MockTriggerException("Trigger creation failed")
+
+ # Use a trigger that will immediately succeed
+ trigger = SuccessTrigger()
+ create_trigger_in_db(session, trigger)
+
+ # Make a TriggererJobRunner and have it retrieve DB tasks
+ job = Job()
+ job_runner = TriggererJobRunner(job)
+ job_runner.trigger_runner = TriggerRunner_()
+ thread = Thread(target=job_runner._execute)
+ thread.start()
+
+ # Wait 4 seconds for the triggerer to stop
+ try:
+ for _ in range(40):
+ time.sleep(0.1)
+ if not thread.is_alive():
+ break
+ else:
+ pytest.fail("TriggererJobRunner did not stop after exception in TriggerRunner")
+
+ if not job_runner.trigger_runner.stop:
+ pytest.fail("TriggerRunner not marked as stopped after exception in TriggerRunner")
+
+ finally:
+ job_runner.trigger_runner.stop = True
+ # with suppress(MockTriggerException):
+ job_runner.trigger_runner.join()
+ thread.join()
+
+
def test_trigger_firing(session):
"""
Checks that when a trigger fires, it correctly makes it into the
[airflow] 42/42: Update trigger url to fix test
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ccee9c5ec1134cedeb3cdd3fc3f945eb78b62d88
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu Jul 6 23:19:24 2023 +0100
Update trigger url to fix test
---
tests/www/views/test_views_trigger_dag.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tests/www/views/test_views_trigger_dag.py b/tests/www/views/test_views_trigger_dag.py
index b6fcd20d3e..a07f73a1d9 100644
--- a/tests/www/views/test_views_trigger_dag.py
+++ b/tests/www/views/test_views_trigger_dag.py
@@ -309,7 +309,7 @@ def test_trigger_dag_params_array_value_none_render(admin_client, dag_maker, ses
def test_dag_run_id_pattern(session, admin_client, pattern, run_id, result):
with conf_vars({("scheduler", "allowed_run_id_pattern"): pattern}):
test_dag_id = "example_bash_operator"
- admin_client.post(f"dags/{test_dag_id}/trigger?&run_id={run_id}")
+ admin_client.post(f"trigger?dag_id={test_dag_id}&run_id={run_id}")
run = session.query(DagRun).filter(DagRun.dag_id == test_dag_id).first()
if result:
assert run is not None
[airflow] 09/42: Render list items in rendered fields view (#32042)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e0e8d71b0e02cc55cbd2d170dc287d0429903a25
Author: Clemens <cl...@gmail.com>
AuthorDate: Tue Jul 4 05:47:27 2023 +0800
Render list items in rendered fields view (#32042)
---------
Co-authored-by: clemens.valiente <cl...@grabtaxi.com>
(cherry picked from commit 2b5d4e319b82267349aa20cb3230f3b2ad35adce)
---
airflow/www/utils.py | 32 ++++++++++----
airflow/www/views.py | 6 +--
tests/www/views/test_views_rendered.py | 78 +++++++++++++++++++++++++++++++++-
3 files changed, 101 insertions(+), 15 deletions(-)
diff --git a/airflow/www/utils.py b/airflow/www/utils.py
index 0f17984afa..8c382abc61 100644
--- a/airflow/www/utils.py
+++ b/airflow/www/utils.py
@@ -20,7 +20,7 @@ from __future__ import annotations
import json
import textwrap
import time
-from typing import TYPE_CHECKING, Any, Sequence
+from typing import TYPE_CHECKING, Any, Callable, Sequence
from urllib.parse import urlencode
from flask import request, url_for
@@ -36,6 +36,7 @@ from markupsafe import Markup
from pendulum.datetime import DateTime
from pygments import highlight, lexers
from pygments.formatters import HtmlFormatter
+from pygments.lexer import Lexer
from sqlalchemy import func, types
from sqlalchemy.ext.associationproxy import AssociationProxy
@@ -546,20 +547,35 @@ def pygment_html_render(s, lexer=lexers.TextLexer):
return highlight(s, lexer(), HtmlFormatter(linenos=True))
-def render(obj, lexer):
- """Render a given Python object with a given Pygments lexer"""
- out = ""
+def render(obj: Any, lexer: Lexer, handler: Callable[[Any], str] | None = None):
+ """Render a given Python object with a given Pygments lexer."""
if isinstance(obj, str):
- out = Markup(pygment_html_render(obj, lexer))
+ return Markup(pygment_html_render(obj, lexer))
+
elif isinstance(obj, (tuple, list)):
+ out = ""
for i, text_to_render in enumerate(obj):
+ if lexer is lexers.PythonLexer:
+ text_to_render = repr(text_to_render)
out += Markup("<div>List item #{}</div>").format(i)
out += Markup("<div>" + pygment_html_render(text_to_render, lexer) + "</div>")
+ return out
+
elif isinstance(obj, dict):
+ out = ""
for k, v in obj.items():
+ if lexer is lexers.PythonLexer:
+ v = repr(v)
out += Markup('<div>Dict item "{}"</div>').format(k)
out += Markup("<div>" + pygment_html_render(v, lexer) + "</div>")
- return out
+ return out
+
+ elif handler is not None and obj is not None:
+ return Markup(pygment_html_render(handler(obj), lexer))
+
+ else:
+ # Return empty string otherwise
+ return ""
def json_render(obj, lexer):
@@ -600,8 +616,8 @@ def get_attr_renderer():
"mysql": lambda x: render(x, lexers.MySqlLexer),
"postgresql": lambda x: render(x, lexers.PostgresLexer),
"powershell": lambda x: render(x, lexers.PowerShellLexer),
- "py": lambda x: render(get_python_source(x), lexers.PythonLexer),
- "python_callable": lambda x: render(get_python_source(x), lexers.PythonLexer),
+ "py": lambda x: render(x, lexers.PythonLexer, get_python_source),
+ "python_callable": lambda x: render(x, lexers.PythonLexer, get_python_source),
"rst": lambda x: render(x, lexers.RstLexer),
"sql": lambda x: render(x, lexers.SqlLexer),
"tsql": lambda x: render(x, lexers.TransactSqlLexer),
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 8eaac3eec5..37b756d332 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1393,11 +1393,7 @@ class Airflow(AirflowBaseView):
content = getattr(task, template_field)
renderer = task.template_fields_renderers.get(template_field, template_field)
if renderer in renderers:
- if isinstance(content, (dict, list)):
- json_content = json.dumps(content, sort_keys=True, indent=4)
- html_dict[template_field] = renderers[renderer](json_content)
- else:
- html_dict[template_field] = renderers[renderer](content)
+ html_dict[template_field] = renderers[renderer](content)
else:
html_dict[template_field] = Markup("<pre><code>{}</pre></code>").format(pformat(content))
diff --git a/tests/www/views/test_views_rendered.py b/tests/www/views/test_views_rendered.py
index 6a2cb9717f..1557a083d5 100644
--- a/tests/www/views/test_views_rendered.py
+++ b/tests/www/views/test_views_rendered.py
@@ -24,7 +24,9 @@ import pytest
from markupsafe import escape
from airflow.models import DAG, RenderedTaskInstanceFields, Variable
+from airflow.models.baseoperator import BaseOperator
from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
from airflow.serialization.serialized_objects import SerializedDAG
from airflow.utils import timezone
from airflow.utils.session import create_session
@@ -64,6 +66,39 @@ def task2(dag):
)
+@pytest.fixture()
+def task3(dag):
+ class TestOperator(BaseOperator):
+ template_fields = ("sql",)
+
+ def __init__(self, *, sql, **kwargs):
+ super().__init__(**kwargs)
+ self.sql = sql
+
+ def execute(self, context):
+ pass
+
+ return TestOperator(
+ task_id="task3",
+ sql=["SELECT 1;", "SELECT 2;"],
+ dag=dag,
+ )
+
+
+@pytest.fixture()
+def task4(dag):
+ def func(*op_args):
+ pass
+
+ return PythonOperator(
+ task_id="task4",
+ python_callable=func,
+ op_args=["{{ task_instance_key_str }}_args"],
+ op_kwargs={"0": "{{ task_instance_key_str }}_kwargs"},
+ dag=dag,
+ )
+
+
@pytest.fixture()
def task_secret(dag):
return BashOperator(
@@ -85,7 +120,7 @@ def init_blank_db():
@pytest.fixture(autouse=True)
-def reset_db(dag, task1, task2, task_secret):
+def reset_db(dag, task1, task2, task3, task4, task_secret):
yield
clear_db_dags()
clear_db_runs()
@@ -93,7 +128,7 @@ def reset_db(dag, task1, task2, task_secret):
@pytest.fixture()
-def create_dag_run(dag, task1, task2, task_secret):
+def create_dag_run(dag, task1, task2, task3, task4, task_secret):
def _create_dag_run(*, execution_date, session):
dag_run = dag.create_dagrun(
state=DagRunState.RUNNING,
@@ -108,6 +143,10 @@ def create_dag_run(dag, task1, task2, task_secret):
ti2.state = TaskInstanceState.SCHEDULED
ti3 = dag_run.get_task_instance(task_secret.task_id, session=session)
ti3.state = TaskInstanceState.QUEUED
+ ti4 = dag_run.get_task_instance(task3.task_id, session=session)
+ ti4.state = TaskInstanceState.SUCCESS
+ ti5 = dag_run.get_task_instance(task4.task_id, session=session)
+ ti5.state = TaskInstanceState.SUCCESS
session.flush()
return dag_run
@@ -290,3 +329,38 @@ def test_rendered_task_detail_env_secret(patch_app, admin_client, request, env,
if request.node.callspec.id.endswith("-tpld-var"):
Variable.delete("plain_var")
Variable.delete("secret_var")
+
+
+@pytest.mark.usefixtures("patch_app")
+def test_rendered_template_view_for_list_template_field_args(admin_client, create_dag_run, task3):
+ """
+ Test that the Rendered View can show a list of syntax-highlighted SQL statements
+ """
+ assert task3.sql == ["SELECT 1;", "SELECT 2;"]
+
+ with create_session() as session:
+ create_dag_run(execution_date=DEFAULT_DATE, session=session)
+
+ url = f"rendered-templates?task_id=task3&dag_id=testdag&execution_date={quote_plus(str(DEFAULT_DATE))}"
+
+ resp = admin_client.get(url, follow_redirects=True)
+ check_content_in_response("List item #0", resp)
+ check_content_in_response("List item #1", resp)
+
+
+@pytest.mark.usefixtures("patch_app")
+def test_rendered_template_view_for_op_args(admin_client, create_dag_run, task4):
+ """
+ Test that the Rendered View can show rendered values in op_args and op_kwargs
+ """
+ assert task4.op_args == ["{{ task_instance_key_str }}_args"]
+ assert list(task4.op_kwargs.values()) == ["{{ task_instance_key_str }}_kwargs"]
+
+ with create_session() as session:
+ create_dag_run(execution_date=DEFAULT_DATE, session=session)
+
+ url = f"rendered-templates?task_id=task4&dag_id=testdag&execution_date={quote_plus(str(DEFAULT_DATE))}"
+
+ resp = admin_client.get(url, follow_redirects=True)
+ check_content_in_response("testdag__task4__20200301_args", resp)
+ check_content_in_response("testdag__task4__20200301_kwargs", resp)
[airflow] 16/42: Use mapIndex to display extra links per mapped task. (#32154)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 028af0cafc75eb6a1e37e4eceddbe62686adc39f
Author: Karthikeyan Singaravelan <ti...@gmail.com>
AuthorDate: Wed Jun 28 20:52:09 2023 +0530
Use mapIndex to display extra links per mapped task. (#32154)
(cherry picked from commit 98c47f48e1b292d535d39cc3349660aa736d76cd)
---
airflow/www/static/js/api/useExtraLinks.ts | 60 ++++++++++++----------
.../js/dag/details/taskInstance/ExtraLinks.tsx | 10 +++-
.../static/js/dag/details/taskInstance/index.tsx | 9 ++++
3 files changed, 51 insertions(+), 28 deletions(-)
diff --git a/airflow/www/static/js/api/useExtraLinks.ts b/airflow/www/static/js/api/useExtraLinks.ts
index a76cb2540c..302d58aa53 100644
--- a/airflow/www/static/js/api/useExtraLinks.ts
+++ b/airflow/www/static/js/api/useExtraLinks.ts
@@ -32,38 +32,44 @@ export default function useExtraLinks({
dagId,
taskId,
executionDate,
+ mapIndex,
extraLinks,
}: {
dagId: string;
taskId: string;
executionDate: string;
+ mapIndex?: number | undefined;
extraLinks: string[];
}) {
- return useQuery(["extraLinks", dagId, taskId, executionDate], async () => {
- const data = await Promise.all(
- extraLinks.map(async (link) => {
- const url = `${extraLinksUrl}?task_id=${encodeURIComponent(
- taskId
- )}&dag_id=${encodeURIComponent(
- dagId
- )}&execution_date=${encodeURIComponent(
- executionDate
- )}&link_name=${encodeURIComponent(link)}`;
- try {
- const datum = await axios.get<AxiosResponse, LinkData>(url);
- return {
- name: link,
- url: datum.url,
- };
- } catch (e) {
- console.error(e);
- return {
- name: link,
- url: "",
- };
- }
- })
- );
- return data;
- });
+ return useQuery(
+ ["extraLinks", dagId, taskId, executionDate, mapIndex],
+ async () => {
+ const data = await Promise.all(
+ extraLinks.map(async (link) => {
+ mapIndex ??= -1;
+ const url = `${extraLinksUrl}?task_id=${encodeURIComponent(
+ taskId
+ )}&dag_id=${encodeURIComponent(
+ dagId
+ )}&execution_date=${encodeURIComponent(
+ executionDate
+ )}&link_name=${encodeURIComponent(link)}&map_index=${mapIndex}`;
+ try {
+ const datum = await axios.get<AxiosResponse, LinkData>(url);
+ return {
+ name: link,
+ url: datum.url,
+ };
+ } catch (e) {
+ console.error(e);
+ return {
+ name: link,
+ url: "",
+ };
+ }
+ })
+ );
+ return data;
+ }
+ );
}
diff --git a/airflow/www/static/js/dag/details/taskInstance/ExtraLinks.tsx b/airflow/www/static/js/dag/details/taskInstance/ExtraLinks.tsx
index a5e2a37539..1121459b4b 100644
--- a/airflow/www/static/js/dag/details/taskInstance/ExtraLinks.tsx
+++ b/airflow/www/static/js/dag/details/taskInstance/ExtraLinks.tsx
@@ -26,14 +26,22 @@ interface Props {
dagId: string;
taskId: string;
executionDate: string;
+ mapIndex?: number | undefined;
extraLinks: string[];
}
-const ExtraLinks = ({ dagId, taskId, executionDate, extraLinks }: Props) => {
+const ExtraLinks = ({
+ dagId,
+ taskId,
+ executionDate,
+ mapIndex,
+ extraLinks,
+}: Props) => {
const { data: links } = useExtraLinks({
dagId,
taskId,
executionDate,
+ mapIndex,
extraLinks,
});
diff --git a/airflow/www/static/js/dag/details/taskInstance/index.tsx b/airflow/www/static/js/dag/details/taskInstance/index.tsx
index e22e346610..204e900ea8 100644
--- a/airflow/www/static/js/dag/details/taskInstance/index.tsx
+++ b/airflow/www/static/js/dag/details/taskInstance/index.tsx
@@ -99,6 +99,15 @@ const TaskInstance = ({ taskId, runId, mapIndex }: Props) => {
key={dagId + runId + taskId + instance.mapIndex}
/>
)}
+ {isMapped && group.extraLinks && isMapIndexDefined && (
+ <ExtraLinks
+ taskId={taskId}
+ dagId={dagId}
+ mapIndex={mapIndex}
+ executionDate={executionDate}
+ extraLinks={group?.extraLinks}
+ />
+ )}
{!isMapped && group.extraLinks && (
<ExtraLinks
taskId={taskId}
[airflow] 36/42: Catch the exception that triggerer initialization failed (#31999)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f10204b59fc1e2927f558b4fbc79cbbef59cc8fa
Author: Wei Lee <we...@gmail.com>
AuthorDate: Wed Jun 21 02:55:04 2023 +0800
Catch the exception that triggerer initialization failed (#31999)
* fix(jobs/triggerer_job_runner): catch the exception that triggerer initialization failed
* test(jobs/triggerer_job_runner): add test case for checking TriggerRunner.update_trigger catch triggerer change exception
* refactor(jobs/triggerer_job_runner): catch multiple exception cases in the same try catch block when initialing trigger
* refactor(jobs/triggerer_job_runner): remove unnecessary TypeError catching
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
* Revert "refactor(jobs/triggerer_job_runner): remove unnecessary TypeError catching"
This reverts commit ac561a2416f16a8f9a3b211715fd4a7e012b47ad.
* Revert "refactor(jobs/triggerer_job_runner): catch multiple exception cases in the same try catch block when initialing trigger"
This reverts commit 4f3b24b8a7698973fe319618b6ff9f459dc0155a.
---------
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
(cherry picked from commit a1ba15570219d2fe77466367e84fafa92cbdb24e)
---
airflow/jobs/triggerer_job_runner.py | 9 ++++++++-
tests/jobs/test_triggerer_job.py | 22 +++++++++++++++++++++-
2 files changed, 29 insertions(+), 2 deletions(-)
diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py
index 7ca3fc2e94..8cbc17b228 100644
--- a/airflow/jobs/triggerer_job_runner.py
+++ b/airflow/jobs/triggerer_job_runner.py
@@ -688,7 +688,14 @@ class TriggerRunner(threading.Thread, LoggingMixin):
# Either the trigger code or the path to it is bad. Fail the trigger.
self.failed_triggers.append((new_id, e))
continue
- new_trigger_instance = trigger_class(**new_trigger_orm.kwargs)
+
+ try:
+ new_trigger_instance = trigger_class(**new_trigger_orm.kwargs)
+ except TypeError as err:
+ self.log.error("Trigger failed; message=%s", err)
+ self.failed_triggers.append((new_id, err))
+ continue
+
self.set_trigger_logging_metadata(new_trigger_orm.task_instance, new_id, new_trigger_instance)
self.to_create.append((new_id, new_trigger_instance))
# Enqueue orphaned triggers for cancellation
diff --git a/tests/jobs/test_triggerer_job.py b/tests/jobs/test_triggerer_job.py
index d0ce887a57..3e43d8f520 100644
--- a/tests/jobs/test_triggerer_job.py
+++ b/tests/jobs/test_triggerer_job.py
@@ -37,7 +37,7 @@ from airflow.models.baseoperator import BaseOperator
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.triggers.base import TriggerEvent
-from airflow.triggers.temporal import TimeDeltaTrigger
+from airflow.triggers.temporal import DateTimeTrigger, TimeDeltaTrigger
from airflow.triggers.testing import FailureTrigger, SuccessTrigger
from airflow.utils import timezone
from airflow.utils.log.logging_mixin import RedirectStdHandler
@@ -281,6 +281,26 @@ class TestTriggerRunner:
await trigger_runner.run_trigger(1, mock_trigger)
assert "Trigger cancelled due to timeout" in caplog.text
+ @patch("airflow.models.trigger.Trigger.bulk_fetch")
+ @patch(
+ "airflow.jobs.triggerer_job_runner.TriggerRunner.get_trigger_by_classpath",
+ return_value=DateTimeTrigger,
+ )
+ def test_update_trigger_with_triggerer_argument_change(
+ self, mock_bulk_fetch, mock_get_trigger_by_classpath, session, caplog
+ ) -> None:
+ trigger_runner = TriggerRunner()
+ mock_trigger_orm = MagicMock()
+ mock_trigger_orm.kwargs = {"moment": ..., "not_exists_arg": ...}
+ mock_get_trigger_by_classpath.return_value = {1: mock_trigger_orm}
+
+ trigger_runner.update_triggers({1})
+
+ assert (
+ "Trigger failed; message=__init__() got an unexpected keyword argument 'not_exists_arg'"
+ in caplog.text
+ )
+
def test_trigger_create_race_condition_18392(session, tmp_path):
"""
[airflow] 34/42: Fix ts/eslint warnings (#32321)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 04a3a0ab27091909cfaf52b6a59b3b47bb4af268
Author: Pierre Jeambrun <pi...@gmail.com>
AuthorDate: Mon Jul 3 09:42:26 2023 +0200
Fix ts/eslint warnings (#32321)
(cherry picked from commit b6eeccdfb67409f5035adc3ab8af97c4c03a40dc)
---
airflow/www/alias-rest-types.js | 1 +
airflow/www/babel.config.js | 2 +-
airflow/www/jest-setup.js | 2 ++
airflow/www/package.json | 2 +-
airflow/www/static/js/api/useDatasets.ts | 2 +-
airflow/www/static/js/api/useExtraLinks.ts | 7 +++++--
airflow/www/static/js/callModal.js | 3 +++
airflow/www/static/js/components/Table/Cells.tsx | 2 ++
airflow/www/static/js/components/Table/Table.test.tsx | 2 +-
airflow/www/static/js/components/Table/index.tsx | 6 ++++--
airflow/www/static/js/components/Tooltip.tsx | 3 ++-
airflow/www/static/js/connection_form.js | 4 ++++
airflow/www/static/js/dag/details/graph/utils.ts | 2 +-
airflow/www/static/js/dag/grid/dagRuns/index.test.tsx | 1 +
airflow/www/static/js/dag/grid/index.test.tsx | 1 +
airflow/www/static/js/dag_dependencies.js | 2 ++
airflow/www/static/js/dags.js | 1 +
airflow/www/static/js/datasets/List.test.tsx | 12 +++++++++---
airflow/www/static/js/datasets/List.tsx | 2 ++
airflow/www/static/js/datetime_utils.js | 1 +
airflow/www/static/js/gantt.js | 13 ++++++++++++-
airflow/www/static/js/graph.js | 1 +
airflow/www/static/js/main.js | 19 ++++++++++---------
airflow/www/static/js/task.js | 2 ++
airflow/www/static/js/task_instances.js | 2 ++
airflow/www/static/js/ti_log.js | 9 ++++++---
airflow/www/static/js/types/react-table-config.d.ts | 4 ++--
airflow/www/static/js/utils/URLSearchParamWrapper.ts | 3 ++-
airflow/www/static/js/utils/graph.ts | 14 +++++++++++---
29 files changed, 93 insertions(+), 32 deletions(-)
diff --git a/airflow/www/alias-rest-types.js b/airflow/www/alias-rest-types.js
index 83ddd89aae..c2767acca6 100644
--- a/airflow/www/alias-rest-types.js
+++ b/airflow/www/alias-rest-types.js
@@ -225,6 +225,7 @@ function generate(file) {
fs.writeFileSync(file, finalText, (err) => {
if (err) {
+ // eslint-disable-next-line no-console
console.error(err);
}
});
diff --git a/airflow/www/babel.config.js b/airflow/www/babel.config.js
index f38c60b91f..7f40a9fdf6 100644
--- a/airflow/www/babel.config.js
+++ b/airflow/www/babel.config.js
@@ -17,7 +17,7 @@
* under the License.
*/
-module.exports = function (api) {
+module.exports = (api) => {
api.cache(true);
const presets = [
diff --git a/airflow/www/jest-setup.js b/airflow/www/jest-setup.js
index 0a13d7c644..1323c26098 100644
--- a/airflow/www/jest-setup.js
+++ b/airflow/www/jest-setup.js
@@ -31,8 +31,10 @@ axios.defaults.adapter = require("axios/lib/adapters/http");
axios.interceptors.response.use((res) => res.data || res);
setLogger({
+ /* eslint-disable no-console */
log: console.log,
warn: console.warn,
+ /* eslint-enable no-console */
// ✅ no more errors on the console
error: () => {},
});
diff --git a/airflow/www/package.json b/airflow/www/package.json
index b15568593c..72fe94b4fc 100644
--- a/airflow/www/package.json
+++ b/airflow/www/package.json
@@ -7,7 +7,7 @@
"dev": "NODE_ENV=development webpack --watch --progress --devtool eval-cheap-source-map --mode development",
"prod": "NODE_ENV=production node --max_old_space_size=4096 ./node_modules/webpack/bin/webpack.js --mode production --progress",
"build": "NODE_ENV=production webpack --progress --mode production",
- "lint": "eslint --ignore-path=.eslintignore --ext .js,.jsx,.ts,.tsx . && tsc",
+ "lint": "eslint --ignore-path=.eslintignore --max-warnings=0 --ext .js,.jsx,.ts,.tsx . && tsc",
"lint:fix": "eslint --fix --ignore-path=.eslintignore --ext .js,.jsx,.ts,.tsx . && tsc",
"format": "yarn prettier --write .",
"generate-api-types": "npx openapi-typescript \"../api_connexion/openapi/v1.yaml\" --output static/js/types/api-generated.ts && node alias-rest-types.js static/js/types/api-generated.ts"
diff --git a/airflow/www/static/js/api/useDatasets.ts b/airflow/www/static/js/api/useDatasets.ts
index 379d3c3e61..62d2583d90 100644
--- a/airflow/www/static/js/api/useDatasets.ts
+++ b/airflow/www/static/js/api/useDatasets.ts
@@ -24,7 +24,7 @@ import { getMetaValue } from "src/utils";
import type { DatasetListItem } from "src/types";
import type { unitOfTime } from "moment";
-interface DatasetsData {
+export interface DatasetsData {
datasets: DatasetListItem[];
totalEntries: number;
}
diff --git a/airflow/www/static/js/api/useExtraLinks.ts b/airflow/www/static/js/api/useExtraLinks.ts
index 302d58aa53..3d654e8cbc 100644
--- a/airflow/www/static/js/api/useExtraLinks.ts
+++ b/airflow/www/static/js/api/useExtraLinks.ts
@@ -46,14 +46,16 @@ export default function useExtraLinks({
async () => {
const data = await Promise.all(
extraLinks.map(async (link) => {
- mapIndex ??= -1;
+ const definedMapIndex = mapIndex ?? -1;
const url = `${extraLinksUrl}?task_id=${encodeURIComponent(
taskId
)}&dag_id=${encodeURIComponent(
dagId
)}&execution_date=${encodeURIComponent(
executionDate
- )}&link_name=${encodeURIComponent(link)}&map_index=${mapIndex}`;
+ )}&link_name=${encodeURIComponent(
+ link
+ )}&map_index=${definedMapIndex}`;
try {
const datum = await axios.get<AxiosResponse, LinkData>(url);
return {
@@ -61,6 +63,7 @@ export default function useExtraLinks({
url: datum.url,
};
} catch (e) {
+ // eslint-disable-next-line no-console
console.error(e);
return {
name: link,
diff --git a/airflow/www/static/js/callModal.js b/airflow/www/static/js/callModal.js
index a2cac9f26b..050c7eebff 100644
--- a/airflow/www/static/js/callModal.js
+++ b/airflow/www/static/js/callModal.js
@@ -56,6 +56,7 @@ const showExternalLogRedirect =
const buttons = Array.from(
document.querySelectorAll('a[id^="btn_"][data-base-url]')
).reduce((obj, elm) => {
+ // eslint-disable-next-line no-param-reassign
obj[elm.id.replace("btn_", "")] = elm;
return obj;
}, {});
@@ -64,12 +65,14 @@ function updateButtonUrl(elm, params) {
let url = elm.dataset.baseUrl;
if (params.dag_id && elm.dataset.baseUrl.indexOf(dagId) !== -1) {
url = url.replace(dagId, params.dag_id);
+ // eslint-disable-next-line no-param-reassign
delete params.dag_id;
}
if (
Object.prototype.hasOwnProperty.call(params, "map_index") &&
params.map_index === undefined
) {
+ // eslint-disable-next-line no-param-reassign
delete params.map_index;
}
elm.setAttribute("href", `${url}?${$.param(params)}`);
diff --git a/airflow/www/static/js/components/Table/Cells.tsx b/airflow/www/static/js/components/Table/Cells.tsx
index 8084eb917c..9e86a74935 100644
--- a/airflow/www/static/js/components/Table/Cells.tsx
+++ b/airflow/www/static/js/components/Table/Cells.tsx
@@ -41,8 +41,10 @@ import { SimpleStatus } from "src/dag/StatusBox";
interface CellProps {
cell: {
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
value: any;
row: {
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
original: Record<string, any>;
};
};
diff --git a/airflow/www/static/js/components/Table/Table.test.tsx b/airflow/www/static/js/components/Table/Table.test.tsx
index a3664d49a2..36d9b22456 100644
--- a/airflow/www/static/js/components/Table/Table.test.tsx
+++ b/airflow/www/static/js/components/Table/Table.test.tsx
@@ -28,7 +28,7 @@ import type { SortingRule } from "react-table";
import { ChakraWrapper } from "src/utils/testUtils";
import { Table } from ".";
-const data: Record<string, any>[] = [
+const data: Record<string, string>[] = [
{ firstName: "Lamont", lastName: "Grimes", country: "United States" },
{ firstName: "Alysa", lastName: "Armstrong", country: "Spain" },
{ firstName: "Petra", lastName: "Blick", country: "France" },
diff --git a/airflow/www/static/js/components/Table/index.tsx b/airflow/www/static/js/components/Table/index.tsx
index a2450707b3..753035da07 100644
--- a/airflow/www/static/js/components/Table/index.tsx
+++ b/airflow/www/static/js/components/Table/index.tsx
@@ -89,7 +89,7 @@ interface TableProps {
pageSize?: number;
isLoading?: boolean;
selectRows?: (selectedRows: number[]) => void;
- onRowClicked?: (row: Row<object>, e: any) => void;
+ onRowClicked?: (row: Row<object>, e: unknown) => void;
}
export const Table = ({
@@ -244,7 +244,9 @@ export const Table = ({
}
}
onClick={
- onRowClicked ? (e: any) => onRowClicked(row, e) : undefined
+ onRowClicked
+ ? (e: unknown) => onRowClicked(row, e)
+ : undefined
}
>
{row.cells.map((cell) => (
diff --git a/airflow/www/static/js/components/Tooltip.tsx b/airflow/www/static/js/components/Tooltip.tsx
index fc09cefc1f..72a55b0862 100644
--- a/airflow/www/static/js/components/Tooltip.tsx
+++ b/airflow/www/static/js/components/Tooltip.tsx
@@ -136,7 +136,7 @@ const Tooltip = forwardRef<TooltipProps, "div">((props, ref) => {
* Ensure tooltip has only one child node
*/
const child = React.Children.only(children) as React.ReactElement & {
- ref?: React.Ref<any>;
+ ref?: React.Ref<HTMLElement>;
};
const trigger: React.ReactElement = React.cloneElement(
child,
@@ -168,6 +168,7 @@ const Tooltip = forwardRef<TooltipProps, "div">((props, ref) => {
>
<StyledTooltip
variants={scale}
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
{...(tooltipProps as any)}
initial="exit"
animate="enter"
diff --git a/airflow/www/static/js/connection_form.js b/airflow/www/static/js/connection_form.js
index 456e7d036e..41df55dc07 100644
--- a/airflow/www/static/js/connection_form.js
+++ b/airflow/www/static/js/connection_form.js
@@ -70,12 +70,15 @@ function getControlsContainer() {
function restoreFieldBehaviours() {
Array.from(document.querySelectorAll("label[data-orig-text]")).forEach(
(elem) => {
+ // eslint-disable-next-line no-param-reassign
elem.innerText = elem.dataset.origText;
+ // eslint-disable-next-line no-param-reassign
delete elem.dataset.origText;
}
);
Array.from(document.querySelectorAll(".form-control")).forEach((elem) => {
+ // eslint-disable-next-line no-param-reassign
elem.placeholder = "";
elem.parentElement.parentElement.classList.remove("hide");
});
@@ -259,6 +262,7 @@ $(document).ready(() => {
- All other custom form fields (i.e. fields that are named ``extra__...``) in
alphabetical order
*/
+ // eslint-disable-next-line func-names
$.each(inArray, function () {
if (this.name === "conn_id") {
outObj.connection_id = this.value;
diff --git a/airflow/www/static/js/dag/details/graph/utils.ts b/airflow/www/static/js/dag/details/graph/utils.ts
index 703fb38b9c..8e9c4d1934 100644
--- a/airflow/www/static/js/dag/details/graph/utils.ts
+++ b/airflow/www/static/js/dag/details/graph/utils.ts
@@ -132,7 +132,7 @@ export const nodeColor = ({
export const nodeStrokeColor = (
{ data: { isSelected } }: ReactFlowNode<CustomNodeProps>,
- colors: any
+ colors: Record<string, string>
) => (isSelected ? colors.blue[500] : "");
interface BuildEdgesProps {
diff --git a/airflow/www/static/js/dag/grid/dagRuns/index.test.tsx b/airflow/www/static/js/dag/grid/dagRuns/index.test.tsx
index d77f30af32..2ac20ea3cc 100644
--- a/airflow/www/static/js/dag/grid/dagRuns/index.test.tsx
+++ b/airflow/www/static/js/dag/grid/dagRuns/index.test.tsx
@@ -48,6 +48,7 @@ const generateRuns = (length: number): DagRun[] =>
note: "someRandomValue",
}));
+/* eslint-disable @typescript-eslint/no-explicit-any */
describe("Test DagRuns", () => {
test("Durations and manual run arrow render correctly, but without any date ticks", () => {
const dagRuns: DagRun[] = [
diff --git a/airflow/www/static/js/dag/grid/index.test.tsx b/airflow/www/static/js/dag/grid/index.test.tsx
index 1b3888ffae..66e1cafaeb 100644
--- a/airflow/www/static/js/dag/grid/index.test.tsx
+++ b/airflow/www/static/js/dag/grid/index.test.tsx
@@ -135,6 +135,7 @@ describe("Test ToggleGroups", () => {
const returnValue = {
data: mockGridData,
isSuccess: true,
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
} as any;
jest
diff --git a/airflow/www/static/js/dag_dependencies.js b/airflow/www/static/js/dag_dependencies.js
index ddd40c248a..a3c4a713ea 100644
--- a/airflow/www/static/js/dag_dependencies.js
+++ b/airflow/www/static/js/dag_dependencies.js
@@ -97,6 +97,7 @@ function setUpZoomSupport() {
}
function setUpNodeHighlighting(focusItem = null) {
+ // eslint-disable-next-line func-names
d3.selectAll("g.node").on("mouseover", function (d) {
d3.select(this).selectAll("rect").style("stroke", highlightColor);
highlightNodes(g.predecessors(d), upstreamColor, highlightStrokeWidth);
@@ -114,6 +115,7 @@ function setUpNodeHighlighting(focusItem = null) {
});
});
+ // eslint-disable-next-line func-names
d3.selectAll("g.node").on("mouseout", function (d) {
d3.select(this).selectAll("rect,circle").style("stroke", null);
highlightNodes(g.predecessors(d), null, initialStrokeWidth);
diff --git a/airflow/www/static/js/dags.js b/airflow/www/static/js/dags.js
index 1854ae53a2..ea53be18ac 100644
--- a/airflow/www/static/js/dags.js
+++ b/airflow/www/static/js/dags.js
@@ -320,6 +320,7 @@ function getDagIds({ activeDagsOnly = false } = {}) {
dagIds = dagIds.filter(":checked");
}
dagIds = dagIds
+ // eslint-disable-next-line func-names
.map(function () {
return $(this).data("dag-id");
})
diff --git a/airflow/www/static/js/datasets/List.test.tsx b/airflow/www/static/js/datasets/List.test.tsx
index 34e70f70b0..c1cd0da344 100644
--- a/airflow/www/static/js/datasets/List.test.tsx
+++ b/airflow/www/static/js/datasets/List.test.tsx
@@ -25,6 +25,8 @@ import { render } from "@testing-library/react";
import * as useDatasetsModule from "src/api/useDatasets";
import { Wrapper } from "src/utils/testUtils";
+import type { UseQueryResult } from "react-query";
+import type { DatasetListItem } from "src/types";
import DatasetsList from "./List";
const datasets = [
@@ -57,22 +59,26 @@ const datasets = [
},
];
+type UseDatasetsReturn = UseQueryResult<useDatasetsModule.DatasetsData> & {
+ data: useDatasetsModule.DatasetsData;
+};
+
const returnValue = {
data: {
datasets,
totalEntries: datasets.length,
},
isSuccess: true,
-} as any;
+} as UseDatasetsReturn;
const emptyReturnValue = {
data: {
- datasets: [],
+ datasets: [] as DatasetListItem[],
totalEntries: 0,
},
isSuccess: true,
isLoading: false,
-} as any;
+} as UseDatasetsReturn;
describe("Test Datasets List", () => {
test("Displays a list of datasets", () => {
diff --git a/airflow/www/static/js/datasets/List.tsx b/airflow/www/static/js/datasets/List.tsx
index e201e1d2d4..6bbfd26497 100644
--- a/airflow/www/static/js/datasets/List.tsx
+++ b/airflow/www/static/js/datasets/List.tsx
@@ -49,8 +49,10 @@ interface Props {
interface CellProps {
cell: {
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
value: any;
row: {
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
original: Record<string, any>;
};
};
diff --git a/airflow/www/static/js/datetime_utils.js b/airflow/www/static/js/datetime_utils.js
index b9e4b76eeb..c23f8d665e 100644
--- a/airflow/www/static/js/datetime_utils.js
+++ b/airflow/www/static/js/datetime_utils.js
@@ -104,6 +104,7 @@ export function updateAllDateTimes() {
// Since we have set the default timezone for moment, it will automatically
// convert it to the new target for us
$(".datetime input").each((_, el) => {
+ // eslint-disable-next-line no-param-reassign
el.value = moment(el.value).format();
});
}
diff --git a/airflow/www/static/js/gantt.js b/airflow/www/static/js/gantt.js
index 048ba741e6..2b7cf10c81 100644
--- a/airflow/www/static/js/gantt.js
+++ b/airflow/www/static/js/gantt.js
@@ -74,7 +74,7 @@ const replacements = {
"%": "%",
};
-moment.fn.strftime = function (format) {
+moment.fn.strftime = function formatTime(format) {
// Break up format string based on strftime tokens
const tokens = format.split(/(%-?.)/);
const momentFormat = tokens
@@ -173,12 +173,15 @@ d3.gantt = () => {
tasks.forEach((a) => {
if (!(a.start_date instanceof moment)) {
+ // eslint-disable-next-line no-param-reassign
a.start_date = moment(a.start_date);
}
if (!(a.end_date instanceof moment)) {
+ // eslint-disable-next-line no-param-reassign
a.end_date = moment(a.end_date);
}
if (a.queued_dttm && !(a.queued_dttm instanceof moment)) {
+ // eslint-disable-next-line no-param-reassign
a.queued_dttm = moment(a.queued_dttm);
}
});
@@ -349,12 +352,14 @@ d3.gantt = () => {
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.margin = function (value) {
if (!arguments.length) return margin;
margin = value;
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.timeDomain = function (value) {
if (!arguments.length) return [timeDomainStart, timeDomainEnd];
timeDomainStart = +value[0];
@@ -367,36 +372,42 @@ d3.gantt = () => {
* vale The value can be "fit" - the domain fits the data or
* "fixed" - fixed domain.
*/
+ // eslint-disable-next-line func-names
gantt.timeDomainMode = function (value) {
if (!arguments.length) return timeDomainMode;
timeDomainMode = value;
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.taskTypes = function (value) {
if (!arguments.length) return taskTypes;
taskTypes = value;
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.width = function (value) {
if (!arguments.length) return width;
width = +value;
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.height = function (value) {
if (!arguments.length) return height;
height = +value;
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.tickFormat = function (value) {
if (!arguments.length) return tickFormat;
tickFormat = value;
return gantt;
};
+ // eslint-disable-next-line func-names
gantt.selector = function (value) {
if (!arguments.length) return selector;
selector = value;
diff --git a/airflow/www/static/js/graph.js b/airflow/www/static/js/graph.js
index 38b1ae6211..b61639d3a8 100644
--- a/airflow/www/static/js/graph.js
+++ b/airflow/www/static/js/graph.js
@@ -153,6 +153,7 @@ function collapseGroup(nodeId, node) {
getChildrenIds(node).forEach((childId) => mapTaskToNode.set(childId, nodeId));
+ // eslint-disable-next-line no-param-reassign
node = g.node(nodeId);
// Set children edges onto the group edge
diff --git a/airflow/www/static/js/main.js b/airflow/www/static/js/main.js
index 7350a5cfff..95861ea6c1 100644
--- a/airflow/www/static/js/main.js
+++ b/airflow/www/static/js/main.js
@@ -69,19 +69,20 @@ export function escapeHtml(text) {
window.escapeHtml = escapeHtml;
export function convertSecsToHumanReadable(seconds) {
+ let processedSeconds = seconds;
const oriSeconds = seconds;
const floatingPart = oriSeconds - Math.floor(oriSeconds);
- seconds = Math.floor(seconds);
+ processedSeconds = Math.floor(processedSeconds);
const secondsPerHour = 60 * 60;
const secondsPerMinute = 60;
- const hours = Math.floor(seconds / secondsPerHour);
- seconds -= hours * secondsPerHour;
+ const hours = Math.floor(processedSeconds / secondsPerHour);
+ processedSeconds -= hours * secondsPerHour;
- const minutes = Math.floor(seconds / secondsPerMinute);
- seconds -= minutes * secondsPerMinute;
+ const minutes = Math.floor(processedSeconds / secondsPerMinute);
+ processedSeconds -= minutes * secondsPerMinute;
let readableFormat = "";
if (hours > 0) {
@@ -90,12 +91,12 @@ export function convertSecsToHumanReadable(seconds) {
if (minutes > 0) {
readableFormat += `${minutes}Min `;
}
- if (seconds + floatingPart > 0) {
+ if (processedSeconds + floatingPart > 0) {
if (Math.floor(oriSeconds) === oriSeconds) {
- readableFormat += `${seconds}Sec`;
+ readableFormat += `${processedSeconds}Sec`;
} else {
- seconds += floatingPart;
- readableFormat += `${seconds.toFixed(3)}Sec`;
+ processedSeconds += floatingPart;
+ readableFormat += `${processedSeconds.toFixed(3)}Sec`;
}
}
return readableFormat;
diff --git a/airflow/www/static/js/task.js b/airflow/www/static/js/task.js
index df54602729..d4c03d1848 100644
--- a/airflow/www/static/js/task.js
+++ b/airflow/www/static/js/task.js
@@ -27,6 +27,7 @@ document.addEventListener("DOMContentLoaded", () => {
const value = attr.innerHTML;
if (value.length === 32 && moment(value, "YYYY-MM-DD").isValid()) {
// 32 is the length of our timestamps
+ // eslint-disable-next-line no-param-reassign
attr.innerHTML = "";
const timeElement = document.createElement("time");
timeElement.setAttribute("datetime", value);
@@ -35,6 +36,7 @@ document.addEventListener("DOMContentLoaded", () => {
attr.appendChild(timeElement);
} else if (validator.isURL(value)) {
// very basic url detection
+ // eslint-disable-next-line no-param-reassign
attr.innerHTML = "";
const linkElement = document.createElement("a");
linkElement.setAttribute("href", value);
diff --git a/airflow/www/static/js/task_instances.js b/airflow/www/static/js/task_instances.js
index 8df8b46841..e3ad1a4633 100644
--- a/airflow/www/static/js/task_instances.js
+++ b/airflow/www/static/js/task_instances.js
@@ -119,12 +119,14 @@ export default function tiTooltip(ti, task, { includeTryNumber = false } = {}) {
if (ti.state === "running") {
const startDate =
ti.start_date instanceof moment ? ti.start_date : moment(ti.start_date);
+ // eslint-disable-next-line no-param-reassign
ti.duration = moment().diff(startDate, "second");
} else if (!ti.duration && ti.end_date) {
const startDate =
ti.start_date instanceof moment ? ti.start_date : moment(ti.start_date);
const endDate =
ti.end_date instanceof moment ? ti.end_date : moment(ti.end_date);
+ // eslint-disable-next-line no-param-reassign
ti.duration = moment(endDate).diff(startDate, "second");
}
diff --git a/airflow/www/static/js/ti_log.js b/airflow/www/static/js/ti_log.js
index d2e223b103..23fd0af769 100644
--- a/airflow/www/static/js/ti_log.js
+++ b/airflow/www/static/js/ti_log.js
@@ -37,6 +37,7 @@ function recurse(delay = DELAY) {
setTimeout(resolve, delay);
});
}
+/* eslint-disable no-console */
// Enable auto tailing only when users scroll down to the bottom
// of the page. This prevent auto tailing the page if users want
@@ -166,15 +167,16 @@ function autoTailingLog(tryNumber, metadata = null, autoTailing = false) {
}
function setDownloadUrl(tryNumber) {
- if (!tryNumber) {
+ let tryNumberData = tryNumber;
+ if (!tryNumberData) {
// default to the currently selected tab
- tryNumber = $("#ti_log_try_number_list .active a").data("try-number");
+ tryNumberData = $("#ti_log_try_number_list .active a").data("try-number");
}
const query = new URLSearchParams({
dag_id: dagId,
task_id: taskId,
execution_date: executionDate,
- try_number: tryNumber,
+ try_number: tryNumberData,
metadata: "null",
format: "file",
});
@@ -187,6 +189,7 @@ $(document).ready(() => {
autoTailingLog(TOTAL_ATTEMPTS, null, true);
setDownloadUrl();
+ // eslint-disable-next-line func-names
$("#ti_log_try_number_list a").click(function () {
const tryNumber = $(this).data("try-number");
diff --git a/airflow/www/static/js/types/react-table-config.d.ts b/airflow/www/static/js/types/react-table-config.d.ts
index 0b62ceb16f..2bc503ed18 100644
--- a/airflow/www/static/js/types/react-table-config.d.ts
+++ b/airflow/www/static/js/types/react-table-config.d.ts
@@ -87,7 +87,7 @@ declare module "react-table" {
// this matches the spirit of the underlying js library,
// but might be cleaner if it's replaced by a more specific type that matches your
// feature set, this is a safe default.
- Record<string, any> {}
+ Record<string, unknown> {}
export interface Hooks<
D extends Record<string, unknown> = Record<string, unknown>
@@ -139,7 +139,7 @@ declare module "react-table" {
export interface Cell<
D extends Record<string, unknown> = Record<string, unknown>,
// eslint-disable-next-line @typescript-eslint/no-unused-vars
- V = any
+ V = unknown
> extends UseGroupByCellProps<D>,
UseRowStateCellProps<D> {}
diff --git a/airflow/www/static/js/utils/URLSearchParamWrapper.ts b/airflow/www/static/js/utils/URLSearchParamWrapper.ts
index befeadcaf5..476db61b90 100644
--- a/airflow/www/static/js/utils/URLSearchParamWrapper.ts
+++ b/airflow/www/static/js/utils/URLSearchParamWrapper.ts
@@ -18,7 +18,8 @@
*/
class URLSearchParamsWrapper extends URLSearchParams {
- constructor(init?: { [keys: string]: any }) {
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
+ constructor(init?: Record<string, any>) {
if (init) {
const stringValues: { [keys: string]: string } = {};
Object.keys(init).forEach((key) => {
diff --git a/airflow/www/static/js/utils/graph.ts b/airflow/www/static/js/utils/graph.ts
index d2d2923c5a..50b8896cec 100644
--- a/airflow/www/static/js/utils/graph.ts
+++ b/airflow/www/static/js/utils/graph.ts
@@ -84,13 +84,21 @@ const generateGraph = ({
}: GenerateProps) => {
const closedGroupIds: string[] = [];
- const formatChildNode = (node: any) => {
+ const formatChildNode = (
+ node: DepNode
+ ): DepNode & {
+ label: string;
+ layoutOptions?: Record<string, string>;
+ width?: number;
+ height?: number;
+ } => {
const { id, value, children } = node;
const isOpen = openGroupIds?.includes(value.label);
const childCount =
- children?.filter((c: any) => !c.id.includes("join_id")).length || 0;
- if (isOpen && children.length) {
+ children?.filter((c: DepNode) => !c.id.includes("join_id")).length || 0;
+ if (isOpen && children?.length) {
return {
+ ...node,
id,
value: {
...value,
[airflow] 33/42: Hide sensitive values from extra in connection edit form (#32309)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 191daab7a4f9e632e95e506724d053a3a7cf4606
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Wed Jul 5 21:39:27 2023 +0200
Hide sensitive values from extra in connection edit form (#32309)
The fields that are sensitive (i.e password field is used
to show them in Connection edit view) should also be hidden
when they are stored as "extra" in the form extra field.
This PR handles both - replacing such values with a
placeholder as well as not updating the value if the
placeholder has not been modified.
(cherry picked from commit d01248382fe45a5f5a7fdeed4082a80c5f814ad8)
---
airflow/providers_manager.py | 8 +++-
airflow/www/views.py | 57 ++++++++++++++++++++-----
tests/www/views/test_views_connection.py | 71 ++++++++++++++++++++++++++++----
3 files changed, 117 insertions(+), 19 deletions(-)
diff --git a/airflow/providers_manager.py b/airflow/providers_manager.py
index dd214c755c..cbba074750 100644
--- a/airflow/providers_manager.py
+++ b/airflow/providers_manager.py
@@ -215,6 +215,7 @@ class ConnectionFormWidgetInfo(NamedTuple):
package_name: str
field: Any
field_name: str
+ is_sensitive: bool
T = TypeVar("T", bound=Callable)
@@ -873,7 +874,12 @@ class ProvidersManager(LoggingMixin, metaclass=Singleton):
# In case of inherited hooks this might be happening several times
continue
self._connection_form_widgets[prefixed_field_name] = ConnectionFormWidgetInfo(
- hook_class.__name__, package_name, field, field_identifier
+ hook_class.__name__,
+ package_name,
+ field,
+ field_identifier,
+ hasattr(field.field_class.widget, "input_type")
+ and field.field_class.widget.input_type == "password",
)
def _add_customized_fields(self, package_name: str, hook_class: type, customized_fields: dict):
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 18188d3de5..0e06a1ff29 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -87,8 +87,14 @@ from airflow.api.common.mark_tasks import (
from airflow.compat.functools import cached_property
from airflow.configuration import AIRFLOW_CONFIG, conf
from airflow.datasets import Dataset
-from airflow.exceptions import AirflowException, ParamValidationError, RemovedInAirflow3Warning
+from airflow.exceptions import (
+ AirflowException,
+ AirflowNotFoundException,
+ ParamValidationError,
+ RemovedInAirflow3Warning,
+)
from airflow.executors.executor_loader import ExecutorLoader
+from airflow.hooks.base import BaseHook
from airflow.jobs.job import Job
from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
from airflow.jobs.triggerer_job_runner import TriggererJobRunner
@@ -144,6 +150,8 @@ LINECHART_X_AXIS_TICKFORMAT = (
"} else {xLabel = d3.time.format('%H:%M, %d %b')(new Date(parseInt(d)));} return xLabel;}"
)
+SENSITIVE_FIELD_PLACEHOLDER = "RATHER_LONG_SENSITIVE_FIELD_PLACEHOLDER"
+
def truncate_task_duration(task_duration):
"""
@@ -4420,14 +4428,22 @@ class ConnectionModelView(AirflowModelView):
base_order = ("conn_id", "asc")
- def _iter_extra_field_names(self) -> Iterator[tuple[str, str]]:
+ def _iter_extra_field_names_and_sensitivity(self) -> Iterator[tuple[str, str, bool]]:
"""Iterate through provider-backed connection fields.
Note that this cannot be a property (including a cached property)
because Flask-Appbuilder attempts to access all members on startup, and
using a property would initialize the providers manager too eagerly.
+
+ Returns tuple of:
+
+ * key
+ * field_name
+ * whether the field is sensitive
"""
- return ((k, v.field_name) for k, v in ProvidersManager().connection_form_widgets.items())
+ return (
+ (k, v.field_name, v.is_sensitive) for k, v in ProvidersManager().connection_form_widgets.items()
+ )
@property
def add_columns(self) -> list[str]:
@@ -4440,7 +4456,10 @@ class ConnectionModelView(AirflowModelView):
superfuluous checks done by Flask-Appbuilder on startup).
"""
if self._add_columns is type(self)._add_columns and has_request_context():
- self._add_columns = [*self._add_columns, *(k for k, _ in self._iter_extra_field_names())]
+ self._add_columns = [
+ *self._add_columns,
+ *(k for k, _, _ in self._iter_extra_field_names_and_sensitivity()),
+ ]
return self._add_columns
@property
@@ -4454,7 +4473,10 @@ class ConnectionModelView(AirflowModelView):
superfuluous checks done by Flask-Appbuilder on startup).
"""
if self._edit_columns is type(self)._edit_columns and has_request_context():
- self._edit_columns = [*self._edit_columns, *(k for k, _ in self._iter_extra_field_names())]
+ self._edit_columns = [
+ *self._edit_columns,
+ *(k for k, _, _ in self._iter_extra_field_names_and_sensitivity()),
+ ]
return self._edit_columns
@action("muldelete", "Delete", "Are you sure you want to delete selected records?", single=False)
@@ -4543,7 +4565,6 @@ class ConnectionModelView(AirflowModelView):
"""Process form data."""
conn_id = form.data["conn_id"]
conn_type = form.data["conn_type"]
-
# The extra value is the combination of custom fields for this conn_type and the Extra field.
# The extra form field with all extra values (including custom fields) is in the form being processed
# so we start with those values, and override them with anything in the custom fields.
@@ -4568,8 +4589,7 @@ class ConnectionModelView(AirflowModelView):
)
del form.extra
del extra_json
-
- for key, field_name in self._iter_extra_field_names():
+ for key, field_name, is_sensitive in self._iter_extra_field_names_and_sensitivity():
if key in form.data and key.startswith("extra__"):
conn_type_from_extra_field = key.split("__")[1]
if conn_type_from_extra_field == conn_type:
@@ -4578,8 +4598,21 @@ class ConnectionModelView(AirflowModelView):
# value isn't an empty string.
if value != "":
extra[field_name] = value
-
if extra.keys():
+ sensitive_unchanged_keys = set()
+ for key, value in extra.items():
+ if value == SENSITIVE_FIELD_PLACEHOLDER:
+ sensitive_unchanged_keys.add(key)
+ if sensitive_unchanged_keys:
+ try:
+ conn = BaseHook.get_connection(conn_id)
+ except AirflowNotFoundException:
+ conn = None
+ for key in sensitive_unchanged_keys:
+ if conn and conn.extra_dejson.get(key):
+ extra[key] = conn.extra_dejson.get(key)
+ else:
+ del extra[key]
form.extra.data = json.dumps(extra)
def prefill_form(self, form, pk):
@@ -4597,7 +4630,7 @@ class ConnectionModelView(AirflowModelView):
logging.warning("extra field for %s is not a dictionary", form.data.get("conn_id", "<unknown>"))
return
- for field_key, field_name in self._iter_extra_field_names():
+ for field_key, field_name, is_sensitive in self._iter_extra_field_names_and_sensitivity():
value = extra_dictionary.get(field_name, "")
if not value:
@@ -4607,6 +4640,10 @@ class ConnectionModelView(AirflowModelView):
if value:
field = getattr(form, field_key)
field.data = value
+ if is_sensitive and field_name in extra_dictionary:
+ extra_dictionary[field_name] = SENSITIVE_FIELD_PLACEHOLDER
+ # form.data is a property that builds the dictionary from fields so we have to modify the fields
+ form.extra.data = json.dumps(extra_dictionary)
class PluginView(AirflowBaseView):
diff --git a/tests/www/views/test_views_connection.py b/tests/www/views/test_views_connection.py
index fb6566d936..e57a8d484c 100644
--- a/tests/www/views/test_views_connection.py
+++ b/tests/www/views/test_views_connection.py
@@ -78,10 +78,29 @@ def test_prefill_form_null_extra():
mock_form.data = {"conn_id": "test", "extra": None, "conn_type": "test"}
cmv = ConnectionModelView()
- cmv._iter_extra_field_names = mock.Mock(return_value=())
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(return_value=())
cmv.prefill_form(form=mock_form, pk=1)
+def test_prefill_form_sensitive_fields_extra():
+ mock_form = mock.Mock()
+ mock_form.data = {
+ "conn_id": "test",
+ "extra": json.dumps({"sensitive_extra": "TEST1", "non_sensitive_extra": "TEST2"}),
+ "conn_type": "test",
+ }
+
+ cmv = ConnectionModelView()
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(
+ return_value=[("sensitive_extra_key", "sensitive_extra", True)]
+ )
+ cmv.prefill_form(form=mock_form, pk=1)
+ assert json.loads(mock_form.extra.data) == {
+ "sensitive_extra": "RATHER_LONG_SENSITIVE_FIELD_PLACEHOLDER",
+ "non_sensitive_extra": "TEST2",
+ }
+
+
@pytest.mark.parametrize(
"extras, expected",
[
@@ -105,7 +124,9 @@ def test_prefill_form_backcompat(extras, expected):
mock_form.data = {"conn_id": "test", "extra": json.dumps(extras), "conn_type": "test"}
cmv = ConnectionModelView()
- cmv._iter_extra_field_names = mock.Mock(return_value=[("extra__test__my_param", "my_param")])
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(
+ return_value=[("extra__test__my_param", "my_param", False)]
+ )
cmv.prefill_form(form=mock_form, pk=1)
assert mock_form.extra__test__my_param.data == expected
@@ -133,7 +154,9 @@ def test_process_form_extras_both(mock_pm_hooks, mock_import_str, field_name):
}
cmv = ConnectionModelView()
- cmv._iter_extra_field_names = mock.Mock(return_value=[("extra__test__custom_field", field_name)])
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(
+ return_value=[("extra__test__custom_field", field_name, False)]
+ )
cmv.process_form(form=mock_form, is_created=True)
assert json.loads(mock_form.extra.data) == {
field_name: "custom_field_val",
@@ -159,7 +182,7 @@ def test_process_form_extras_extra_only(mock_pm_hooks, mock_import_str):
}
cmv = ConnectionModelView()
- cmv._iter_extra_field_names = mock.Mock(return_value=())
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(return_value=())
cmv.process_form(form=mock_form, is_created=True)
assert json.loads(mock_form.extra.data) == {"param2": "param2_val"}
@@ -185,10 +208,10 @@ def test_process_form_extras_custom_only(mock_pm_hooks, mock_import_str, field_n
}
cmv = ConnectionModelView()
- cmv._iter_extra_field_names = mock.Mock(
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(
return_value=[
- ("extra__test3__custom_field", field_name),
- ("extra__test3__custom_bool_field", False),
+ ("extra__test3__custom_field", field_name, False),
+ ("extra__test3__custom_bool_field", False, False),
],
)
cmv.process_form(form=mock_form, is_created=True)
@@ -216,7 +239,9 @@ def test_process_form_extras_updates(mock_pm_hooks, mock_import_str, field_name)
}
cmv = ConnectionModelView()
- cmv._iter_extra_field_names = mock.Mock(return_value=[("extra__test4__custom_field", field_name)])
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(
+ return_value=[("extra__test4__custom_field", field_name, False)]
+ )
cmv.process_form(form=mock_form, is_created=True)
if field_name == "custom_field":
@@ -228,6 +253,36 @@ def test_process_form_extras_updates(mock_pm_hooks, mock_import_str, field_name)
assert json.loads(mock_form.extra.data) == {"extra__test4__custom_field": "custom_field_val4"}
+@mock.patch("airflow.utils.module_loading.import_string")
+@mock.patch("airflow.providers_manager.ProvidersManager.hooks", new_callable=PropertyMock)
+@mock.patch("airflow.www.views.BaseHook")
+def test_process_form_extras_updates_sensitive_placeholder_unchanged(
+ mock_base_hook, mock_pm_hooks, mock_import_str
+):
+ """
+ Test the handling of sensitive unchanged field (where placeholder has not been modified).
+ """
+
+ # Testing parameters set in both extra and custom fields (connection updates).
+ mock_form = mock.Mock()
+ mock_form.data = {
+ "conn_type": "test4",
+ "conn_id": "extras_test4",
+ "extra": '{"sensitive_extra": "RATHER_LONG_SENSITIVE_FIELD_PLACEHOLDER", "extra__custom": "value"}',
+ }
+ mock_base_hook.get_connection.return_value = Connection(extra='{"sensitive_extra": "old_value"}')
+ cmv = ConnectionModelView()
+ cmv._iter_extra_field_names_and_sensitivity = mock.Mock(
+ return_value=[("sensitive_extra_key", "sensitive_extra", True)]
+ )
+ cmv.process_form(form=mock_form, is_created=True)
+
+ assert json.loads(mock_form.extra.data) == {
+ "extra__custom": "value",
+ "sensitive_extra": "old_value",
+ }
+
+
def test_duplicate_connection(admin_client):
"""Test Duplicate multiple connection with suffix"""
conn1 = Connection(
[airflow] 38/42: Fix triggerers alive check and add a new conf for triggerer heartbeat rate (#32123)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cf31c6f1368b69957d7a1e84538c5e336c1bd28d
Author: Hussein Awala <hu...@awala.fr>
AuthorDate: Fri Jun 30 22:19:06 2023 +0200
Fix triggerers alive check and add a new conf for triggerer heartbeat rate (#32123)
Signed-off-by: Hussein Awala <hu...@awala.fr>
(cherry picked from commit d117728cd6f337266bebcf4916325d5de815fe03)
---
airflow/cli/commands/triggerer_command.py | 3 +-
airflow/config_templates/config.yml | 7 +++
airflow/config_templates/default_airflow.cfg | 3 ++
airflow/jobs/triggerer_job_runner.py | 2 +-
airflow/models/trigger.py | 13 +++---
tests/models/test_trigger.py | 64 ++++++++++++++++++++++++++--
6 files changed, 81 insertions(+), 11 deletions(-)
diff --git a/airflow/cli/commands/triggerer_command.py b/airflow/cli/commands/triggerer_command.py
index 7bfd77bf8e..d1a63e9c22 100644
--- a/airflow/cli/commands/triggerer_command.py
+++ b/airflow/cli/commands/triggerer_command.py
@@ -55,7 +55,8 @@ def triggerer(args):
"""Starts Airflow Triggerer."""
settings.MASK_SECRETS_IN_LOGS = True
print(settings.HEADER)
- triggerer_job_runner = TriggererJobRunner(job=Job(), capacity=args.capacity)
+ triggerer_heartrate = conf.getfloat("triggerer", "JOB_HEARTBEAT_SEC")
+ triggerer_job_runner = TriggererJobRunner(job=Job(heartrate=triggerer_heartrate), capacity=args.capacity)
if args.daemon:
pid, stdout, stderr, log_file = setup_locations(
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index 4653a82416..897ed3b402 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -2459,6 +2459,13 @@ triggerer:
type: string
example: ~
default: "1000"
+ job_heartbeat_sec:
+ description: |
+ How often to heartbeat the Triggerer job to ensure it hasn't been killed.
+ version_added: 2.6.3
+ type: float
+ example: ~
+ default: "5"
kerberos:
description: ~
options:
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index c765f17ead..1908c39bf5 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -1254,6 +1254,9 @@ allowed_run_id_pattern = ^[A-Za-z0-9_.~:+-]+$
# How many triggers a single Triggerer will run at once, by default.
default_capacity = 1000
+# How often to heartbeat the Triggerer job to ensure it hasn't been killed.
+job_heartbeat_sec = 5
+
[kerberos]
ccache = /tmp/airflow_krb5_ccache
diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py
index 8cbc17b228..c1168a09b1 100644
--- a/airflow/jobs/triggerer_job_runner.py
+++ b/airflow/jobs/triggerer_job_runner.py
@@ -374,7 +374,7 @@ class TriggererJobRunner(BaseJobRunner["Job | JobPydantic"], LoggingMixin):
adds them to our runner, and then removes ones from it we no longer
need.
"""
- Trigger.assign_unassigned(self.job.id, self.capacity)
+ Trigger.assign_unassigned(self.job.id, self.capacity, self.job.heartrate)
ids = Trigger.ids_for_triggerer(self.job.id)
self.trigger_runner.update_triggers(set(ids))
diff --git a/airflow/models/trigger.py b/airflow/models/trigger.py
index adad7ba1c3..e65a65505a 100644
--- a/airflow/models/trigger.py
+++ b/airflow/models/trigger.py
@@ -196,10 +196,11 @@ class Trigger(Base):
@classmethod
@internal_api_call
@provide_session
- def assign_unassigned(cls, triggerer_id, capacity, session: Session = NEW_SESSION) -> None:
+ def assign_unassigned(cls, triggerer_id, capacity, heartrate, session: Session = NEW_SESSION) -> None:
"""
- Takes a triggerer_id and the capacity for that triggerer and assigns unassigned
- triggers until that capacity is reached, or there are no more unassigned triggers.
+ Takes a triggerer_id, the capacity for that triggerer and the Triggerer job heartrate,
+ and assigns unassigned triggers until that capacity is reached, or there are no more
+ unassigned triggers.
"""
from airflow.jobs.job import Job # To avoid circular import
@@ -208,12 +209,14 @@ class Trigger(Base):
if capacity <= 0:
return
-
+ # we multiply heartrate by a grace_multiplier to give the triggerer
+ # a chance to heartbeat before we consider it dead
+ health_check_threshold = heartrate * 2.1
alive_triggerer_ids = [
row[0]
for row in session.query(Job.id).filter(
Job.end_date.is_(None),
- Job.latest_heartbeat > timezone.utcnow() - datetime.timedelta(seconds=30),
+ Job.latest_heartbeat > timezone.utcnow() - datetime.timedelta(seconds=health_check_threshold),
Job.job_type == "TriggererJob",
)
]
diff --git a/tests/models/test_trigger.py b/tests/models/test_trigger.py
index ac5bba4606..6b434933dd 100644
--- a/tests/models/test_trigger.py
+++ b/tests/models/test_trigger.py
@@ -140,16 +140,17 @@ def test_assign_unassigned(session, create_task_instance):
"""
Tests that unassigned triggers of all appropriate states are assigned.
"""
- finished_triggerer = Job(heartrate=10, state=State.SUCCESS)
+ triggerer_heartrate = 10
+ finished_triggerer = Job(heartrate=triggerer_heartrate, state=State.SUCCESS)
TriggererJobRunner(finished_triggerer)
finished_triggerer.end_date = timezone.utcnow() - datetime.timedelta(hours=1)
session.add(finished_triggerer)
assert not finished_triggerer.is_alive()
- healthy_triggerer = Job(heartrate=10, state=State.RUNNING)
+ healthy_triggerer = Job(heartrate=triggerer_heartrate, state=State.RUNNING)
TriggererJobRunner(healthy_triggerer)
session.add(healthy_triggerer)
assert healthy_triggerer.is_alive()
- new_triggerer = Job(heartrate=10, state=State.RUNNING)
+ new_triggerer = Job(heartrate=triggerer_heartrate, state=State.RUNNING)
TriggererJobRunner(new_triggerer)
session.add(new_triggerer)
assert new_triggerer.is_alive()
@@ -168,7 +169,7 @@ def test_assign_unassigned(session, create_task_instance):
session.add(trigger_unassigned_to_triggerer)
session.commit()
assert session.query(Trigger).count() == 3
- Trigger.assign_unassigned(new_triggerer.id, 100, session=session)
+ Trigger.assign_unassigned(new_triggerer.id, 100, session=session, heartrate=triggerer_heartrate)
session.expire_all()
# Check that trigger on killed triggerer and unassigned trigger are assigned to new triggerer
assert (
@@ -184,3 +185,58 @@ def test_assign_unassigned(session, create_task_instance):
session.query(Trigger).filter(Trigger.id == trigger_on_healthy_triggerer.id).one().triggerer_id
== healthy_triggerer.id
)
+
+
+@pytest.mark.parametrize("check_triggerer_heartrate", [10, 60, 300])
+def test_assign_unassigned_missing_heartbeat(session, create_task_instance, check_triggerer_heartrate):
+ """
+ Tests that the triggers assigned to a dead triggers are considered as unassigned
+ and they are assigned to an alive triggerer.
+ """
+ import time_machine
+
+ block_triggerer_heartrate = 9999
+ with time_machine.travel(datetime.datetime.utcnow(), tick=False) as t:
+ first_triggerer = Job(heartrate=block_triggerer_heartrate, state=State.RUNNING)
+ TriggererJobRunner(first_triggerer)
+ session.add(first_triggerer)
+ assert first_triggerer.is_alive()
+ second_triggerer = Job(heartrate=block_triggerer_heartrate, state=State.RUNNING)
+ TriggererJobRunner(second_triggerer)
+ session.add(second_triggerer)
+ assert second_triggerer.is_alive()
+ session.commit()
+ trigger_on_first_triggerer = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
+ trigger_on_first_triggerer.id = 1
+ trigger_on_first_triggerer.triggerer_id = first_triggerer.id
+ trigger_on_second_triggerer = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
+ trigger_on_second_triggerer.id = 2
+ trigger_on_second_triggerer.triggerer_id = second_triggerer.id
+ session.add(trigger_on_first_triggerer)
+ session.add(trigger_on_second_triggerer)
+ session.commit()
+ assert session.query(Trigger).count() == 2
+ triggers_ids = [
+ (first_triggerer.id, second_triggerer.id),
+ (first_triggerer.id, second_triggerer.id),
+ (first_triggerer.id, second_triggerer.id),
+ # Check that after more than 2.1 heartrates, the first triggerer is considered dead
+ # and the first trigger is assigned to the second triggerer
+ (second_triggerer.id, second_triggerer.id),
+ ]
+ for i in range(4):
+ Trigger.assign_unassigned(
+ second_triggerer.id, 100, session=session, heartrate=check_triggerer_heartrate
+ )
+ session.expire_all()
+ # Check that trigger on killed triggerer and unassigned trigger are assigned to new triggerer
+ assert (
+ session.query(Trigger).filter(Trigger.id == trigger_on_first_triggerer.id).one().triggerer_id
+ == triggers_ids[i][0]
+ )
+ assert (
+ session.query(Trigger).filter(Trigger.id == trigger_on_second_triggerer.id).one().triggerer_id
+ == triggers_ids[i][1]
+ )
+ t.shift(datetime.timedelta(seconds=check_triggerer_heartrate))
+ second_triggerer.latest_heartbeat += datetime.timedelta(seconds=check_triggerer_heartrate)
[airflow] 19/42: fix bugged singleton implementation (#32218)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7d0d67198ea9ff8457b8550f1a54e22214d50e26
Author: Maciej Obuchowski <ob...@gmail.com>
AuthorDate: Thu Jun 29 14:02:58 2023 +0200
fix bugged singleton implementation (#32218)
* fix bugged singleton implementation
Signed-off-by: Maciej Obuchowski <ob...@gmail.com>
(cherry picked from commit e2e707ca2720bf066228a136c36c17fd6eda802c)
---
airflow/executors/kubernetes_executor.py | 9 +----
airflow/providers_manager.py | 9 +----
airflow/utils/singleton.py | 33 ++++++++++++++++
tests/always/test_providers_manager.py | 5 +++
tests/utils/test_singleton.py | 65 ++++++++++++++++++++++++++++++++
5 files changed, 107 insertions(+), 14 deletions(-)
diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py
index 4f243f598a..ff7a48bcd4 100644
--- a/airflow/executors/kubernetes_executor.py
+++ b/airflow/executors/kubernetes_executor.py
@@ -49,6 +49,7 @@ from airflow.kubernetes.pod_generator import PodGenerator
from airflow.utils.event_scheduler import EventScheduler
from airflow.utils.log.logging_mixin import LoggingMixin, remove_escape_codes
from airflow.utils.session import NEW_SESSION, provide_session
+from airflow.utils.singleton import Singleton
from airflow.utils.state import State, TaskInstanceState
if TYPE_CHECKING:
@@ -69,17 +70,11 @@ ALL_NAMESPACES = "ALL_NAMESPACES"
POD_EXECUTOR_DONE_KEY = "airflow_executor_done"
-class ResourceVersion:
+class ResourceVersion(metaclass=Singleton):
"""Singleton for tracking resourceVersion from Kubernetes."""
- _instance: ResourceVersion | None = None
resource_version: dict[str, str] = {}
- def __new__(cls):
- if cls._instance is None:
- cls._instance = super().__new__(cls)
- return cls._instance
-
class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin):
"""Watches for Kubernetes jobs."""
diff --git a/airflow/providers_manager.py b/airflow/providers_manager.py
index 2cab30d89c..84fd957a0e 100644
--- a/airflow/providers_manager.py
+++ b/airflow/providers_manager.py
@@ -39,6 +39,7 @@ from airflow.utils import yaml
from airflow.utils.entry_points import entry_points_with_dist
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.module_loading import import_string
+from airflow.utils.singleton import Singleton
log = logging.getLogger(__name__)
@@ -348,7 +349,7 @@ def provider_info_cache(cache_name: str) -> Callable[[T], T]:
return provider_info_cache_decorator
-class ProvidersManager(LoggingMixin):
+class ProvidersManager(LoggingMixin, metaclass=Singleton):
"""
Manages all provider packages.
@@ -357,14 +358,8 @@ class ProvidersManager(LoggingMixin):
local source folders (if airflow is run from sources).
"""
- _instance = None
resource_version = "0"
- def __new__(cls):
- if cls._instance is None:
- cls._instance = super().__new__(cls)
- return cls._instance
-
def __init__(self):
"""Initializes the manager."""
super().__init__()
diff --git a/airflow/utils/singleton.py b/airflow/utils/singleton.py
new file mode 100644
index 0000000000..cfc97eddbf
--- /dev/null
+++ b/airflow/utils/singleton.py
@@ -0,0 +1,33 @@
+#
+# 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 annotations
+
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class Singleton(type, Generic[T]):
+ """Metaclass that allows to implement singleton pattern."""
+
+ _instances: dict[Singleton[T], T] = {}
+
+ def __call__(cls: Singleton[T], *args, **kwargs) -> T:
+ if cls not in cls._instances:
+ cls._instances[cls] = super().__call__(*args, **kwargs)
+ return cls._instances[cls]
diff --git a/tests/always/test_providers_manager.py b/tests/always/test_providers_manager.py
index 663eea393b..31a6f70823 100644
--- a/tests/always/test_providers_manager.py
+++ b/tests/always/test_providers_manager.py
@@ -36,6 +36,11 @@ class TestProviderManager:
def inject_fixtures(self, caplog):
self._caplog = caplog
+ @pytest.fixture(autouse=True, scope="function")
+ def clean(self):
+ """The tests depend on a clean state of a ProvidersManager."""
+ ProvidersManager().__init__()
+
def test_providers_are_loaded(self):
with self._caplog.at_level(logging.WARNING):
provider_manager = ProvidersManager()
diff --git a/tests/utils/test_singleton.py b/tests/utils/test_singleton.py
new file mode 100644
index 0000000000..57145fe7b9
--- /dev/null
+++ b/tests/utils/test_singleton.py
@@ -0,0 +1,65 @@
+# 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 annotations
+
+from airflow.utils.singleton import Singleton
+
+
+class A(metaclass=Singleton):
+ pass
+
+
+class Counter(metaclass=Singleton):
+ """Singleton class that counts how much __init__ and count was called."""
+
+ counter = 0
+
+ def __init__(self):
+ self.counter += 1
+
+ def count(self):
+ self.counter += 1
+
+
+def test_singleton_refers_to_same_instance():
+ a, b = A(), A()
+ assert a is b
+
+
+def test_singleton_after_out_of_context_does_refer_to_same_instance():
+ # check if setting something on singleton is preserved after instance goes out of context
+ def x():
+ a = A()
+ a.a = "a"
+
+ x()
+ b = A()
+ assert b.a == "a"
+
+
+def test_singleton_does_not_call_init_second_time():
+ # first creation of Counter, check if __init__ is called
+ c = Counter()
+ assert c.counter == 1
+
+ # check if "new instance" calls __init__ - it shouldn't
+ d = Counter()
+ assert c.counter == 1
+
+ # check if incrementing "new instance" increments counter on previous one
+ d.count()
+ assert c.counter == 2