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 2022/11/09 20:22:43 UTC
[airflow] branch v2-4-test updated (9af3a9ce0c -> c7ee64fe11)
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a change to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
discard 9af3a9ce0c Fix mini scheduler expansion of mapped task (#27506)
discard 4cc75e82c3 SLAMiss is nullable and not always given back when pulling task instances (#27423)
discard b7d724154f Fix behavior of `_` when searching for DAGs (#27448)
discard 3d26efd080 Bump loader-utils from 1.4.0 to 1.4.1 in /airflow/www (#27552)
discard 2850e4109e Fix getting the dag/task ids from base executor (#27550)
discard 6e64cdbbfd Fix sqlalchemy primary key black-out error on DDRQ (#27538)
discard cb0cd0f39f Update index.rst (#27529)
discard 4e2a9ce6bc Fix wording in DAG Runs description(#27470)
discard 8fb9087c82 Document that KubernetesExecutor overwrites container args (#27450)
discard e17acd9c26 Fix BaseOperator link (#27441)
discard f1073dbcdb Correct timer units to seconds from milliseconds. (#27360)
discard 38b6ad0cbb fIx failing masking tests for python < 3.10 (#27337)
discard 1b4697ac07 fIx failing masking tests for python < 3.10 (#27337)
discard 8dd61d2b0f Fix typo (#27327)
discard b8a0ca046f Add missed import in the Trigger Rules example (#27309)
discard 6515eb966d Fix IntegrityError during webserver startup (#27297)
discard 51194c72c3 Add case insensitive constraint to username (#27266)
discard 9f6c9e47be Fix python external template keys (#27256)
discard 1700c2c4ed reduce extraneous task log requests (#27233)
discard cabe5a5228 Make RotatingFilehandler used in DagProcessor non-caching (#27223)
discard 16d7c3ef25 Listener: Set task on sqlalchemy taskinstance object (#27167)
discard 8b555ce117 Fix some bug in web ui dags list page (auto-refresh & jump search null state) (#27141)
discard 968704f717 Fix typo in set-up-database.rst (#27137)
discard e90f5091be Update SLA wording to reflect it is relative to Dag Run start. (#27111)
discard 6d7881e6c6 Update dags.rst (#27110)
discard dc3817cc2c Move TriggerDagRun conf check to execute (#27035)
new 3d15859add Update dags.rst (#27110)
new 8e58db81ff Update SLA wording to reflect it is relative to Dag Run start. (#27111)
new 8c7e2c3563 Fix typo in set-up-database.rst (#27137)
new 66125651b6 Fix some bug in web ui dags list page (auto-refresh & jump search null state) (#27141)
new 1eb047dbfb Listener: Set task on sqlalchemy taskinstance object (#27167)
new 4706f09699 Make RotatingFilehandler used in DagProcessor non-caching (#27223)
new 9c3ee045ac reduce extraneous task log requests (#27233)
new d3a450c3f9 Fix python external template keys (#27256)
new 1f74f02f42 Add case insensitive constraint to username (#27266)
new 2c5b336db2 Fix IntegrityError during webserver startup (#27297)
new dc33e07741 Add missed import in the Trigger Rules example (#27309)
new b0ff69b02b Fix typo (#27327)
new 076c1419e7 fIx failing masking tests for python < 3.10 (#27337)
new 910827c3f1 fIx failing masking tests for python < 3.10 (#27337)
new 6c5e5aa9db Correct timer units to seconds from milliseconds. (#27360)
new 7990b04423 Fix BaseOperator link (#27441)
new 1c6de53b7f Document that KubernetesExecutor overwrites container args (#27450)
new e3caa27ad5 Fix wording in DAG Runs description(#27470)
new 3b4ff89b34 Update index.rst (#27529)
new cafad9d7fb Fix sqlalchemy primary key black-out error on DDRQ (#27538)
new ef86a455ab Fix getting the dag/task ids from base executor (#27550)
new b928af0595 Bump loader-utils from 1.4.0 to 1.4.1 in /airflow/www (#27552)
new 6d0d162fe3 Fix behavior of `_` when searching for DAGs (#27448)
new c30180fc20 SLAMiss is nullable and not always given back when pulling task instances (#27423)
new 64ec6c7639 Fix mini scheduler expansion of mapped task (#27506)
new f0f67e8bc9 Replace FAB url filtering function with Airflows (#27576)
new 676c1ee9b3 Remove self.using_mysql attribute (#27571)
new f21f38ef61 Update version to 2.4.3
new c7ee64fe11 Add release notes
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 (9af3a9ce0c)
\
N -- N -- N refs/heads/v2-4-test (c7ee64fe11)
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 29 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:
README.md | 14 +++----
RELEASE_NOTES.rst | 47 ++++++++++++++++++++++
airflow/jobs/scheduler_job.py | 1 -
airflow/operators/trigger_dagrun.py | 10 ++---
airflow/utils/db.py | 1 +
airflow/www/extensions/init_appbuilder.py | 12 ++++++
.../installation/supported-versions.rst | 2 +-
docs/docker-stack/README.md | 10 ++---
.../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 +++----
newsfragments/27065.misc.rst | 1 -
.../ci/pre_commit/pre_commit_supported_versions.py | 2 +-
setup.py | 2 +-
tests/operators/test_trigger_dagrun.py | 14 +++----
21 files changed, 102 insertions(+), 44 deletions(-)
delete mode 100644 newsfragments/27065.misc.rst
[airflow] 13/29: fIx failing masking tests for python < 3.10 (#27337)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 076c1419e79dd3438f42d4f12b24fd6489bc360c
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Fri Oct 28 09:36:27 2022 +0200
fIx failing masking tests for python < 3.10 (#27337)
Seems that the number of times user is printed in stack trace
depend on Python version. The fix in #27335 seems to only have
worked for Python 3.10 with the 1.0.0 of exceptiongroup the
stack trace has less stack levels.
(cherry picked from commit fcec804ff513d8f148d5069a48e7adf703855a96)
---
setup.py | 2 +-
tests/utils/log/test_secrets_masker.py | 27 ++-------------------------
2 files changed, 3 insertions(+), 26 deletions(-)
diff --git a/setup.py b/setup.py
index d1eea7e550..8d1b19af1d 100644
--- a/setup.py
+++ b/setup.py
@@ -202,7 +202,7 @@ def write_version(filename: str = str(AIRFLOW_SOURCES_ROOT / "airflow" / "git_ve
"""
Write the Semver version + git hash to file, e.g. ".dev0+2f635dc265e78db6708f59f68e8009abb92c1e65".
- :param str filename: Destination file to write
+ :param str filename: Destination file to write.
"""
text = f"{git_version(version)}"
with open(filename, 'w') as file:
diff --git a/tests/utils/log/test_secrets_masker.py b/tests/utils/log/test_secrets_masker.py
index 6de30eb574..edc516f04a 100644
--- a/tests/utils/log/test_secrets_masker.py
+++ b/tests/utils/log/test_secrets_masker.py
@@ -169,31 +169,8 @@ class TestSecretsMasker:
except RuntimeError:
logger.exception("Err")
- line = lineno() - 8
-
- assert caplog.text == textwrap.dedent(
- f"""\
- ERROR Err
- Traceback (most recent call last):
- File ".../test_secrets_masker.py", line {line}, in test_masking_in_implicit_context_exceptions
- raise RuntimeError(f"Cannot connect to user:{{p}}")
- RuntimeError: Cannot connect to user:***
-
- During handling of the above exception, another exception occurred:
-
- Traceback (most recent call last):
- File ".../test_secrets_masker.py", line {line+2}, in test_masking_in_implicit_context_exceptions
- raise RuntimeError(f'Exception: {{ex1}}')
- RuntimeError: Exception: Cannot connect to user:***
-
- During handling of the above exception, another exception occurred:
-
- Traceback (most recent call last):
- File ".../test_secrets_masker.py", line {line+4}, in test_masking_in_implicit_context_exceptions
- raise RuntimeError(f'Exception: {{ex2}}')
- RuntimeError: Exception: Exception: Cannot connect to user:***
- """
- )
+ assert "user:password" not in caplog.text
+ assert caplog.text.count("user:***") >= 2
def test_masking_in_explicit_context_exceptions(self, logger, caplog):
"""
[airflow] 08/29: Fix python external template keys (#27256)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d3a450c3f952ed9e280bba510a8ec61e91a2fa21
Author: enisnazif <en...@gmail.com>
AuthorDate: Mon Oct 31 04:34:30 2022 +0000
Fix python external template keys (#27256)
(cherry picked from commit 27a92fecc9be30c9b1268beb60db44d2c7b3628f)
---
airflow/operators/python.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow/operators/python.py b/airflow/operators/python.py
index 538e805cce..133779f088 100644
--- a/airflow/operators/python.py
+++ b/airflow/operators/python.py
@@ -621,7 +621,7 @@ class ExternalPythonOperator(_BasePythonVirtualenvOperator):
macros when starting.
"""
- template_fields: Sequence[str] = tuple({'python_path'} | set(PythonOperator.template_fields))
+ template_fields: Sequence[str] = tuple({'python'} | set(PythonOperator.template_fields))
def __init__(
self,
[airflow] 12/29: Fix typo (#27327)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit b0ff69b02b932f74ac3d90c974b66f11e59b90d2
Author: bmtKIA6 <bm...@users.noreply.github.com>
AuthorDate: Thu Oct 27 14:04:23 2022 -0400
Fix typo (#27327)
(cherry picked from commit 9c73b3f7fc1d18925d0ed09e8719f53b8147b0f2)
---
RELEASE_NOTES.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/RELEASE_NOTES.rst b/RELEASE_NOTES.rst
index f126faf57c..46d9a702f3 100644
--- a/RELEASE_NOTES.rst
+++ b/RELEASE_NOTES.rst
@@ -158,7 +158,7 @@ pass a list of 1 or more Datasets:
.. code-block:: python
- with DAG(dag_id='dataset-consmer', schedule=[dataset]):
+ with DAG(dag_id='dataset-consumer', schedule=[dataset]):
...
And to mark a task as producing a dataset pass the dataset(s) to the ``outlets`` attribute:
[airflow] 09/29: Add case insensitive constraint to username (#27266)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1f74f02f424c4e160e39073167d8368c9d20cd68
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Mon Oct 31 18:03:37 2022 +0100
Add case insensitive constraint to username (#27266)
This helps us to recognize usernames properly
(cherry picked from commit 1d25105189db09df53327240a01f39e09e10b15a)
---
...e_insensitive_unique_constraint_for_username.py | 89 ++++++++++++++++++++++
airflow/utils/db.py | 33 +++++++-
airflow/www/fab_security/sqla/manager.py | 2 +-
airflow/www/fab_security/sqla/models.py | 36 ++++++++-
docs/apache-airflow/img/airflow_erd.sha256 | 2 +-
docs/apache-airflow/migrations-ref.rst | 4 +-
tests/www/test_security.py | 21 ++++-
7 files changed, 179 insertions(+), 8 deletions(-)
diff --git a/airflow/migrations/versions/0119_2_4_3_add_case_insensitive_unique_constraint_for_username.py b/airflow/migrations/versions/0119_2_4_3_add_case_insensitive_unique_constraint_for_username.py
new file mode 100644
index 0000000000..7648be8cc1
--- /dev/null
+++ b/airflow/migrations/versions/0119_2_4_3_add_case_insensitive_unique_constraint_for_username.py
@@ -0,0 +1,89 @@
+#
+# 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.
+
+"""Add case-insensitive unique constraint for username
+
+Revision ID: e07f49787c9d
+Revises: b0d31815b5a6
+Create Date: 2022-10-25 17:29:46.432326
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'e07f49787c9d'
+down_revision = 'b0d31815b5a6'
+branch_labels = None
+depends_on = None
+airflow_version = '2.4.3'
+
+
+def upgrade():
+ """Apply Add case-insensitive unique constraint"""
+ conn = op.get_bind()
+ if conn.dialect.name == 'postgresql':
+ op.create_index('idx_ab_user_username', 'ab_user', [sa.text('LOWER(username)')], unique=True)
+ op.create_index(
+ "idx_ab_register_user_username", 'ab_register_user', [sa.text('LOWER(username)')], unique=True
+ )
+ elif conn.dialect.name == 'sqlite':
+ with op.batch_alter_table('ab_user') as batch_op:
+ batch_op.alter_column(
+ 'username',
+ existing_type=sa.String(64),
+ _type=sa.String(64, collation='NOCASE'),
+ unique=True,
+ nullable=False,
+ )
+ with op.batch_alter_table('ab_register_user') as batch_op:
+ batch_op.alter_column(
+ 'username',
+ existing_type=sa.String(64),
+ _type=sa.String(64, collation='NOCASE'),
+ unique=True,
+ nullable=False,
+ )
+
+
+def downgrade():
+ """Unapply Add case-insensitive unique constraint"""
+ conn = op.get_bind()
+ if conn.dialect.name == 'postgresql':
+ op.drop_index('idx_ab_user_username', table_name='ab_user')
+ op.drop_index('idx_ab_register_user_username', table_name='ab_register_user')
+ elif conn.dialect.name == 'sqlite':
+ with op.batch_alter_table('ab_user') as batch_op:
+ batch_op.alter_column(
+ 'username',
+ existing_type=sa.String(64, collation='NOCASE'),
+ _type=sa.String(64),
+ unique=True,
+ nullable=False,
+ )
+ with op.batch_alter_table('ab_register_user') as batch_op:
+ batch_op.alter_column(
+ 'username',
+ existing_type=sa.String(64, collation='NOCASE'),
+ _type=sa.String(64),
+ unique=True,
+ nullable=False,
+ )
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index 08066a59f7..ab9a9c4d11 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -919,6 +919,36 @@ def check_conn_id_duplicates(session: Session) -> Iterable[str]:
)
+def check_username_duplicates(session: Session) -> Iterable[str]:
+ """
+ Check unique username in User & RegisterUser table
+
+ :param session: session of the sqlalchemy
+ :rtype: str
+ """
+ from airflow.www.fab_security.sqla.models import RegisterUser, User
+
+ for model in [User, RegisterUser]:
+ dups = []
+ try:
+ dups = (
+ session.query(model.username) # type: ignore[attr-defined]
+ .group_by(model.username) # type: ignore[attr-defined]
+ .having(func.count() > 1)
+ .all()
+ )
+ except (exc.OperationalError, exc.ProgrammingError):
+ # fallback if tables hasn't been created yet
+ session.rollback()
+ if dups:
+ yield (
+ f'Seems you have mixed case usernames in {model.__table__.name} table.\n' # type: ignore
+ 'You have to rename or delete those mixed case usernames '
+ 'before upgrading the database.\n'
+ f'usernames with mixed cases: {[dup.username for dup in dups]}'
+ )
+
+
def reflect_tables(tables: list[Base | str] | None, session):
"""
When running checks prior to upgrades, we use reflection to determine current state of the
@@ -1393,6 +1423,7 @@ def _check_migration_errors(session: Session = NEW_SESSION) -> Iterable[str]:
check_conn_type_null,
check_run_id_null,
check_bad_references,
+ check_username_duplicates,
)
for check_fn in check_functions:
log.debug("running check function %s", check_fn.__name__)
@@ -1679,7 +1710,7 @@ def drop_flask_models(connection):
:param connection: SQLAlchemy Connection
:return: None
"""
- from flask_appbuilder.models.sqla import Base
+ from airflow.www.fab_security.sqla.models import Base
Base.metadata.drop_all(connection)
diff --git a/airflow/www/fab_security/sqla/manager.py b/airflow/www/fab_security/sqla/manager.py
index bbc44296fa..7300687f8a 100644
--- a/airflow/www/fab_security/sqla/manager.py
+++ b/airflow/www/fab_security/sqla/manager.py
@@ -168,7 +168,7 @@ class SecurityManager(BaseSecurityManager):
else:
return (
self.get_session.query(self.user_model)
- .filter(self.user_model.username == username)
+ .filter(func.lower(self.user_model.username) == func.lower(username))
.one_or_none()
)
except MultipleResultsFound:
diff --git a/airflow/www/fab_security/sqla/models.py b/airflow/www/fab_security/sqla/models.py
index a0b78d6083..b7a35d1d45 100644
--- a/airflow/www/fab_security/sqla/models.py
+++ b/airflow/www/fab_security/sqla/models.py
@@ -25,7 +25,19 @@ from typing import TYPE_CHECKING
from flask import current_app, g
from flask_appbuilder.models.sqla import Model
-from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Table, UniqueConstraint
+from sqlalchemy import (
+ Boolean,
+ Column,
+ DateTime,
+ ForeignKey,
+ Index,
+ Integer,
+ String,
+ Table,
+ UniqueConstraint,
+ event,
+ func,
+)
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import backref, relationship
@@ -135,7 +147,9 @@ class User(Model):
id = Column(Integer, primary_key=True)
first_name = Column(String(64), nullable=False)
last_name = Column(String(64), nullable=False)
- username = Column(String(256), unique=True, nullable=False)
+ username = Column(
+ String(256).with_variant(String(256, collation='NOCASE'), "sqlite"), unique=True, nullable=False
+ )
password = Column(String(256))
active = Column(Boolean)
email = Column(String(256), unique=True, nullable=False)
@@ -228,8 +242,24 @@ class RegisterUser(Model):
id = Column(Integer, primary_key=True)
first_name = Column(String(64), nullable=False)
last_name = Column(String(64), nullable=False)
- username = Column(String(256), unique=True, nullable=False)
+ username = Column(
+ String(256).with_variant(String(256, collation='NOCASE'), "sqlite"), unique=True, nullable=False
+ )
password = Column(String(256))
email = Column(String(256), nullable=False)
registration_date = Column(DateTime, default=datetime.datetime.now, nullable=True)
registration_hash = Column(String(256))
+
+
+@event.listens_for(User.__table__, "before_create")
+def add_index_on_ab_user_username_postgres(table, conn, **kw):
+ if conn.dialect.name != "postgresql":
+ return
+ table.indexes.add(Index("idx_ab_user_username", func.lower(table.c.username), unique=True))
+
+
+@event.listens_for(RegisterUser.__table__, "before_create")
+def add_index_on_ab_register_user_username_postgres(table, conn, **kw):
+ if conn.dialect.name != "postgresql":
+ return
+ table.indexes.add(Index("idx_ab_register_user_username", func.lower(table.c.username), unique=True))
diff --git a/docs/apache-airflow/img/airflow_erd.sha256 b/docs/apache-airflow/img/airflow_erd.sha256
index da12a349e2..6fa283f045 100644
--- a/docs/apache-airflow/img/airflow_erd.sha256
+++ b/docs/apache-airflow/img/airflow_erd.sha256
@@ -1 +1 @@
-a522ff773bc6403318e4a15be19f25d48323208de3acc8a970f7f40ad4782563
\ No newline at end of file
+88fe8ef077e673c69080ac260baa0bab33d9189035c0dea5cc652c990ef44ee8
\ No newline at end of file
diff --git a/docs/apache-airflow/migrations-ref.rst b/docs/apache-airflow/migrations-ref.rst
index 0a42c7e5d3..75694f4ded 100644
--- a/docs/apache-airflow/migrations-ref.rst
+++ b/docs/apache-airflow/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
+---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description |
+=================================+===================+===================+==============================================================+
-| ``b0d31815b5a6`` (head) | ``ecb43d2a1842`` | ``2.4.2`` | Add missing auto-increment to columns on FAB tables |
+| ``e07f49787c9d`` (head) | ``b0d31815b5a6`` | ``2.4.3`` | Add case-insensitive unique constraint for username |
++---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
+| ``b0d31815b5a6`` | ``ecb43d2a1842`` | ``2.4.2`` | Add missing auto-increment to columns on FAB tables |
+---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
| ``ecb43d2a1842`` | ``1486deb605b4`` | ``2.4.0`` | Add processor_subdir column to DagModel, SerializedDagModel |
| | | | and CallbackRequest tables. |
diff --git a/tests/www/test_security.py b/tests/www/test_security.py
index df008acc30..3c74dc9c45 100644
--- a/tests/www/test_security.py
+++ b/tests/www/test_security.py
@@ -37,7 +37,13 @@ from airflow.www import app as application
from airflow.www.fab_security.manager import AnonymousUser
from airflow.www.fab_security.sqla.models import User, assoc_permission_role
from airflow.www.utils import CustomSQLAInterface
-from tests.test_utils.api_connexion_utils import create_user_scope, delete_role, set_user_single_role
+from tests.test_utils.api_connexion_utils import (
+ create_user,
+ create_user_scope,
+ delete_role,
+ delete_user,
+ set_user_single_role,
+)
from tests.test_utils.asserts import assert_queries_count
from tests.test_utils.db import clear_db_dags, clear_db_runs
from tests.test_utils.mock_security_manager import MockSecurityManager
@@ -935,3 +941,16 @@ def test_update_user_auth_stat_subsequent_unsuccessful_auth(mock_security_manage
assert old_user.fail_login_count == 10
assert old_user.last_login == datetime.datetime(1984, 12, 1, 0, 0, 0)
assert mock_security_manager.update_user.called_once
+
+
+def test_users_can_be_found(app, security_manager, session, caplog):
+ """Test that usernames are case insensitive"""
+ create_user(app, "Test")
+ create_user(app, "test")
+ create_user(app, "TEST")
+ create_user(app, "TeSt")
+ assert security_manager.find_user("Test")
+ users = security_manager.get_all_users()
+ assert len(users) == 1
+ delete_user(app, "Test")
+ assert "Error adding new user to database" in caplog.text
[airflow] 16/29: Fix BaseOperator link (#27441)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7990b0442310d62b009cf620cbc4a562dcb90781
Author: Ruben Laguna <ru...@gmail.com>
AuthorDate: Tue Nov 1 19:45:43 2022 +0100
Fix BaseOperator link (#27441)
(cherry picked from commit 8829e999077040248ab302faa76897915430bfd3)
---
docs/apache-airflow/executor/celery.rst | 2 +-
docs/apache-airflow/python-api-ref.rst | 8 ++++----
2 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/docs/apache-airflow/executor/celery.rst b/docs/apache-airflow/executor/celery.rst
index f61cff47ad..a904b2d1f4 100644
--- a/docs/apache-airflow/executor/celery.rst
+++ b/docs/apache-airflow/executor/celery.rst
@@ -182,7 +182,7 @@ Two databases are also available:
During this process, two 2 process are created:
- LocalTaskJobProcess - It logic is described by LocalTaskJob. It is monitoring RawTaskProcess. New processes are started using TaskRunner.
-- RawTaskProcess - It is process with the user code e.g. :meth:`~airflow.models.BaseOperator.execute`.
+- RawTaskProcess - It is process with the user code e.g. :meth:`~airflow.models.baseoperator.BaseOperator.execute`.
| [1] **SchedulerProcess** processes the tasks and when it finds a task that needs to be done, sends it to the **QueueBroker**.
| [2] **SchedulerProcess** also begins to periodically query **ResultBackend** for the status of the task.
diff --git a/docs/apache-airflow/python-api-ref.rst b/docs/apache-airflow/python-api-ref.rst
index dc04c6a93b..9e385f846d 100644
--- a/docs/apache-airflow/python-api-ref.rst
+++ b/docs/apache-airflow/python-api-ref.rst
@@ -31,7 +31,7 @@ The DAG is Airflow's core model that represents a recurring workflow. Check out
Operators
---------
Operators allow for generation of certain types of tasks that become nodes in
-the DAG when instantiated. All operators derive from :class:`~airflow.models.BaseOperator` and
+the DAG when instantiated. All operators derive from :class:`~airflow.models.baseoperator.BaseOperator` and
inherit many attributes and methods that way.
There are 3 main types of operators:
@@ -47,16 +47,16 @@ There are 3 main types of operators:
BaseOperator
''''''''''''
-All operators are derived from :class:`~airflow.models.BaseOperator` and acquire much
+All operators are derived from :class:`~airflow.models.baseoperator.BaseOperator` and acquire much
functionality through inheritance. Since this is the core of the engine,
-it's worth taking the time to understand the parameters of :class:`~airflow.models.BaseOperator`
+it's worth taking the time to understand the parameters of :class:`~airflow.models.baseoperator.BaseOperator`
to understand the primitive features that can be leveraged in your
DAGs.
BaseSensorOperator
''''''''''''''''''
All sensors are derived from :class:`~airflow.sensors.base.BaseSensorOperator`. All sensors inherit
-the :attr:`~airflow.sensors.base.BaseSensorOperator.timeout` and :attr:`~airflow.sensors.base.BaseSensorOperator.poke_interval` on top of the :class:`~airflow.models.BaseOperator`
+the :attr:`~airflow.sensors.base.BaseSensorOperator.timeout` and :attr:`~airflow.sensors.base.BaseSensorOperator.poke_interval` on top of the :class:`~airflow.models.baseoperator.BaseOperator`
attributes.
Operators packages
[airflow] 25/29: Fix mini scheduler expansion of mapped task (#27506)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 64ec6c7639323cec6fd8b3525fbf7e6a5537e5bd
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Nov 9 15:05:59 2022 +0100
Fix mini scheduler expansion of mapped task (#27506)
We have a case where the mini scheduler tries to expand a mapped task even when the downstream tasks are not yet done.
The mini scheduler extracts a partial subset of a dag and in the process, some upstream tasks are dropped.
If the task happens to be a mapped task, the expansion will fail since it needs the upstream output to make the expansion. When the expansion fails, the task is marked as `upstream_failed`. This leads to other downstream tasks being marked as upstream failed.
The solution was to ignore this error and not mark the mapped task as upstream_failed when the expansion fails and the dag is a partial subset
Co-authored-by: Ash Berlin-Taylor <as...@firemirror.com>
(cherry picked from commit ed92e5d521f958642615b038ec13068b527db1c4)
---
airflow/jobs/local_task_job.py | 59 +---------------------------
airflow/models/mappedoperator.py | 30 +++++++++-----
airflow/models/taskinstance.py | 61 ++++++++++++++++++++++++++++
tests/jobs/test_local_task_job.py | 1 -
tests/models/test_taskinstance.py | 83 +++++++++++++++++++++++++++++++++++++++
5 files changed, 165 insertions(+), 69 deletions(-)
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 1881511f9b..698c469dbb 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -18,25 +18,20 @@
from __future__ import annotations
import signal
-from typing import TYPE_CHECKING
import psutil
-from sqlalchemy.exc import OperationalError
from airflow.configuration import conf
from airflow.exceptions import AirflowException
from airflow.jobs.base_job import BaseJob
from airflow.listeners.events import register_task_instance_state_events
from airflow.listeners.listener import get_listener_manager
-from airflow.models.dagrun import DagRun
from airflow.models.taskinstance import TaskInstance
-from airflow.sentry import Sentry
from airflow.stats import Stats
from airflow.task.task_runner import get_task_runner
from airflow.utils import timezone
from airflow.utils.net import get_hostname
from airflow.utils.session import provide_session
-from airflow.utils.sqlalchemy import with_row_locks
from airflow.utils.state import State
@@ -165,7 +160,7 @@ class LocalTaskJob(BaseJob):
if not self.task_instance.test_mode:
if conf.getboolean('scheduler', 'schedule_after_task_execution', fallback=True):
- self._run_mini_scheduler_on_child_tasks()
+ self.task_instance.schedule_downstream_tasks()
def on_kill(self):
self.task_runner.terminate()
@@ -230,58 +225,6 @@ class LocalTaskJob(BaseJob):
self.terminating = True
self._state_change_checks += 1
- @provide_session
- @Sentry.enrich_errors
- def _run_mini_scheduler_on_child_tasks(self, session=None) -> None:
- try:
- # Re-select the row with a lock
- dag_run = with_row_locks(
- session.query(DagRun).filter_by(
- dag_id=self.dag_id,
- run_id=self.task_instance.run_id,
- ),
- session=session,
- ).one()
-
- task = self.task_instance.task
- if TYPE_CHECKING:
- assert task.dag
-
- # Get a partial DAG with just the specific tasks we want to examine.
- # In order for dep checks to work correctly, we include ourself (so
- # TriggerRuleDep can check the state of the task we just executed).
- partial_dag = task.dag.partial_subset(
- task.downstream_task_ids,
- include_downstream=True,
- include_upstream=False,
- include_direct_upstream=True,
- )
-
- dag_run.dag = partial_dag
- info = dag_run.task_instance_scheduling_decisions(session)
-
- skippable_task_ids = {
- task_id for task_id in partial_dag.task_ids if task_id not in task.downstream_task_ids
- }
-
- schedulable_tis = [ti for ti in info.schedulable_tis if ti.task_id not in skippable_task_ids]
- for schedulable_ti in schedulable_tis:
- if not hasattr(schedulable_ti, "task"):
- schedulable_ti.task = task.dag.get_task(schedulable_ti.task_id)
-
- num = dag_run.schedule_tis(schedulable_tis)
- self.log.info("%d downstream tasks scheduled from follow-on schedule check", num)
-
- session.commit()
- except OperationalError as e:
- # Any kind of DB error here is _non fatal_ as this block is just an optimisation.
- self.log.info(
- "Skipping mini scheduling run due to exception: %s",
- e.statement,
- exc_info=True,
- )
- session.rollback()
-
@staticmethod
def _enable_task_listeners():
"""
diff --git a/airflow/models/mappedoperator.py b/airflow/models/mappedoperator.py
index 62cc22f379..9c591bf364 100644
--- a/airflow/models/mappedoperator.py
+++ b/airflow/models/mappedoperator.py
@@ -619,13 +619,18 @@ class MappedOperator(AbstractOperator):
try:
total_length = self._get_specified_expand_input().get_total_map_length(run_id, session=session)
except NotFullyPopulated as e:
- self.log.info(
- "Cannot expand %r for run %s; missing upstream values: %s",
- self,
- run_id,
- sorted(e.missing),
- )
total_length = None
+ # partial dags comes from the mini scheduler. It's
+ # possible that the upstream tasks are not yet done,
+ # but we don't have upstream of upstreams in partial dags,
+ # so we ignore this exception.
+ if not self.dag or not self.dag.partial:
+ self.log.error(
+ "Cannot expand %r for run %s; missing upstream values: %s",
+ self,
+ run_id,
+ sorted(e.missing),
+ )
state: TaskInstanceState | None = None
unmapped_ti: TaskInstance | None = (
@@ -646,10 +651,15 @@ class MappedOperator(AbstractOperator):
# The unmapped task instance still exists and is unfinished, i.e. we
# haven't tried to run it before.
if total_length is None:
- # If the map length cannot be calculated (due to unavailable
- # upstream sources), fail the unmapped task.
- unmapped_ti.state = TaskInstanceState.UPSTREAM_FAILED
- indexes_to_map: Iterable[int] = ()
+ if self.dag and self.dag.partial:
+ # If the DAG is partial, it's likely that the upstream tasks
+ # are not done yet, so we do nothing
+ indexes_to_map: Iterable[int] = ()
+ else:
+ # If the map length cannot be calculated (due to unavailable
+ # upstream sources), fail the unmapped task.
+ unmapped_ti.state = TaskInstanceState.UPSTREAM_FAILED
+ indexes_to_map = ()
elif total_length < 1:
# If the upstream maps this to a zero-length value, simply mark
# the unmapped task instance as SKIPPED (if needed).
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index d4453ca842..4388d592e9 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -2572,6 +2572,67 @@ class TaskInstance(Base, LoggingMixin):
return filters[0]
return or_(*filters)
+ @Sentry.enrich_errors
+ @provide_session
+ def schedule_downstream_tasks(self, session=None):
+ """
+ The mini-scheduler for scheduling downstream tasks of this task instance
+ :meta: private
+ """
+ from sqlalchemy.exc import OperationalError
+
+ from airflow.models import DagRun
+
+ try:
+ # Re-select the row with a lock
+ dag_run = with_row_locks(
+ session.query(DagRun).filter_by(
+ dag_id=self.dag_id,
+ run_id=self.run_id,
+ ),
+ session=session,
+ ).one()
+
+ task = self.task
+ if TYPE_CHECKING:
+ assert task.dag
+
+ # Get a partial DAG with just the specific tasks we want to examine.
+ # In order for dep checks to work correctly, we include ourself (so
+ # TriggerRuleDep can check the state of the task we just executed).
+ partial_dag = task.dag.partial_subset(
+ task.downstream_task_ids,
+ include_downstream=True,
+ include_upstream=False,
+ include_direct_upstream=True,
+ )
+
+ dag_run.dag = partial_dag
+ info = dag_run.task_instance_scheduling_decisions(session)
+
+ skippable_task_ids = {
+ task_id for task_id in partial_dag.task_ids if task_id not in task.downstream_task_ids
+ }
+
+ schedulable_tis = [ti for ti in info.schedulable_tis if ti.task_id not in skippable_task_ids]
+ for schedulable_ti in schedulable_tis:
+ if not hasattr(schedulable_ti, "task"):
+ schedulable_ti.task = task.dag.get_task(schedulable_ti.task_id)
+
+ num = dag_run.schedule_tis(schedulable_tis, session=session)
+ self.log.info("%d downstream tasks scheduled from follow-on schedule check", num)
+
+ session.flush()
+
+ except OperationalError as e:
+ # Any kind of DB error here is _non fatal_ as this block is just an optimisation.
+ self.log.info(
+ "Skipping mini scheduling run due to exception: %s",
+ e.statement,
+ exc_info=True,
+ )
+ session.rollback()
+
# State of the task instance.
# Stores string version of the task state.
diff --git a/tests/jobs/test_local_task_job.py b/tests/jobs/test_local_task_job.py
index 61bce800c8..26f5e629a7 100644
--- a/tests/jobs/test_local_task_job.py
+++ b/tests/jobs/test_local_task_job.py
@@ -739,7 +739,6 @@ class TestLocalTaskJob:
ti2_l.refresh_from_db()
assert ti2_k.state == State.SUCCESS
assert ti2_l.state == State.NONE
- assert "0 downstream tasks scheduled from follow-on schedule" in caplog.text
failed_deps = list(ti2_l.get_failed_dep_statuses())
assert len(failed_deps) == 1
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index f4edebcb08..7f7cca0da2 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -3614,3 +3614,86 @@ def test_expand_non_templated_field(dag_maker, session):
echo_task = dag.get_task("echo")
assert "get_extra_env" in echo_task.upstream_task_ids
+
+
+def test_mapped_task_does_not_error_in_mini_scheduler_if_upstreams_are_not_done(dag_maker, caplog, session):
+ """
+ This tests that when scheduling child tasks of a task and there's a mapped downstream task,
+ if the mapped downstream task has upstreams that are not yet done, the mapped downstream task is
+ not marked as `upstream_failed'
+ """
+ with dag_maker() as dag:
+
+ @dag.task
+ def second_task():
+ return [0, 1, 2]
+
+ @dag.task
+ def first_task():
+ print(2)
+
+ @dag.task
+ def middle_task(id):
+ return id
+
+ middle = middle_task.expand(id=second_task())
+
+ @dag.task
+ def last_task():
+ print(3)
+
+ [first_task(), middle] >> last_task()
+
+ dag_run = dag_maker.create_dagrun()
+ first_ti = dag_run.get_task_instance(task_id="first_task")
+ second_ti = dag_run.get_task_instance(task_id="second_task")
+ first_ti.state = State.SUCCESS
+ second_ti.state = State.RUNNING
+ session.merge(first_ti)
+ session.merge(second_ti)
+ session.commit()
+ first_ti.schedule_downstream_tasks(session=session)
+ middle_ti = dag_run.get_task_instance(task_id="middle_task")
+ assert middle_ti.state != State.UPSTREAM_FAILED
+ assert "0 downstream tasks scheduled from follow-on schedule" in caplog.text
+
+
+def test_mapped_task_expands_in_mini_scheduler_if_upstreams_are_done(dag_maker, caplog, session):
+ """Test that mini scheduler expands mapped task"""
+ with dag_maker() as dag:
+
+ @dag.task
+ def second_task():
+ return [0, 1, 2]
+
+ @dag.task
+ def first_task():
+ print(2)
+
+ @dag.task
+ def middle_task(id):
+ return id
+
+ middle = middle_task.expand(id=second_task())
+
+ @dag.task
+ def last_task():
+ print(3)
+
+ [first_task(), middle] >> last_task()
+
+ dr = dag_maker.create_dagrun()
+
+ first_ti = dr.get_task_instance(task_id="first_task")
+ first_ti.state = State.SUCCESS
+ session.merge(first_ti)
+ session.commit()
+ second_task = dag.get_task("second_task")
+ second_ti = dr.get_task_instance(task_id="second_task")
+ second_ti.refresh_from_task(second_task)
+ second_ti.run()
+ second_ti.schedule_downstream_tasks(session=session)
+ for i in range(3):
+ middle_ti = dr.get_task_instance(task_id="middle_task", map_index=i)
+ assert middle_ti.state == State.SCHEDULED
+ assert "3 downstream tasks scheduled from follow-on schedule" in caplog.text
[airflow] 23/29: Fix behavior of `_` when searching for DAGs (#27448)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6d0d162fe344e1687e6a28a72cd54feec38f9f8e
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Tue Nov 1 16:55:02 2022 -0700
Fix behavior of `_` when searching for DAGs (#27448)
On the homepage, when you search for DAGs, `_` wasn't behaving as most
people might expect since it is a single character wildcard in SQL. We
will automatically escape it now.
(cherry picked from commit eda089334864547e18b71cb22fa231872edd4e8b)
---
airflow/www/views.py | 5 +++--
tests/www/views/test_views_home.py | 10 +++++++++-
2 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 640adf84df..4b9c7c4bb5 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -647,9 +647,10 @@ class Airflow(AirflowBaseView):
dags_query = session.query(DagModel).filter(~DagModel.is_subdag, DagModel.is_active)
if arg_search_query:
+ escaped_arg_search_query = arg_search_query.replace("_", r"\_")
dags_query = dags_query.filter(
- DagModel.dag_id.ilike('%' + arg_search_query + '%')
- | DagModel.owners.ilike('%' + arg_search_query + '%')
+ DagModel.dag_id.ilike("%" + escaped_arg_search_query + "%", escape="\\")
+ | DagModel.owners.ilike("%" + escaped_arg_search_query + "%", escape="\\")
)
if arg_tags_filter:
diff --git a/tests/www/views/test_views_home.py b/tests/www/views/test_views_home.py
index 0d1719e89b..369e09ea14 100644
--- a/tests/www/views/test_views_home.py
+++ b/tests/www/views/test_views_home.py
@@ -118,7 +118,7 @@ def client_single_dag(app, user_single_dag):
)
-TEST_FILTER_DAG_IDS = ['filter_test_1', 'filter_test_2', 'a_first_dag_id_asc']
+TEST_FILTER_DAG_IDS = ["filter_test_1", "filter_test_2", "a_first_dag_id_asc", "filter.test"]
def _process_file(file_path, session):
@@ -185,6 +185,14 @@ def test_home_dag_list_filtered_singledag_user(working_dags, client_single_dag):
check_content_not_in_response(f"dag_id={dag_id}", resp)
+def test_home_dag_list_search(working_dags, user_client):
+ resp = user_client.get("home?search=filter_test", follow_redirects=True)
+ check_content_in_response("dag_id=filter_test_1", resp)
+ check_content_in_response("dag_id=filter_test_2", resp)
+ check_content_not_in_response("dag_id=filter.test", resp)
+ check_content_not_in_response("dag_id=a_first_dag_id_asc", resp)
+
+
def test_home_robots_header_in_response(user_client):
# Responses should include X-Robots-Tag header
resp = user_client.get('home', follow_redirects=True)
[airflow] 02/29: Update SLA wording to reflect it is relative to Dag Run start. (#27111)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 8e58db81ffeb8eada0df4a36b903bb3d2c0a33ad
Author: Damian Shaw <11...@users.noreply.github.com>
AuthorDate: Thu Oct 27 10:34:57 2022 -0400
Update SLA wording to reflect it is relative to Dag Run start. (#27111)
(cherry picked from commit 639210a7e0bfc3f04f28c7d7278292d2cae7234b)
---
docs/apache-airflow/concepts/tasks.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/concepts/tasks.rst b/docs/apache-airflow/concepts/tasks.rst
index 63fbe818e0..c3f9d1de3b 100644
--- a/docs/apache-airflow/concepts/tasks.rst
+++ b/docs/apache-airflow/concepts/tasks.rst
@@ -158,7 +158,7 @@ If you merely want to be notified if a task runs over but still let it run to co
SLAs
----
-An SLA, or a Service Level Agreement, is an expectation for the maximum time a Task should take. If a task takes longer than this to run, it is then visible in the "SLA Misses" part of the user interface, as well as going out in an email of all tasks that missed their SLA.
+An SLA, or a Service Level Agreement, is an expectation for the maximum time a Task should be completed relative to the Dag Run start time. If a task takes longer than this to run, it is then visible in the "SLA Misses" part of the user interface, as well as going out in an email of all tasks that missed their SLA.
Tasks over their SLA are not cancelled, though - they are allowed to run to completion. If you want to cancel a task after a certain runtime is reached, you want :ref:`concepts:timeouts` instead.
[airflow] 19/29: Update index.rst (#27529)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3b4ff89b344ad81df9c2fdab27ec98ca0d558408
Author: Cons Sh <um...@yandex.ru>
AuthorDate: Sun Nov 6 20:40:02 2022 +0400
Update index.rst (#27529)
docker-compose doesn'to work, we need to use docker compose now
(cherry picked from commit 2a4f81c1782044f3e9feb2dc968e56ebc0525a1a)
---
docs/apache-airflow/howto/docker-compose/index.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/howto/docker-compose/index.rst b/docs/apache-airflow/howto/docker-compose/index.rst
index 5cde8f7bc1..f9ec22d74f 100644
--- a/docs/apache-airflow/howto/docker-compose/index.rst
+++ b/docs/apache-airflow/howto/docker-compose/index.rst
@@ -125,7 +125,7 @@ On **all operating systems**, you need to run database migrations and create the
.. code-block:: bash
- docker-compose up airflow-init
+ docker compose up airflow-init
After initialization is complete, you should see a message like this:
[airflow] 15/29: Correct timer units to seconds from milliseconds. (#27360)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6c5e5aa9db046f10058a8228bb8a2a3c6ee98aae
Author: Jacob <ja...@gmail.com>
AuthorDate: Mon Nov 7 16:25:17 2022 -0600
Correct timer units to seconds from milliseconds. (#27360)
``dagrun.duration.success.<dag_id>``
``dagrun.schedule_delay.<dag_id>``
``dagrun.<dag_id>.first_task_scheduling_delay`` are reported to statsd in seconds, not milliseconds. Other timers in this list might be suspect, but I haven't checked them yet.
(cherry picked from commit 25d7ca918e2d382cccd7d70386a1055d33bb517b)
---
docs/apache-airflow/logging-monitoring/metrics.rst | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/docs/apache-airflow/logging-monitoring/metrics.rst b/docs/apache-airflow/logging-monitoring/metrics.rst
index d92327905f..a5baa902d0 100644
--- a/docs/apache-airflow/logging-monitoring/metrics.rst
+++ b/docs/apache-airflow/logging-monitoring/metrics.rst
@@ -155,12 +155,12 @@ Name Description
``dagrun.dependency-check.<dag_id>`` Milliseconds taken to check DAG dependencies
``dag.<dag_id>.<task_id>.duration`` Milliseconds taken to finish a task
``dag_processing.last_duration.<dag_file>`` Milliseconds taken to load the given DAG file
-``dagrun.duration.success.<dag_id>`` Milliseconds taken for a DagRun to reach success state
+``dagrun.duration.success.<dag_id>`` Seconds taken for a DagRun to reach success state
``dagrun.duration.failed.<dag_id>`` Milliseconds taken for a DagRun to reach failed state
-``dagrun.schedule_delay.<dag_id>`` Milliseconds of delay between the scheduled DagRun
+``dagrun.schedule_delay.<dag_id>`` Seconds of delay between the scheduled DagRun
start date and the actual DagRun start date
``scheduler.critical_section_duration`` Milliseconds spent in the critical section of scheduler loop --
only a single scheduler can enter this loop at a time
-``dagrun.<dag_id>.first_task_scheduling_delay`` Milliseconds elapsed between first task start_date and dagrun expected start
+``dagrun.<dag_id>.first_task_scheduling_delay`` Seconds elapsed between first task start_date and dagrun expected start
``collect_db_dags`` Milliseconds taken for fetching all Serialized Dags from DB
=================================================== ========================================================================
[airflow] 18/29: Fix wording in DAG Runs description(#27470)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e3caa27ad5dbe61b36b08dd6b5eb245b5771c6d9
Author: SnowOracle <33...@users.noreply.github.com>
AuthorDate: Thu Nov 3 15:17:12 2022 -0400
Fix wording in DAG Runs description(#27470)
Co-authored-by: Jed Cunningham <66...@users.noreply.github.com>
(cherry picked from commit 104ce1e7d81457a768366bd1ba319ecad934a921)
---
docs/apache-airflow/dag-run.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/dag-run.rst b/docs/apache-airflow/dag-run.rst
index 6124cfc90e..641b12ec12 100644
--- a/docs/apache-airflow/dag-run.rst
+++ b/docs/apache-airflow/dag-run.rst
@@ -19,7 +19,7 @@ DAG Runs
=========
A DAG Run is an object representing an instantiation of the DAG in time.
Any time the DAG is executed, a DAG Run is created and all tasks inside it are executed. The status of the DAG Run depends on the tasks states.
-Each DAG Run is run separately from another, meaning that you can have running DAG many times at the same time.
+Each DAG Run is run separately from one another, meaning that you can have many runs of a DAG at the same time.
.. _dag-run:dag-run-status:
[airflow] 06/29: Make RotatingFilehandler used in DagProcessor non-caching (#27223)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 4706f09699653383da68e58bbd666e9beda79aa7
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Tue Oct 25 10:38:40 2022 +0200
Make RotatingFilehandler used in DagProcessor non-caching (#27223)
The RotatingFileHandler is used when you enable it via
`CONFIG_PROCESSOR_MANAGER_LOGGER=True` and it exhibits similar
behaviour as the FileHandler had when it comes to caching
the file on the Kernel level. While it is harmless (the cache
will be freed when needed), it is also misleading for those who
are trying to understand memory usage by Airlfow.
The fix is to add a custom non-caching RotatingFileHandler similarly
as in #18054.
Note that it will require to manually modify local settings if
the settings were created before this change.
Fixes: #27065
(cherry picked from commit 126b7b8a073f75096d24378ffd749ce166267826)
---
airflow/config_templates/airflow_local_settings.py | 2 +-
airflow/utils/log/non_caching_file_handler.py | 47 ++++++++++++++++------
newsfragments/27065.misc.rst | 1 +
3 files changed, 37 insertions(+), 13 deletions(-)
diff --git a/airflow/config_templates/airflow_local_settings.py b/airflow/config_templates/airflow_local_settings.py
index 317544ca7e..aa5791a324 100644
--- a/airflow/config_templates/airflow_local_settings.py
+++ b/airflow/config_templates/airflow_local_settings.py
@@ -158,7 +158,7 @@ if EXTRA_LOGGER_NAMES:
DEFAULT_DAG_PARSING_LOGGING_CONFIG: dict[str, dict[str, dict[str, Any]]] = {
'handlers': {
'processor_manager': {
- 'class': 'logging.handlers.RotatingFileHandler',
+ 'class': 'airflow.utils.log.non_caching_file_handler.NonCachingRotatingFileHandler',
'formatter': 'airflow',
'filename': DAG_PROCESSOR_MANAGER_LOG_LOCATION,
'mode': 'a',
diff --git a/airflow/utils/log/non_caching_file_handler.py b/airflow/utils/log/non_caching_file_handler.py
index 5fbed3a1e1..46fe69a99b 100644
--- a/airflow/utils/log/non_caching_file_handler.py
+++ b/airflow/utils/log/non_caching_file_handler.py
@@ -16,11 +16,25 @@
# under the License.
from __future__ import annotations
-import logging
import os
+from logging import FileHandler
+from logging.handlers import RotatingFileHandler
+from typing import IO
-class NonCachingFileHandler(logging.FileHandler):
+def make_file_io_non_caching(io: IO[str]) -> IO[str]:
+ try:
+ fd = io.fileno()
+ os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED)
+ except Exception:
+ # in case either file descriptor cannot be retrieved or fadvise is not available
+ # we should simply return the wrapper retrieved by FileHandler's open method
+ # the advice to the kernel is just an advice and if we cannot give it, we won't
+ pass
+ return io
+
+
+class NonCachingFileHandler(FileHandler):
"""
This is an extension of the python FileHandler that advises the Kernel to not cache the file
in PageCache when it is written. While there is nothing wrong with such cache (it will be cleaned
@@ -35,13 +49,22 @@ class NonCachingFileHandler(logging.FileHandler):
"""
def _open(self):
- wrapper = super()._open()
- try:
- fd = wrapper.fileno()
- os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED)
- except Exception:
- # in case either file descriptor cannot be retrieved or fadvise is not available
- # we should simply return the wrapper retrieved by FileHandler's open method
- # the advise to the kernel is just an advise and if we cannot give it, we won't
- pass
- return wrapper
+ return make_file_io_non_caching(super()._open())
+
+
+class NonCachingRotatingFileHandler(RotatingFileHandler):
+ """
+ This is an extension of the python RotatingFileHandler that advises the Kernel to not cache the file
+ in PageCache when it is written. While there is nothing wrong with such cache (it will be cleaned
+ when memory is needed), it causes ever-growing memory usage when scheduler is running as it keeps
+ on writing new log files and the files are not rotated later on. This might lead to confusion
+ for our users, who are monitoring memory usage of Scheduler - without realising that it is
+ harmless and expected in this case.
+
+ See https://github.com/apache/airflow/issues/27065
+
+ Adding the advice to Kernel might help with not generating the cache memory growth in the first place.
+ """
+
+ def _open(self):
+ return make_file_io_non_caching(super()._open())
diff --git a/newsfragments/27065.misc.rst b/newsfragments/27065.misc.rst
new file mode 100644
index 0000000000..215bc8321d
--- /dev/null
+++ b/newsfragments/27065.misc.rst
@@ -0,0 +1 @@
+In case you want to decrease cache memory when ``CONFIG_PROCESSOR_MANAGER_LOGGER=True``, and you have your local settings created before, you can update ``processor_manager_handler`` to use ``airflow.utils.log.non_caching_file_handler.NonCachingRotatingFileHandler`` handler instead of ``logging.RotatingFileHandler``.
[airflow] 22/29: Bump loader-utils from 1.4.0 to 1.4.1 in /airflow/www (#27552)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit b928af0595d28b2255fd749d907af76ef49f66b8
Author: Dependabot [bot] <49...@users.noreply.github.com>
AuthorDate: Tue Nov 8 11:42:12 2022 -0500
Bump loader-utils from 1.4.0 to 1.4.1 in /airflow/www (#27552)
Bumps [loader-utils](https://github.com/webpack/loader-utils) from 1.4.0 to 1.4.1.
- [Release notes](https://github.com/webpack/loader-utils/releases)
- [Changelog](https://github.com/webpack/loader-utils/blob/v1.4.1/CHANGELOG.md)
- [Commits](https://github.com/webpack/loader-utils/compare/v1.4.0...v1.4.1)
---
updated-dependencies:
- dependency-name: loader-utils
dependency-type: indirect
...
Signed-off-by: dependabot[bot] <su...@github.com>
Signed-off-by: dependabot[bot] <su...@github.com>
Co-authored-by: dependabot[bot] <49...@users.noreply.github.com>
(cherry picked from commit 9936d61f7b1e4e122bdbdf099df75c48aa9bc033)
---
airflow/www/yarn.lock | 22 ++++++++++------------
1 file changed, 10 insertions(+), 12 deletions(-)
diff --git a/airflow/www/yarn.lock b/airflow/www/yarn.lock
index 14abfaf89a..d47debe4f4 100644
--- a/airflow/www/yarn.lock
+++ b/airflow/www/yarn.lock
@@ -7161,14 +7161,7 @@ json5@^1.0.1:
dependencies:
minimist "^1.2.0"
-json5@^2.1.2:
- version "2.2.0"
- resolved "https://registry.yarnpkg.com/json5/-/json5-2.2.0.tgz#2dfefe720c6ba525d9ebd909950f0515316c89a3"
- integrity sha512-f+8cldu7X/y7RAJurMEJmdoKXGB/X550w2Nr3tTbezL6RwEE/iMcm+tZnXeoZtKuOq6ft8+CqzEkrIgx1fPoQA==
- dependencies:
- minimist "^1.2.5"
-
-json5@^2.2.1:
+json5@^2.1.2, json5@^2.2.1:
version "2.2.1"
resolved "https://registry.yarnpkg.com/json5/-/json5-2.2.1.tgz#655d50ed1e6f95ad1a3caababd2b0efda10b395c"
integrity sha512-1hqLFMSrGHRHxav9q9gNjJ5EXznIxGVO09xQRrwplcS8qs28pZ8s8hupZAmqDwZUmVZ2Qb2jnyPOWcDH8m8dlA==
@@ -7253,9 +7246,9 @@ loader-runner@^4.2.0:
integrity sha512-3R/1M+yS3j5ou80Me59j7F9IMs4PXs3VqRrm0TU3AbKPxlmpoY1TNscJV/oGJXo8qCatFGTfDbY6W6ipGOYXfg==
loader-utils@^1.4.0:
- version "1.4.0"
- resolved "https://registry.yarnpkg.com/loader-utils/-/loader-utils-1.4.0.tgz#c579b5e34cb34b1a74edc6c1fb36bfa371d5a613"
- integrity sha512-qH0WSMBtn/oHuwjy/NucEgbx5dbxxnxup9s4PVXJUDHZBQY+s0NWA9rJf53RBnQZxfch7euUui7hpoAPvALZdA==
+ version "1.4.1"
+ resolved "https://registry.yarnpkg.com/loader-utils/-/loader-utils-1.4.1.tgz#278ad7006660bccc4d2c0c1578e17c5c78d5c0e0"
+ integrity sha512-1Qo97Y2oKaU+Ro2xnDMR26g1BwMT29jNbem1EvcujW2jqt+j5COXyscjM7bLQkM9HaxI7pkWeW7gnI072yMI9Q==
dependencies:
big.js "^5.2.2"
emojis-list "^3.0.0"
@@ -7601,7 +7594,12 @@ minimist-options@4.1.0:
is-plain-obj "^1.1.0"
kind-of "^6.0.3"
-minimist@^1.2.0, minimist@^1.2.5, minimist@^1.2.6:
+minimist@^1.2.0:
+ version "1.2.7"
+ resolved "https://registry.yarnpkg.com/minimist/-/minimist-1.2.7.tgz#daa1c4d91f507390437c6a8bc01078e7000c4d18"
+ integrity sha512-bzfL1YUZsP41gmu/qjrEk0Q6i2ix/cVeAhbCbqH9u3zYutS1cLg00qhrD0M2MVdCcx4Sc0UpP2eBWo9rotpq6g==
+
+minimist@^1.2.5, minimist@^1.2.6:
version "1.2.6"
resolved "https://registry.yarnpkg.com/minimist/-/minimist-1.2.6.tgz#8637a5b759ea0d6e98702cfb3a9283323c93af44"
integrity sha512-Jsjnk4bw3YJqYzbdyBiNsPWHPfO++UGG749Cxs6peCu5Xg4nrena6OVxOYxrQTqww0Jmwt+Ref8rggumkTLz9Q==
[airflow] 28/29: Update version to 2.4.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-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f21f38ef612b20c48c3e8f782f95cf11ed46a5dd
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Nov 9 18:55:14 2022 +0100
Update version to 2.4.3
---
README.md | 14 +++++++-------
airflow/utils/db.py | 1 +
docs/apache-airflow/installation/supported-versions.rst | 2 +-
docs/docker-stack/README.md | 10 +++++-----
.../docker-examples/extending/add-apt-packages/Dockerfile | 2 +-
.../extending/add-build-essential-extend/Dockerfile | 2 +-
.../docker-examples/extending/add-providers/Dockerfile | 2 +-
.../docker-examples/extending/add-pypi-packages/Dockerfile | 2 +-
.../extending/add-requirement-packages/Dockerfile | 2 +-
.../docker-examples/extending/custom-providers/Dockerfile | 2 +-
.../docker-examples/extending/embedding-dags/Dockerfile | 2 +-
.../extending/writable-directory/Dockerfile | 2 +-
docs/docker-stack/entrypoint.rst | 14 +++++++-------
scripts/ci/pre_commit/pre_commit_supported_versions.py | 2 +-
setup.py | 2 +-
15 files changed, 31 insertions(+), 30 deletions(-)
diff --git a/README.md b/README.md
index a2bfe6bd00..bdc6e6f62f 100644
--- a/README.md
+++ b/README.md
@@ -85,7 +85,7 @@ 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.4.2) |
+| | Main version (dev) | Stable version (2.4.3) |
|---------------------|------------------------------|------------------------------|
| Python | 3.7, 3.8, 3.9, 3.10 | 3.7, 3.8, 3.9, 3.10 |
| Platform | AMD64/ARM64(\*) | AMD64/ARM64(\*) |
@@ -158,15 +158,15 @@ them to the appropriate format and workflow that your tool requires.
```bash
-pip install 'apache-airflow==2.4.2' \
- --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.4.2/constraints-3.7.txt"
+pip install 'apache-airflow==2.4.3' \
+ --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.4.3/constraints-3.7.txt"
```
2. Installing with extras (i.e., postgres, google)
```bash
-pip install 'apache-airflow[postgres,google]==2.4.2' \
- --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.4.2/constraints-3.7.txt"
+pip install 'apache-airflow[postgres,google]==2.4.3' \
+ --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.4.3/constraints-3.7.txt"
```
For information on installing provider packages, check
@@ -271,7 +271,7 @@ Apache Airflow version life cycle:
| Version | Current Patch/Minor | State | First Release | Limited Support | EOL/Terminated |
|-----------|-----------------------|-----------|-----------------|-------------------|------------------|
-| 2 | 2.4.2 | Supported | Dec 17, 2020 | TBD | TBD |
+| 2 | 2.4.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 |
@@ -301,7 +301,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.4.2` images are Python 3.7 images. This means that default reference image will
+ and `apache/airflow:2.4.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/utils/db.py b/airflow/utils/db.py
index ab9a9c4d11..bb9302e0ba 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -74,6 +74,7 @@ REVISION_HEADS_MAP = {
"2.4.0": "ecb43d2a1842",
"2.4.1": "ecb43d2a1842",
"2.4.2": "b0d31815b5a6",
+ "2.4.3": "e07f49787c9d",
}
diff --git a/docs/apache-airflow/installation/supported-versions.rst b/docs/apache-airflow/installation/supported-versions.rst
index 98da679914..a296dc3844 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.4.2 Supported Dec 17, 2020 TBD TBD
+2 2.4.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 02118bd63b..8c5b2631d1 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.4.2`):
+You can find the following images there (Assuming Airflow version `2.4.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.4.2` - the versioned Airflow image with default Python version (3.7 currently)
-* `apache/airflow:2.4.2-pythonX.Y` - the versioned Airflow image with specific Python version
+* `apache/airflow:2.4.3` - the versioned Airflow image with default Python version (3.7 currently)
+* `apache/airflow:2.4.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.4.2` - the versioned Airflow image with default Python version (3.7 currently)
-* `apache/airflow:slim-2.4.2-pythonX.Y` - the versioned Airflow image with specific Python version
+* `apache/airflow:slim-2.4.3` - the versioned Airflow image with default Python version (3.7 currently)
+* `apache/airflow:slim-2.4.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-apt-packages/Dockerfile b/docs/docker-stack/docker-examples/extending/add-apt-packages/Dockerfile
index f7e4fa667e..8161acccb2 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.4.2
+FROM apache/airflow:2.4.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 86d6a14def..f470efd9b5 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.4.2
+FROM apache/airflow:2.4.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 0fc059cfb6..ac3f9676d2 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.4.2
+FROM apache/airflow:2.4.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 f6b2f7c220..8c278361fb 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.4.2
+FROM apache/airflow:2.4.3
RUN pip install --no-cache-dir 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 bac10687b0..95a7e6e15d 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.4.2
+FROM apache/airflow:2.4.3
COPY requirements.txt /
RUN pip install --no-cache-dir -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 1e32be7b9d..14837550d4 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.4.2
+FROM apache/airflow:2.4.3
RUN pip install --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 b073eee633..29e4c3aad9 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.4.2
+FROM apache/airflow:2.4.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 571041d39c..317a1a067a 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.4.2
+FROM apache/airflow:2.4.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 9c21b72702..54663effc3 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.4.2-python3.6 bash -c "ls -la"
+ docker run -it apache/airflow:2.4.3-python3.6 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.4.2-python3.6 python -c "print('test')"
+ > docker run -it apache/airflow:2.4.3-python3.6 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.4.2-python3.6 airflow webserver
+ docker run -it apache/airflow:2.4.3-python3.6 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.4.2-python3.6 help
+ > docker run -it apache/airflow:2.4.3-python3.6 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.4.2-python3.8 webserver
+ apache/airflow:2.4.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.4.2-python3.8 webserver
+ apache/airflow:2.4.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.4.2-python3.8 webserver
+ apache/airflow:2.4.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 cfaef5dcb6..07f57e3c47 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.4.2", "Supported", "Dec 17, 2020", "TBD", "TBD"),
+ ("2", "2.4.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/setup.py b/setup.py
index 8d1b19af1d..936071d2dc 100644
--- a/setup.py
+++ b/setup.py
@@ -49,7 +49,7 @@ PY39 = sys.version_info >= (3, 9)
logger = logging.getLogger(__name__)
-version = '2.4.2'
+version = '2.4.3'
AIRFLOW_SOURCES_ROOT = Path(__file__).parent.resolve()
PROVIDERS_ROOT = AIRFLOW_SOURCES_ROOT / "airflow" / "providers"
[airflow] 29/29: Add release notes
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit c7ee64fe11b12a466cd16773179c5f982425eb69
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Nov 9 21:20:19 2022 +0100
Add release notes
---
RELEASE_NOTES.rst | 47 ++++++++++++++++++++++++++++++++++++++++++++
newsfragments/27065.misc.rst | 1 -
2 files changed, 47 insertions(+), 1 deletion(-)
diff --git a/RELEASE_NOTES.rst b/RELEASE_NOTES.rst
index 46d9a702f3..bdf4d2cb1c 100644
--- a/RELEASE_NOTES.rst
+++ b/RELEASE_NOTES.rst
@@ -21,6 +21,53 @@
.. towncrier release notes start
+Airflow 2.4.3 (2022-11-13)
+--------------------------
+
+Significant Changes
+^^^^^^^^^^^^^^^^^^^
+
+Make ``RotatingFilehandler`` used in ``DagProcessor`` non-caching (#27223)
+""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""
+
+ In case you want to decrease cache memory when ``CONFIG_PROCESSOR_MANAGER_LOGGER=True``, and you have your local settings created before,
+ you can update ``processor_manager_handler`` to use ``airflow.utils.log.non_caching_file_handler.NonCachingRotatingFileHandler`` handler instead of ``logging.RotatingFileHandler``. (#27065)
+
+Bug Fixes
+^^^^^^^^^
+- Replace FAB url filtering function with Airflows (#27576)
+- Fix mini scheduler expansion of mapped task (#27506)
+- ``SLAMiss`` is nullable and not always given back when pulling task instances (#27423)
+- Fix behavior of ``_`` when searching for DAGs (#27448)
+- Fix getting the ``dag/task`` ids from base executor (#27550)
+- Fix sqlalchemy primary key black-out error on DDRQ (#27538)
+- Fix IntegrityError during webserver startup (#27297)
+- Add case insensitive constraint to username (#27266)
+- Fix python external template keys (#27256)
+- reduce extraneous task log requests (#27233)
+- Make ``RotatingFilehandler`` used in ``DagProcessor`` non-caching (#27223)
+- Listener: Set task on sqlalchemy taskinstance object (#27167)
+- Fix some bug in web ui dags list page (auto-refresh & jump search null state) (#27141)
+- set ``executor.job_id`` to ``BackfillJob.id`` for backfills (#27020)
+
+Misc/Internal
+^^^^^^^^^^^^^
+- Bump loader-utils from ``1.4.0`` to ``1.4.1`` in ``/airflow/www`` (#27552)
+- Reduce log level for k8s ``TCP_KEEPALIVE`` etc warnings (#26981)
+
+Doc only changes
+^^^^^^^^^^^^^^^^
+- Update ``index.rst`` (#27529)
+- Fix wording in DAG Runs description(#27470)
+- Document that ``KubernetesExecutor`` overwrites container args (#27450)
+- Fix ``BaseOperator`` link (#27441)
+- Correct timer units to seconds from milliseconds. (#27360)
+- Add missed import in the Trigger Rules example (#27309)
+- Fix typo in ``set-up-database.rst`` (#27137)
+- Update SLA wording to reflect it is relative to ``Dag Run`` start. (#27111)
+- Update ``dags.rst`` (#27110)
+- Add ``kerberos`` environment variables to the docs (#27028)
+
Airflow 2.4.2 (2022-10-23)
--------------------------
diff --git a/newsfragments/27065.misc.rst b/newsfragments/27065.misc.rst
deleted file mode 100644
index 215bc8321d..0000000000
--- a/newsfragments/27065.misc.rst
+++ /dev/null
@@ -1 +0,0 @@
-In case you want to decrease cache memory when ``CONFIG_PROCESSOR_MANAGER_LOGGER=True``, and you have your local settings created before, you can update ``processor_manager_handler`` to use ``airflow.utils.log.non_caching_file_handler.NonCachingRotatingFileHandler`` handler instead of ``logging.RotatingFileHandler``.
[airflow] 05/29: Listener: Set task on sqlalchemy taskinstance object (#27167)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1eb047dbfb62086da73a0225fd25e949b6c3fd8f
Author: Kaxil Naik <ka...@apache.org>
AuthorDate: Thu Oct 20 21:19:50 2022 +0100
Listener: Set task on sqlalchemy taskinstance object (#27167)
same as https://github.com/apache/airflow/pull/21157
TaskListener API's contract promises to pass TaskInstance object to listener plugin. However, what happens is not 100% true - the object being passed is one that maps to current SQLAlchemy session.
`_run_raw_task` before merging the TI operates on detached TaskInstance object, then merges it to current session. Since there is no attached object in the SQLAlchemy identity map, SQLAlchemy creates it, and it's this object that's being passed to the SQLAlchemy event listeners.
The problem with that is that when creating new SQLAlchemy object, SQLAlchemy takes care about setting only database-mapped fields. The ones that are purely on the python side, like task aren't being set on the new object.
This PR manually sets `task` on the new SQLAlchemy object, so that `on_task_instance_success` receives proper TaskInstance with task field set.
(cherry picked from commit 395ad7110e53a30a5d33f648d1dd797482eb268c)
---
airflow/models/taskinstance.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 48e12e7a62..d4453ca842 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1527,7 +1527,7 @@ class TaskInstance(Base, LoggingMixin):
if not test_mode:
session.add(Log(self.state, self))
- session.merge(self)
+ session.merge(self).task = self.task
if self.state == TaskInstanceState.SUCCESS:
self._register_dataset_changes(session=session)
session.commit()
[airflow] 10/29: Fix IntegrityError during webserver startup (#27297)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 2c5b336db2a24837f1cfee44a96c063eff4ee9c9
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu Oct 27 05:25:44 2022 +0100
Fix IntegrityError during webserver startup (#27297)
(cherry picked from commit 8f99c793ec4289f7fc28d890b6c2887f0951e09b)
---
airflow/cli/commands/webserver_command.py | 6 ++++++
tests/cli/commands/test_webserver_command.py | 1 +
2 files changed, 7 insertions(+)
diff --git a/airflow/cli/commands/webserver_command.py b/airflow/cli/commands/webserver_command.py
index 9d15d44b7d..ce45705ee0 100644
--- a/airflow/cli/commands/webserver_command.py
+++ b/airflow/cli/commands/webserver_command.py
@@ -414,6 +414,12 @@ def webserver(args):
run_args += ["airflow.www.app:cached_app()"]
+ # To prevent different workers creating the web app and
+ # all writing to the database at the same time, we use the --preload option.
+ # With the preload option, the app is loaded before the workers are forked, and each worker will
+ # then have a copy of the app
+ run_args += ['--preload']
+
gunicorn_master_proc = None
def kill_proc(signum, _):
diff --git a/tests/cli/commands/test_webserver_command.py b/tests/cli/commands/test_webserver_command.py
index 9aa8e97885..0e15dda077 100644
--- a/tests/cli/commands/test_webserver_command.py
+++ b/tests/cli/commands/test_webserver_command.py
@@ -406,6 +406,7 @@ class TestCliWebServer:
'--access-logformat',
'custom_log_format',
'airflow.www.app:cached_app()',
+ "--preload",
],
close_fds=True,
)
[airflow] 20/29: Fix sqlalchemy primary key black-out error on DDRQ (#27538)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cafad9d7fbb0c57ddbe8c0f67f67ed8a7401457b
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Mon Nov 7 14:03:10 2022 +0100
Fix sqlalchemy primary key black-out error on DDRQ (#27538)
closes https://github.com/apache/airflow/issues/27509
(cherry picked from commit fc59b02cfac7fd691602edc92a7abac38ed51531)
---
airflow/models/dataset.py | 1 +
tests/models/test_taskinstance.py | 37 +++++++++++++++++++++++++++++++++++++
2 files changed, 38 insertions(+)
diff --git a/airflow/models/dataset.py b/airflow/models/dataset.py
index b1a58e5442..c9cc3eca2a 100644
--- a/airflow/models/dataset.py
+++ b/airflow/models/dataset.py
@@ -116,6 +116,7 @@ class DagScheduleDatasetReference(Base):
DagScheduleDatasetReference.dataset_id == foreign(DatasetDagRunQueue.dataset_id),
DagScheduleDatasetReference.dag_id == foreign(DatasetDagRunQueue.target_dag_id),
)""",
+ cascade="all, delete, delete-orphan",
)
__tablename__ = "dag_schedule_dataset_reference"
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index 4eff50092b..f4edebcb08 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -1916,6 +1916,43 @@ class TestTaskInstance:
# check that no dataset events were generated
assert session.query(DatasetEvent).count() == 0
+ def test_changing_of_dataset_when_ddrq_is_already_populated(self, dag_maker, session):
+ """
+ Test that when a task that produces dataset has ran, that changing the consumer
+ dag dataset will not cause primary key blank-out
+ """
+ from airflow import Dataset
+
+ with dag_maker(schedule=None, serialized=True) as dag1:
+
+ @task(outlets=Dataset("test/1"))
+ def test_task1():
+ print(1)
+
+ test_task1()
+
+ dr1 = dag_maker.create_dagrun()
+ test_task1 = dag1.get_task("test_task1")
+
+ with dag_maker(dag_id="testdag", schedule=[Dataset("test/1")], serialized=True):
+
+ @task
+ def test_task2():
+ print(1)
+
+ test_task2()
+
+ ti = dr1.get_task_instance(task_id="test_task1")
+ ti.run()
+ # Change the dataset.
+ with dag_maker(dag_id="testdag", schedule=[Dataset("test2/1")], serialized=True):
+
+ @task
+ def test_task2():
+ print(1)
+
+ test_task2()
+
@staticmethod
def _test_previous_dates_setup(
schedule_interval: str | datetime.timedelta | None,
[airflow] 03/29: Fix typo in set-up-database.rst (#27137)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 8c7e2c356360268e17e86e9b0a16060b42d4be15
Author: Roman <lk...@gmail.com>
AuthorDate: Tue Oct 25 11:10:34 2022 +0300
Fix typo in set-up-database.rst (#27137)
(cherry picked from commit 131d339696e9568a2a2dc55c2a6963897cdc82b7)
---
docs/apache-airflow/howto/set-up-database.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/howto/set-up-database.rst b/docs/apache-airflow/howto/set-up-database.rst
index 01368b2f80..f5db806f28 100644
--- a/docs/apache-airflow/howto/set-up-database.rst
+++ b/docs/apache-airflow/howto/set-up-database.rst
@@ -118,7 +118,7 @@ An example URI for the sqlite database:
AmazonLinux SQLite can only be upgraded to v3.7 using the source repos. Airflow requires v3.15 or higher. Use the
following instructions to setup the base image (or AMI) with latest SQLite3
-Pre-requisite: You will need ``wget``, ``tar``, ``gzip``,`` gcc``, ``make``, and ``expect`` to get the upgrade process working.
+Pre-requisite: You will need ``wget``, ``tar``, ``gzip``, ``gcc``, ``make``, and ``expect`` to get the upgrade process working.
.. code-block:: bash
[airflow] 27/29: Remove self.using_mysql attribute (#27571)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 676c1ee9b30b6bffcfa21379548cd39386230299
Author: TreyYi <ct...@gmail.com>
AuthorDate: Thu Nov 10 02:58:41 2022 +0900
Remove self.using_mysql attribute (#27571)
This is to remove `self.using_mysql` attribute since it is not used in scheduler_job.py.
(cherry picked from commit 2ac45b011d04ac141a15abfc24cc054687cadbc2)
---
airflow/jobs/scheduler_job.py | 1 -
1 file changed, 1 deletion(-)
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index 74541a1cf8..03a7935ed9 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -151,7 +151,6 @@ class SchedulerJob(BaseJob):
# Check what SQL backend we use
sql_conn: str = conf.get_mandatory_value('database', 'sql_alchemy_conn').lower()
self.using_sqlite = sql_conn.startswith('sqlite')
- self.using_mysql = sql_conn.startswith('mysql')
# Dag Processor agent - not used in Dag Processor standalone mode.
self.processor_agent: DagFileProcessorAgent | None = None
[airflow] 14/29: fIx failing masking tests for python < 3.10 (#27337)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 910827c3f1c9c10c4cb6d2516fcfa030320b6fcc
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Fri Oct 28 09:36:27 2022 +0200
fIx failing masking tests for python < 3.10 (#27337)
Seems that the number of times user is printed in stack trace
depend on Python version. The fix in #27335 seems to only have
worked for Python 3.10 with the 1.0.0 of exceptiongroup the
stack trace has less stack levels.
(cherry picked from commit fcec804ff513d8f148d5069a48e7adf703855a96)
[airflow] 04/29: Fix some bug in web ui dags list page (auto-refresh & jump search null state) (#27141)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 66125651b619a4129c66f36f5d705d399166ff62
Author: Bob Du <i...@bobdu.cc>
AuthorDate: Fri Oct 21 22:39:59 2022 +0800
Fix some bug in web ui dags list page (auto-refresh & jump search null state) (#27141)
* Fix some bug in web ui dags list page
Signed-off-by: BobDu <i...@bobdu.cc>
* use switch case & fix remove loading dot
Signed-off-by: BobDu <i...@bobdu.cc>
* rename drawDagStats & selectors constants
Signed-off-by: BobDu <i...@bobdu.cc>
* Further clean up of duplicate code
Signed-off-by: BobDu <i...@bobdu.cc>
Signed-off-by: BobDu <i...@bobdu.cc>
(cherry picked from commit ebd34cbd9f62d086bb39727d85fd58a0758edba3)
---
airflow/www/static/js/dags.js | 145 ++++++++------------------------
airflow/www/templates/airflow/dags.html | 6 +-
2 files changed, 37 insertions(+), 114 deletions(-)
diff --git a/airflow/www/static/js/dags.js b/airflow/www/static/js/dags.js
index db8f19b203..3635da841f 100644
--- a/airflow/www/static/js/dags.js
+++ b/airflow/www/static/js/dags.js
@@ -43,6 +43,9 @@ const gridUrl = getMetaValue('grid_url');
const nextDatasets = {};
let nextDatasetsError;
+const DAG_RUN = 'dag-run';
+const TASK_INSTANCE = 'task-instance';
+
// auto refresh interval in milliseconds
// (x2 the interval in tree/graph view since this page can take longer to refresh )
const refreshIntervalMs = 2000;
@@ -188,10 +191,10 @@ d3.selectAll('.js-last-run-tooltip')
d3.select(this).attr('data-original-title', tiTooltip(lastRunData));
});
-function drawDagStatsForDag(dagId, states) {
- const g = d3.select(`svg#dag-run-${dagId.replace(/\./g, '__dot__')}`)
+function drawDagStats(selector, dagId, states) {
+ const g = d3.select(`svg#${selector}-${dagId.replace(/\./g, '__dot__')}`)
.attr('height', diameter + (strokeWidthHover * 2))
- .attr('width', '120px')
+ .attr('width', (states.length * (diameter + circleMargin)) + circleMargin)
.selectAll('g')
.data(states)
.enter()
@@ -203,103 +206,41 @@ function drawDagStatsForDag(dagId, states) {
});
g.append('svg:a')
- .attr('href', (d) => `${dagRunUrl}?_flt_3_dag_id=${dagId}&_flt_3_state=${d.state}`)
- .append('circle')
- .attr('id', (d) => `run-${dagId.replace(/\./g, '_')}${d.state || 'none'}`)
- .attr('class', 'has-svg-tooltip')
- .attr('stroke-width', (d) => {
- if (d.count > 0) return strokeWidth;
-
- return 1;
- })
- .attr('stroke', (d) => {
- if (d.count > 0) return STATE_COLOR[d.state];
-
- return 'gainsboro';
- })
- .attr('fill', '#fff')
- .attr('r', diameter / 2)
- .attr('title', (d) => d.state)
- .on('mouseover', (d) => {
- if (d.count > 0) {
- d3.select(this).transition().duration(400)
- .attr('fill', '#e2e2e2')
- .style('stroke-width', strokeWidthHover);
+ .attr('href', (d) => {
+ const params = new URLSearchParams();
+ params.append('_flt_3_dag_id', dagId);
+ /* eslint no-unused-expressions: ["error", { "allowTernary": true }] */
+ d.state ? params.append('_flt_3_state', d.state) : params.append('_flt_8_state', '');
+ switch (selector) {
+ case DAG_RUN: return `${dagRunUrl}?${params.toString()}`;
+ case TASK_INSTANCE: return `${taskInstanceUrl}?${params.toString()}`;
+ default: return '';
}
})
- .on('mouseout', (d) => {
- if (d.count > 0) {
- d3.select(this).transition().duration(400)
- .attr('fill', '#fff')
- .style('stroke-width', strokeWidth);
- }
- })
- .style('opacity', 0)
- .transition()
- .duration(300)
- .delay((d, i) => i * 50)
- .style('opacity', 1);
- d3.select('.js-loading-dag-stats').remove();
-
- g.append('text')
- .attr('fill', '#51504f')
- .attr('text-anchor', 'middle')
- .attr('vertical-align', 'middle')
- .attr('font-size', 9)
- .attr('y', 3)
- .style('pointer-events', 'none')
- .text((d) => (d.count > 0 ? d.count : ''));
-}
-
-function dagStatsHandler(error, json) {
- Object.keys(json).forEach((dagId) => {
- const states = json[dagId];
- drawDagStatsForDag(dagId, states);
- });
-}
-
-function drawTaskStatsForDag(dagId, states) {
- const g = d3.select(`svg#task-run-${dagId.replace(/\./g, '__dot__')}`)
- .attr('height', diameter + (strokeWidthHover * 2))
- .attr('width', (states.length * (diameter + circleMargin)) + circleMargin)
- .selectAll('g')
- .data(states)
- .enter()
- .append('g')
- .attr('transform', (d, i) => {
- const x = (i * (diameter + circleMargin)) + (diameter / 2 + circleMargin);
- const y = (diameter / 2) + strokeWidthHover;
- return `translate(${x},${y})`;
- });
-
- g.append('svg:a')
- .attr('href', (d) => `${taskInstanceUrl}?_flt_3_dag_id=${dagId}&_flt_3_state=${d.state}`)
.append('circle')
- .attr('id', (d) => `task-${dagId.replace(/\./g, '_')}${d.state || 'none'}`)
+ .attr('id', (d) => `${selector}-${dagId.replace(/\./g, '_')}-${d.state || 'none'}`)
.attr('class', 'has-svg-tooltip')
.attr('stroke-width', (d) => {
if (d.count > 0) return strokeWidth;
-
return 1;
})
.attr('stroke', (d) => {
if (d.count > 0) return STATE_COLOR[d.state];
-
return 'gainsboro';
})
.attr('fill', '#fff')
.attr('r', diameter / 2)
.attr('title', (d) => d.state || 'none')
- .on('mouseover', function mouseOver(d) {
+ .on('mouseover', (d) => {
if (d.count > 0) {
- d3.select(this).transition().duration(400)
+ d3.select(d3.event.currentTarget).transition().duration(400)
.attr('fill', '#e2e2e2')
.style('stroke-width', strokeWidthHover);
}
})
- .on('mouseout', function mouseOut(d) {
+ .on('mouseout', (d) => {
if (d.count > 0) {
- d3.select(this).transition().duration(400)
+ d3.select(d3.event.currentTarget).transition().duration(400)
.attr('fill', '#fff')
.style('stroke-width', strokeWidth);
}
@@ -310,7 +251,7 @@ function drawTaskStatsForDag(dagId, states) {
.delay((d, i) => i * 50)
.style('opacity', 1);
- d3.select('.js-loading-task-stats').remove();
+ d3.select(`.js-loading-${selector}-stats`).remove();
g.append('text')
.attr('fill', '#51504f')
@@ -322,10 +263,10 @@ function drawTaskStatsForDag(dagId, states) {
.text((d) => (d.count > 0 ? d.count : ''));
}
-function taskStatsHandler(error, json) {
+function dagStatsHandler(selector, json) {
Object.keys(json).forEach((dagId) => {
const states = json[dagId];
- drawTaskStatsForDag(dagId, states);
+ drawDagStats(selector, dagId, states);
});
}
@@ -355,14 +296,14 @@ function getDagStats() {
.post(params, lastDagRunsHandler);
d3.json(dagStatsUrl)
.header('X-CSRFToken', csrfToken)
- .post(params, dagStatsHandler);
+ .post(params, (error, json) => dagStatsHandler(DAG_RUN, json));
d3.json(taskStatsUrl)
.header('X-CSRFToken', csrfToken)
- .post(params, taskStatsHandler);
+ .post(params, (error, json) => dagStatsHandler(TASK_INSTANCE, json));
} else {
// no dags, hide the loading dots
- $('.js-loading-task-stats').remove();
- $('.js-loading-dag-stats').remove();
+ $(`.js-loading-${DAG_RUN}-stats`).remove();
+ $(`.js-loading-${TASK_INSTANCE}-stats`).remove();
}
}
@@ -381,7 +322,7 @@ function hideSvgTooltip() {
$('#svg-tooltip').css('display', 'none');
}
-function refreshDagRunsAndTasks(selector, dagId, states) {
+function refreshDagStats(selector, dagId, states) {
d3.select(`svg#${selector}-${dagId.replace(/\./g, '__dot__')}`)
.selectAll('circle')
.data(states)
@@ -391,19 +332,9 @@ function refreshDagRunsAndTasks(selector, dagId, states) {
})
.attr('stroke', (d) => {
if (d.count > 0) return STATE_COLOR[d.state];
-
return 'gainsboro';
- })
- .attr('fill', '#fff')
- .attr('r', diameter / 2)
- .attr('title', (d) => d.state)
- .on('mouseover', (d) => {
- if (d.count > 0) {
- d3.select(this).transition().duration(400)
- .attr('fill', '#e2e2e2')
- .style('stroke-width', strokeWidthHover);
- }
});
+
d3.select(`svg#${selector}-${dagId.replace(/\./g, '__dot__')}`)
.selectAll('text')
.data(states)
@@ -415,13 +346,6 @@ function refreshDagRunsAndTasks(selector, dagId, states) {
});
}
-function refreshTaskStateHandler(error, ts) {
- Object.keys(ts).forEach((dagId) => {
- const states = ts[dagId];
- refreshDagRunsAndTasks('task-run', dagId, states);
- });
-}
-
let refreshInterval;
function checkActiveRuns(json) {
@@ -437,12 +361,11 @@ function checkActiveRuns(json) {
}
}
-function refreshDagRuns(error, json) {
- checkActiveRuns(json);
+function refreshDagStatsHandler(selector, json) {
+ if (selector === DAG_RUN) checkActiveRuns(json);
Object.keys(json).forEach((dagId) => {
const states = json[dagId];
- drawDagStatsForDag(dagId, states);
- refreshDagRunsAndTasks('dag-run', dagId, states);
+ refreshDagStats(selector, dagId, states);
});
}
@@ -459,10 +382,10 @@ function handleRefresh({ activeDagsOnly = false } = {}) {
.post(params, lastDagRunsHandler);
d3.json(dagStatsUrl)
.header('X-CSRFToken', csrfToken)
- .post(params, refreshDagRuns);
+ .post(params, (error, json) => refreshDagStatsHandler(DAG_RUN, json));
d3.json(taskStatsUrl)
.header('X-CSRFToken', csrfToken)
- .post(params, refreshTaskStateHandler);
+ .post(params, (error, json) => refreshDagStatsHandler(TASK_INSTANCE, json));
}
setTimeout(() => {
$('#loading-dots').css('display', 'none');
diff --git a/airflow/www/templates/airflow/dags.html b/airflow/www/templates/airflow/dags.html
index c675b4a05a..91d234d5e0 100644
--- a/airflow/www/templates/airflow/dags.html
+++ b/airflow/www/templates/airflow/dags.html
@@ -268,7 +268,7 @@
{% endfor %}
</td>
<td style="padding:0; width:130px;">
- {{ loading_dots(classes='js-loading-dag-stats text-muted') }}
+ {{ loading_dots(classes='js-loading-dag-run-stats text-muted') }}
<svg height="10" width="10" id="dag-run-{{ dag.safe_dag_id }}" style="display: block;"></svg>
</td>
<td>
@@ -331,8 +331,8 @@
{% endif %}
</td>
<td style="padding:0; width:323px; height:10px;">
- {{ loading_dots(classes='js-loading-task-stats text-muted') }}
- <svg height="10" width="10" id='task-run-{{ dag.safe_dag_id }}' style="display: block;"></svg>
+ {{ loading_dots(classes='js-loading-task-instance-stats text-muted') }}
+ <svg height="10" width="10" id='task-instance-{{ dag.safe_dag_id }}' style="display: block;"></svg>
</td>
<td class="text-center">
<div class="btn-group">
[airflow] 21/29: Fix getting the dag/task ids from base executor (#27550)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ef86a455ab63fc38cb47e23b7c9a177d571dc5d6
Author: jordanjeremy <72...@users.noreply.github.com>
AuthorDate: Mon Nov 7 17:07:58 2022 -0600
Fix getting the dag/task ids from base executor (#27550)
(cherry picked from commit f304df7857eea344a7c274dcf41bbe5271a0083f)
---
airflow/executors/base_executor.py | 2 +-
tests/executors/test_base_executor.py | 7 +++++++
2 files changed, 8 insertions(+), 1 deletion(-)
diff --git a/airflow/executors/base_executor.py b/airflow/executors/base_executor.py
index 1d1b0e2cf7..70b65f40e1 100644
--- a/airflow/executors/base_executor.py
+++ b/airflow/executors/base_executor.py
@@ -364,7 +364,7 @@ class BaseExecutor(LoggingMixin):
if len(command) > 3 and "--help" not in command:
dag_id: str | None = None
task_id: str | None = None
- for arg in command[4:]:
+ for arg in command[3:]:
if not arg.startswith("--"):
if dag_id is None:
dag_id = arg
diff --git a/tests/executors/test_base_executor.py b/tests/executors/test_base_executor.py
index 5d0f22dab4..fee53d35ab 100644
--- a/tests/executors/test_base_executor.py
+++ b/tests/executors/test_base_executor.py
@@ -125,3 +125,10 @@ def test_trigger_running_tasks(dag_maker, change_state_attempt):
assert len(executor.execute_async.mock_calls) == len(dagrun.task_instances) + 1
else:
assert len(executor.execute_async.mock_calls) == len(dagrun.task_instances)
+
+
+def test_validate_airflow_tasks_run_command(dag_maker):
+ dagrun = setup_dagrun(dag_maker)
+ tis = dagrun.task_instances
+ dag_id, task_id = BaseExecutor.validate_airflow_tasks_run_command(tis[0].command_as_list())
+ assert dag_id == dagrun.dag_id and task_id == tis[0].task_id
[airflow] 17/29: Document that KubernetesExecutor overwrites container args (#27450)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1c6de53b7f1a7d89ac1010633a8012a64a6d6afe
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Tue Nov 1 23:08:45 2022 -0700
Document that KubernetesExecutor overwrites container args (#27450)
The executor will overwrite the args for the container so that the
worker runs the right task.
(cherry picked from commit 8f5e100f30764e7b1818a336feaa8bb390cbb327)
---
docs/apache-airflow/executor/kubernetes.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/executor/kubernetes.rst b/docs/apache-airflow/executor/kubernetes.rst
index e24582c7df..e5c1fc3d09 100644
--- a/docs/apache-airflow/executor/kubernetes.rst
+++ b/docs/apache-airflow/executor/kubernetes.rst
@@ -126,7 +126,7 @@ pod_override
When using the KubernetesExecutor, Airflow offers the ability to override system defaults on a per-task basis.
To utilize this functionality, create a Kubernetes V1pod object and fill in your desired overrides.
-Please note that the scheduler will override the ``metadata.name`` of the V1pod before launching it.
+Please note that the scheduler will override the ``metadata.name`` and ``containers[0].args`` of the V1pod before launching it.
To overwrite the base container of the pod launched by the KubernetesExecutor,
create a V1pod with a single container, and overwrite the fields as follows:
[airflow] 01/29: Update dags.rst (#27110)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3d15859add37d9639275b5c2676ad579e25802f7
Author: Dylan Turnbull <10...@users.noreply.github.com>
AuthorDate: Tue Oct 25 09:34:26 2022 -0500
Update dags.rst (#27110)
Corrects the use of an incorrect word when explaining that pickling can't be enabled for serialization.
(cherry picked from commit 7fcdb490b4932e55fd0d92443e7ade451802a564)
---
docs/apache-airflow/concepts/dags.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/concepts/dags.rst b/docs/apache-airflow/concepts/dags.rst
index aa73c66992..d34f7529b1 100644
--- a/docs/apache-airflow/concepts/dags.rst
+++ b/docs/apache-airflow/concepts/dags.rst
@@ -709,7 +709,7 @@ You can either do this all inside of the ``DAG_FOLDER``, with a standard filesys
Note that packaged DAGs come with some caveats:
-* They cannot be used if you have picking enabled for serialization
+* They cannot be used if you have pickling enabled for serialization
* They cannot contain compiled libraries (e.g. ``libz.so``), only pure Python
* They will be inserted into Python's ``sys.path`` and importable by any other code in the Airflow process, so ensure the package names don't clash with other packages already installed on your system.
[airflow] 24/29: SLAMiss is nullable and not always given back when pulling task instances (#27423)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit c30180fc2014ef087520e2762265f88501a1d933
Author: Charles Machalow <cm...@linkedin.com>
AuthorDate: Wed Nov 2 11:37:36 2022 -0700
SLAMiss is nullable and not always given back when pulling task instances (#27423)
* SLAMiss is nullable and not always given back when pulling task instances
See https://github.com/apache/airflow-client-python/issues/50 for an example of the related error
* SLAMiss is nullable and not always given back when pulling task instances
See https://github.com/apache/airflow-client-python/issues/50 for an example of the related error
(cherry picked from commit 63638cd2162219bd0a67caface4b1f1f8b88cc10)
---
airflow/api_connexion/openapi/v1.yaml | 1 +
airflow/www/static/js/types/api-generated.ts | 2 +-
2 files changed, 2 insertions(+), 1 deletion(-)
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
index 9967bf1ed1..49f61e3c16 100644
--- a/airflow/api_connexion/openapi/v1.yaml
+++ b/airflow/api_connexion/openapi/v1.yaml
@@ -2928,6 +2928,7 @@ components:
nullable: true
notification_sent:
type: boolean
+ nullable: true
TaskInstance:
type: object
diff --git a/airflow/www/static/js/types/api-generated.ts b/airflow/www/static/js/types/api-generated.ts
index da7fc40783..74f92a79aa 100644
--- a/airflow/www/static/js/types/api-generated.ts
+++ b/airflow/www/static/js/types/api-generated.ts
@@ -1146,7 +1146,7 @@ export interface components {
timestamp?: string;
description?: string | null;
notification_sent?: boolean;
- };
+ } | null;
TaskInstance: {
task_id?: string;
dag_id?: string;
[airflow] 07/29: reduce extraneous task log requests (#27233)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9c3ee045accbbef9f1b1bbee523d86afbf57e6c1
Author: Brent Bovenzi <br...@astronomer.io>
AuthorDate: Tue Oct 25 10:26:23 2022 -0400
reduce extraneous task log requests (#27233)
(cherry picked from commit e73e90e388f7916ae5eea48ba39687d99f7a94b1)
---
airflow/www/static/js/api/useTaskLog.ts | 23 +++++++++++++++++-----
.../js/dag/details/taskInstance/Logs/index.tsx | 4 +++-
.../static/js/dag/details/taskInstance/index.tsx | 3 ++-
3 files changed, 23 insertions(+), 7 deletions(-)
diff --git a/airflow/www/static/js/api/useTaskLog.ts b/airflow/www/static/js/api/useTaskLog.ts
index 1c8b999de8..580c5e0ab4 100644
--- a/airflow/www/static/js/api/useTaskLog.ts
+++ b/airflow/www/static/js/api/useTaskLog.ts
@@ -20,15 +20,19 @@
import axios, { AxiosResponse } from 'axios';
import { useQuery } from 'react-query';
import { useAutoRefresh } from 'src/context/autorefresh';
-import type { API } from 'src/types';
+import type { API, TaskInstance } from 'src/types';
import { getMetaValue } from 'src/utils';
const taskLogApi = getMetaValue('task_log_api');
+interface Props extends API.GetLogVariables {
+ state?: TaskInstance['state'];
+}
+
const useTaskLog = ({
- dagId, dagRunId, taskId, taskTryNumber, mapIndex, fullContent,
-}: API.GetLogVariables) => {
+ dagId, dagRunId, taskId, taskTryNumber, mapIndex, fullContent, state,
+}: Props) => {
let url: string = '';
if (taskLogApi) {
url = taskLogApi.replace('_DAG_RUN_ID_', dagRunId).replace('_TASK_ID_', taskId).replace(/-1$/, taskTryNumber.toString());
@@ -36,12 +40,21 @@ const useTaskLog = ({
const { isRefreshOn } = useAutoRefresh();
+ // Only refresh is the state is pending
+ const isStatePending = state === 'deferred'
+ || state === 'scheduled'
+ || state === 'running'
+ || state === 'up_for_reschedule'
+ || state === 'up_for_retry'
+ || state === 'queued'
+ || state === 'restarting';
+
return useQuery(
- ['taskLogs', dagId, dagRunId, taskId, mapIndex, taskTryNumber, fullContent],
+ ['taskLogs', dagId, dagRunId, taskId, mapIndex, taskTryNumber, fullContent, state],
() => axios.get<AxiosResponse, string>(url, { headers: { Accept: 'text/plain' }, params: { map_index: mapIndex, full_content: fullContent } }),
{
placeholderData: '',
- refetchInterval: isRefreshOn && (autoRefreshInterval || 1) * 1000,
+ refetchInterval: isStatePending && isRefreshOn && (autoRefreshInterval || 1) * 1000,
},
);
};
diff --git a/airflow/www/static/js/dag/details/taskInstance/Logs/index.tsx b/airflow/www/static/js/dag/details/taskInstance/Logs/index.tsx
index 8b35c3bc3a..ec6249cb27 100644
--- a/airflow/www/static/js/dag/details/taskInstance/Logs/index.tsx
+++ b/airflow/www/static/js/dag/details/taskInstance/Logs/index.tsx
@@ -87,6 +87,7 @@ interface Props {
mapIndex?: TaskInstance['mapIndex'];
executionDate: DagRun['executionDate'];
tryNumber: TaskInstance['tryNumber'];
+ state?: TaskInstance['state'];
}
const Logs = ({
@@ -96,6 +97,7 @@ const Logs = ({
mapIndex,
executionDate,
tryNumber,
+ state,
}: Props) => {
const [internalIndexes, externalIndexes] = getLinkIndexes(tryNumber);
const [selectedTryNumber, setSelectedTryNumber] = useState<number | undefined>();
@@ -105,7 +107,6 @@ const Logs = ({
const [fileSourceFilters, setFileSourceFilters] = useState<Array<FileSourceOption>>([]);
const { timezone } = useTimezone();
- //
const taskTryNumber = selectedTryNumber || tryNumber || 1;
const { data, isSuccess } = useTaskLog({
dagId,
@@ -114,6 +115,7 @@ const Logs = ({
mapIndex,
taskTryNumber,
fullContent: shouldRequestFullContent,
+ state,
});
const params = new URLSearchParamsWrapper({
diff --git a/airflow/www/static/js/dag/details/taskInstance/index.tsx b/airflow/www/static/js/dag/details/taskInstance/index.tsx
index fd6a59c8ba..0061be5983 100644
--- a/airflow/www/static/js/dag/details/taskInstance/index.tsx
+++ b/airflow/www/static/js/dag/details/taskInstance/index.tsx
@@ -124,7 +124,7 @@ const TaskInstance = ({
operator={operator}
/>
)}
- <Tabs size="lg" index={selectedTabIndex} onChange={handleTabsChange}>
+ <Tabs size="lg" index={selectedTabIndex} onChange={handleTabsChange} isLazy>
<TabList>
<Tab>
<Text as="strong">Details</Text>
@@ -183,6 +183,7 @@ const TaskInstance = ({
mapIndex={mapIndex}
executionDate={executionDate}
tryNumber={instance?.tryNumber}
+ state={instance?.state}
/>
</TabPanel>
)}
[airflow] 26/29: Replace FAB url filtering function with Airflows (#27576)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f0f67e8bc9dcb9444cfc5b88ee075191785469b7
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Wed Nov 9 11:02:42 2022 -0800
Replace FAB url filtering function with Airflows (#27576)
Use our url filtering util function so there is consistency between
FAB and Airflow routes.
(cherry picked from commit b33d22c2243c552f4d1372bbe6b75e5fb097f567)
---
airflow/www/extensions/init_appbuilder.py | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/airflow/www/extensions/init_appbuilder.py b/airflow/www/extensions/init_appbuilder.py
index 2a1ee904f8..fd66314550 100644
--- a/airflow/www/extensions/init_appbuilder.py
+++ b/airflow/www/extensions/init_appbuilder.py
@@ -215,6 +215,7 @@ class AirflowAppBuilder:
else:
self.post_init()
self._init_extension(app)
+ self._swap_url_filter()
def _init_extension(self, app):
app.appbuilder = self
@@ -222,6 +223,17 @@ class AirflowAppBuilder:
app.extensions = {}
app.extensions["appbuilder"] = self
+ def _swap_url_filter(self):
+ """
+ Use our url filtering util function so there is consistency between
+ FAB and Airflow routes
+ """
+ from flask_appbuilder.security import views as fab_sec_views
+
+ from airflow.www.views import get_safe_url
+
+ fab_sec_views.get_safe_redirect = get_safe_url
+
def post_init(self):
for baseview in self.baseviews:
# instantiate the views and add session
[airflow] 11/29: Add missed import in the Trigger Rules example (#27309)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit dc33e0774134fa794ab5c97b8c7db4c4a4afbd7d
Author: pigidser <38...@users.noreply.github.com>
AuthorDate: Thu Oct 27 14:26:39 2022 +0400
Add missed import in the Trigger Rules example (#27309)
* Add missed import in the Trigger Rules example
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
(cherry picked from commit 426f527690f7713bc1e9a48b3c91a659d7b338bc)
---
docs/apache-airflow/concepts/dags.rst | 1 +
1 file changed, 1 insertion(+)
diff --git a/docs/apache-airflow/concepts/dags.rst b/docs/apache-airflow/concepts/dags.rst
index d34f7529b1..aba17ee1a8 100644
--- a/docs/apache-airflow/concepts/dags.rst
+++ b/docs/apache-airflow/concepts/dags.rst
@@ -402,6 +402,7 @@ You can also combine this with the :ref:`concepts:depends-on-past` functionality
# dags/branch_without_trigger.py
import pendulum
+ from airflow.decorators import task
from airflow.models import DAG
from airflow.operators.empty import EmptyOperator