You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by jh...@apache.org on 2021/08/13 18:20:20 UTC
[airflow] branch v2-1-test updated (c37de28 -> 7e79747)
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a change to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git.
discard c37de28 Add State types for tasks and DAGs (#15285)
discard 8f96385 Add State types for tasks and DAGs (#15285)
discard 5ea02cc Fix external elasticsearch logs link (#16357)
discard 3d6db6d Fixed broken json_client (#17529)
discard fe70111 Add 'queued' to DagRunState (#16854)
discard 1be2003 Fix task retries when they receive sigkill and have retries and properly handle sigterm (#16301)
discard 26e8d2d Fix bug that log can't be shown when task runs failed (#16768)
discard 20ed40b Better diagnostics and self-healing of docker-compose (#17484)
discard bc79182 Install providers from sources in prod image only on main (#17458)
discard dcfc23e Attempt to reduce flakiness of PythonVirtualeEnv test_airflow_context (#17486)
discard 005a5d0 Disable Helm tests when branch is not main
discard 31005cb Fixes some static check errors (#17432)
discard d5f0b1a Switches to "/" convention in ghcr.io images (#17356)
discard 9d2ad2a Fix failing static checks in main (#17424)
discard 2763e8e Add timeout when asking whether to rebuild image (#17412)
discard 70629ca Optimizes structure of the Dockerfiles and use latest tools (#17418)
discard f2da764 Improve image building documentation for new users (#17409)
discard 403a087 Enhancement to bash scripts (#17098)
discard 205d6d4 Run mini scheduler in LocalTaskJob during task exit (#16289)
discard 38d4bab Fix race condition with dagrun callbacks (#16741)
discard f7bece3 Add 'queued' state to DagRun (#16401)
discard 0bd9558 Proper warning message when recorded PID is different from current PID (#17411)
discard 67b4ff0 commiting dagPickle session when the airflow tasks run --ship-dag --interactive command is executed FIXES: 15748 (#15890)
discard f479eff fix: change graph focus to top of view instead of center (#16484)
discard 4d2db2e Redact conn secrets in webserver logs (#16579)
discard 88fd2db Fix ``AttributeError``: ``datetime.timezone`` object has no attribute ``name`` (#16599)
discard 7288621 Set Process title for Worker when using ``LocalExecutor`` (#16623)
discard 55a4645 Fix slow (cleared) tasks being be adopted by Celery worker. (#16718)
discard 12c46c9 BugFix: Correctly handle custom `deps` and `task_group` during DAG Serialization (#16734)
discard 479a34d Fix unchecked indexing in _build_metrics (#16744)
discard 60938f0 fix: instance name env var (#16749)
discard 3750e5e Don't check execution_date in refresh_from_db (#16809)
discard 77ce4a2 Fix minor typo in configuration.py (#16832)
discard 36f0ecb Fixed task instance retrieval in XCom view (#16923)
discard 3beb37e Fix UPDATING.md (#16933)
discard 120cbb9 Core: Enable the use of __init_subclass__ in subclasses of BaseOperator (#17027)
discard 66897eb #16976 Add json.dumps() for templated fields objects: 'dict' and 'list' (#17082)
discard aaa9a4b Adds more explanatory message when SecretsMasker is not configured (#17101)
discard 1b4af0b Fix: ``TaskInstance`` does not show ``queued_by_job_id`` & ``external_executor_id`` (#17179)
discard 24e75e7 Do not seek error file when it is closed (#17187)
discard 4a4959f Fix running tasks with default_impersonation config (#17229)
discard ed76d11 Show serialization exceptions in DAG parsing log (#17277)
discard 798237a Handle and log exceptions raised during task callback (#17347)
discard 40f4388 Improve diagnostics message when users have secret_key misconfigured (#17410)
discard eaa21db Add missing permissions to varimport (#17468)
discard cf64187 Add back missing permissions to UserModelView controls. (#17431)
discard 4678cf54 Validate type of `priority_weight` during parsing (#16765)
discard b558205 Fix calculating duration in tree view (#16695)
discard 8e8b58a Fix CLI 'kubernetes cleanup-pods' which fails on invalid label key (#17298)
discard 012321b Fail tasks in scheduler when executor reports they failed (#15929)
discard 673d78a fix(smart_sensor): Unbound variable errors (#14774)
discard ac832a2 Updates to FlaskAppBuilder 3.3.2+ (#17208)
discard f14860d Update alias for field_mask in Google Memmcache (#16975)
discard 4aef457 AIRFLOW-5529 Add Apache Drill provider. (#16884)
discard ab46af9 bump dnspython (#16698)
discard f3b74d8 Add type annotations to setup.py (#16658)
discard c967d29 Bump Jinja2 upper-bound from 2.12.0 to 4.0.0 (#16595)
discard e8058eb Remove SQLAlchemy <1.4 constraint (#16630)
discard 88e6305 Switch back http provider after requests removes LGPL dependency (#16974)
discard 3d2a5cf Switches to "/" convention in ghcr.io images with optimisations
add ab3634a Switch back http provider after requests removes LGPL dependency (#16974)
add 68977b9 Remove SQLAlchemy <1.4 constraint (#16630)
add fe5920e Bump Jinja2 upper-bound from 2.12.0 to 4.0.0 (#16595)
add dfeb73d Add type annotations to setup.py (#16658)
add 867adda bump dnspython (#16698)
add 42334b0 AIRFLOW-5529 Add Apache Drill provider. (#16884)
add 37c935d Update alias for field_mask in Google Memmcache (#16975)
add 3c0f4d9 Updates to FlaskAppBuilder 3.3.2+ (#17208)
add 7a43960 Enhancement to bash scripts (#17098)
add 70794b0 Do not pull CI image for ownership fixing on first, fresh breeze run (#17419)
add 38c7115 Increases timeout for helm chart builds (#17417)
add 8609c82 Improve image building documentation for new users (#17409)
add 106d9f0 Optimizes structure of the Dockerfiles and use latest tools (#17418)
add 7371f98 Add timeout when asking whether to rebuild image (#17412)
add 2df7e6e Improve diagnostics message when users have secret_key misconfigured (#17410)
add f196690 Fix failing static checks in main (#17424)
add 8fc7697 Switches to "/" convention in ghcr.io images (#17356)
add 2ba71c9 Fixes some static check errors (#17432)
add 562be8c Disable Helm tests when branch is not main
add 8bd1a7d Attempt to reduce flakiness of PythonVirtualeEnv test_airflow_context (#17486)
add 0948366 Install providers from sources in prod image only on main (#17458)
add 3833fb7 Better diagnostics and self-healing of docker-compose (#17484)
new b82782c Add back missing permissions to UserModelView controls. (#17431)
new be09eef Path correction in docs for airflow core (#17567)
new e2be8d3 fix: instance name env var (#16749)
new 0d0f4e8 Fix minor typo in configuration.py (#16832)
new 6ebbf82 fix(smart_sensor): Unbound variable errors (#14774)
new dd94647 Add State types for tasks and DAGs (#15285)
new dd3c06f commiting dagPickle session when the airflow tasks run --ship-dag --interactive command is executed FIXES: 15748 (#15890)
new c6d09ed Fail tasks in scheduler when executor reports they failed (#15929)
new f3d431d fix: change graph focus to top of view instead of center (#16484)
new 6eec6d0 Redact conn secrets in webserver logs (#16579)
new e593671 Fix ``AttributeError``: ``datetime.timezone`` object has no attribute ``name`` (#16599)
new 600309b Set Process title for Worker when using ``LocalExecutor`` (#16623)
new 748c662 Fix calculating duration in tree view (#16695)
new 3211672 Fix slow (cleared) tasks being be adopted by Celery worker. (#16718)
new 943d3c5 BugFix: Correctly handle custom `deps` and `task_group` during DAG Serialization (#16734)
new 1aa156a Fix unchecked indexing in _build_metrics (#16744)
new d04fe50 Validate type of `priority_weight` during parsing (#16765)
new ae4ab58 Don't check execution_date in refresh_from_db (#16809)
new 84bb043 Fixed task instance retrieval in XCom view (#16923)
new e7a87e5 Fix UPDATING.md (#16933)
new 75c3e75 Core: Enable the use of __init_subclass__ in subclasses of BaseOperator (#17027)
new f5ef0dd #16976 Add json.dumps() for templated fields objects: 'dict' and 'list' (#17082)
new 4709c65 Adds more explanatory message when SecretsMasker is not configured (#17101)
new 18c6baf Switch test_backfill_job.py from unittest to pytest style (#17112)
new 0656f10 Fix: ``TaskInstance`` does not show ``queued_by_job_id`` & ``external_executor_id`` (#17179)
new 20f6edf Do not seek error file when it is closed (#17187)
new db96ee5 Show serialization exceptions in DAG parsing log (#17277)
new 5739fed Fix CLI 'kubernetes cleanup-pods' which fails on invalid label key (#17298)
new 0678a94 Handle and log exceptions raised during task callback (#17347)
new 93b4c7b Optimize context sent for docker build (#17415)
new 65667b5 Add missing permissions to varimport (#17468)
new 42b6e2f Fixed broken json_client (#17529)
new 84df5ff Rescue if a DagRun's DAG was removed from db (#17544)
new 43a31e1 Fix running tasks with default_impersonation config (#17229)
new 6b8398a Proper warning message when recorded PID is different from current PID (#17411)
new effbee1 Revert "Fix DAG run state not updated while DAG is paused (#16343)"
new 3386b12 Run mini scheduler in LocalTaskJob during task exit (#16289)
new 7e79747 Fix DAG run state not updated while DAG is paused (#16343)
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 (c37de28)
\
N -- N -- N refs/heads/v2-1-test (7e79747)
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 38 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:
Dockerfile.ci | 4 +-
airflow/api_connexion/openapi/v1.yaml | 4 +-
airflow/jobs/local_task_job.py | 40 +-
airflow/jobs/scheduler_job.py | 182 +++--
...93827b8_add_queued_at_column_to_dagrun_table.py | 49 --
airflow/models/dag.py | 10 +-
airflow/models/dagrun.py | 31 +-
airflow/models/taskinstance.py | 8 +-
.../providers/elasticsearch/log/es_task_handler.py | 9 +-
airflow/utils/log/log_reader.py | 7 +-
airflow/utils/log/logging_mixin.py | 5 -
airflow/utils/state.py | 2 -
airflow/www/static/js/tree.js | 4 +-
airflow/www/views.py | 12 +-
docs/apache-airflow/installation.rst | 2 +-
docs/apache-airflow/migrations-ref.rst | 4 +-
tests/api/common/experimental/test_mark_tasks.py | 4 +-
.../endpoints/test_dag_run_endpoint.py | 14 +-
tests/api_connexion/schemas/test_dag_run_schema.py | 3 -
tests/dag_processing/test_processor.py | 773 ---------------------
tests/jobs/test_backfill_job.py | 45 +-
tests/jobs/test_local_task_job.py | 206 +-----
tests/jobs/test_scheduler_job.py | 278 ++++----
tests/models/test_cleartasks.py | 37 -
tests/models/test_dagrun.py | 25 +-
tests/models/test_taskinstance.py | 32 -
.../elasticsearch/log/test_es_task_handler.py | 10 -
tests/sensors/test_external_task_sensor.py | 8 +-
tests/utils/log/test_log_reader.py | 21 -
tests/utils/test_dag_processing.py | 11 +-
tests/www/views/test_views_blocked.py | 84 +++
tests/www/views/test_views_log.py | 16 +-
tests/www/views/test_views_tasks.py | 118 +---
33 files changed, 461 insertions(+), 1597 deletions(-)
delete mode 100644 airflow/migrations/versions/97cdd93827b8_add_queued_at_column_to_dagrun_table.py
delete mode 100644 tests/dag_processing/test_processor.py
create mode 100644 tests/www/views/test_views_blocked.py
[airflow] 24/38: Switch test_backfill_job.py from unittest to
pytest style (#17112)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 18c6baff95c4af77d72aefc0f4c8e5efed34416b
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Tue Jul 20 21:43:01 2021 +0100
Switch test_backfill_job.py from unittest to pytest style (#17112)
Prep work to use pytest fixtures in these tests
(cherry picked from commit 960da8a9074a4fb58881ea79f7dc9bc8fd58a5c4)
---
tests/jobs/test_backfill_job.py | 45 +++++++++++++++++++----------------------
1 file changed, 21 insertions(+), 24 deletions(-)
diff --git a/tests/jobs/test_backfill_job.py b/tests/jobs/test_backfill_job.py
index 154b485..62fe153 100644
--- a/tests/jobs/test_backfill_job.py
+++ b/tests/jobs/test_backfill_job.py
@@ -21,12 +21,10 @@ import datetime
import json
import logging
import threading
-import unittest
from unittest.mock import patch
import pytest
import sqlalchemy
-from parameterized import parameterized
from airflow import settings
from airflow.cli import cli_parser
@@ -56,7 +54,23 @@ logger = logging.getLogger(__name__)
DEFAULT_DATE = timezone.datetime(2016, 1, 1)
-class TestBackfillJob(unittest.TestCase):
+@pytest.fixture(scope="module")
+def dag_bag():
+ return DagBag(include_examples=True)
+
+
+class TestBackfillJob:
+ @staticmethod
+ def clean_db():
+ clear_db_runs()
+ clear_db_pools()
+
+ @pytest.fixture(autouse=True)
+ def set_instance_attrs(self, dag_bag):
+ self.clean_db()
+ self.parser = cli_parser.get_parser()
+ self.dagbag = dag_bag
+
def _get_dummy_dag(self, dag_id, pool=Pool.DEFAULT_POOL_NAME, task_concurrency=None):
dag = DAG(dag_id=dag_id, start_date=DEFAULT_DATE, schedule_interval='@daily')
@@ -73,22 +87,6 @@ class TestBackfillJob(unittest.TestCase):
count += 1
return count
- @classmethod
- def setUpClass(cls):
- cls.dagbag = DagBag(include_examples=True)
-
- @staticmethod
- def clean_db():
- clear_db_runs()
- clear_db_pools()
-
- def setUp(self):
- self.clean_db()
- self.parser = cli_parser.get_parser()
-
- def tearDown(self) -> None:
- self.clean_db()
-
def test_unfinished_dag_runs_set_to_failed(self):
dag = self._get_dummy_dag('dummy_dag')
@@ -210,7 +208,8 @@ class TestBackfillJob(unittest.TestCase):
session.close()
@pytest.mark.backend("postgres", "mysql")
- @parameterized.expand(
+ @pytest.mark.parametrize(
+ "dag_id, expected_execution_order",
[
[
"example_branch_operator",
@@ -246,7 +245,7 @@ class TestBackfillJob(unittest.TestCase):
),
],
["latest_only", ("latest_only", "task1")],
- ]
+ ],
)
def test_backfill_examples(self, dag_id, expected_execution_order):
"""
@@ -471,8 +470,6 @@ class TestBackfillJob(unittest.TestCase):
except AirflowException:
return
- self.fail()
-
@patch('airflow.jobs.backfill_job.BackfillJob.log')
def test_backfill_respect_pool_limit(self, mock_log):
session = settings.Session()
@@ -727,7 +724,7 @@ class TestBackfillJob(unittest.TestCase):
start_date=DEFAULT_DATE,
end_date=DEFAULT_DATE,
)
- with self.assertRaises(BackfillUnfinished):
+ with pytest.raises(BackfillUnfinished):
job.run()
def test_backfill_ordered_concurrent_execute(self):
[airflow] 03/38: fix: instance name env var (#16749)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e2be8d3b854d263ec734d24352c498ad9c17a7f1
Author: Rikhil Raithatha <23...@users.noreply.github.com>
AuthorDate: Thu Jul 1 11:03:35 2021 +0100
fix: instance name env var (#16749)
(cherry picked from commit fa811057a6ae0fc6c5e4bff1e18971c262a42a4c)
---
docs/apache-airflow/howto/customize-dag-ui-page-instance-name.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/howto/customize-dag-ui-page-instance-name.rst b/docs/apache-airflow/howto/customize-dag-ui-page-instance-name.rst
index e54a066..f67c193 100644
--- a/docs/apache-airflow/howto/customize-dag-ui-page-instance-name.rst
+++ b/docs/apache-airflow/howto/customize-dag-ui-page-instance-name.rst
@@ -38,7 +38,7 @@ To make this change, simply:
.. code-block::
- AIRFLOW__WEBSERVER__SITE_TITLE = "DevEnv"
+ AIRFLOW__WEBSERVER__INSTANCE_NAME = "DevEnv"
Screenshots
[airflow] 27/38: Show serialization exceptions in DAG parsing log
(#17277)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit db96ee5d6ac158fdb07f1d5a0cb66f0a15ff2aae
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Wed Jul 28 10:51:21 2021 -0600
Show serialization exceptions in DAG parsing log (#17277)
Make sure that any exceptions that happen when writing serialized DAGs
to the db get written to the DAG parsing log, instead of only being added
to `import_errors` for consumption via the UI.
(cherry picked from commit 9cd5a97654fa82f1d4d8f599e8eb81957b3f7286)
---
airflow/models/dagbag.py | 1 +
tests/models/test_dagbag.py | 4 +++-
2 files changed, 4 insertions(+), 1 deletion(-)
diff --git a/airflow/models/dagbag.py b/airflow/models/dagbag.py
index 5d84ba2..b37482a 100644
--- a/airflow/models/dagbag.py
+++ b/airflow/models/dagbag.py
@@ -595,6 +595,7 @@ class DagBag(LoggingMixin):
except OperationalError:
raise
except Exception:
+ self.log.exception("Failed to write serialized DAG: %s", dag.full_filepath)
return [(dag.fileloc, traceback.format_exc(limit=-self.dagbag_import_error_traceback_depth))]
# Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' in case
diff --git a/tests/models/test_dagbag.py b/tests/models/test_dagbag.py
index b4edc0c..a065318 100644
--- a/tests/models/test_dagbag.py
+++ b/tests/models/test_dagbag.py
@@ -703,7 +703,9 @@ class TestDagBag(unittest.TestCase):
)
assert dagbag.import_errors == {}
- dagbag.sync_to_db(session=session)
+ with self.assertLogs(level="ERROR") as cm:
+ dagbag.sync_to_db(session=session)
+ self.assertIn("SerializationError", "\n".join(cm.output))
assert path in dagbag.import_errors
err = dagbag.import_errors[path]
[airflow] 12/38: Set Process title for Worker when using
``LocalExecutor`` (#16623)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 600309b5f623fa41203d982c88bfeb33f3ef51e9
Author: Kaxil Naik <ka...@gmail.com>
AuthorDate: Thu Jun 24 09:18:52 2021 +0100
Set Process title for Worker when using ``LocalExecutor`` (#16623)
This has annoyed me for a long time. When using ``LocalExecutor``, it was difficult to see which process is a worker as it just showed up as below -- which had same title as parent scheduler process. This PR/commit adds a title for idle workers and when a task is running it has the "command" that is running in the title, similar to our supervising process
Before:
```
root 124 0.0 0.0 6676 4636 pts/1 Ss Jun23 0:00 \_ -bash
root 1449 0.8 2.6 988356 326312 pts/1 Sl+ Jun23 0:16 | \_ /usr/local/bin/python /usr/local/bin/airflow webserver
root 1584 0.0 0.4 121068 56864 pts/1 S+ Jun23 0:01 | \_ gunicorn: master [airflow-webserver]
root 1587 0.6 2.5 986144 318712 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1588 0.6 2.5 984776 317672 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1589 0.6 2.5 985688 318148 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1590 0.6 2.5 985200 317776 pts/1 Sl+ Jun23 0:11 | \_ [ready] gunicorn: worker [airflow-webserver]
root 128 0.0 0.0 6676 4552 pts/2 Ss Jun23 0:00 \_ -bash
root 13933 31.0 0.9 466596 117656 pts/2 S+ 00:22 0:01 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13941 0.0 0.7 466340 97988 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13942 3.2 0.8 1392072 100136 pts/2 Sl+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13950 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13952 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13955 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13958 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13962 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13966 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13969 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13975 0.0 0.8 466340 98404 pts/2 S+ 00:22 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13979 6.5 0.8 466596 99956 pts/2 S 00:22 0:00 \_ airflow scheduler -- DagFileProcessorManager
```
After (with no running tasks - idle workers):
```
root 124 0.0 0.0 6676 4636 pts/1 Ss Jun23 0:00 \_ -bash
root 1449 0.8 2.6 988356 326312 pts/1 Sl+ Jun23 0:16 | \_ /usr/local/bin/python /usr/local/bin/airflow webserver
root 1584 0.0 0.4 121068 56864 pts/1 S+ Jun23 0:01 | \_ gunicorn: master [airflow-webserver]
root 1587 0.6 2.5 985752 318184 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1588 0.6 2.5 984776 317672 pts/1 Sl+ Jun23 0:11 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1589 0.6 2.5 985688 318148 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1590 0.6 2.5 985200 317776 pts/1 Sl+ Jun23 0:11 | \_ [ready] gunicorn: worker [airflow-webserver]
root 128 0.0 0.0 6676 4552 pts/2 Ss Jun23 0:00 \_ -bash
root 13237 25.7 0.9 466596 117692 pts/2 S+ 00:20 0:02 \_ airflow worker -- LocalExecutor
root 13245 0.1 0.7 466340 97804 pts/2 S+ 00:20 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13246 2.1 0.8 1318340 100104 pts/2 Sl+ 00:20 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 13254 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13256 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13259 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13263 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13267 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13271 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13274 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13276 0.0 0.8 466340 98396 pts/2 S+ 00:20 0:00 \_ airflow worker -- LocalExecutor
root 13282 4.1 0.8 466596 99952 pts/2 S 00:20 0:00 \_ airflow scheduler -- DagFileProcessorManager
```
After (with running tasks):
```
root@a7c8aa590704:/opt/airflow# ps auxf
USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND
root 6434 0.0 0.0 6652 4584 pts/3 Ss 00:01 0:00 /bin/bash
root 19250 0.0 0.0 9556 3064 pts/3 R+ 00:39 0:00 \_ ps auxf
root 1 0.0 0.0 2148 720 ? Ss Jun23 0:00 /usr/bin/dumb-init -- /entrypoint
root 7 0.0 0.0 6656 4400 pts/0 Ss Jun23 0:00 /bin/bash
root 121 0.0 0.0 8220 3228 pts/0 S+ Jun23 0:00 \_ tmux
root 101 0.0 0.0 15856 4272 ? Ss Jun23 0:00 /usr/sbin/sshd
root 123 0.0 0.0 10176 5148 ? Ss Jun23 0:00 tmux
root 124 0.0 0.0 6676 4636 pts/1 Ss Jun23 0:00 \_ -bash
root 1449 0.6 2.6 988356 326312 pts/1 Sl+ Jun23 0:20 | \_ /usr/local/bin/python /usr/local/bin/airflow webserver
root 1584 0.0 0.4 121068 56864 pts/1 S+ Jun23 0:01 | \_ gunicorn: master [airflow-webserver]
root 1587 0.4 2.5 986144 318712 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1588 0.4 2.5 984776 317672 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1589 0.4 2.5 985848 318600 pts/1 Sl+ Jun23 0:13 | \_ [ready] gunicorn: worker [airflow-webserver]
root 1590 0.4 2.5 985628 318424 pts/1 Sl+ Jun23 0:12 | \_ [ready] gunicorn: worker [airflow-webserver]
root 128 0.0 0.0 6676 4552 pts/2 Ss Jun23 0:00 \_ -bash
root 19030 17.9 0.9 467108 118628 pts/2 S+ 00:38 0:02 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 19038 0.0 0.7 466084 97776 pts/2 S+ 00:38 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 19039 1.4 0.8 1318084 99804 pts/2 Sl+ 00:38 0:00 \_ /usr/local/bin/python /usr/local/bin/airflow scheduler
root 19047 0.0 0.8 466084 98692 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'runme_2', '2021-06-24T00:39:06.539715+00:00', '--local', '
root 19240 25.3 0.8 470820 104400 pts/2 S+ 00:39 0:00 | \_ airflow task supervisor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'runme_2', '2021-06-24T00:39:06.539715+00:00', '--local', '--po
root 19246 0.0 0.8 470956 103980 pts/2 S 00:39 0:00 | \_ airflow task runner: example_bash_operator runme_2 2021-06-24T00:39:06.539715+00:00 91
root 19049 0.1 0.8 466084 98696 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'runme_1', '2021-06-24T00:39:06.539715+00:00', '--local', '
root 19241 26.0 0.8 470824 104408 pts/2 S+ 00:39 0:00 | \_ airflow task supervisor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'runme_1', '2021-06-24T00:39:06.539715+00:00', '--local', '--po
root 19248 0.0 0.8 470824 103720 pts/2 S 00:39 0:00 | \_ airflow task runner: example_bash_operator runme_1 2021-06-24T00:39:06.539715+00:00 93
root 19052 0.1 0.8 466084 98760 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'runme_0', '2021-06-24T00:39:06.539715+00:00', '--local', '
root 19244 26.0 0.8 470824 104404 pts/2 S+ 00:39 0:00 | \_ airflow task supervisor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'runme_0', '2021-06-24T00:39:06.539715+00:00', '--local', '--po
root 19245 0.0 0.8 471212 104032 pts/2 S 00:39 0:00 | \_ airflow task runner: example_bash_operator runme_0 2021-06-24T00:39:06.539715+00:00 90
root 19056 0.1 0.8 466084 98760 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'this_will_skip', '2021-06-24T00:39:06.539715+00:00', '--lo
root 19243 24.6 0.8 470824 104400 pts/2 S+ 00:39 0:00 | \_ airflow task supervisor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'this_will_skip', '2021-06-24T00:39:06.539715+00:00', '--local'
root 19247 0.0 0.8 470956 103712 pts/2 S 00:39 0:00 | \_ airflow task runner: example_bash_operator this_will_skip 2021-06-24T00:39:06.539715+00:00 92
root 19057 0.1 0.8 466084 98760 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'also_run_this', '2021-06-24T00:39:06.539715+00:00', '--loc
root 19242 26.6 0.8 470824 104404 pts/2 R+ 00:39 0:00 | \_ airflow task supervisor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'also_run_this', '2021-06-24T00:39:06.539715+00:00', '--local',
root 19249 0.0 0.8 470824 101976 pts/2 S 00:39 0:00 | \_ airflow task supervisor: ['airflow', 'tasks', 'run', 'example_bash_operator', 'also_run_this', '2021-06-24T00:39:06.539715+00:00', '--loc
root 19062 0.0 0.8 466084 98300 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor
root 19066 0.0 0.8 466084 98300 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor
root 19069 0.0 0.8 466084 98300 pts/2 S+ 00:38 0:00 \_ airflow worker -- LocalExecutor
root 19075 2.7 0.8 466596 100144 pts/2 S 00:38 0:00 \_ airflow scheduler -- DagFileProcessorManager
```
Once the worker is done executing a task, the worker is renamed back to `airflow worker -- LocalExecutor`
(cherry picked from commit c8a628abf484f0bd9805f44dd37e284d2b5ee7db)
---
airflow/executors/local_executor.py | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/airflow/executors/local_executor.py b/airflow/executors/local_executor.py
index ab9356f..7789f28 100644
--- a/airflow/executors/local_executor.py
+++ b/airflow/executors/local_executor.py
@@ -63,6 +63,7 @@ class LocalWorkerBase(Process, LoggingMixin):
# We know we've just started a new process, so lets disconnect from the metadata db now
settings.engine.pool.dispose()
settings.engine.dispose()
+ setproctitle("airflow worker -- LocalExecutor")
return super().run()
def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None:
@@ -76,12 +77,15 @@ class LocalWorkerBase(Process, LoggingMixin):
return
self.log.info("%s running %s", self.__class__.__name__, command)
+ setproctitle(f"airflow worker -- LocalExecutor: {command}")
if settings.EXECUTE_TASKS_NEW_PYTHON_INTERPRETER:
state = self._execute_work_in_subprocess(command)
else:
state = self._execute_work_in_fork(command)
self.result_queue.put((key, state))
+ # Remove the command since the worker is done executing the task
+ setproctitle("airflow worker -- LocalExecutor")
def _execute_work_in_subprocess(self, command: CommandType) -> str:
try:
[airflow] 32/38: Fixed broken json_client (#17529)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 42b6e2f64174f397b6c59fcf942c815f7354c2be
Author: sw1010 <sw...@interia.pl>
AuthorDate: Tue Aug 10 13:26:54 2021 +0200
Fixed broken json_client (#17529)
The json_client depends on httpx and thus the response doesn't contain
the attribute 'ok'. In consequence, the missing attribute has been changed
to is_error.
closes: #17527
Co-authored-by: Szymon Wojciechowski <sz...@here.com>
(cherry picked from commit fbddefe5cbc0fe163c42ab070bc0d7e44c5cef5b)
---
airflow/api/client/json_client.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow/api/client/json_client.py b/airflow/api/client/json_client.py
index 2d5fd02..d87a3ae 100644
--- a/airflow/api/client/json_client.py
+++ b/airflow/api/client/json_client.py
@@ -32,7 +32,7 @@ class Client(api_client.Client):
if json is not None:
params['json'] = json
resp = getattr(self._session, method.lower())(**params)
- if not resp.ok:
+ if resp.is_error:
# It is justified here because there might be many resp types.
try:
data = resp.json()
[airflow] 13/38: Fix calculating duration in tree view (#16695)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 748c662d8f3a7cb8ed04ca01a900242c1aeeea9c
Author: Brent Bovenzi <br...@gmail.com>
AuthorDate: Mon Jun 28 11:23:19 2021 -0400
Fix calculating duration in tree view (#16695)
Make sure moment doesn't default the end_date to now and show the wrong duration
(cherry picked from commit f0b3345ddc489627d73d190a1401804e7b0d9c4e)
---
airflow/www/static/js/tree.js | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/airflow/www/static/js/tree.js b/airflow/www/static/js/tree.js
index 07daf0e..4bf366a 100644
--- a/airflow/www/static/js/tree.js
+++ b/airflow/www/static/js/tree.js
@@ -305,7 +305,11 @@ document.addEventListener('DOMContentLoaded', () => {
.style('stroke-opacity', (d) => (d.external_trigger ? '0' : '1'))
.on('mouseover', function (d) {
// Calculate duration if it doesn't exist
- const tt = tiTooltip({ ...d, duration: d.duration || moment(d.end_date).diff(d.start_date, 'seconds') });
+ const tt = tiTooltip({
+ ...d,
+ // if end_date is undefined then moment will default to now instead of null
+ duration: d.duration || d.end_date ? moment(d.end_date).diff(d.start_date, 'seconds') : null,
+ });
taskTip.direction('n');
taskTip.show(tt, this);
d3.select(this).transition().duration(duration)
[airflow] 33/38: Rescue if a DagRun's DAG was removed from db
(#17544)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 84df5ffd8870cb2a315f3e5007e6c392ced0c96b
Author: Tzu-ping Chung <tp...@astronomer.io>
AuthorDate: Fri Aug 13 07:44:21 2021 +0800
Rescue if a DagRun's DAG was removed from db (#17544)
Fix #17442.
The exception happens when a DAG is removed from the database (via web UI or something else), but there are still unfinished runs associated to it. This catches the scenario and use the existing fallback setting `max_active_runs` to zero.
(cherry picked from commit 60ddcd10bbe5c2375b14307456b8e5f76c1d0dcd)
---
airflow/www/views.py | 7 ++-
tests/www/views/test_views_blocked.py | 84 +++++++++++++++++++++++++++++++++++
2 files changed, 89 insertions(+), 2 deletions(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 09d27e0..150ecc9 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -89,7 +89,7 @@ from airflow.api.common.experimental.mark_tasks import (
set_dag_run_state_to_success,
)
from airflow.configuration import AIRFLOW_CONFIG, conf
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, SerializedDagNotFound
from airflow.executors.executor_loader import ExecutorLoader
from airflow.jobs.base_job import BaseJob
from airflow.jobs.scheduler_job import SchedulerJob
@@ -1677,7 +1677,10 @@ class Airflow(AirflowBaseView):
payload = []
for dag_id, active_dag_runs in dags:
max_active_runs = 0
- dag = current_app.dag_bag.get_dag(dag_id)
+ try:
+ dag = current_app.dag_bag.get_dag(dag_id)
+ except SerializedDagNotFound:
+ dag = None
if dag:
# TODO: Make max_active_runs a column so we can query for it directly
max_active_runs = dag.max_active_runs
diff --git a/tests/www/views/test_views_blocked.py b/tests/www/views/test_views_blocked.py
new file mode 100644
index 0000000..26ad7bd
--- /dev/null
+++ b/tests/www/views/test_views_blocked.py
@@ -0,0 +1,84 @@
+#
+# 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.
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.operators.dummy import DummyOperator
+from airflow.operators.subdag import SubDagOperator
+from airflow.utils import timezone
+from airflow.utils.session import create_session
+from airflow.utils.state import State
+from tests.test_utils.db import clear_db_runs
+
+
+@pytest.fixture()
+def running_subdag(admin_client, dag_maker):
+ with dag_maker(dag_id="running_dag.subdag") as subdag:
+ DummyOperator(task_id="dummy")
+
+ with dag_maker(dag_id="running_dag") as dag:
+ SubDagOperator(task_id="subdag", subdag=subdag)
+
+ dag_bag = DagBag(include_examples=False, include_smart_sensor=False)
+ dag_bag.bag_dag(dag, root_dag=dag)
+
+ with create_session() as session:
+ # This writes both DAGs to DagModel, but only serialize the parent DAG.
+ dag_bag.sync_to_db(session=session)
+
+ # Simulate triggering the SubDagOperator to run the subdag.
+ subdag.create_dagrun(
+ run_id="blocked_run_example_bash_operator",
+ state=State.RUNNING,
+ execution_date=timezone.datetime(2016, 1, 1),
+ start_date=timezone.datetime(2016, 1, 1),
+ session=session,
+ )
+
+ # Now delete the parent DAG but leave the subdag.
+ session.query(DagModel).filter(DagModel.dag_id == dag.dag_id).delete()
+ session.query(SerializedDagModel).filter(SerializedDagModel.dag_id == dag.dag_id).delete()
+
+ yield subdag
+
+ with create_session() as session:
+ session.query(DagModel).filter(DagModel.dag_id == subdag.dag_id).delete()
+ clear_db_runs()
+
+
+def test_blocked_subdag_success(admin_client, running_subdag):
+ """Test the /blocked endpoint works when a DAG is deleted.
+
+ When a DAG is bagged, it is written to both DagModel and SerializedDagModel,
+ but its subdags are only written to DagModel (without serialization). Thus,
+ ``DagBag.get_dag(subdag_id)`` would raise ``SerializedDagNotFound`` if the
+ subdag was not previously bagged in the dagbag (perhaps due to its root DAG
+ being deleted). ``DagBag.get_dag()`` calls should catch the exception and
+ properly handle this situation.
+ """
+ resp = admin_client.post("/blocked", data={"dag_ids": [running_subdag.dag_id]})
+ assert resp.status_code == 200
+ assert resp.json == [
+ {
+ "dag_id": running_subdag.dag_id,
+ "active_dag_run": 1,
+ "max_active_runs": 0, # Default value for an unserialized DAG.
+ },
+ ]
[airflow] 23/38: Adds more explanatory message when SecretsMasker
is not configured (#17101)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 4709c65b5189371d93003d48bf256ac4c754cdc0
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Tue Jul 20 10:51:59 2021 +0200
Adds more explanatory message when SecretsMasker is not configured (#17101)
The secrets masker added in 2.1.0 introduced requirement that
at least one SecretsMasker needs to be configured per task.
However this introduced problems for several users who migrated
from Airflow 1.10 and had their custom logging configuration
done without first copying the base airflow configuration.
The message about missing SecretsMasker was pretty cryptic for the
users. This PR changes the message to be much more descriptive
and pointing the user to the right place in documentation
explaining how advanced logging configuration should be done.
(cherry picked from commit 3a2e162387b5d73f1badda8fcf027fbc2caa0f28)
---
airflow/utils/log/secrets_masker.py | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
diff --git a/airflow/utils/log/secrets_masker.py b/airflow/utils/log/secrets_masker.py
index 1e8ad16..0ce0424 100644
--- a/airflow/utils/log/secrets_masker.py
+++ b/airflow/utils/log/secrets_masker.py
@@ -102,7 +102,12 @@ def _secrets_masker() -> "SecretsMasker":
for flt in logging.getLogger('airflow.task').filters:
if isinstance(flt, SecretsMasker):
return flt
- raise RuntimeError("No SecretsMasker found!")
+ raise RuntimeError(
+ "Logging Configuration Error! No SecretsMasker found! If you have custom logging, please make "
+ "sure you configure it taking airflow configuration as a base as explained at "
+ "https://airflow.apache.org/docs/apache-airflow/stable/logging-monitoring/logging-tasks.html"
+ "#advanced-configuration"
+ )
class SecretsMasker(logging.Filter):
[airflow] 01/38: Add back missing permissions to UserModelView
controls. (#17431)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit b82782c68c98d2c193e4cca1ba1e94e1dbac2609
Author: James Timmins <ja...@astronomer.io>
AuthorDate: Mon Aug 9 07:46:05 2021 -0700
Add back missing permissions to UserModelView controls. (#17431)
Currently, on user model views except from UserDBModelView, view controls don't show up on /users/list. This fixes that issue by adding the missing views back to all user model views.
Additionally, Edit User on the different Show User views all redirect to the logged in user's profile views. This fixes that issue by removing the Edit User view from Show User.
closes: #16202
(cherry picked from commit c1e2af4dd2bf868307caae9f2fa825562319a4f8)
---
airflow/www/views.py | 143 +++++++++--------
tests/www/views/test_views_custom_user_views.py | 199 ++++++++++++++++++++++++
2 files changed, 277 insertions(+), 65 deletions(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index e21b823..fa89cb1 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -55,6 +55,7 @@ from flask_appbuilder import BaseView, ModelView, expose
from flask_appbuilder.actions import action
from flask_appbuilder.fieldwidgets import Select2Widget
from flask_appbuilder.models.sqla.filters import BaseFilter
+from flask_appbuilder.security.decorators import has_access
from flask_appbuilder.security.views import (
PermissionModelView,
PermissionViewModelView,
@@ -4112,7 +4113,72 @@ class CustomViewMenuModelView(ViewMenuModelView):
]
-class CustomUserDBModelView(UserDBModelView):
+class CustomUserInfoEditView(UserInfoEditView):
+ """Customize permission names for FAB's builtin UserInfoEditView."""
+
+ class_permission_name = permissions.RESOURCE_MY_PROFILE
+ route_base = "/userinfoeditview"
+ method_permission_name = {
+ 'this_form_get': 'edit',
+ 'this_form_post': 'edit',
+ }
+ base_permissions = [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]
+
+
+class CustomUserStatsChartView(UserStatsChartView):
+ """Customize permission names for FAB's builtin UserStatsChartView."""
+
+ class_permission_name = permissions.RESOURCE_USER_STATS_CHART
+ route_base = "/userstatschartview"
+ method_permission_name = {
+ 'chart': 'read',
+ 'list': 'read',
+ }
+ base_permissions = [permissions.ACTION_CAN_READ]
+
+
+class MultiResourceUserMixin:
+ """Remaps UserModelView permissions to new resources and actions."""
+
+ _class_permission_name = permissions.RESOURCE_USER
+
+ class_permission_name_mapping = {
+ 'userinfoedit': permissions.RESOURCE_MY_PROFILE,
+ 'userinfo': permissions.RESOURCE_MY_PROFILE,
+ }
+
+ method_permission_name = {
+ 'userinfo': 'read',
+ 'download': 'read',
+ 'show': 'read',
+ 'list': 'read',
+ 'edit': 'edit',
+ 'userinfoedit': 'edit',
+ 'delete': 'delete',
+ }
+
+ base_permissions = [
+ permissions.ACTION_CAN_READ,
+ permissions.ACTION_CAN_EDIT,
+ permissions.ACTION_CAN_DELETE,
+ ]
+
+ @expose("/show/<pk>", methods=["GET"])
+ @has_access
+ def show(self, pk):
+ pk = self._deserialize_pk_if_composite(pk)
+ widgets = self._show(pk)
+ widgets['show'].template_args['actions'].pop('userinfoedit')
+ return self.render_template(
+ self.show_template,
+ pk=pk,
+ title=self.show_title,
+ widgets=widgets,
+ related_views=self._related_views,
+ )
+
+
+class CustomUserDBModelView(MultiResourceUserMixin, UserDBModelView):
"""Customize permission names for FAB's builtin UserDBModelView."""
_class_permission_name = permissions.RESOURCE_USER
@@ -4126,15 +4192,15 @@ class CustomUserDBModelView(UserDBModelView):
method_permission_name = {
'add': 'create',
- 'userinfo': 'read',
'download': 'read',
'show': 'read',
'list': 'read',
'edit': 'edit',
+ 'delete': 'delete',
'resetmypassword': 'read',
'resetpasswords': 'read',
- 'userinfoedit': 'edit',
- 'delete': 'delete',
+ 'userinfo': 'read',
+ 'userinfoedit': 'read',
}
base_permissions = [
@@ -4154,7 +4220,6 @@ class CustomUserDBModelView(UserDBModelView):
return self.class_permission_name_mapping.get(action_name, self._class_permission_name)
if method_name:
return self.class_permission_name_mapping.get(method_name, self._class_permission_name)
-
return self._class_permission_name
@class_permission_name.setter
@@ -4162,77 +4227,25 @@ class CustomUserDBModelView(UserDBModelView):
self._class_permission_name = name
-class CustomUserInfoEditView(UserInfoEditView):
- """Customize permission names for FAB's builtin UserInfoEditView."""
-
- class_permission_name = permissions.RESOURCE_MY_PROFILE
- route_base = "/userinfoeditview"
- method_permission_name = {
- 'this_form_get': 'read',
- 'this_form_post': 'edit',
- }
- base_permissions = [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]
-
-
-class CustomUserStatsChartView(UserStatsChartView):
- """Customize permission names for FAB's builtin UserStatsChartView."""
-
- class_permission_name = permissions.RESOURCE_USER_STATS_CHART
- route_base = "/userstatschartview"
- method_permission_name = {
- 'chart': 'read',
- 'list': 'read',
- }
- base_permissions = [permissions.ACTION_CAN_READ]
-
-
-class CustomUserLDAPModelView(UserLDAPModelView):
+class CustomUserLDAPModelView(MultiResourceUserMixin, UserLDAPModelView):
"""Customize permission names for FAB's builtin UserLDAPModelView."""
- class_permission_name = permissions.RESOURCE_MY_PROFILE
- method_permission_name = {
- 'userinfo': 'read',
- 'list': 'read',
- }
- base_permissions = [
- permissions.ACTION_CAN_READ,
- ]
+ pass
-class CustomUserOAuthModelView(UserOAuthModelView):
+class CustomUserOAuthModelView(MultiResourceUserMixin, UserOAuthModelView):
"""Customize permission names for FAB's builtin UserOAuthModelView."""
- class_permission_name = permissions.RESOURCE_MY_PROFILE
- method_permission_name = {
- 'userinfo': 'read',
- 'list': 'read',
- }
- base_permissions = [
- permissions.ACTION_CAN_READ,
- ]
+ pass
-class CustomUserOIDModelView(UserOIDModelView):
+class CustomUserOIDModelView(MultiResourceUserMixin, UserOIDModelView):
"""Customize permission names for FAB's builtin UserOIDModelView."""
- class_permission_name = permissions.RESOURCE_MY_PROFILE
- method_permission_name = {
- 'userinfo': 'read',
- 'list': 'read',
- }
- base_permissions = [
- permissions.ACTION_CAN_READ,
- ]
+ pass
-class CustomUserRemoteUserModelView(UserRemoteUserModelView):
+class CustomUserRemoteUserModelView(MultiResourceUserMixin, UserRemoteUserModelView):
"""Customize permission names for FAB's builtin UserRemoteUserModelView."""
- class_permission_name = permissions.RESOURCE_MY_PROFILE
- method_permission_name = {
- 'userinfo': 'read',
- 'list': 'read',
- }
- base_permissions = [
- permissions.ACTION_CAN_READ,
- ]
+ pass
diff --git a/tests/www/views/test_views_custom_user_views.py b/tests/www/views/test_views_custom_user_views.py
new file mode 100644
index 0000000..a031472
--- /dev/null
+++ b/tests/www/views/test_views_custom_user_views.py
@@ -0,0 +1,199 @@
+#
+# 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.
+
+import unittest
+
+from flask_appbuilder import SQLA
+from parameterized import parameterized
+
+from airflow import settings
+from airflow.security import permissions
+from airflow.www import app as application
+from airflow.www.views import CustomUserDBModelView
+from tests.test_utils.api_connexion_utils import create_user, delete_role
+from tests.test_utils.www import check_content_in_response, check_content_not_in_response, client_with_login
+
+
+class TestSecurity(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ settings.configure_orm()
+ cls.session = settings.Session
+ cls.app = application.create_app(testing=True)
+ cls.appbuilder = cls.app.appbuilder
+ cls.app.config['WTF_CSRF_ENABLED'] = False
+ cls.security_manager = cls.appbuilder.sm
+
+ cls.delete_roles()
+
+ def setUp(self):
+ self.db = SQLA(self.app)
+ self.appbuilder.add_view(CustomUserDBModelView, "CustomUserDBModelView", category="ModelViews")
+ self.client = self.app.test_client() # type:ignore
+
+ @classmethod
+ def delete_roles(cls):
+ for role_name in ['role_edit_one_dag']:
+ delete_role(cls.app, role_name)
+
+ @parameterized.expand(
+ [
+ (
+ "/resetpassword/form?pk={user.id}",
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_PASSWORD),
+ 'Reset Password Form',
+ ),
+ (
+ "/resetmypassword/form",
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PASSWORD),
+ 'Reset Password Form',
+ ),
+ (
+ "/users/userinfo/",
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PROFILE),
+ 'Your user information',
+ ),
+ (
+ "/userinfoeditview/form",
+ (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_MY_PROFILE),
+ 'Edit User',
+ ),
+ ("/users/add", (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_USER), 'Add User'),
+ ("/users/list/", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'List Users'),
+ ("/users/show/{user.id}", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'Show User'),
+ ("/users/edit/{user.id}", (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_USER), 'Edit User'),
+ ]
+ )
+ def test_user_model_view_with_access(self, url, permission, expected_text):
+ user_without_access = create_user(
+ self.app,
+ username="no_access",
+ role_name="role_no_access",
+ permissions=[
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+ ],
+ )
+ client = client_with_login(
+ self.app,
+ username="no_access",
+ password="no_access",
+ )
+ response = client.get(url.replace("{user.id}", str(user_without_access.id)), follow_redirects=True)
+ check_content_not_in_response(expected_text, response)
+
+ @parameterized.expand(
+ [
+ (
+ "/resetpassword/form?pk={user.id}",
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_PASSWORD),
+ 'Reset Password Form',
+ ),
+ (
+ "/resetmypassword/form",
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PASSWORD),
+ 'Reset Password Form',
+ ),
+ (
+ "/users/userinfo/",
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PROFILE),
+ 'Your user information',
+ ),
+ (
+ "/userinfoeditview/form",
+ (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_MY_PROFILE),
+ 'Edit User',
+ ),
+ ("/users/add", (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_USER), 'Add User'),
+ ("/users/list/", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'List Users'),
+ ("/users/show/{user.id}", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'Show User'),
+ ("/users/edit/{user.id}", (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_USER), 'Edit User'),
+ ]
+ )
+ def test_user_model_view_without_access(self, url, permission, expected_text):
+
+ user_with_access = create_user(
+ self.app,
+ username="has_access",
+ role_name="role_has_access",
+ permissions=[(permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE), permission],
+ )
+
+ client = client_with_login(
+ self.app,
+ username="has_access",
+ password="has_access",
+ )
+ response = client.get(url.replace("{user.id}", str(user_with_access.id)), follow_redirects=True)
+ check_content_in_response(expected_text, response)
+
+ def test_user_model_view_without_delete_access(self):
+
+ user_to_delete = create_user(
+ self.app,
+ username="user_to_delete",
+ role_name="user_to_delete",
+ )
+
+ create_user(
+ self.app,
+ username="no_access",
+ role_name="role_no_access",
+ permissions=[
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+ ],
+ )
+
+ client = client_with_login(
+ self.app,
+ username="no_access",
+ password="no_access",
+ )
+
+ response = client.post(f"/users/delete/{user_to_delete.id}", follow_redirects=True)
+
+ check_content_not_in_response("Deleted Row", response)
+ assert bool(self.security_manager.get_user_by_id(user_to_delete.id)) is True
+
+ def test_user_model_view_with_delete_access(self):
+
+ user_to_delete = create_user(
+ self.app,
+ username="user_to_delete",
+ role_name="user_to_delete",
+ )
+
+ create_user(
+ self.app,
+ username="has_access",
+ role_name="role_has_access",
+ permissions=[
+ (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+ (permissions.ACTION_CAN_DELETE, permissions.RESOURCE_USER),
+ ],
+ )
+
+ client = client_with_login(
+ self.app,
+ username="has_access",
+ password="has_access",
+ )
+
+ response = client.post(f"/users/delete/{user_to_delete.id}", follow_redirects=True)
+ check_content_in_response("Deleted Row", response)
+ check_content_not_in_response(user_to_delete.username, response)
+ assert bool(self.security_manager.get_user_by_id(user_to_delete.id)) is False
[airflow] 17/38: Validate type of `priority_weight` during parsing
(#16765)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d04fe50a3f76402517e042c69ea4029b62502105
Author: Kaxil Naik <ka...@gmail.com>
AuthorDate: Fri Jul 2 01:52:50 2021 +0100
Validate type of `priority_weight` during parsing (#16765)
closes https://github.com/apache/airflow/issues/16762
Without this the scheduler crashes as validation does not happen at DAG Parsing time.
(cherry picked from commit 9d170279a60d9d4ed513bae1c35999926f042662)
---
airflow/models/baseoperator.py | 7 ++++++-
tests/models/test_baseoperator.py | 5 +++++
2 files changed, 11 insertions(+), 1 deletion(-)
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
index 10e8bfd..1fec8cf 100644
--- a/airflow/models/baseoperator.py
+++ b/airflow/models/baseoperator.py
@@ -586,10 +586,15 @@ class BaseOperator(Operator, LoggingMixin, TaskMixin, metaclass=BaseOperatorMeta
if isinstance(max_retry_delay, timedelta):
self.max_retry_delay = max_retry_delay
else:
- self.log.debug("Max_retry_delay isn't timedelta object, assuming secs")
+ self.log.debug("max_retry_delay isn't a timedelta object, assuming secs")
self.max_retry_delay = timedelta(seconds=max_retry_delay)
self.params = params or {} # Available in templates!
+ if priority_weight is not None and not isinstance(priority_weight, int):
+ raise AirflowException(
+ f"`priority_weight` for task '{self.task_id}' only accepts integers, "
+ f"received '{type(priority_weight)}'."
+ )
self.priority_weight = priority_weight
if not WeightRule.is_valid(weight_rule):
raise AirflowException(
diff --git a/tests/models/test_baseoperator.py b/tests/models/test_baseoperator.py
index fa02b4e..04d3f54 100644
--- a/tests/models/test_baseoperator.py
+++ b/tests/models/test_baseoperator.py
@@ -109,6 +109,11 @@ class TestBaseOperator(unittest.TestCase):
with pytest.raises(AirflowException, match='Argument.*test_param.*required'):
DummyClass(default_args=default_args)
+ def test_incorrect_priority_weight(self):
+ error_msg = "`priority_weight` for task 'test_op' only accepts integers, received '<class 'str'>'."
+ with pytest.raises(AirflowException, match=error_msg):
+ DummyOperator(task_id="test_op", priority_weight="2")
+
@parameterized.expand(
[
("{{ foo }}", {"foo": "bar"}, "bar"),
[airflow] 11/38: Fix ``AttributeError``: ``datetime.timezone``
object has no attribute ``name`` (#16599)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e59367192dbe14d15b536983bc17f56a0087a136
Author: Ruben Laguna <ru...@gmail.com>
AuthorDate: Wed Jun 23 16:31:45 2021 +0200
Fix ``AttributeError``: ``datetime.timezone`` object has no attribute ``name`` (#16599)
closes: #16551
Previous implementation tried to force / coerce the provided timezone (from the dag's `start_date`) into a `pendulum.tz.timezone.*` that only worked if the provided timezone was already a pendulum's timezone and it specifically failed when with `datetime.timezone.utc` as timezone.
(cherry picked from commit 86c20910aed48f7d5b2ebaa91fa40d47c52d7db3)
---
airflow/models/dag.py | 4 +--
tests/models/test_dag.py | 40 +++++++++++++++++++++++++++
tests/serialization/test_dag_serialization.py | 2 ++
3 files changed, 44 insertions(+), 2 deletions(-)
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 1861eda..13d69c2 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -485,7 +485,7 @@ class DAG(LoggingMixin):
else:
# absolute (e.g. 3 AM)
naive = cron.get_next(datetime)
- tz = pendulum.timezone(self.timezone.name)
+ tz = self.timezone
following = timezone.make_aware(naive, tz)
return timezone.convert_to_utc(following)
elif self.normalized_schedule_interval is not None:
@@ -513,7 +513,7 @@ class DAG(LoggingMixin):
else:
# absolute (e.g. 3 AM)
naive = cron.get_prev(datetime)
- tz = pendulum.timezone(self.timezone.name)
+ tz = self.timezone
previous = timezone.make_aware(naive, tz)
return timezone.convert_to_utc(previous)
elif self.normalized_schedule_interval is not None:
diff --git a/tests/models/test_dag.py b/tests/models/test_dag.py
index 939baf4..34d761d 100644
--- a/tests/models/test_dag.py
+++ b/tests/models/test_dag.py
@@ -621,6 +621,46 @@ class TestDag(unittest.TestCase):
_next = dag.following_schedule(_next)
assert _next.isoformat() == "2015-01-02T02:00:00+00:00"
+ def test_previous_schedule_datetime_timezone(self):
+ # Check that we don't get an AttributeError 'name' for self.timezone
+
+ start = datetime.datetime(2018, 3, 25, 2, tzinfo=datetime.timezone.utc)
+ dag = DAG('tz_dag', start_date=start, schedule_interval='@hourly')
+ when = dag.previous_schedule(start)
+ assert when.isoformat() == "2018-03-25T01:00:00+00:00"
+
+ def test_following_schedule_datetime_timezone(self):
+ # Check that we don't get an AttributeError 'name' for self.timezone
+
+ start = datetime.datetime(2018, 3, 25, 2, tzinfo=datetime.timezone.utc)
+ dag = DAG('tz_dag', start_date=start, schedule_interval='@hourly')
+ when = dag.following_schedule(start)
+ assert when.isoformat() == "2018-03-25T03:00:00+00:00"
+
+ def test_following_schedule_datetime_timezone_utc0530(self):
+ # Check that we don't get an AttributeError 'name' for self.timezone
+ class UTC0530(datetime.tzinfo):
+ """tzinfo derived concrete class named "+0530" with offset of 19800"""
+
+ # can be configured here
+ _offset = datetime.timedelta(seconds=19800)
+ _dst = datetime.timedelta(0)
+ _name = "+0530"
+
+ def utcoffset(self, dt):
+ return self.__class__._offset
+
+ def dst(self, dt):
+ return self.__class__._dst
+
+ def tzname(self, dt):
+ return self.__class__._name
+
+ start = datetime.datetime(2018, 3, 25, 10, tzinfo=UTC0530())
+ dag = DAG('tz_dag', start_date=start, schedule_interval='@hourly')
+ when = dag.following_schedule(start)
+ assert when.isoformat() == "2018-03-25T05:30:00+00:00"
+
def test_dagtag_repr(self):
clear_db_dags()
dag = DAG('dag-test-dagtag', start_date=DEFAULT_DATE, tags=['tag-1', 'tag-2'])
diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index f24e862..29e6ac0 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -27,6 +27,7 @@ from datetime import datetime, timedelta, timezone
from glob import glob
from unittest import mock
+import pendulum
import pytest
from dateutil.relativedelta import FR, relativedelta
from kubernetes.client import models as k8s
@@ -444,6 +445,7 @@ class TestStringifiedDAGs(unittest.TestCase):
datetime(2019, 7, 30, tzinfo=timezone.utc),
datetime(2019, 8, 1, tzinfo=timezone.utc),
),
+ (pendulum.datetime(2019, 8, 1, tz='UTC'), None, pendulum.datetime(2019, 8, 1, tz='UTC')),
]
)
def test_deserialization_start_date(self, dag_start_date, task_start_date, expected_task_start_date):
[airflow] 02/38: Path correction in docs for airflow core (#17567)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit be09eef3f437e3793a6370783aa9ccea797c225b
Author: Amit Talreja <ss...@gmail.com>
AuthorDate: Thu Aug 12 19:20:41 2021 +0530
Path correction in docs for airflow core (#17567)
Airflow core constraint path was wrong.
(cherry picked from commit 951006e1ca36db02c05b05a749edd0b88b607f6e)
---
docs/apache-airflow/installation.rst | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/apache-airflow/installation.rst b/docs/apache-airflow/installation.rst
index 8d1624c..574d7ec 100644
--- a/docs/apache-airflow/installation.rst
+++ b/docs/apache-airflow/installation.rst
@@ -259,7 +259,7 @@ If you don't want to install any extra providers, initially you can use the comm
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
# For example: 3.6
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-no-providers-${PYTHON_VERSION}.txt"
- # For example: https://raw.githubusercontent.com/apache/airflow/constraints-no-providers-|version|/constraints-3.6.txt
+ # For example: https://raw.githubusercontent.com/apache/airflow/constraints-|version|/constraints-no-providers-3.6.txt
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"
[airflow] 28/38: Fix CLI 'kubernetes cleanup-pods' which fails on
invalid label key (#17298)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 5739fedc550ef2dc708dec719b0dab5b048cde60
Author: Damir Lampa <60...@users.noreply.github.com>
AuthorDate: Thu Jul 29 14:17:51 2021 -0600
Fix CLI 'kubernetes cleanup-pods' which fails on invalid label key (#17298)
Fix for #16013 - CLI 'kubernetes cleanup-pods' fails on invalid label key
(cherry picked from commit 36bdfe8d0ef7e5fc428434f8313cf390ee9acc8f)
---
airflow/cli/commands/kubernetes_command.py | 12 ++----------
tests/cli/commands/test_kubernetes_command.py | 8 ++------
2 files changed, 4 insertions(+), 16 deletions(-)
diff --git a/airflow/cli/commands/kubernetes_command.py b/airflow/cli/commands/kubernetes_command.py
index 3c3c8e6..2660dae 100644
--- a/airflow/cli/commands/kubernetes_command.py
+++ b/airflow/cli/commands/kubernetes_command.py
@@ -96,16 +96,8 @@ def cleanup_pods(args):
'try_number',
'airflow_version',
]
- list_kwargs = {
- "namespace": namespace,
- "limit": 500,
- "label_selector": client.V1LabelSelector(
- match_expressions=[
- client.V1LabelSelectorRequirement(key=label, operator="Exists")
- for label in airflow_pod_labels
- ]
- ),
- }
+ list_kwargs = {"namespace": namespace, "limit": 500, "label_selector": ','.join(airflow_pod_labels)}
+
while True:
pod_list = kube_client.list_namespaced_pod(**list_kwargs)
for pod in pod_list.items:
diff --git a/tests/cli/commands/test_kubernetes_command.py b/tests/cli/commands/test_kubernetes_command.py
index f2a8605..490c7fa 100644
--- a/tests/cli/commands/test_kubernetes_command.py
+++ b/tests/cli/commands/test_kubernetes_command.py
@@ -55,12 +55,8 @@ class TestGenerateDagYamlCommand(unittest.TestCase):
class TestCleanUpPodsCommand(unittest.TestCase):
- label_selector = kubernetes.client.V1LabelSelector(
- match_expressions=[
- kubernetes.client.V1LabelSelectorRequirement(key=label, operator="Exists")
- for label in ['dag_id', 'task_id', 'execution_date', 'try_number', 'airflow_version']
- ]
- )
+
+ label_selector = ','.join(['dag_id', 'task_id', 'execution_date', 'try_number', 'airflow_version'])
@classmethod
def setUpClass(cls):
[airflow] 22/38: #16976 Add json.dumps() for templated fields
objects: 'dict' and 'list' (#17082)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f5ef0dda988bdfef75a396f0cae1b8f7e031818e
Author: Vladimir Aranovsky <66...@users.noreply.github.com>
AuthorDate: Mon Jul 19 22:22:42 2021 +0300
#16976 Add json.dumps() for templated fields objects: 'dict' and 'list' (#17082)
(cherry picked from commit 1a0730a08f2d72cd71447b6d6549ec10d266dd6a)
---
airflow/www/views.py | 2 ++
1 file changed, 2 insertions(+)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 6b2b2a4..f77e66c 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -972,6 +972,8 @@ class Airflow(AirflowBaseView):
content = getattr(task, template_field)
renderer = task.template_fields_renderers.get(template_field, template_field)
if renderer in renderers:
+ if isinstance(content, (dict, list)):
+ content = json.dumps(content, sort_keys=True, indent=4)
html_dict[template_field] = renderers[renderer](content)
else:
html_dict[template_field] = Markup("<pre><code>{}</pre></code>").format(pformat(content))
[airflow] 14/38: Fix slow (cleared) tasks being be adopted by
Celery worker. (#16718)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3211672835cb13e305305ff956a524a118196175
Author: Jorrick Sleijster <jo...@gmail.com>
AuthorDate: Sat Jul 3 00:02:01 2021 +0200
Fix slow (cleared) tasks being be adopted by Celery worker. (#16718)
Celery executor is currently adopting anything that has ever run before and has been cleared since then.
**Example of the issue:**
We have a DAG that runs over 150 sensor tasks and 50 ETL tasks while having a concurrency of 3 and max_active_runs of 16. This setup is required because we want to divide the resources and we don't want this DAG to take up all the resources. What will happen is that many tasks will be in scheduled for a bit as it can't queue them due to the concurrency of 3. However, because of the current implementations, if these tasks ever run before, they would get adopted by the schedulers execut [...]
**Contents of the PR**:
1. Tasks that are in scheduled should never have arrived at an executor. Hence, we remove the task state scheduled from the option to be adopted.
2. Given this task instance `external_executor_id` is quite important in deciding whether it is adopted, we will also reset this when we reset the state of the TaskInstance.
(cherry picked from commit 554a23928efb4ff1d87d115ae2664edec3a9408c)
---
airflow/jobs/scheduler_job.py | 2 +-
airflow/models/taskinstance.py | 1 +
tests/executors/test_celery_executor.py | 2 ++
tests/jobs/test_scheduler_job.py | 14 +++++++-------
tests/models/test_cleartasks.py | 24 ++++++++++++++++++++++++
5 files changed, 35 insertions(+), 8 deletions(-)
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index 1758ae1..fe8e0b0 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -1859,7 +1859,7 @@ class SchedulerJob(BaseJob):
self.log.info("Marked %d SchedulerJob instances as failed", num_failed)
Stats.incr(self.__class__.__name__.lower() + '_end', num_failed)
- resettable_states = [State.SCHEDULED, State.QUEUED, State.RUNNING]
+ resettable_states = [State.QUEUED, State.RUNNING]
query = (
session.query(TI)
.filter(TI.state.in_(resettable_states))
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 76cb079..aeb2a22 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -170,6 +170,7 @@ def clear_task_instances(
# original max_tries or the last attempted try number.
ti.max_tries = max(ti.max_tries, ti.prev_attempted_tries)
ti.state = State.NONE
+ ti.external_executor_id = None
session.merge(ti)
task_id_by_key[ti.dag_id][ti.execution_date][ti.try_number].add(ti.task_id)
diff --git a/tests/executors/test_celery_executor.py b/tests/executors/test_celery_executor.py
index e24a3dd..88ea95c 100644
--- a/tests/executors/test_celery_executor.py
+++ b/tests/executors/test_celery_executor.py
@@ -329,9 +329,11 @@ class TestCeleryExecutor(unittest.TestCase):
ti1 = TaskInstance(task=task_1, execution_date=exec_date)
ti1.external_executor_id = '231'
ti1.queued_dttm = queued_dttm
+ ti1.state = State.QUEUED
ti2 = TaskInstance(task=task_2, execution_date=exec_date)
ti2.external_executor_id = '232'
ti2.queued_dttm = queued_dttm
+ ti2.state = State.QUEUED
tis = [ti1, ti2]
executor = celery_executor.CeleryExecutor()
diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
index 37ae65b..fe0b257 100644
--- a/tests/jobs/test_scheduler_job.py
+++ b/tests/jobs/test_scheduler_job.py
@@ -2124,9 +2124,9 @@ class TestSchedulerJob(unittest.TestCase):
session=session,
)
ti = dr.get_task_instance(task_id=op1.task_id, session=session)
- ti.state = State.SCHEDULED
+ ti.state = State.QUEUED
ti2 = dr2.get_task_instance(task_id=op1.task_id, session=session)
- ti2.state = State.SCHEDULED
+ ti2.state = State.QUEUED
session.commit()
processor = mock.MagicMock()
@@ -2140,7 +2140,7 @@ class TestSchedulerJob(unittest.TestCase):
assert ti.state == State.NONE
ti2 = dr2.get_task_instance(task_id=op1.task_id, session=session)
- assert ti2.state == State.SCHEDULED, "Tasks run by Backfill Jobs should not be reset"
+ assert ti2.state == State.QUEUED, "Tasks run by Backfill Jobs should not be reset"
@parameterized.expand(
[
@@ -3654,7 +3654,7 @@ class TestSchedulerJob(unittest.TestCase):
session=session,
)
ti = dr1.get_task_instances(session=session)[0]
- ti.state = State.SCHEDULED
+ ti.state = State.QUEUED
session.merge(ti)
session.merge(dr1)
session.commit()
@@ -3799,12 +3799,12 @@ class TestSchedulerJob(unittest.TestCase):
ti1, ti2 = dr1.get_task_instances(session=session)
dr1.state = State.RUNNING
- ti1.state = State.SCHEDULED
+ ti1.state = State.QUEUED
ti1.queued_by_job_id = old_job.id
session.merge(dr1)
session.merge(ti1)
- ti2.state = State.SCHEDULED
+ ti2.state = State.QUEUED
ti2.queued_by_job_id = self.scheduler_job.id
session.merge(ti2)
session.flush()
@@ -3816,7 +3816,7 @@ class TestSchedulerJob(unittest.TestCase):
session.refresh(ti1)
assert ti1.state is None
session.refresh(ti2)
- assert State.SCHEDULED == ti2.state
+ assert ti2.state == State.QUEUED
session.rollback()
if old_job.processor_agent:
old_job.processor_agent.end()
diff --git a/tests/models/test_cleartasks.py b/tests/models/test_cleartasks.py
index 1c5606e..9b8fbd0 100644
--- a/tests/models/test_cleartasks.py
+++ b/tests/models/test_cleartasks.py
@@ -56,6 +56,7 @@ class TestClearTasks(unittest.TestCase):
ti0.run()
ti1.run()
+
with create_session() as session:
qry = session.query(TI).filter(TI.dag_id == dag.dag_id).all()
clear_task_instances(qry, session, dag=dag)
@@ -68,6 +69,29 @@ class TestClearTasks(unittest.TestCase):
assert ti1.try_number == 2
assert ti1.max_tries == 3
+ def test_clear_task_instances_external_executor_id(self):
+ dag = DAG(
+ 'test_clear_task_instances_external_executor_id',
+ start_date=DEFAULT_DATE,
+ end_date=DEFAULT_DATE + datetime.timedelta(days=10),
+ )
+ task0 = DummyOperator(task_id='task0', owner='test', dag=dag)
+ ti0 = TI(task=task0, execution_date=DEFAULT_DATE)
+ ti0.state = State.SUCCESS
+ ti0.external_executor_id = "some_external_executor_id"
+
+ with create_session() as session:
+ session.add(ti0)
+ session.commit()
+
+ qry = session.query(TI).filter(TI.dag_id == dag.dag_id).all()
+ clear_task_instances(qry, session, dag=dag)
+
+ ti0.refresh_from_db()
+
+ assert ti0.state is None
+ assert ti0.external_executor_id is None
+
def test_clear_task_instances_without_task(self):
dag = DAG(
'test_clear_task_instances_without_task',
[airflow] 21/38: Core: Enable the use of __init_subclass__ in
subclasses of BaseOperator (#17027)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 75c3e752667ef3bb0fe82d8edefafb444531b0e8
Author: Henry Zhang <me...@henry.dev>
AuthorDate: Thu Jul 22 15:23:51 2021 -0700
Core: Enable the use of __init_subclass__ in subclasses of BaseOperator (#17027)
This fixes a regression in 2.1 where subclasses of BaseOperator could no
longer use `__init_subclass__` to allow class instantiation time
customization.
Related BPO: https://bugs.python.org/issue29581
Fixes: https://github.com/apache/airflow/issues/17014
(cherry picked from commit 901513203f287d4f8152f028e9070a2dec73ad74)
---
airflow/models/baseoperator.py | 4 ++--
tests/models/test_baseoperator.py | 26 ++++++++++++++++++++++++++
2 files changed, 28 insertions(+), 2 deletions(-)
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
index 1fec8cf..360f264 100644
--- a/airflow/models/baseoperator.py
+++ b/airflow/models/baseoperator.py
@@ -186,8 +186,8 @@ class BaseOperatorMeta(abc.ABCMeta):
return cast(T, apply_defaults)
- def __new__(cls, name, bases, namespace):
- new_cls = super().__new__(cls, name, bases, namespace)
+ def __new__(cls, name, bases, namespace, **kwargs):
+ new_cls = super().__new__(cls, name, bases, namespace, **kwargs)
new_cls.__init__ = cls._apply_defaults(new_cls.__init__)
return new_cls
diff --git a/tests/models/test_baseoperator.py b/tests/models/test_baseoperator.py
index 04d3f54..87a9247 100644
--- a/tests/models/test_baseoperator.py
+++ b/tests/models/test_baseoperator.py
@@ -516,3 +516,29 @@ class TestXComArgsRelationsAreResolved:
with pytest.raises(AirflowException):
op1 = DummyOperator(task_id="op1")
CustomOp(task_id="op2", field=op1.output)
+
+
+class InitSubclassOp(DummyOperator):
+ def __init_subclass__(cls, class_arg=None, **kwargs) -> None:
+ cls._class_arg = class_arg
+ super().__init_subclass__(**kwargs)
+
+ def execute(self, context):
+ self.context_arg = context
+
+
+class TestInitSubclassOperator:
+ def test_init_subclass_args(self):
+ class_arg = "foo"
+ context = {"key": "value"}
+
+ class ConcreteSubclassOp(InitSubclassOp, class_arg=class_arg):
+ pass
+
+ task = ConcreteSubclassOp(task_id="op1")
+ task_copy = task.prepare_for_execution()
+
+ task_copy.execute(context)
+
+ assert task_copy._class_arg == class_arg
+ assert task_copy.context_arg == context
[airflow] 36/38: Revert "Fix DAG run state not updated while DAG is
paused (#16343)"
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit effbee143fb5f1afdff25b6ec6065808297ad04b
Author: James Timmins <ja...@astronomer.io>
AuthorDate: Fri Aug 13 11:01:58 2021 -0700
Revert "Fix DAG run state not updated while DAG is paused (#16343)"
This reverts commit 446e66b052a116f8371be331624c1e1b03299380.
---
airflow/jobs/local_task_job.py | 15 --------------
tests/jobs/test_local_task_job.py | 42 ++-------------------------------------
2 files changed, 2 insertions(+), 55 deletions(-)
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 3afc801..3dca97e 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -158,8 +158,6 @@ class LocalTaskJob(BaseJob):
if self.task_instance.state != State.SUCCESS:
error = self.task_runner.deserialize_run_error()
self.task_instance._run_finished_callback(error=error)
- if not self.task_instance.test_mode:
- self._update_dagrun_state_for_paused_dag()
def on_kill(self):
self.task_runner.terminate()
@@ -213,16 +211,3 @@ class LocalTaskJob(BaseJob):
error = self.task_runner.deserialize_run_error() or "task marked as failed externally"
ti._run_finished_callback(error=error)
self.terminating = True
-
- @provide_session
- def _update_dagrun_state_for_paused_dag(self, session=None):
- """
- Checks for paused dags with DagRuns in the running state and
- update the DagRun state if possible
- """
- dag = self.task_instance.task.dag
- if dag.get_is_paused():
- dag_run = self.task_instance.get_dagrun(session=session)
- if dag_run:
- dag_run.dag = dag
- dag_run.update_state(session=session, execute_callbacks=True)
diff --git a/tests/jobs/test_local_task_job.py b/tests/jobs/test_local_task_job.py
index ed43198..17a7285 100644
--- a/tests/jobs/test_local_task_job.py
+++ b/tests/jobs/test_local_task_job.py
@@ -44,7 +44,6 @@ from airflow.utils.net import get_hostname
from airflow.utils.session import create_session
from airflow.utils.state import State
from airflow.utils.timeout import timeout
-from airflow.utils.types import DagRunType
from tests.test_utils.asserts import assert_queries_count
from tests.test_utils.db import clear_db_jobs, clear_db_runs
from tests.test_utils.mock_executor import MockExecutor
@@ -668,43 +667,6 @@ class TestLocalTaskJob(unittest.TestCase):
assert task_terminated_externally.value == 1
assert not process.is_alive()
- def test_task_exit_should_update_state_of_finished_dagruns_with_dag_paused(self):
- """Test that with DAG paused, DagRun state will update when the tasks finishes the run"""
- dag = DAG(dag_id='test_dags', start_date=DEFAULT_DATE)
- op1 = PythonOperator(task_id='dummy', dag=dag, owner='airflow', python_callable=lambda: True)
-
- session = settings.Session()
- orm_dag = DagModel(
- dag_id=dag.dag_id,
- has_task_concurrency_limits=False,
- next_dagrun=dag.start_date,
- next_dagrun_create_after=dag.following_schedule(DEFAULT_DATE),
- is_active=True,
- is_paused=True,
- )
- session.add(orm_dag)
- session.flush()
- # Write Dag to DB
- dagbag = DagBag(dag_folder="/dev/null", include_examples=False, read_dags_from_db=False)
- dagbag.bag_dag(dag, root_dag=dag)
- dagbag.sync_to_db()
-
- dr = dag.create_dagrun(
- run_type=DagRunType.SCHEDULED,
- state=State.RUNNING,
- execution_date=DEFAULT_DATE,
- start_date=DEFAULT_DATE,
- session=session,
- )
- assert dr.state == State.RUNNING
- ti = TaskInstance(op1, dr.execution_date)
- job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
- job1.task_runner = StandardTaskRunner(job1)
- job1.run()
- session.add(dr)
- session.refresh(dr)
- assert dr.state == State.SUCCESS
-
@pytest.fixture()
def clean_db_helper():
@@ -723,12 +685,12 @@ class TestLocalTaskJobPerformance:
task = DummyOperator(task_id='test_state_succeeded1', dag=dag)
dag.clear()
- dag.create_dagrun(run_id=unique_prefix, execution_date=DEFAULT_DATE, state=State.NONE)
+ dag.create_dagrun(run_id=unique_prefix, state=State.NONE)
ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
mock_get_task_runner.return_value.return_code.side_effects = return_codes
job = LocalTaskJob(task_instance=ti, executor=MockExecutor())
- with assert_queries_count(16):
+ with assert_queries_count(15):
job.run()
[airflow] 09/38: fix: change graph focus to top of view instead of
center (#16484)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f3d431ddd3dffaaff2cf3d9a922472f07ba413e3
Author: Brent Bovenzi <br...@gmail.com>
AuthorDate: Wed Jun 16 13:20:19 2021 -0500
fix: change graph focus to top of view instead of center (#16484)
(cherry picked from commit f1675853a5ed9b779ee2fc13bb9aa97185472bc7)
---
airflow/www/static/js/graph.js | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/airflow/www/static/js/graph.js b/airflow/www/static/js/graph.js
index 6482d91..5d8ab8e 100644
--- a/airflow/www/static/js/graph.js
+++ b/airflow/www/static/js/graph.js
@@ -553,7 +553,6 @@ function focusedGroupKey() {
function focusGroup(nodeId) {
if (nodeId != null && zoom != null) {
const { x } = g.node(nodeId);
- const { y } = g.node(nodeId);
// This is the total canvas size.
const { width, height } = svg.node().getBoundingClientRect();
@@ -572,7 +571,8 @@ function focusGroup(nodeId) {
1.5, // cap zoom level to 1.5 so nodes are not too large
) * 0.9;
- const [deltaX, deltaY] = [width / 2 - x * scale, height / 2 - y * scale];
+ // deltaY of 5 keeps the zoom at the top of the view but with a slight margin
+ const [deltaX, deltaY] = [width / 2 - x * scale, 5];
zoom.translate([deltaX, deltaY]);
zoom.scale(scale);
zoom.event(innerSvg.transition().duration(duration));
[airflow] 30/38: Optimize context sent for docker build (#17415)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 93b4c7b7b4dd4696753d918c1394d605100b0170
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Wed Aug 4 15:56:52 2021 +0200
Optimize context sent for docker build (#17415)
The `provider_packages` folder is not needed during docker build
and it often contains copied sources when building the provider
packages. Also _doctree folder is prepared during documentation
building. This makes `docker build` wait for a few seconds when
those directories have a lot of files.
Excluding those decreases docker build overhead significantly.
(cherry picked from commit dfefc66dfb9ae907dc5f49bb4ec7dd65daecd044)
---
.dockerignore | 4 +---
1 file changed, 1 insertion(+), 3 deletions(-)
diff --git a/.dockerignore b/.dockerignore
index d10cfbc..f6113e2 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -40,9 +40,6 @@
!scripts/in_container
!scripts/docker
-# Add provider packages to the context
-!provider_packages
-
# Add tests and kubernetes_tests to context.
!tests
!kubernetes_tests
@@ -129,3 +126,4 @@ airflow/www/static/docs
# Exclude docs generated files
docs/_build/
docs/_api/
+docs/_doctrees/
[airflow] 07/38: commiting dagPickle session when the airflow tasks
run --ship-dag --interactive command is executed FIXES: 15748 (#15890)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit dd3c06f21a71d8b4a3e485895ca26c543adc9523
Author: vikram Jadhav <vi...@gmail.com>
AuthorDate: Thu Jun 24 22:57:20 2021 +0530
commiting dagPickle session when the airflow tasks run --ship-dag --interactive command is executed FIXES: 15748 (#15890)
(cherry picked from commit 86d0a96bf796fd767cf50a7224be060efa402d94)
---
airflow/cli/commands/task_command.py | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
index 0302d4d..6dd36db 100644
--- a/airflow/cli/commands/task_command.py
+++ b/airflow/cli/commands/task_command.py
@@ -80,9 +80,9 @@ def _run_task_by_executor(args, dag, ti):
with create_session() as session:
pickle = DagPickle(dag)
session.add(pickle)
- pickle_id = pickle.id
- # TODO: This should be written to a log
- print(f'Pickled dag {dag} as pickle_id: {pickle_id}')
+ pickle_id = pickle.id
+ # TODO: This should be written to a log
+ print(f'Pickled dag {dag} as pickle_id: {pickle_id}')
except Exception as e:
print('Could not pickle the DAG')
print(e)
[airflow] 05/38: fix(smart_sensor): Unbound variable errors (#14774)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6ebbf8272f5cfc5a952731a6c70c528504015193
Author: Shivansh Saini <sh...@gmail.com>
AuthorDate: Thu Jun 24 03:52:27 2021 +0530
fix(smart_sensor): Unbound variable errors (#14774)
Signed-off-by: Shivansh Saini <sh...@gmail.com>
Closes #14770
(cherry picked from commit 4aec25a80e3803238cf658c416c8e6d3975a30f6)
---
airflow/sensors/smart_sensor.py | 1 +
1 file changed, 1 insertion(+)
diff --git a/airflow/sensors/smart_sensor.py b/airflow/sensors/smart_sensor.py
index c8c5ba7..8755eb5 100644
--- a/airflow/sensors/smart_sensor.py
+++ b/airflow/sensors/smart_sensor.py
@@ -435,6 +435,7 @@ class SmartSensorOperator(BaseOperator, SkipMixin):
TI = TaskInstance
count_marked = 0
+ query_result = []
try:
query_result = (
session.query(TI, SI)
[airflow] 38/38: Fix DAG run state not updated while DAG is paused
(#16343)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7e79747f53d5501a3786639336abda6ca9d104d3
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Fri Jun 18 00:29:00 2021 +0100
Fix DAG run state not updated while DAG is paused (#16343)
The state of a DAG run does not update while the DAG is paused.
The tasks continue to run if the DAG run was kicked off before
the DAG was paused and eventually finish and are marked correctly.
The DAG run state does not get updated and stays in Running state until the DAG is unpaused.
This change fixes it by running a check on task exit to update state(if possible)
of the DagRun if the task was able to finish the DagRun while the DAG is paused
Co-authored-by: Ash Berlin-Taylor <as...@firemirror.com>
(cherry picked from commit 3834df6ade22b33addd47e3ab2165a0b282926fa)
---
airflow/jobs/local_task_job.py | 14 +++++++++++++
tests/jobs/test_local_task_job.py | 42 +++++++++++++++++++++++++++++++++++++--
2 files changed, 54 insertions(+), 2 deletions(-)
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 33c2f59..cce4e64 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -164,6 +164,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._update_dagrun_state_for_paused_dag()
def on_kill(self):
self.task_runner.terminate()
@@ -268,3 +269,16 @@ class LocalTaskJob(BaseJob):
exc_info=True,
)
session.rollback()
+
+ @provide_session
+ def _update_dagrun_state_for_paused_dag(self, session=None):
+ """
+ Checks for paused dags with DagRuns in the running state and
+ update the DagRun state if possible
+ """
+ dag = self.task_instance.task.dag
+ if dag.get_is_paused():
+ dag_run = self.task_instance.get_dagrun(session=session)
+ if dag_run:
+ dag_run.dag = dag
+ dag_run.update_state(session=session, execute_callbacks=True)
diff --git a/tests/jobs/test_local_task_job.py b/tests/jobs/test_local_task_job.py
index 48047ad..11e9adf 100644
--- a/tests/jobs/test_local_task_job.py
+++ b/tests/jobs/test_local_task_job.py
@@ -45,6 +45,7 @@ from airflow.utils.net import get_hostname
from airflow.utils.session import create_session
from airflow.utils.state import State
from airflow.utils.timeout import timeout
+from airflow.utils.types import DagRunType
from tests.test_utils import db
from tests.test_utils.asserts import assert_queries_count
from tests.test_utils.config import conf_vars
@@ -781,6 +782,43 @@ class TestLocalTaskJob(unittest.TestCase):
if scheduler_job.processor_agent:
scheduler_job.processor_agent.end()
+ def test_task_exit_should_update_state_of_finished_dagruns_with_dag_paused(self):
+ """Test that with DAG paused, DagRun state will update when the tasks finishes the run"""
+ dag = DAG(dag_id='test_dags', start_date=DEFAULT_DATE)
+ op1 = PythonOperator(task_id='dummy', dag=dag, owner='airflow', python_callable=lambda: True)
+
+ session = settings.Session()
+ orm_dag = DagModel(
+ dag_id=dag.dag_id,
+ has_task_concurrency_limits=False,
+ next_dagrun=dag.start_date,
+ next_dagrun_create_after=dag.following_schedule(DEFAULT_DATE),
+ is_active=True,
+ is_paused=True,
+ )
+ session.add(orm_dag)
+ session.flush()
+ # Write Dag to DB
+ dagbag = DagBag(dag_folder="/dev/null", include_examples=False, read_dags_from_db=False)
+ dagbag.bag_dag(dag, root_dag=dag)
+ dagbag.sync_to_db()
+
+ dr = dag.create_dagrun(
+ run_type=DagRunType.SCHEDULED,
+ state=State.RUNNING,
+ execution_date=DEFAULT_DATE,
+ start_date=DEFAULT_DATE,
+ session=session,
+ )
+ assert dr.state == State.RUNNING
+ ti = TaskInstance(op1, dr.execution_date)
+ job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
+ job1.task_runner = StandardTaskRunner(job1)
+ job1.run()
+ session.add(dr)
+ session.refresh(dr)
+ assert dr.state == State.SUCCESS
+
@pytest.fixture()
def clean_db_helper():
@@ -799,12 +837,12 @@ class TestLocalTaskJobPerformance:
task = DummyOperator(task_id='test_state_succeeded1', dag=dag)
dag.clear()
- dag.create_dagrun(run_id=unique_prefix, state=State.NONE)
+ dag.create_dagrun(run_id=unique_prefix, execution_date=DEFAULT_DATE, state=State.NONE)
ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
mock_get_task_runner.return_value.return_code.side_effects = return_codes
job = LocalTaskJob(task_instance=ti, executor=MockExecutor())
- with assert_queries_count(15):
+ with assert_queries_count(16):
job.run()
[airflow] 35/38: Proper warning message when recorded PID is
different from current PID (#17411)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6b8398aa2f30c60e7ca4fff983611670da361bd6
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Aug 4 13:32:05 2021 +0100
Proper warning message when recorded PID is different from current PID (#17411)
Currently, when the recorded PID is different from the current PID, in
the case of run_as_user, the warning is not clear because ti.pid is used
as the recorded PID instead of parent process of ti.pid. In this case,
users would see that the PIDs are the same but there was a warning that
they are not the same
This change fixes it.
(cherry picked from commit a4b6f1c1c5ab92fd5b623f119263d83bd46ab2e6)
---
airflow/jobs/local_task_job.py | 13 ++++++++-----
1 file changed, 8 insertions(+), 5 deletions(-)
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index af7b317..3afc801 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -187,14 +187,17 @@ class LocalTaskJob(BaseJob):
)
raise AirflowException("Hostname of job runner does not match")
current_pid = self.task_runner.process.pid
-
- same_process = ti.pid == current_pid
+ recorded_pid = ti.pid
+ same_process = recorded_pid == current_pid
if ti.run_as_user or self.task_runner.run_as_user:
- same_process = psutil.Process(ti.pid).ppid() == current_pid
+ recorded_pid = psutil.Process(ti.pid).ppid()
+ same_process = recorded_pid == current_pid
- if ti.pid is not None and not same_process:
- self.log.warning("Recorded pid %s does not match " "the current pid %s", ti.pid, current_pid)
+ if recorded_pid is not None and not same_process:
+ self.log.warning(
+ "Recorded pid %s does not match the current pid %s", recorded_pid, current_pid
+ )
raise AirflowException("PID of job runner does not match")
elif self.task_runner.return_code() is None and hasattr(self.task_runner, 'process'):
self.log.warning(
[airflow] 37/38: Run mini scheduler in LocalTaskJob during task
exit (#16289)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3386b12fb6600d10a233254fd4c6e219d542532d
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu Jun 10 14:29:30 2021 +0100
Run mini scheduler in LocalTaskJob during task exit (#16289)
Currently, the chances of tasks being killed by the LocalTaskJob heartbeat is high.
This is because, after marking a task successful/failed in Taskinstance.py and mini scheduler is enabled,
we start running the mini scheduler. Whenever the mini scheduling takes time and meet the next job heartbeat,
the heartbeat detects that this task has succeeded with no return code because LocalTaskJob.handle_task_exit
was not called after the task succeeded. Hence, the heartbeat thinks that this task was externally marked failed/successful.
This change resolves this by moving the mini scheduler to LocalTaskJob at the handle_task_exit method ensuring
that the task will no longer be killed by the next heartbeat
(cherry picked from commit 408bd26c22913af93d05aa70abc3c66c52cd4588)
---
airflow/jobs/local_task_job.py | 61 ++++++++++++++-
airflow/models/taskinstance.py | 60 +--------------
tests/cli/commands/test_task_command.py | 4 +-
tests/jobs/test_local_task_job.py | 130 ++++++++++++++++++++++++++++++--
tests/models/test_taskinstance.py | 103 -------------------------
5 files changed, 184 insertions(+), 174 deletions(-)
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 3dca97e..33c2f59 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -16,21 +16,24 @@
# specific language governing permissions and limitations
# under the License.
#
-
import signal
from typing import Optional
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.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
@@ -157,7 +160,10 @@ class LocalTaskJob(BaseJob):
self.task_instance.set_state(State.FAILED)
if self.task_instance.state != State.SUCCESS:
error = self.task_runner.deserialize_run_error()
- self.task_instance._run_finished_callback(error=error)
+ self.task_instance._run_finished_callback(error=error) # pylint: disable=protected-access
+ if not self.task_instance.test_mode:
+ if conf.getboolean('scheduler', 'schedule_after_task_execution', fallback=True):
+ self._run_mini_scheduler_on_child_tasks()
def on_kill(self):
self.task_runner.terminate()
@@ -211,3 +217,54 @@ class LocalTaskJob(BaseJob):
error = self.task_runner.deserialize_run_error() or "task marked as failed externally"
ti._run_finished_callback(error=error)
self.terminating = True
+
+ @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,
+ execution_date=self.task_instance.execution_date,
+ ),
+ session=session,
+ ).one()
+
+ # 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)
+ task = self.task_instance.task
+
+ partial_dag = task.dag.partial_subset(
+ task.downstream_task_ids,
+ include_downstream=False,
+ 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()
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index b99fa34..5fb8155 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -35,7 +35,6 @@ import lazy_object_proxy
import pendulum
from jinja2 import TemplateAssertionError, UndefinedError
from sqlalchemy import Column, Float, Index, Integer, PickleType, String, and_, func, or_
-from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import reconstructor, relationship
from sqlalchemy.orm.session import Session
from sqlalchemy.sql.elements import BooleanClauseList
@@ -70,7 +69,7 @@ from airflow.utils.net import get_hostname
from airflow.utils.operator_helpers import context_to_airflow_vars
from airflow.utils.platform import getuser
from airflow.utils.session import provide_session
-from airflow.utils.sqlalchemy import UtcDateTime, with_row_locks
+from airflow.utils.sqlalchemy import UtcDateTime
from airflow.utils.state import State
from airflow.utils.timeout import timeout
@@ -1223,62 +1222,6 @@ class TaskInstance(Base, LoggingMixin):
session.commit()
- if not test_mode:
- self._run_mini_scheduler_on_child_tasks(session)
-
- @provide_session
- @Sentry.enrich_errors
- def _run_mini_scheduler_on_child_tasks(self, session=None) -> None:
- if conf.getboolean('scheduler', 'schedule_after_task_execution', fallback=True):
- from airflow.models.dagrun import DagRun # Avoid circular import
-
- try:
- # Re-select the row with a lock
- dag_run = with_row_locks(
- session.query(DagRun).filter_by(
- dag_id=self.dag_id,
- execution_date=self.execution_date,
- ),
- session=session,
- ).one()
-
- # 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 = self.task.dag.partial_subset(
- self.task.downstream_task_ids,
- include_downstream=False,
- 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 self.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 = self.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(
- f"Skipping mini scheduling run due to exception: {e.statement}",
- exc_info=True,
- )
- session.rollback()
-
def _prepare_and_execute_task_with_callbacks(self, context, task):
"""Prepare Task for Execution"""
from airflow.models.renderedtifields import RenderedTaskInstanceFields
@@ -1440,6 +1383,7 @@ class TaskInstance(Base, LoggingMixin):
session=session,
)
if not res:
+ self.log.info("CHECK AND CHANGE")
return
try:
diff --git a/tests/cli/commands/test_task_command.py b/tests/cli/commands/test_task_command.py
index f50ddbc..2b93e6d 100644
--- a/tests/cli/commands/test_task_command.py
+++ b/tests/cli/commands/test_task_command.py
@@ -71,8 +71,7 @@ class TestCliTasks(unittest.TestCase):
args = self.parser.parse_args(['tasks', 'list', 'example_bash_operator', '--tree'])
task_command.task_list(args)
- @mock.patch("airflow.models.taskinstance.TaskInstance._run_mini_scheduler_on_child_tasks")
- def test_test(self, mock_run_mini_scheduler):
+ def test_test(self):
"""Test the `airflow test` command"""
args = self.parser.parse_args(
["tasks", "test", "example_python_operator", 'print_the_context', '2018-01-01']
@@ -81,7 +80,6 @@ class TestCliTasks(unittest.TestCase):
with redirect_stdout(io.StringIO()) as stdout:
task_command.task_test(args)
- mock_run_mini_scheduler.assert_not_called()
# Check that prints, and log messages, are shown
assert "'example_python_operator__print_the_context__20180101'" in stdout.getvalue()
diff --git a/tests/jobs/test_local_task_job.py b/tests/jobs/test_local_task_job.py
index 17a7285..48047ad 100644
--- a/tests/jobs/test_local_task_job.py
+++ b/tests/jobs/test_local_task_job.py
@@ -33,6 +33,7 @@ from airflow import settings
from airflow.exceptions import AirflowException, AirflowFailException
from airflow.executors.sequential_executor import SequentialExecutor
from airflow.jobs.local_task_job import LocalTaskJob
+from airflow.jobs.scheduler_job import SchedulerJob
from airflow.models.dag import DAG, DagModel
from airflow.models.dagbag import DagBag
from airflow.models.taskinstance import TaskInstance
@@ -44,8 +45,9 @@ from airflow.utils.net import get_hostname
from airflow.utils.session import create_session
from airflow.utils.state import State
from airflow.utils.timeout import timeout
+from tests.test_utils import db
from tests.test_utils.asserts import assert_queries_count
-from tests.test_utils.db import clear_db_jobs, clear_db_runs
+from tests.test_utils.config import conf_vars
from tests.test_utils.mock_executor import MockExecutor
# pylint: skip-file
@@ -56,15 +58,25 @@ TEST_DAG_FOLDER = os.environ['AIRFLOW__CORE__DAGS_FOLDER']
class TestLocalTaskJob(unittest.TestCase):
def setUp(self):
- clear_db_jobs()
- clear_db_runs()
+ db.clear_db_dags()
+ db.clear_db_jobs()
+ db.clear_db_runs()
+ db.clear_db_task_fail()
patcher = patch('airflow.jobs.base_job.sleep')
self.addCleanup(patcher.stop)
self.mock_base_job_sleep = patcher.start()
def tearDown(self) -> None:
- clear_db_jobs()
- clear_db_runs()
+ db.clear_db_dags()
+ db.clear_db_jobs()
+ db.clear_db_runs()
+ db.clear_db_task_fail()
+
+ def validate_ti_states(self, dag_run, ti_state_mapping, error_message):
+ for task_id, expected_state in ti_state_mapping.items():
+ task_instance = dag_run.get_task_instance(task_id=task_id)
+ task_instance.refresh_from_db()
+ assert task_instance.state == expected_state, error_message
def test_localtaskjob_essential_attr(self):
"""
@@ -659,20 +671,122 @@ class TestLocalTaskJob(unittest.TestCase):
if ti.state == State.RUNNING and ti.pid is not None:
break
time.sleep(0.2)
- assert ti.state == State.RUNNING
assert ti.pid is not None
+ assert ti.state == State.RUNNING
os.kill(ti.pid, signal_type)
process.join(timeout=10)
assert failure_callback_called.value == 1
assert task_terminated_externally.value == 1
assert not process.is_alive()
+ @parameterized.expand(
+ [
+ (
+ {('scheduler', 'schedule_after_task_execution'): 'True'},
+ {'A': 'B', 'B': 'C'},
+ {'A': State.QUEUED, 'B': State.NONE, 'C': State.NONE},
+ {'A': State.SUCCESS, 'B': State.SCHEDULED, 'C': State.NONE},
+ {'A': State.SUCCESS, 'B': State.SUCCESS, 'C': State.SCHEDULED},
+ "A -> B -> C, with fast-follow ON when A runs, B should be QUEUED. Same for B and C.",
+ ),
+ (
+ {('scheduler', 'schedule_after_task_execution'): 'False'},
+ {'A': 'B', 'B': 'C'},
+ {'A': State.QUEUED, 'B': State.NONE, 'C': State.NONE},
+ {'A': State.SUCCESS, 'B': State.NONE, 'C': State.NONE},
+ None,
+ "A -> B -> C, with fast-follow OFF, when A runs, B shouldn't be QUEUED.",
+ ),
+ (
+ {('scheduler', 'schedule_after_task_execution'): 'True'},
+ {'A': 'B', 'C': 'B', 'D': 'C'},
+ {'A': State.QUEUED, 'B': State.NONE, 'C': State.NONE, 'D': State.NONE},
+ {'A': State.SUCCESS, 'B': State.NONE, 'C': State.NONE, 'D': State.NONE},
+ None,
+ "D -> C -> B & A -> B, when A runs but C isn't QUEUED yet, B shouldn't be QUEUED.",
+ ),
+ (
+ {('scheduler', 'schedule_after_task_execution'): 'True'},
+ {'A': 'C', 'B': 'C'},
+ {'A': State.QUEUED, 'B': State.FAILED, 'C': State.NONE},
+ {'A': State.SUCCESS, 'B': State.FAILED, 'C': State.UPSTREAM_FAILED},
+ None,
+ "A -> C & B -> C, when A is QUEUED but B has FAILED, C is marked UPSTREAM_FAILED.",
+ ),
+ ]
+ )
+ def test_fast_follow(
+ self, conf, dependencies, init_state, first_run_state, second_run_state, error_message
+ ):
+ # pylint: disable=too-many-locals
+ with conf_vars(conf):
+ session = settings.Session()
+
+ dag = DAG('test_dagrun_fast_follow', start_date=DEFAULT_DATE)
+
+ dag_model = DagModel(
+ dag_id=dag.dag_id,
+ next_dagrun=dag.start_date,
+ is_active=True,
+ )
+ session.add(dag_model)
+ session.flush()
+
+ python_callable = lambda: True
+ with dag:
+ task_a = PythonOperator(task_id='A', python_callable=python_callable)
+ task_b = PythonOperator(task_id='B', python_callable=python_callable)
+ task_c = PythonOperator(task_id='C', python_callable=python_callable)
+ if 'D' in init_state:
+ task_d = PythonOperator(task_id='D', python_callable=python_callable)
+ for upstream, downstream in dependencies.items():
+ dag.set_dependency(upstream, downstream)
+
+ scheduler_job = SchedulerJob(subdir=os.devnull)
+ scheduler_job.dagbag.bag_dag(dag, root_dag=dag)
+
+ dag_run = dag.create_dagrun(run_id='test_dagrun_fast_follow', state=State.RUNNING)
+
+ task_instance_a = TaskInstance(task_a, dag_run.execution_date, init_state['A'])
+
+ task_instance_b = TaskInstance(task_b, dag_run.execution_date, init_state['B'])
+
+ task_instance_c = TaskInstance(task_c, dag_run.execution_date, init_state['C'])
+
+ if 'D' in init_state:
+ task_instance_d = TaskInstance(task_d, dag_run.execution_date, init_state['D'])
+ session.merge(task_instance_d)
+
+ session.merge(task_instance_a)
+ session.merge(task_instance_b)
+ session.merge(task_instance_c)
+ session.flush()
+
+ job1 = LocalTaskJob(
+ task_instance=task_instance_a, ignore_ti_state=True, executor=SequentialExecutor()
+ )
+ job1.task_runner = StandardTaskRunner(job1)
+
+ job2 = LocalTaskJob(
+ task_instance=task_instance_b, ignore_ti_state=True, executor=SequentialExecutor()
+ )
+ job2.task_runner = StandardTaskRunner(job2)
+
+ settings.engine.dispose()
+ job1.run()
+ self.validate_ti_states(dag_run, first_run_state, error_message)
+ if second_run_state:
+ job2.run()
+ self.validate_ti_states(dag_run, second_run_state, error_message)
+ if scheduler_job.processor_agent:
+ scheduler_job.processor_agent.end()
+
@pytest.fixture()
def clean_db_helper():
yield
- clear_db_jobs()
- clear_db_runs()
+ db.clear_db_jobs()
+ db.clear_db_runs()
@pytest.mark.usefixtures("clean_db_helper")
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index 021809b..c1882e1 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -34,10 +34,8 @@ from sqlalchemy.orm.session import Session
from airflow import models, settings
from airflow.exceptions import AirflowException, AirflowFailException, AirflowSkipException
-from airflow.jobs.scheduler_job import SchedulerJob
from airflow.models import (
DAG,
- DagModel,
DagRun,
Pool,
RenderedTaskInstanceFields,
@@ -1963,107 +1961,6 @@ class TestTaskInstance(unittest.TestCase):
with create_session() as session:
session.query(RenderedTaskInstanceFields).delete()
- def validate_ti_states(self, dag_run, ti_state_mapping, error_message):
- for task_id, expected_state in ti_state_mapping.items():
- task_instance = dag_run.get_task_instance(task_id=task_id)
- assert task_instance.state == expected_state, error_message
-
- @parameterized.expand(
- [
- (
- {('scheduler', 'schedule_after_task_execution'): 'True'},
- {'A': 'B', 'B': 'C'},
- {'A': State.QUEUED, 'B': State.NONE, 'C': State.NONE},
- {'A': State.SUCCESS, 'B': State.SCHEDULED, 'C': State.NONE},
- {'A': State.SUCCESS, 'B': State.SUCCESS, 'C': State.SCHEDULED},
- "A -> B -> C, with fast-follow ON when A runs, B should be QUEUED. Same for B and C.",
- ),
- (
- {('scheduler', 'schedule_after_task_execution'): 'False'},
- {'A': 'B', 'B': 'C'},
- {'A': State.QUEUED, 'B': State.NONE, 'C': State.NONE},
- {'A': State.SUCCESS, 'B': State.NONE, 'C': State.NONE},
- None,
- "A -> B -> C, with fast-follow OFF, when A runs, B shouldn't be QUEUED.",
- ),
- (
- {('scheduler', 'schedule_after_task_execution'): 'True'},
- {'A': 'B', 'C': 'B', 'D': 'C'},
- {'A': State.QUEUED, 'B': State.NONE, 'C': State.NONE, 'D': State.NONE},
- {'A': State.SUCCESS, 'B': State.NONE, 'C': State.NONE, 'D': State.NONE},
- None,
- "D -> C -> B & A -> B, when A runs but C isn't QUEUED yet, B shouldn't be QUEUED.",
- ),
- (
- {('scheduler', 'schedule_after_task_execution'): 'True'},
- {'A': 'C', 'B': 'C'},
- {'A': State.QUEUED, 'B': State.FAILED, 'C': State.NONE},
- {'A': State.SUCCESS, 'B': State.FAILED, 'C': State.UPSTREAM_FAILED},
- None,
- "A -> C & B -> C, when A is QUEUED but B has FAILED, C is marked UPSTREAM_FAILED.",
- ),
- ]
- )
- def test_fast_follow(
- self, conf, dependencies, init_state, first_run_state, second_run_state, error_message
- ):
- with conf_vars(conf):
- session = settings.Session()
-
- dag = DAG('test_dagrun_fast_follow', start_date=DEFAULT_DATE)
-
- dag_model = DagModel(
- dag_id=dag.dag_id,
- next_dagrun=dag.start_date,
- is_active=True,
- )
- session.add(dag_model)
- session.flush()
-
- python_callable = lambda: True
- with dag:
- task_a = PythonOperator(task_id='A', python_callable=python_callable)
- task_b = PythonOperator(task_id='B', python_callable=python_callable)
- task_c = PythonOperator(task_id='C', python_callable=python_callable)
- if 'D' in init_state:
- task_d = PythonOperator(task_id='D', python_callable=python_callable)
- for upstream, downstream in dependencies.items():
- dag.set_dependency(upstream, downstream)
-
- scheduler_job = SchedulerJob(subdir=os.devnull)
- scheduler_job.dagbag.bag_dag(dag, root_dag=dag)
-
- dag_run = dag.create_dagrun(run_id='test_dagrun_fast_follow', state=State.RUNNING)
-
- task_instance_a = dag_run.get_task_instance(task_id=task_a.task_id)
- task_instance_a.task = task_a
- task_instance_a.set_state(init_state['A'])
-
- task_instance_b = dag_run.get_task_instance(task_id=task_b.task_id)
- task_instance_b.task = task_b
- task_instance_b.set_state(init_state['B'])
-
- task_instance_c = dag_run.get_task_instance(task_id=task_c.task_id)
- task_instance_c.task = task_c
- task_instance_c.set_state(init_state['C'])
-
- if 'D' in init_state:
- task_instance_d = dag_run.get_task_instance(task_id=task_d.task_id)
- task_instance_d.task = task_d
- task_instance_d.state = init_state['D']
-
- session.commit()
- task_instance_a.run()
-
- self.validate_ti_states(dag_run, first_run_state, error_message)
-
- if second_run_state:
- scheduler_job._critical_section_execute_task_instances(session=session)
- task_instance_b.run()
- self.validate_ti_states(dag_run, second_run_state, error_message)
- if scheduler_job.processor_agent:
- scheduler_job.processor_agent.end()
-
def test_set_state_up_for_retry(self):
dag = DAG('dag', start_date=DEFAULT_DATE)
op1 = DummyOperator(task_id='op_1', owner='test', dag=dag)
[airflow] 25/38: Fix: ``TaskInstance`` does not show
``queued_by_job_id`` & ``external_executor_id`` (#17179)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 0656f10339875e2920611ffe4ecdf49093c2d2ab
Author: Jorrick Sleijster <jo...@gmail.com>
AuthorDate: Fri Jul 23 00:25:08 2021 +0200
Fix: ``TaskInstance`` does not show ``queued_by_job_id`` & ``external_executor_id`` (#17179)
**Problem discovery:**
I was debugging a bug with the `external_executor_id` Airflow after which this UI bug caught my eye and I got annoyed by it. I figured to fix this one first so my other testing can go a bit smoother :)
**Description of the problem:**
Currently there is a BUG inside the Task Instance details (/task) view.
It loads the TaskInstance by calling `TI(task, execution_date)` and then uses `refresh_from_db()` to refresh many fields that are no filled in yet.
However, the assumption is made in that case that it refreshes all values, which it does not.
`external_executor_id` and `queued_by_job_id` are not updated at all and `executor_config` is only instantiated by the original `TI(task, execution_date)` call but also not updated in `refresh_from_db()`.
This also shows in the UI where these values are always showing None, while the TaskInstance view shows you these values are not None.
**The changes in the PR:**
1. Changes to the `update_from_db()` method to include the missing three values.
2. A new test that checks we are really updating ALL values in `update_from_db()`
3. Removal of an incorrect comment as we do need the `execution_date` for that view.
(cherry picked from commit 759c76d7a5d23cc6f6ef4f724a1a322d2445bbd2)
---
airflow/models/taskinstance.py | 3 +++
airflow/www/views.py | 2 --
tests/models/test_taskinstance.py | 54 +++++++++++++++++++++++++++++++++++++++
3 files changed, 57 insertions(+), 2 deletions(-)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index aeb2a22..38d7e22 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -643,7 +643,10 @@ class TaskInstance(Base, LoggingMixin):
self.priority_weight = ti.priority_weight
self.operator = ti.operator
self.queued_dttm = ti.queued_dttm
+ self.queued_by_job_id = ti.queued_by_job_id
self.pid = ti.pid
+ self.executor_config = ti.executor_config
+ self.external_executor_id = ti.external_executor_id
else:
self.state = None
diff --git a/airflow/www/views.py b/airflow/www/views.py
index f77e66c..e519ebf 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1231,8 +1231,6 @@ class Airflow(AirflowBaseView):
"""Retrieve task."""
dag_id = request.args.get('dag_id')
task_id = request.args.get('task_id')
- # Carrying execution_date through, even though it's irrelevant for
- # this context
execution_date = request.args.get('execution_date')
dttm = timezone.parse(execution_date)
form = DateTimeForm(data={'execution_date': dttm})
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index b6cd8b8..0b063f9 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -2026,6 +2026,60 @@ class TestTaskInstance(unittest.TestCase):
assert ti.start_date < ti.end_date
assert ti.duration > 0
+ def test_refresh_from_db(self):
+ run_date = timezone.utcnow()
+
+ expected_values = {
+ "task_id": "test_refresh_from_db_task",
+ "dag_id": "test_refresh_from_db_dag",
+ "execution_date": run_date,
+ "start_date": run_date + datetime.timedelta(days=1),
+ "end_date": run_date + datetime.timedelta(days=1, seconds=1, milliseconds=234),
+ "duration": 1.234,
+ "state": State.SUCCESS,
+ "_try_number": 1,
+ "max_tries": 1,
+ "hostname": "some_unique_hostname",
+ "unixname": "some_unique_unixname",
+ "job_id": 1234,
+ "pool": "some_fake_pool_id",
+ "pool_slots": 25,
+ "queue": "some_queue_id",
+ "priority_weight": 123,
+ "operator": "some_custom_operator",
+ "queued_dttm": run_date + datetime.timedelta(hours=1),
+ "queued_by_job_id": 321,
+ "pid": 123,
+ "executor_config": {"Some": {"extra": "information"}},
+ "external_executor_id": "some_executor_id",
+ }
+ # Make sure we aren't missing any new value in our expected_values list.
+ expected_keys = {f"task_instance.{key.lstrip('_')}" for key in expected_values.keys()}
+ assert {str(c) for c in TI.__table__.columns} == expected_keys, (
+ "Please add all non-foreign values of TaskInstance to this list. "
+ "This prevents refresh_from_db() from missing a field."
+ )
+
+ operator = DummyOperator(task_id=expected_values['task_id'])
+ ti = TI(task=operator, execution_date=expected_values['execution_date'])
+ for key, expected_value in expected_values.items():
+ setattr(ti, key, expected_value)
+ with create_session() as session:
+ session.merge(ti)
+ session.commit()
+
+ mock_task = mock.MagicMock()
+ mock_task.task_id = expected_values["task_id"]
+ mock_task.dag_id = expected_values["dag_id"]
+
+ ti = TI(task=mock_task, execution_date=run_date)
+ ti.refresh_from_db()
+ for key, expected_value in expected_values.items():
+ assert hasattr(ti, key), f"Key {key} is missing in the TaskInstance."
+ assert (
+ getattr(ti, key) == expected_value
+ ), f"Key: {key} had different values. Make sure it loads it in the refresh refresh_from_db()"
+
@pytest.mark.parametrize("pool_override", [None, "test_pool2"])
def test_refresh_from_task(pool_override):
[airflow] 29/38: Handle and log exceptions raised during task
callback (#17347)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 0678a94220f90940a3d84b5c89b17e3c7e9d2dae
Author: Sam Wheating <sa...@shopify.com>
AuthorDate: Fri Aug 6 03:35:13 2021 -0700
Handle and log exceptions raised during task callback (#17347)
Add missing exception handling in success/retry/failure callbacks
(cherry picked from commit faf9f731fa8810e05f868ffec989ea042381ada4)
---
airflow/models/taskinstance.py | 15 ++++++++++++---
tests/models/test_taskinstance.py | 39 +++++++++++++++++++++++++++++++++++++++
2 files changed, 51 insertions(+), 3 deletions(-)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 41fa661..b99fa34 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1390,18 +1390,27 @@ class TaskInstance(Base, LoggingMixin):
if task.on_failure_callback is not None:
context = self.get_template_context()
context["exception"] = error
- task.on_failure_callback(context)
+ try:
+ task.on_failure_callback(context)
+ except Exception:
+ self.log.exception("Error when executing on_failure_callback")
elif self.state == State.SUCCESS:
task = self.task
if task.on_success_callback is not None:
context = self.get_template_context()
- task.on_success_callback(context)
+ try:
+ task.on_success_callback(context)
+ except Exception:
+ self.log.exception("Error when executing on_success_callback")
elif self.state == State.UP_FOR_RETRY:
task = self.task
if task.on_retry_callback is not None:
context = self.get_template_context()
context["exception"] = error
- task.on_retry_callback(context)
+ try:
+ task.on_retry_callback(context)
+ except Exception:
+ self.log.exception("Error when executing on_retry_callback")
@provide_session
def run(
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index 63f0479..021809b 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -1657,6 +1657,45 @@ class TestTaskInstance(unittest.TestCase):
ti.refresh_from_db()
assert ti.state == State.SUCCESS
+ @parameterized.expand(
+ [
+ (State.SUCCESS, "Error when executing on_success_callback"),
+ (State.UP_FOR_RETRY, "Error when executing on_retry_callback"),
+ (State.FAILED, "Error when executing on_failure_callback"),
+ ]
+ )
+ def test_finished_callbacks_handle_and_log_exception(self, finished_state, expected_message):
+ called = completed = False
+
+ def on_finish_callable(context):
+ nonlocal called, completed
+ called = True
+ raise KeyError
+ completed = True
+
+ dag = DAG(
+ 'test_success_callback_handles_exception',
+ start_date=DEFAULT_DATE,
+ end_date=DEFAULT_DATE + datetime.timedelta(days=10),
+ )
+ task = DummyOperator(
+ task_id='op',
+ email='test@test.test',
+ on_success_callback=on_finish_callable,
+ on_retry_callback=on_finish_callable,
+ on_failure_callback=on_finish_callable,
+ dag=dag,
+ )
+
+ ti = TI(task=task, execution_date=datetime.datetime.now())
+ ti._log = mock.Mock()
+ ti.state = finished_state
+ ti._run_finished_callback()
+
+ assert called
+ assert not completed
+ ti.log.exception.assert_called_once_with(expected_message)
+
def test_handle_failure(self):
start_date = timezone.datetime(2016, 6, 1)
dag = models.DAG(dag_id="test_handle_failure", schedule_interval=None, start_date=start_date)
[airflow] 04/38: Fix minor typo in configuration.py (#16832)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 0d0f4e8e8cfb4d2cda949a4968e841f8cbd101c6
Author: sanjayp <sa...@gmail.com>
AuthorDate: Thu Jul 8 19:45:20 2021 +0530
Fix minor typo in configuration.py (#16832)
(cherry picked from commit 2b10680a2f5188400459089713876c1ce49f1caf)
---
airflow/configuration.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow/configuration.py b/airflow/configuration.py
index 23cfb23..452f127 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -454,7 +454,7 @@ class AirflowConfigParser(ConfigParser):
"""
Reads options, imports the full qualified name, and returns the object.
- In case of failure, it throws an exception a clear message with the key aad the section names
+ In case of failure, it throws an exception with the key and section names
:return: The object or None, if the option is empty
"""
[airflow] 34/38: Fix running tasks with default_impersonation
config (#17229)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 43a31e1c515ef6676970b2f2a097a83572af89d4
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Jul 28 16:00:25 2021 +0100
Fix running tasks with default_impersonation config (#17229)
When default_impersonation is set in the configuration, airflow fails
to run task due to PID mismatch between the recorded PID and the current PID
This change fixes it by checking if task_runner.run_as_user is True and use the
same way we check when ti.run_as_user is true to check the PID
(cherry picked from commit 40419dd371c7be53e6c8017b0c4d1bc7f75d0fb6)
---
airflow/jobs/local_task_job.py | 5 ++++-
tests/jobs/test_local_task_job.py | 42 +++++++++++++++++++++++++++++++++++++++
2 files changed, 46 insertions(+), 1 deletion(-)
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 06a2f57..af7b317 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -187,9 +187,12 @@ class LocalTaskJob(BaseJob):
)
raise AirflowException("Hostname of job runner does not match")
current_pid = self.task_runner.process.pid
+
same_process = ti.pid == current_pid
- if ti.run_as_user:
+
+ if ti.run_as_user or self.task_runner.run_as_user:
same_process = psutil.Process(ti.pid).ppid() == current_pid
+
if ti.pid is not None and not same_process:
self.log.warning("Recorded pid %s does not match " "the current pid %s", ti.pid, current_pid)
raise AirflowException("PID of job runner does not match")
diff --git a/tests/jobs/test_local_task_job.py b/tests/jobs/test_local_task_job.py
index 2a50d35..ed43198 100644
--- a/tests/jobs/test_local_task_job.py
+++ b/tests/jobs/test_local_task_job.py
@@ -189,6 +189,48 @@ class TestLocalTaskJob(unittest.TestCase):
with pytest.raises(AirflowException, match='PID of job runner does not match'):
job1.heartbeat_callback()
+ @conf_vars({('core', 'default_impersonation'): 'testuser'})
+ @mock.patch('airflow.jobs.local_task_job.psutil')
+ def test_localtaskjob_heartbeat_with_default_impersonation(self, psutil_mock, dag_maker):
+ session = settings.Session()
+ with dag_maker('test_localtaskjob_heartbeat'):
+ op1 = DummyOperator(task_id='op1')
+ dr = dag_maker.dag_run
+ ti = dr.get_task_instance(task_id=op1.task_id, session=session)
+ ti.state = State.RUNNING
+ ti.pid = 2
+ ti.hostname = get_hostname()
+ session.commit()
+
+ job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
+ ti.task = op1
+ ti.refresh_from_task(op1)
+ job1.task_runner = StandardTaskRunner(job1)
+ job1.task_runner.process = mock.Mock()
+ job1.task_runner.process.pid = 2
+ # Here, ti.pid is 2, the parent process of ti.pid is a mock(different).
+ # And task_runner process is 2. Should fail
+ with pytest.raises(AirflowException, match='PID of job runner does not match'):
+ job1.heartbeat_callback()
+
+ job1.task_runner.process.pid = 1
+ # We make the parent process of ti.pid to equal the task_runner process id
+ psutil_mock.Process.return_value.ppid.return_value = 1
+ ti.state = State.RUNNING
+ ti.pid = 2
+ # The task_runner process id is 1, same as the parent process of ti.pid
+ # as seen above
+ assert job1.task_runner.run_as_user == 'testuser'
+ session.merge(ti)
+ session.commit()
+ job1.heartbeat_callback(session=None)
+
+ # Here the task_runner process id is changed to 2
+ # while parent process of ti.pid is kept at 1, which is different
+ job1.task_runner.process.pid = 2
+ with pytest.raises(AirflowException, match='PID of job runner does not match'):
+ job1.heartbeat_callback()
+
def test_heartbeat_failed_fast(self):
"""
Test that task heartbeat will sleep when it fails fast
[airflow] 26/38: Do not seek error file when it is closed (#17187)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 20f6edf5eb1847cbadef91603389295ed980dba7
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Sun Jul 25 16:19:18 2021 +0100
Do not seek error file when it is closed (#17187)
We do not check if error file is closed before we seek it, which causes exceptions.
Sometimes, this error file does not exist e.g when the task state is changed externally.
This change fixes it by returning None when the file is closed so that custom text can be used for error.
Co-authored-by: Jed Cunningham <66...@users.noreply.github.com>
(cherry picked from commit 01a0aca249eeaf71d182bf537b9d04121257ac09)
---
airflow/models/taskinstance.py | 2 ++
tests/models/test_taskinstance.py | 13 +++++++++++++
2 files changed, 15 insertions(+)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 38d7e22..41fa661 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -110,6 +110,8 @@ def set_current_context(context: Context):
def load_error_file(fd: IO[bytes]) -> Optional[Union[str, Exception]]:
"""Load and return error from error file"""
+ if fd.closed:
+ return None
fd.seek(0, os.SEEK_SET)
data = fd.read()
if not data:
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index 0b063f9..63f0479 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -21,6 +21,7 @@ import os
import time
import unittest
import urllib
+from tempfile import NamedTemporaryFile
from typing import List, Optional, Union, cast
from unittest import mock
from unittest.mock import call, mock_open, patch
@@ -44,6 +45,7 @@ from airflow.models import (
TaskReschedule,
Variable,
)
+from airflow.models.taskinstance import load_error_file, set_error_file
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
@@ -113,6 +115,17 @@ class TestTaskInstance(unittest.TestCase):
def tearDown(self):
self.clean_db()
+ def test_load_error_file_returns_None_for_closed_file(self):
+ error_fd = NamedTemporaryFile()
+ error_fd.close()
+ assert load_error_file(error_fd) is None
+
+ def test_load_error_file_loads_correctly(self):
+ error_message = "some random error message"
+ with NamedTemporaryFile() as error_fd:
+ set_error_file(error_fd.name, error=error_message)
+ assert load_error_file(error_fd) == error_message
+
def test_set_task_dates(self):
"""
Test that tasks properly take start/end dates from DAGs
[airflow] 31/38: Add missing permissions to varimport (#17468)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 65667b5c9a198c91436893e7584b8c1df0b9bb89
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Aug 6 16:22:50 2021 +0100
Add missing permissions to varimport (#17468)
(cherry picked from commit eb6af07f5bc8958efd06818e84a5273a079304e1)
---
airflow/www/views.py | 2 +-
tests/www/views/test_views_variable.py | 13 +++++++++++++
2 files changed, 14 insertions(+), 1 deletion(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index e519ebf..09d27e0 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -3276,7 +3276,6 @@ class VariableModelView(AirflowModelView):
'delete': 'delete',
'action_muldelete': 'delete',
'action_varexport': 'read',
- 'varimport': 'create',
}
base_permissions = [
permissions.ACTION_CAN_CREATE,
@@ -3339,6 +3338,7 @@ class VariableModelView(AirflowModelView):
return response
@expose('/varimport', methods=["POST"])
+ @auth.has_access([(permissions.ACTION_CAN_CREATE, permissions.RESOURCE_VARIABLE)])
@action_logging
def varimport(self):
"""Import variables"""
diff --git a/tests/www/views/test_views_variable.py b/tests/www/views/test_views_variable.py
index f9b3744..292e971 100644
--- a/tests/www/views/test_views_variable.py
+++ b/tests/www/views/test_views_variable.py
@@ -96,6 +96,19 @@ def test_import_variables_success(session, admin_client):
check_content_in_response('4 variable(s) successfully updated.', resp)
+def test_import_variables_anon(session, app):
+ assert session.query(Variable).count() == 0
+
+ content = '{"str_key": "str_value}'
+ bytes_content = io.BytesIO(bytes(content, encoding='utf-8'))
+
+ resp = app.test_client().post(
+ '/variable/varimport', data={'file': (bytes_content, 'test.json')}, follow_redirects=True
+ )
+ check_content_not_in_response('variable(s) successfully updated.', resp)
+ check_content_in_response('Sign In', resp)
+
+
def test_description_retrieval(session, admin_client):
# create valid variable
admin_client.post('/variable/add', data=VARIABLE, follow_redirects=True)
[airflow] 16/38: Fix unchecked indexing in _build_metrics (#16744)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1aa156a3635092b072809d50b8883430858ab16b
Author: Emil Ejbyfeldt <ee...@liveintent.com>
AuthorDate: Thu Jul 1 09:24:16 2021 +0200
Fix unchecked indexing in _build_metrics (#16744)
I am not sure if this can happend in regular uses, but when running test
cases `sys.argv` can be that args passed to the pytest. When this is the
case it is definently possible for argv to only contain a single
element.
(cherry picked from commit a2d6aa07c9de83278eef14a228eb4a02ad43205c)
---
airflow/utils/cli.py | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/airflow/utils/cli.py b/airflow/utils/cli.py
index 2b5eb59..acdb291 100644
--- a/airflow/utils/cli.py
+++ b/airflow/utils/cli.py
@@ -115,7 +115,8 @@ def _build_metrics(func_name, namespace):
sub_commands_to_check = {'users', 'connections'}
sensitive_fields = {'-p', '--password', '--conn-password'}
full_command = list(sys.argv)
- if full_command[1] in sub_commands_to_check:
+ sub_command = full_command[1] if len(full_command) > 1 else None
+ if sub_command in sub_commands_to_check:
for idx, command in enumerate(full_command):
if command in sensitive_fields:
# For cases when password is passed as "--password xyz" (with space between key and value)
[airflow] 15/38: BugFix: Correctly handle custom `deps` and
`task_group` during DAG Serialization (#16734)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 943d3c514ead4f983fdbfef56120e16c5ce42e03
Author: luoyuliuyin <lu...@gmail.com>
AuthorDate: Tue Jul 6 20:44:32 2021 +0800
BugFix: Correctly handle custom `deps` and `task_group` during DAG Serialization (#16734)
We check if the dag changed or not via dag_hash, so we need to correctly handle deps and task_group during DAG serialization to ensure that the generation of dag_hash is stable.
closes https://github.com/apache/airflow/issues/16690
(cherry picked from commit 0632ecf6f56214c78deea2a4b54ea0daebb4e95d)
---
airflow/serialization/serialized_objects.py | 16 +++--
tests/serialization/test_dag_serialization.py | 99 +++++++++++++++++++++++++++
2 files changed, 110 insertions(+), 5 deletions(-)
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index d2f456d..bdbaea8 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -424,7 +424,10 @@ class SerializedBaseOperator(BaseOperator, BaseSerialization):
)
deps.append(f'{module_name}.{klass.__name__}')
- serialize_op['deps'] = deps
+ # deps needs to be sorted here, because op.deps is a set, which is unstable when traversing,
+ # and the same call may get different results.
+ # When calling json.dumps(self.data, sort_keys=True) to generate dag_hash, misjudgment will occur
+ serialize_op['deps'] = sorted(deps)
# Store all template_fields as they are if there are JSON Serializable
# If not, store them as strings
@@ -796,6 +799,9 @@ class SerializedTaskGroup(TaskGroup, BaseSerialization):
if not task_group:
return None
+ # task_group.xxx_ids needs to be sorted here, because task_group.xxx_ids is a set,
+ # when converting set to list, the order is uncertain.
+ # When calling json.dumps(self.data, sort_keys=True) to generate dag_hash, misjudgment will occur
serialize_group = {
"_group_id": task_group._group_id,
"prefix_group_id": task_group.prefix_group_id,
@@ -808,10 +814,10 @@ class SerializedTaskGroup(TaskGroup, BaseSerialization):
else (DAT.TASK_GROUP, SerializedTaskGroup.serialize_task_group(child))
for label, child in task_group.children.items()
},
- "upstream_group_ids": cls._serialize(list(task_group.upstream_group_ids)),
- "downstream_group_ids": cls._serialize(list(task_group.downstream_group_ids)),
- "upstream_task_ids": cls._serialize(list(task_group.upstream_task_ids)),
- "downstream_task_ids": cls._serialize(list(task_group.downstream_task_ids)),
+ "upstream_group_ids": cls._serialize(sorted(task_group.upstream_group_ids)),
+ "downstream_group_ids": cls._serialize(sorted(task_group.downstream_group_ids)),
+ "upstream_task_ids": cls._serialize(sorted(task_group.upstream_task_ids)),
+ "downstream_task_ids": cls._serialize(sorted(task_group.downstream_task_ids)),
}
return serialize_group
diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index 29e6ac0..7b0b476 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -965,6 +965,105 @@ class TestStringifiedDAGs(unittest.TestCase):
check_task_group(serialized_dag.task_group)
+ def test_deps_sorted(self):
+ """
+ Tests serialize_operator, make sure the deps is in order
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.sensors.external_task import ExternalTaskSensor
+
+ execution_date = datetime(2020, 1, 1)
+ with DAG(dag_id="test_deps_sorted", start_date=execution_date) as dag:
+ task1 = ExternalTaskSensor(
+ task_id="task1",
+ external_dag_id="external_dag_id",
+ mode="reschedule",
+ )
+ task2 = DummyOperator(task_id="task2")
+ task1 >> task2
+
+ serialize_op = SerializedBaseOperator.serialize_operator(dag.task_dict["task1"])
+ deps = serialize_op["deps"]
+ assert deps == [
+ 'airflow.ti_deps.deps.not_in_retry_period_dep.NotInRetryPeriodDep',
+ 'airflow.ti_deps.deps.not_previously_skipped_dep.NotPreviouslySkippedDep',
+ 'airflow.ti_deps.deps.prev_dagrun_dep.PrevDagrunDep',
+ 'airflow.ti_deps.deps.ready_to_reschedule.ReadyToRescheduleDep',
+ 'airflow.ti_deps.deps.trigger_rule_dep.TriggerRuleDep',
+ ]
+
+ def test_task_group_sorted(self):
+ """
+ Tests serialize_task_group, make sure the list is in order
+ """
+ from airflow.operators.dummy import DummyOperator
+ from airflow.serialization.serialized_objects import SerializedTaskGroup
+ from airflow.utils.task_group import TaskGroup
+
+ """
+ start
+ ╱ ╲
+ ╱ ╲
+ task_group_up1 task_group_up2
+ (task_up1) (task_up2)
+ ╲ ╱
+ task_group_middle
+ (task_middle)
+ ╱ ╲
+ task_group_down1 task_group_down2
+ (task_down1) (task_down2)
+ ╲ ╱
+ ╲ ╱
+ end
+ """
+ execution_date = datetime(2020, 1, 1)
+ with DAG(dag_id="test_task_group_sorted", start_date=execution_date) as dag:
+ start = DummyOperator(task_id="start")
+
+ with TaskGroup("task_group_up1") as task_group_up1:
+ _ = DummyOperator(task_id="task_up1")
+
+ with TaskGroup("task_group_up2") as task_group_up2:
+ _ = DummyOperator(task_id="task_up2")
+
+ with TaskGroup("task_group_middle") as task_group_middle:
+ _ = DummyOperator(task_id="task_middle")
+
+ with TaskGroup("task_group_down1") as task_group_down1:
+ _ = DummyOperator(task_id="task_down1")
+
+ with TaskGroup("task_group_down2") as task_group_down2:
+ _ = DummyOperator(task_id="task_down2")
+
+ end = DummyOperator(task_id='end')
+
+ start >> task_group_up1
+ start >> task_group_up2
+ task_group_up1 >> task_group_middle
+ task_group_up2 >> task_group_middle
+ task_group_middle >> task_group_down1
+ task_group_middle >> task_group_down2
+ task_group_down1 >> end
+ task_group_down2 >> end
+
+ task_group_middle_dict = SerializedTaskGroup.serialize_task_group(
+ dag.task_group.children["task_group_middle"]
+ )
+ upstream_group_ids = task_group_middle_dict["upstream_group_ids"]
+ assert upstream_group_ids == ['task_group_up1', 'task_group_up2']
+
+ upstream_task_ids = task_group_middle_dict["upstream_task_ids"]
+ assert upstream_task_ids == ['task_group_up1.task_up1', 'task_group_up2.task_up2']
+
+ downstream_group_ids = task_group_middle_dict["downstream_group_ids"]
+ assert downstream_group_ids == ['task_group_down1', 'task_group_down2']
+
+ task_group_down1_dict = SerializedTaskGroup.serialize_task_group(
+ dag.task_group.children["task_group_down1"]
+ )
+ downstream_task_ids = task_group_down1_dict["downstream_task_ids"]
+ assert downstream_task_ids == ['end']
+
def test_edge_info_serialization(self):
"""
Tests edge_info serialization/deserialization.
[airflow] 06/38: Add State types for tasks and DAGs (#15285)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit dd946470719fc363159c92438ab0e5c93ebae39e
Author: Andrew Godwin <an...@astronomer.io>
AuthorDate: Tue Jul 6 17:36:01 2021 -0600
Add State types for tasks and DAGs (#15285)
This adds TaskState and DagState enum types that contain all possible states, makes all other core state constants derive their values from them, and adds a couple of initial type hints that use the new enums (with the plan being that we can add signficantly more later).
closes: #9387
(cherry picked from commit 2b7c59619b7dd6fd5031745ade7756466456f803)
---
airflow/models/dagrun.py | 6 +-
airflow/typing_compat.py | 2 +-
airflow/utils/state.py | 169 ++++++++++++++++++++++++++++-------------------
3 files changed, 107 insertions(+), 70 deletions(-)
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index 3bd86ee..f29d6ac 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -48,7 +48,7 @@ from airflow.utils import callback_requests, timezone
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.session import provide_session
from airflow.utils.sqlalchemy import UtcDateTime, nulls_first, skip_locked, with_row_locks
-from airflow.utils.state import State
+from airflow.utils.state import State, TaskInstanceState
from airflow.utils.types import DagRunType
if TYPE_CHECKING:
@@ -302,7 +302,9 @@ class DagRun(Base, LoggingMixin):
return f"{run_type}__{execution_date.isoformat()}"
@provide_session
- def get_task_instances(self, state=None, session=None) -> Iterable[TI]:
+ def get_task_instances(
+ self, state: Optional[Iterable[TaskInstanceState]] = None, session=None
+ ) -> Iterable[TI]:
"""Returns the task instances for this dag run"""
tis = session.query(TI).filter(
TI.dag_id == self.dag_id,
diff --git a/airflow/typing_compat.py b/airflow/typing_compat.py
index 0f3db75..26237c2 100644
--- a/airflow/typing_compat.py
+++ b/airflow/typing_compat.py
@@ -22,7 +22,7 @@ codebase easier.
"""
try:
- # Protocol and TypedDict are only added to typing module starting from
+ # Literal, Protocol and TypedDict are only added to typing module starting from
# python 3.8 we can safely remove this shim import after Airflow drops
# support for <3.8
from typing import Literal, Protocol, TypedDict, runtime_checkable # type: ignore
diff --git a/airflow/utils/state.py b/airflow/utils/state.py
index d5300e1..b1b27d0 100644
--- a/airflow/utils/state.py
+++ b/airflow/utils/state.py
@@ -16,70 +16,101 @@
# specific language governing permissions and limitations
# under the License.
+from enum import Enum
+from typing import Dict, FrozenSet, Tuple
+
from airflow.settings import STATE_COLORS
+from airflow.utils.types import Optional
-class State:
+class TaskInstanceState(str, Enum):
"""
- Static class with task instance states constants and color method to
- avoid hardcoding.
+ Enum that represents all possible states that a Task Instance can be in.
+
+ Note that None is also allowed, so always use this in a type hint with Optional.
"""
- # scheduler
- NONE = None # type: None
- REMOVED = "removed"
- SCHEDULED = "scheduled"
+ # Set by the scheduler
+ # None - Task is created but should not run yet
+ REMOVED = "removed" # Task vanished from DAG before it ran
+ SCHEDULED = "scheduled" # Task should run and will be handed to executor soon
+
+ # Set by the task instance itself
+ QUEUED = "queued" # Executor has enqueued the task
+ RUNNING = "running" # Task is executing
+ SUCCESS = "success" # Task completed
+ SHUTDOWN = "shutdown" # External request to shut down
+ FAILED = "failed" # Task errored out
+ UP_FOR_RETRY = "up_for_retry" # Task failed but has retries left
+ UP_FOR_RESCHEDULE = "up_for_reschedule" # A waiting `reschedule` sensor
+ UPSTREAM_FAILED = "upstream_failed" # One or more upstream deps failed
+ SKIPPED = "skipped" # Skipped by branching or some other mechanism
+ SENSING = "sensing" # Smart sensor offloaded to the sensor DAG
- # set by the executor (t.b.d.)
- # LAUNCHED = "launched"
+ def __str__(self) -> str: # pylint: disable=invalid-str-returned
+ return self.value
+
+
+class DagRunState(str, Enum):
+ """
+ Enum that represents all possible states that a DagRun can be in.
+
+ These are "shared" with TaskInstanceState in some parts of the code,
+ so please ensure that their values always match the ones with the
+ same name in TaskInstanceState.
+ """
- # set by a task
- QUEUED = "queued"
RUNNING = "running"
SUCCESS = "success"
- SHUTDOWN = "shutdown" # External request to shut down
FAILED = "failed"
- UP_FOR_RETRY = "up_for_retry"
- UP_FOR_RESCHEDULE = "up_for_reschedule"
- UPSTREAM_FAILED = "upstream_failed"
- SKIPPED = "skipped"
- SENSING = "sensing"
-
- task_states = (
- SUCCESS,
- RUNNING,
- FAILED,
- UPSTREAM_FAILED,
- SKIPPED,
- UP_FOR_RETRY,
- UP_FOR_RESCHEDULE,
- QUEUED,
- NONE,
- SCHEDULED,
- SENSING,
- REMOVED,
- )
- dag_states = (
- SUCCESS,
- RUNNING,
- FAILED,
+
+class State:
+ """
+ Static class with task instance state constants and color methods to
+ avoid hardcoding.
+ """
+
+ # Backwards-compat constants for code that does not yet use the enum
+ # These first three are shared by DagState and TaskState
+ SUCCESS = TaskInstanceState.SUCCESS
+ RUNNING = TaskInstanceState.RUNNING
+ FAILED = TaskInstanceState.FAILED
+
+ # These are TaskState only
+ NONE = None
+ REMOVED = TaskInstanceState.REMOVED
+ SCHEDULED = TaskInstanceState.SCHEDULED
+ QUEUED = TaskInstanceState.QUEUED
+ SHUTDOWN = TaskInstanceState.SHUTDOWN
+ UP_FOR_RETRY = TaskInstanceState.UP_FOR_RETRY
+ UP_FOR_RESCHEDULE = TaskInstanceState.UP_FOR_RESCHEDULE
+ UPSTREAM_FAILED = TaskInstanceState.UPSTREAM_FAILED
+ SKIPPED = TaskInstanceState.SKIPPED
+ SENSING = TaskInstanceState.SENSING
+
+ task_states: Tuple[Optional[TaskInstanceState], ...] = (None,) + tuple(TaskInstanceState)
+
+ dag_states: Tuple[DagRunState, ...] = (
+ DagRunState.SUCCESS,
+ DagRunState.RUNNING,
+ DagRunState.FAILED,
)
- state_color = {
- QUEUED: 'gray',
- RUNNING: 'lime',
- SUCCESS: 'green',
- SHUTDOWN: 'blue',
- FAILED: 'red',
- UP_FOR_RETRY: 'gold',
- UP_FOR_RESCHEDULE: 'turquoise',
- UPSTREAM_FAILED: 'orange',
- SKIPPED: 'pink',
- REMOVED: 'lightgrey',
- SCHEDULED: 'tan',
- NONE: 'lightblue',
- SENSING: 'lightseagreen',
+ state_color: Dict[Optional[TaskInstanceState], str] = {
+ None: 'lightblue',
+ TaskInstanceState.QUEUED: 'gray',
+ TaskInstanceState.RUNNING: 'lime',
+ TaskInstanceState.SUCCESS: 'green',
+ TaskInstanceState.SHUTDOWN: 'blue',
+ TaskInstanceState.FAILED: 'red',
+ TaskInstanceState.UP_FOR_RETRY: 'gold',
+ TaskInstanceState.UP_FOR_RESCHEDULE: 'turquoise',
+ TaskInstanceState.UPSTREAM_FAILED: 'orange',
+ TaskInstanceState.SKIPPED: 'pink',
+ TaskInstanceState.REMOVED: 'lightgrey',
+ TaskInstanceState.SCHEDULED: 'tan',
+ TaskInstanceState.SENSING: 'lightseagreen',
}
state_color.update(STATE_COLORS) # type: ignore
@@ -96,17 +127,17 @@ class State:
return 'white'
return 'black'
- running = frozenset([RUNNING, SENSING])
+ running: FrozenSet[TaskInstanceState] = frozenset([TaskInstanceState.RUNNING, TaskInstanceState.SENSING])
"""
A list of states indicating that a task is being executed.
"""
- finished = frozenset(
+ finished: FrozenSet[TaskInstanceState] = frozenset(
[
- SUCCESS,
- FAILED,
- SKIPPED,
- UPSTREAM_FAILED,
+ TaskInstanceState.SUCCESS,
+ TaskInstanceState.FAILED,
+ TaskInstanceState.SKIPPED,
+ TaskInstanceState.UPSTREAM_FAILED,
]
)
"""
@@ -118,16 +149,16 @@ class State:
case, it is no longer running.
"""
- unfinished = frozenset(
+ unfinished: FrozenSet[Optional[TaskInstanceState]] = frozenset(
[
- NONE,
- SCHEDULED,
- QUEUED,
- RUNNING,
- SENSING,
- SHUTDOWN,
- UP_FOR_RETRY,
- UP_FOR_RESCHEDULE,
+ None,
+ TaskInstanceState.SCHEDULED,
+ TaskInstanceState.QUEUED,
+ TaskInstanceState.RUNNING,
+ TaskInstanceState.SENSING,
+ TaskInstanceState.SHUTDOWN,
+ TaskInstanceState.UP_FOR_RETRY,
+ TaskInstanceState.UP_FOR_RESCHEDULE,
]
)
"""
@@ -135,12 +166,16 @@ class State:
a run or has not even started.
"""
- failed_states = frozenset([FAILED, UPSTREAM_FAILED])
+ failed_states: FrozenSet[TaskInstanceState] = frozenset(
+ [TaskInstanceState.FAILED, TaskInstanceState.UPSTREAM_FAILED]
+ )
"""
A list of states indicating that a task or dag is a failed state.
"""
- success_states = frozenset([SUCCESS, SKIPPED])
+ success_states: FrozenSet[TaskInstanceState] = frozenset(
+ [TaskInstanceState.SUCCESS, TaskInstanceState.SKIPPED]
+ )
"""
A list of states indicating that a task or dag is a success state.
"""
[airflow] 18/38: Don't check execution_date in refresh_from_db
(#16809)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ae4ab586b8bb12f0827505d05c88ac924734410a
Author: Tzu-ping Chung <tp...@astronomer.io>
AuthorDate: Wed Jul 7 19:04:15 2021 +0800
Don't check execution_date in refresh_from_db (#16809)
The native sqlalchemy DateTime type does not compare well when timezones
don't match. This can happen if the current execution_date on a DagRun
instance is not in UTC (the db entry is always in UTC).
Since DagRun has a unique constraint on (dag_id, run_id), these two
should be able to return one unique result, and the executrion_date
column should not be needed anyway. Let's just remove that filter to
prevent all the datetime comparison trouble.
(cherry picked from commit faffaec73385db3c6910d31ccea9fc4f9f3f9d42)
---
airflow/models/dagrun.py | 29 ++---------------------------
1 file changed, 2 insertions(+), 27 deletions(-)
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index f29d6ac..07f309d 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -18,19 +18,7 @@
from datetime import datetime
from typing import TYPE_CHECKING, Any, Iterable, List, NamedTuple, Optional, Tuple, Union
-from sqlalchemy import (
- Boolean,
- Column,
- DateTime,
- Index,
- Integer,
- PickleType,
- String,
- UniqueConstraint,
- and_,
- func,
- or_,
-)
+from sqlalchemy import Boolean, Column, Index, Integer, PickleType, String, UniqueConstraint, and_, func, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import backref, relationship, synonym
@@ -165,20 +153,7 @@ class DagRun(Base, LoggingMixin):
:param session: database session
:type session: Session
"""
- DR = DagRun
-
- exec_date = func.cast(self.execution_date, DateTime)
-
- dr = (
- session.query(DR)
- .filter(
- DR.dag_id == self.dag_id,
- func.cast(DR.execution_date, DateTime) == exec_date,
- DR.run_id == self.run_id,
- )
- .one()
- )
-
+ dr = session.query(DagRun).filter(DagRun.dag_id == self.dag_id, DagRun.run_id == self.run_id).one()
self.id = dr.id
self.state = dr.state
[airflow] 10/38: Redact conn secrets in webserver logs (#16579)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6eec6d0e37cc27d1743922931dcde2a3b16900bf
Author: Sumit Maheshwari <ms...@users.noreply.github.com>
AuthorDate: Tue Jun 22 14:40:45 2021 +0530
Redact conn secrets in webserver logs (#16579)
(cherry picked from commit 2a59de3e558e3b60caad876dee8fa4b43a7a17cf)
---
airflow/hooks/base.py | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/airflow/hooks/base.py b/airflow/hooks/base.py
index f286a6f..3531bf3 100644
--- a/airflow/hooks/base.py
+++ b/airflow/hooks/base.py
@@ -22,6 +22,7 @@ from typing import TYPE_CHECKING, Any, Dict, List
from airflow.typing_compat import Protocol
from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.log.secrets_masker import redact
if TYPE_CHECKING:
from airflow.models.connection import Connection # Avoid circular imports.
@@ -74,8 +75,8 @@ class BaseHook(LoggingMixin):
conn.port,
conn.schema,
conn.login,
- conn.password,
- conn.extra_dejson,
+ redact(conn.password),
+ redact(conn.extra_dejson),
)
return conn
[airflow] 20/38: Fix UPDATING.md (#16933)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e7a87e5a64ea27cbe0fc6849de5ffc815e37fb89
Author: Kaxil Naik <ka...@gmail.com>
AuthorDate: Mon Jul 12 01:21:01 2021 +0100
Fix UPDATING.md (#16933)
(cherry picked from commit 543c31f76066592d356ef1a283888f3507fd968d)
---
UPDATING.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/UPDATING.md b/UPDATING.md
index 67ec6b8..2361b6c 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -1607,9 +1607,9 @@ https://cloud.google.com/compute/docs/disks/performance
Hence, the default value for `master_disk_size` in `DataprocCreateClusterOperator` has been changed from 500GB to 1TB.
-#### `<airflow class="providers google c"></airflow>loud.operators.bigquery.BigQueryGetDatasetTablesOperator`
+#### `airflow.providers.google.cloud.operators.bigquery.BigQueryGetDatasetTablesOperator`
-We changed signature of BigQueryGetDatasetTablesOperator.
+We changed signature of `BigQueryGetDatasetTablesOperator`.
Before:
[airflow] 19/38: Fixed task instance retrieval in XCom view (#16923)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 84bb0436eae65d2c77415cb129b6072c324a47b4
Author: Pietro <54...@users.noreply.github.com>
AuthorDate: Sun Jul 11 16:54:34 2021 +0200
Fixed task instance retrieval in XCom view (#16923)
The SQL logical "and" was not implemented correctly and hence the filter was returning wrong results
(cherry picked from commit 9a1fc19a627662122dd8dacee2ef61c3edf97b85)
---
airflow/www/views.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow/www/views.py b/airflow/www/views.py
index fa89cb1..6b2b2a4 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1328,7 +1328,7 @@ class Airflow(AirflowBaseView):
dm_db = models.DagModel
ti_db = models.TaskInstance
dag = session.query(dm_db).filter(dm_db.dag_id == dag_id).first()
- ti = session.query(ti_db).filter(ti_db.dag_id == dag_id and ti_db.task_id == task_id).first()
+ ti = session.query(ti_db).filter(and_(ti_db.dag_id == dag_id, ti_db.task_id == task_id)).first()
if not ti:
flash(f"Task [{dag_id}.{task_id}] doesn't seem to exist at the moment", "error")
[airflow] 08/38: Fail tasks in scheduler when executor reports they
failed (#15929)
Posted by jh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit c6d09edf7b77ac4becc7654465c535571f88babf
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu May 20 11:22:01 2021 +0100
Fail tasks in scheduler when executor reports they failed (#15929)
When a task fails in executor while still queued in scheduler, the executor reports
this failure but scheduler doesn't change the task state resulting in the task
being queued until the scheduler is restarted. This commit fixes it by ensuring
that when a task is reported to have failed in the executor, the task is failed
in scheduler
(cherry picked from commit deececcabc080844ca89272a2e4ab1183cd51e3f)
---
airflow/jobs/scheduler_job.py | 4 +++-
tests/jobs/test_scheduler_job.py | 2 +-
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index b99f4b2..1758ae1 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -1252,12 +1252,14 @@ class SchedulerJob(BaseJob):
"task says its %s. (Info: %s) Was the task killed externally?"
)
self.log.error(msg, ti, state, ti.state, info)
+
request = TaskCallbackRequest(
full_filepath=ti.dag_model.fileloc,
simple_task_instance=SimpleTaskInstance(ti),
msg=msg % (ti, state, ti.state, info),
)
-
+ self.log.info('Setting task instance %s state to %s as reported by executor', ti, state)
+ ti.set_state(state)
self.processor_agent.send_callback_to_execute(request)
return len(event_buffer)
diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
index 0d1f530..37ae65b 100644
--- a/tests/jobs/test_scheduler_job.py
+++ b/tests/jobs/test_scheduler_job.py
@@ -907,7 +907,7 @@ class TestSchedulerJob(unittest.TestCase):
self.scheduler_job._process_executor_events(session=session)
ti1.refresh_from_db()
- assert ti1.state == State.QUEUED
+ assert ti1.state == State.FAILED
mock_task_callback.assert_called_once_with(
full_filepath='/test_path1/',
simple_task_instance=mock.ANY,