From cac8b770f9a446883ed5b0abdd78ca1fc5fb8d09 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Thu, 4 Dec 2025 14:14:25 +0800 Subject: [PATCH] Drop LogTemplate DB Model --- airflow-core/src/airflow/models/__init__.py | 1 - airflow-core/src/airflow/models/dagrun.py | 28 ------- airflow-core/src/airflow/models/tasklog.py | 47 ----------- .../serialization/serialized_objects.py | 4 - airflow-core/src/airflow/utils/cli.py | 3 +- airflow-core/src/airflow/utils/db.py | 81 +------------------ airflow-core/src/airflow/utils/helpers.py | 23 ------ .../airflow/utils/log/file_task_handler.py | 2 +- .../src/airflow/utils/log/log_reader.py | 26 ------ .../api_fastapi/common/test_exceptions.py | 2 +- .../core_api/routes/public/test_log.py | 2 +- .../cli/commands/test_scheduler_command.py | 8 +- .../tests/unit/utils/log/test_log_reader.py | 61 -------------- .../tests/unit/utils/test_db_cleanup.py | 1 - airflow-core/tests/unit/utils/test_helpers.py | 19 ----- .../tests/unit/utils/test_log_handlers.py | 17 ++-- .../src/tests_common/pytest_plugin.py | 10 ++- .../aws/log/test_cloudwatch_task_handler.py | 6 +- .../amazon/aws/log/test_s3_task_handler.py | 6 +- .../apache/hdfs/log/test_hdfs_task_handler.py | 3 +- .../elasticsearch/log/es_task_handler.py | 4 +- .../elasticsearch/log/test_es_task_handler.py | 20 ++--- .../google/cloud/log/test_gcs_task_handler.py | 3 +- .../azure/log/test_wasb_task_handler.py | 3 +- .../opensearch/log/os_task_handler.py | 4 +- .../opensearch/log/test_os_task_handler.py | 13 ++- 26 files changed, 53 insertions(+), 344 deletions(-) delete mode 100644 airflow-core/src/airflow/models/tasklog.py diff --git a/airflow-core/src/airflow/models/__init__.py b/airflow-core/src/airflow/models/__init__.py index c395197a158e9..ad8eb57c4f84d 100644 --- a/airflow-core/src/airflow/models/__init__.py +++ b/airflow-core/src/airflow/models/__init__.py @@ -69,7 +69,6 @@ def import_all_models(): import airflow.models.errors import airflow.models.serialized_dag import airflow.models.taskinstancehistory - import airflow.models.tasklog import airflow.models.team import airflow.models.xcom diff --git a/airflow-core/src/airflow/models/dagrun.py b/airflow-core/src/airflow/models/dagrun.py index 07758be44842d..b6298245f7c05 100644 --- a/airflow-core/src/airflow/models/dagrun.py +++ b/airflow-core/src/airflow/models/dagrun.py @@ -66,7 +66,6 @@ from airflow.models.base import Base, StringID from airflow.models.taskinstance import TaskInstance as TI from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH -from airflow.models.tasklog import LogTemplate from airflow.models.taskmap import TaskMap from airflow.sdk.definitions.deadline import DeadlineReference from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set @@ -182,14 +181,6 @@ class DagRun(Base, LoggingMixin): run_after: Mapped[datetime] = mapped_column(UtcDateTime, default=_default_run_after, nullable=False) # When a scheduler last attempted to schedule TIs for this DagRun last_scheduling_decision: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True) - # Foreign key to LogTemplate. DagRun rows created prior to this column's - # existence have this set to NULL. Later rows automatically populate this on - # insert to point to the latest LogTemplate entry. - log_template_id: Mapped[int] = mapped_column( - Integer, - ForeignKey("log_template.id", name="task_instance_log_template_id_fkey", ondelete="NO ACTION"), - default=select(func.max(LogTemplate.__table__.c.id)), - ) updated_at: Mapped[datetime] = mapped_column( UtcDateTime, default=timezone.utcnow, onupdate=timezone.utcnow ) @@ -2117,25 +2108,6 @@ def schedule_tis( return count - @provide_session - def get_log_template(self, *, session: Session = NEW_SESSION) -> LogTemplate: - return DagRun._get_log_template(log_template_id=self.log_template_id, session=session) - - @staticmethod - @provide_session - def _get_log_template(log_template_id: int | None, session: Session = NEW_SESSION) -> LogTemplate: - template: LogTemplate | None - if log_template_id is None: # DagRun created before LogTemplate introduction. - template = session.scalar(select(LogTemplate).order_by(LogTemplate.id).limit(1)) - else: - template = session.get(LogTemplate, log_template_id) - if template is None: - raise AirflowException( - f"No log_template entry found for ID {log_template_id!r}. " - f"Please make sure you set up the metadatabase correctly." - ) - return template - @staticmethod def _get_partial_task_ids(dag: SerializedDAG | None) -> list[str] | None: return dag.task_ids if dag and dag.partial else None diff --git a/airflow-core/src/airflow/models/tasklog.py b/airflow-core/src/airflow/models/tasklog.py deleted file mode 100644 index 7e2c81f4b0c52..0000000000000 --- a/airflow-core/src/airflow/models/tasklog.py +++ /dev/null @@ -1,47 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -from __future__ import annotations - -from datetime import datetime - -from sqlalchemy import Integer, Text -from sqlalchemy.orm import Mapped - -from airflow._shared.timezones import timezone -from airflow.models.base import Base -from airflow.utils.sqlalchemy import UtcDateTime, mapped_column - - -class LogTemplate(Base): - """ - Changes to ``log_filename_template`` and ``elasticsearch_id``. - - This table is automatically populated when Airflow starts up, to store the - config's value if it does not match the last row in the table. - """ - - __tablename__ = "log_template" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - filename: Mapped[str] = mapped_column(Text, nullable=False) - elasticsearch_id: Mapped[str] = mapped_column(Text, nullable=False) - created_at: Mapped[datetime] = mapped_column(UtcDateTime, nullable=False, default=timezone.utcnow) - - def __repr__(self) -> str: - attrs = ", ".join(f"{k}={getattr(self, k)}" for k in ("filename", "elasticsearch_id")) - return f"LogTemplate({attrs})" diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py index d9c1938207744..2c60d9aefbd23 100644 --- a/airflow-core/src/airflow/serialization/serialized_objects.py +++ b/airflow-core/src/airflow/serialization/serialized_objects.py @@ -72,7 +72,6 @@ from airflow.models.deadline import Deadline from airflow.models.expandinput import create_expand_input from airflow.models.taskinstancekey import TaskInstanceKey -from airflow.models.tasklog import LogTemplate from airflow.models.xcom import XComModel from airflow.models.xcom_arg import SchedulerXComArg, deserialize_xcom_arg from airflow.sdk import DAG, Asset, AssetAlias, AssetAll, AssetAny, AssetWatcher, BaseOperator, XComArg @@ -2401,9 +2400,6 @@ def _create_orm_dagrun( bundle_version=bundle_version, partition_key=partition_key, ) - # Load defaults into the following two fields to ensure result can be serialized detached - max_log_template_id = session.scalar(select(func.max(LogTemplate.__table__.c.id))) - run.log_template_id = int(max_log_template_id) if max_log_template_id is not None else 0 run.created_dag_version = dag_version run.consumed_asset_events = [] session.add(run) diff --git a/airflow-core/src/airflow/utils/cli.py b/airflow-core/src/airflow/utils/cli.py index ed9eb7e84f633..105a6882cd308 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -106,11 +106,10 @@ def wrapper(*args, **kwargs): # Check and run migrations if necessary if check_db: from airflow.configuration import conf - from airflow.utils.db import check_and_run_migrations, synchronize_log_template + from airflow.utils.db import check_and_run_migrations if conf.getboolean("database", "check_migrations"): check_and_run_migrations() - synchronize_log_template() return f(*args, **kwargs) except Exception as e: metrics["error"] = e diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 231a74966c868..c211587521bc6 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -46,7 +46,6 @@ func, inspect, literal, - or_, select, text, ) @@ -751,9 +750,8 @@ def initdb(session: Session = NEW_SESSION): _create_db_from_orm(session=session) external_db_manager.initdb(session) - # Add default pool & sync log_template + # Add default pool add_default_pool_if_not_exists(session=session) - synchronize_log_template(session=session) def _get_alembic_config(): @@ -889,82 +887,6 @@ def check_and_run_migrations(): sys.exit(1) -@provide_session -def synchronize_log_template(*, session: Session = NEW_SESSION) -> None: - """ - Synchronize log template configs with table. - - This checks if the last row fully matches the current config values, and - insert a new row if not. - """ - # NOTE: SELECT queries in this function are INTENTIONALLY written with the - # SQL builder style, not the ORM query API. This avoids configuring the ORM - # unless we need to insert something, speeding up CLI in general. - - from airflow.models.tasklog import LogTemplate - - metadata = reflect_tables([LogTemplate], session) - log_template_table: Table | None = metadata.tables.get(LogTemplate.__tablename__) - - if log_template_table is None: - log.info("Log template table does not exist (added in 2.3.0); skipping log template sync.") - return - - filename = conf.get("logging", "log_filename_template") - elasticsearch_id = conf.get("elasticsearch", "log_id_template") - - stored = session.execute( - select( - log_template_table.c.filename, - log_template_table.c.elasticsearch_id, - ) - .order_by(log_template_table.c.id.desc()) - .limit(1) - ).first() - - # If we have an empty table, and the default values exist, we will seed the - # table with values from pre 2.3.0, so old logs will still be retrievable. - if not stored: - is_default_log_id = elasticsearch_id == conf.get_default_value("elasticsearch", "log_id_template") - is_default_filename = filename == conf.get_default_value("logging", "log_filename_template") - if is_default_log_id and is_default_filename: - session.add( - LogTemplate( - filename="{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log", - elasticsearch_id="{dag_id}-{task_id}-{logical_date}-{try_number}", - ) - ) - - # Before checking if the _current_ value exists, we need to check if the old config value we upgraded in - # place exists! - pre_upgrade_filename = conf.upgraded_values.get(("logging", "log_filename_template"), filename) - pre_upgrade_elasticsearch_id = conf.upgraded_values.get( - ("elasticsearch", "log_id_template"), elasticsearch_id - ) - if pre_upgrade_filename != filename or pre_upgrade_elasticsearch_id != elasticsearch_id: - # The previous non-upgraded value likely won't be the _latest_ value (as after we've recorded the - # recorded the upgraded value it will be second-to-newest), so we'll have to just search which is okay - # as this is a table with a tiny number of rows - row = session.execute( - select(log_template_table.c.id) - .where( - or_( - log_template_table.c.filename == pre_upgrade_filename, - log_template_table.c.elasticsearch_id == pre_upgrade_elasticsearch_id, - ) - ) - .order_by(log_template_table.c.id.desc()) - .limit(1) - ).first() - if not row: - session.add( - LogTemplate(filename=pre_upgrade_filename, elasticsearch_id=pre_upgrade_elasticsearch_id) - ) - - if not stored or stored.filename != filename or stored.elasticsearch_id != elasticsearch_id: - session.add(LogTemplate(filename=filename, elasticsearch_id=elasticsearch_id)) - - def reflect_tables(tables: list[MappedClassProtocol | str] | None, session): """ When running checks prior to upgrades, we use reflection to determine current state of the database. @@ -1150,7 +1072,6 @@ def upgradedb( settings.reconfigure_orm() add_default_pool_if_not_exists(session=session) - synchronize_log_template(session=session) @provide_session diff --git a/airflow-core/src/airflow/utils/helpers.py b/airflow-core/src/airflow/utils/helpers.py index 0e4c5f325b131..1f2488d82f8d2 100644 --- a/airflow-core/src/airflow/utils/helpers.py +++ b/airflow-core/src/airflow/utils/helpers.py @@ -170,29 +170,6 @@ def _render_template_to_string(template: jinja2.Template, context: Context) -> s return render_template(template, cast("MutableMapping[str, Any]", context), native=False) -def render_log_filename(ti: TaskInstance, try_number, filename_template) -> str: - """ - Given task instance, try_number, filename_template, return the rendered log filename. - - :param ti: task instance - :param try_number: try_number of the task - :param filename_template: filename template, which can be jinja template or - python string template - """ - filename_template, filename_jinja_template = parse_template_string(filename_template) - if filename_jinja_template: - jinja_context = ti.get_template_context() - jinja_context["try_number"] = try_number - return _render_template_to_string(filename_jinja_template, jinja_context) - - return filename_template.format( - dag_id=ti.dag_id, - task_id=ti.task_id, - logical_date=ti.logical_date.isoformat(), - try_number=try_number, - ) - - def convert_camel_to_snake(camel_str: str) -> str: """Convert CamelCase to snake_case.""" return CAMELCASE_TO_SNAKE_CASE_REGEX.sub(r"_\1", camel_str).lower() diff --git a/airflow-core/src/airflow/utils/log/file_task_handler.py b/airflow-core/src/airflow/utils/log/file_task_handler.py index 44ccf2cde3803..9f1823c1a8218 100644 --- a/airflow-core/src/airflow/utils/log/file_task_handler.py +++ b/airflow-core/src/airflow/utils/log/file_task_handler.py @@ -526,7 +526,7 @@ def _render_filename( date = dag_run.logical_date or dag_run.run_after formatted_date = date.isoformat() - template = dag_run.get_log_template(session=session).filename + template = conf.get("logging", "log_filename_template") str_tpl, jinja_tpl = parse_template_string(template) if jinja_tpl: return render_template( diff --git a/airflow-core/src/airflow/utils/log/log_reader.py b/airflow-core/src/airflow/utils/log/log_reader.py index 9ebc3a9050c52..99576559e96b1 100644 --- a/airflow-core/src/airflow/utils/log/log_reader.py +++ b/airflow-core/src/airflow/utils/log/log_reader.py @@ -25,17 +25,13 @@ from typing import TYPE_CHECKING from airflow.configuration import conf -from airflow.utils.helpers import render_log_filename from airflow.utils.log.file_task_handler import FileTaskHandler, StructuredLogMessage from airflow.utils.log.logging_mixin import ExternalLoggingMixin -from airflow.utils.session import NEW_SESSION, provide_session from airflow.utils.state import TaskInstanceState if TYPE_CHECKING: from typing import TypeAlias - from sqlalchemy.orm.session import Session - from airflow.models.taskinstance import TaskInstance from airflow.models.taskinstancehistory import TaskInstanceHistory from airflow.utils.log.file_task_handler import LogHandlerOutputStream, LogMetadata @@ -190,25 +186,3 @@ def supports_external_link(self) -> bool: return False return self.log_handler.supports_external_link - - @provide_session - def render_log_filename( - self, - ti: TaskInstance | TaskInstanceHistory, - try_number: int | None = None, - *, - session: Session = NEW_SESSION, - ) -> str: - """ - Render the log attachment filename. - - :param ti: The task instance - :param try_number: The task try number - """ - dagrun = ti.get_dagrun(session=session) - attachment_filename = render_log_filename( - ti=ti, - try_number="all" if try_number is None else try_number, - filename_template=dagrun.get_log_template(session=session).filename, - ) - return attachment_filename diff --git a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py index 9d06ad325cf5a..6adc91fbfb1e2 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py @@ -231,7 +231,7 @@ def test_handle_single_column_unique_constraint_error( status_code=status.HTTP_409_CONFLICT, detail={ "reason": "Unique constraint violation", - "statement": "INSERT INTO dag_run (dag_id, queued_at, logical_date, start_date, end_date, state, run_id, creating_job_id, run_type, triggered_by, triggering_user_name, conf, data_interval_start, data_interval_end, run_after, last_scheduling_decision, log_template_id, updated_at, clear_number, backfill_id, bundle_version, scheduled_by_job_id, context_carrier, created_dag_version_id, partition_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, (SELECT max(log_template.id) AS max_1 \nFROM log_template), ?, ?, ?, ?, ?, ?, ?, ?)", + "statement": "INSERT INTO dag_run (dag_id, queued_at, logical_date, start_date, end_date, state, run_id, creating_job_id, run_type, triggered_by, triggering_user_name, conf, data_interval_start, data_interval_end, run_after, last_scheduling_decision, updated_at, clear_number, backfill_id, bundle_version, scheduled_by_job_id, context_carrier, created_dag_version_id, partition_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, (SELECT max(log_template.id) AS max_1 \nFROM log_template), ?, ?, ?, ?, ?, ?, ?, ?)", "orig_error": "UNIQUE constraint failed: dag_run.dag_id, dag_run.run_id", "message": MESSAGE, }, diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_log.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_log.py index 5df670c3a6054..20715e06ded96 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_log.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_log.py @@ -115,7 +115,7 @@ def add_one(x: int): test_client.app.dependency_overrides[dag_bag_from_app] = lambda: dagbag @pytest.fixture - def configure_loggers(self, tmp_path, create_log_template): + def configure_loggers(self, tmp_path): self.log_dir = tmp_path # TASK_ID diff --git a/airflow-core/tests/unit/cli/commands/test_scheduler_command.py b/airflow-core/tests/unit/cli/commands/test_scheduler_command.py index e381f9a754f0e..779fd251b4053 100644 --- a/airflow-core/tests/unit/cli/commands/test_scheduler_command.py +++ b/airflow-core/tests/unit/cli/commands/test_scheduler_command.py @@ -73,28 +73,24 @@ def test_skip_serve_logs(self, mock_process, mock_scheduler_job, executor): assert mock_process.call_count == 0 @mock.patch("airflow.utils.db.check_and_run_migrations") - @mock.patch("airflow.utils.db.synchronize_log_template") @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner") @mock.patch("airflow.cli.commands.scheduler_command.Process") - def test_check_migrations_is_false(self, mock_process, mock_scheduler_job, mock_log, mock_run_migration): + def test_check_migrations_is_false(self, mock_process, mock_scheduler_job, mock_run_migration): mock_scheduler_job.return_value.job_type = "SchedulerJob" args = self.parser.parse_args(["scheduler"]) with conf_vars({("database", "check_migrations"): "False"}): scheduler_command.scheduler(args) mock_run_migration.assert_not_called() - mock_log.assert_called_once() @mock.patch("airflow.utils.db.check_and_run_migrations") - @mock.patch("airflow.utils.db.synchronize_log_template") @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner") @mock.patch("airflow.cli.commands.scheduler_command.Process") - def test_check_migrations_is_true(self, mock_process, mock_scheduler_job, mock_log, mock_run_migration): + def test_check_migrations_is_true(self, mock_process, mock_scheduler_job, mock_run_migration): mock_scheduler_job.return_value.job_type = "SchedulerJob" args = self.parser.parse_args(["scheduler"]) with conf_vars({("database", "check_migrations"): "True"}): scheduler_command.scheduler(args) mock_run_migration.assert_called_once() - mock_log.assert_called_once() @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner") @mock.patch("airflow.cli.commands.scheduler_command.Process") diff --git a/airflow-core/tests/unit/utils/log/test_log_reader.py b/airflow-core/tests/unit/utils/log/test_log_reader.py index f5b4692c085bb..27b75c60b077e 100644 --- a/airflow-core/tests/unit/utils/log/test_log_reader.py +++ b/airflow-core/tests/unit/utils/log/test_log_reader.py @@ -17,23 +17,17 @@ from __future__ import annotations import copy -import datetime import os import sys import tempfile import types -from typing import TYPE_CHECKING from unittest import mock -import pendulum import pytest from airflow import settings from airflow._shared.timezones import timezone from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG -from airflow.models.tasklog import LogTemplate -from airflow.providers.standard.operators.python import PythonOperator -from airflow.timetables.base import DataInterval from airflow.utils.log.log_reader import TaskLogReader from airflow.utils.log.logging_mixin import ExternalLoggingMixin from airflow.utils.state import TaskInstanceState @@ -46,10 +40,6 @@ pytestmark = pytest.mark.db_test -if TYPE_CHECKING: - from airflow.models import DagRun - - class TestLogView: DAG_ID = "dag_log_reader" TASK_ID = "task_log_reader" @@ -112,8 +102,6 @@ def prepare_log_files(self, log_dir): @pytest.fixture(autouse=True) def prepare_db(self, create_task_instance): session = settings.Session() - log_template = LogTemplate(filename=self.FILENAME_TEMPLATE, elasticsearch_id="") - session.add(log_template) session.commit() ti = create_task_instance( dag_id=self.DAG_ID, @@ -129,7 +117,6 @@ def prepare_db(self, create_task_instance): yield clear_db_runs() clear_db_dags() - session.delete(log_template) session.commit() def test_test_read_log_chunks_should_read_one_try(self): @@ -292,54 +279,6 @@ def test_supports_external_link(self): mock_prop.return_value = True assert task_log_reader.supports_external_link - def test_task_log_filename_unique(self, dag_maker): - """ - Ensure the default log_filename_template produces a unique filename. - - See discussion in apache/airflow#19058 [1]_ for how uniqueness may - change in a future Airflow release. For now, the logical date is used - to distinguish DAG runs. This test should be modified when the logical - date is no longer used to ensure uniqueness. - - [1]: https://github.com/apache/airflow/issues/19058 - """ - dag_id = "test_task_log_filename_ts_corresponds_to_logical_date" - task_id = "echo_run_type" - - def echo_run_type(dag_run: DagRun, **kwargs): - print(dag_run.run_type) - - with dag_maker(dag_id, start_date=self.DEFAULT_DATE, schedule="@daily") as dag: - PythonOperator(task_id=task_id, python_callable=echo_run_type) - - start = pendulum.datetime(2021, 1, 1) - end = start + datetime.timedelta(days=1) - trigger_time = end + datetime.timedelta(hours=4, minutes=29) # Arbitrary. - - # Create two DAG runs that have the same data interval, but not the same - # logical date, to check if they correctly use different log files. - scheduled_dagrun: DagRun = dag_maker.create_dagrun( - run_type=DagRunType.SCHEDULED, - logical_date=start, - data_interval=DataInterval(start, end), - ) - manual_dagrun: DagRun = dag_maker.create_dagrun( - run_type=DagRunType.MANUAL, - logical_date=trigger_time, - data_interval=DataInterval(start, end), - ) - - scheduled_ti = scheduled_dagrun.get_task_instance(task_id) - manual_ti = manual_dagrun.get_task_instance(task_id) - assert scheduled_ti is not None - assert manual_ti is not None - - scheduled_ti.refresh_from_task(dag.get_task(task_id)) - manual_ti.refresh_from_task(dag.get_task(task_id)) - - reader = TaskLogReader() - assert reader.render_log_filename(scheduled_ti, 1) != reader.render_log_filename(manual_ti, 1) - @pytest.mark.parametrize( ("state", "try_number", "expected_event", "use_self_ti"), [ diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index 918993d9bcbdf..d05c037e25554 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -452,7 +452,6 @@ def test_no_models_missing(self): "asset_alias", # not good way to know if "stale" "task_map", # keys to TI, so no need "serialized_dag", # handled through FK to Dag - "log_template", # not a significant source of data; age not indicative of staleness "dag_tag", # not a significant source of data; age not indicative of staleness, "dag_owner_attributes", # not a significant source of data; age not indicative of staleness, "dag_code", # self-maintaining diff --git a/airflow-core/tests/unit/utils/test_helpers.py b/airflow-core/tests/unit/utils/test_helpers.py index 6179acadfb94f..8e16d11869843 100644 --- a/airflow-core/tests/unit/utils/test_helpers.py +++ b/airflow-core/tests/unit/utils/test_helpers.py @@ -23,7 +23,6 @@ import pytest -from airflow._shared.timezones import timezone from airflow.exceptions import AirflowException from airflow.jobs.base_job_runner import BaseJobRunner from airflow.serialization.definitions.notset import NOTSET @@ -55,24 +54,6 @@ def clear_db(): class TestHelpers: - @pytest.mark.db_test - @pytest.mark.usefixtures("clear_db") - def test_render_log_filename(self, create_task_instance): - try_number = 1 - dag_id = "test_render_log_filename_dag" - task_id = "test_render_log_filename_task" - logical_date = timezone.datetime(2016, 1, 1) - - ti = create_task_instance(dag_id=dag_id, task_id=task_id, logical_date=logical_date) - filename_template = "{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log" - - ts = ti.get_template_context()["ts"] - expected_filename = f"{dag_id}/{task_id}/{ts}/{try_number}.log" - - rendered_filename = helpers.render_log_filename(ti, try_number, filename_template) - - assert rendered_filename == expected_filename - def test_chunks(self): with pytest.raises(ValueError, match=CHUNK_SIZE_POSITIVE_INT): list(helpers.chunks([1, 2, 3], 0)) diff --git a/airflow-core/tests/unit/utils/test_log_handlers.py b/airflow-core/tests/unit/utils/test_log_handlers.py index 30669ab05997f..26ba2265be7cc 100644 --- a/airflow-core/tests/unit/utils/test_log_handlers.py +++ b/airflow-core/tests/unit/utils/test_log_handlers.py @@ -681,8 +681,7 @@ def setup_mock_aws(): @pytest.mark.parametrize("logical_date", ((None), (DEFAULT_DATE))) class TestFilenameRendering: - def test_python_formatting(self, create_log_template, create_task_instance, logical_date): - create_log_template("{dag_id}/{task_id}/{logical_date}/{try_number}.log") + def test_python_formatting(self, create_task_instance, logical_date): filename_rendering_ti = create_task_instance( dag_id="dag_for_testing_filename_rendering", task_id="task_for_testing_filename_rendering", @@ -700,9 +699,8 @@ def test_python_formatting(self, create_log_template, create_task_instance, logi rendered_filename = fth._render_filename(filename_rendering_ti, 42) assert expected_filename == rendered_filename - def test_python_formatting_catchup_false(self, create_log_template, create_task_instance, logical_date): + def test_python_formatting_catchup_false(self, create_task_instance, logical_date): """Test the filename rendering with catchup=False (the new default behavior)""" - create_log_template("{dag_id}/{task_id}/{logical_date}/{try_number}.log") filename_rendering_ti = create_task_instance( dag_id="dag_for_testing_filename_rendering", task_id="task_for_testing_filename_rendering", @@ -724,8 +722,7 @@ def test_python_formatting_catchup_false(self, create_log_template, create_task_ rendered_filename = fth._render_filename(filename_rendering_ti, 42) assert rendered_filename == expected_filename - def test_jinja_rendering(self, create_log_template, create_task_instance, logical_date): - create_log_template("{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log") + def test_jinja_rendering(self, create_task_instance, logical_date): filename_rendering_ti = create_task_instance( dag_id="dag_for_testing_filename_rendering", task_id="task_for_testing_filename_rendering", @@ -743,9 +740,8 @@ def test_jinja_rendering(self, create_log_template, create_task_instance, logica rendered_filename = fth._render_filename(filename_rendering_ti, 42) assert expected_filename == rendered_filename - def test_jinja_rendering_catchup_false(self, create_log_template, create_task_instance, logical_date): + def test_jinja_rendering_catchup_false(self, create_task_instance, logical_date): """Test the Jinja template rendering with catchup=False (the new default behavior)""" - create_log_template("{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log") filename_rendering_ti = create_task_instance( dag_id="dag_for_testing_filename_rendering", task_id="task_for_testing_filename_rendering", @@ -767,11 +763,8 @@ def test_jinja_rendering_catchup_false(self, create_log_template, create_task_in rendered_filename = fth._render_filename(filename_rendering_ti, 42) assert expected_filename == rendered_filename - def test_jinja_id_in_template_for_history( - self, create_log_template, create_task_instance, logical_date, session - ): + def test_jinja_id_in_template_for_history(self, create_task_instance, logical_date, session): """Test that Jinja template using ti.id works for both TaskInstance and TaskInstanceHistory""" - create_log_template("{{ ti.id }}.log") ti = create_task_instance( dag_id="dag_history_test", task_id="history_task", diff --git a/devel-common/src/tests_common/pytest_plugin.py b/devel-common/src/tests_common/pytest_plugin.py index 27826b6604a83..03c30e924f4a0 100644 --- a/devel-common/src/tests_common/pytest_plugin.py +++ b/devel-common/src/tests_common/pytest_plugin.py @@ -1705,11 +1705,19 @@ def _get(dag_id: str): @pytest.fixture def create_log_template(request): from airflow import settings - from airflow.models.tasklog import LogTemplate session = settings.Session() def _create_log_template(filename_template, elasticsearch_id=""): + from tests_common.test_utils.version_compat import AIRFLOW_V_3_2_PLUS + + if AIRFLOW_V_3_2_PLUS: + # assertion to avoid creating LogTemplate in Airflow 3.2.0+ + raise RuntimeError("LogTemplate is removed in Airflow 3.2.0") + + # only < 3.2.0 are able to import LogTemplate + from airflow.models.tasklog import LogTemplate + log_template = LogTemplate(filename=filename_template, elasticsearch_id=elasticsearch_id) session.add(log_template) session.commit() diff --git a/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py b/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py index 01eebfa0b1f4f..8f8b7c844210e 100644 --- a/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py +++ b/providers/amazon/tests/unit/amazon/aws/log/test_cloudwatch_task_handler.py @@ -187,16 +187,12 @@ def clear_db(self): clear_db_dag_bundles() @pytest.fixture(autouse=True) - def setup(self, create_log_template, tmp_path_factory, session, testing_dag_bundle): + def setup(self, tmp_path_factory, session, testing_dag_bundle): # self.clear_db() with conf_vars({("logging", "remote_log_conn_id"): "aws_default"}): self.remote_log_group = "log_group_name" self.region_name = "us-west-2" self.local_log_location = str(tmp_path_factory.mktemp("local-cloudwatch-log-location")) - if AIRFLOW_V_3_0_PLUS: - create_log_template("{dag_id}/{task_id}/{logical_date}/{try_number}.log") - else: - create_log_template("{dag_id}/{task_id}/{execution_date}/{try_number}.log") self.cloudwatch_task_handler = CloudwatchTaskHandler( self.local_log_location, f"arn:aws:logs:{self.region_name}:11111111:log-group:{self.remote_log_group}", diff --git a/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py b/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py index 65f70ed1b75aa..3ec9ae9f4a86d 100644 --- a/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py +++ b/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py @@ -60,13 +60,12 @@ def clear_db(self): clear_db_dag_bundles() @pytest.fixture(autouse=True) - def setup_tests(self, create_log_template, tmp_path_factory, session, testing_dag_bundle): + def setup_tests(self, tmp_path_factory, session, testing_dag_bundle): with conf_vars({("logging", "remote_log_conn_id"): "aws_default"}): self.remote_log_base = "s3://bucket/remote/log/location" self.remote_log_location = "s3://bucket/remote/log/location/1.log" self.remote_log_key = "remote/log/location/1.log" self.local_log_location = str(tmp_path_factory.mktemp("local-s3-log-location")) - create_log_template("{try_number}.log") self.s3_task_handler = S3TaskHandler(self.local_log_location, self.remote_log_base) # Verify the hook now with the config override self.subject = self.s3_task_handler.io @@ -203,13 +202,12 @@ def clear_db(self): clear_db_dag_bundles() @pytest.fixture(autouse=True) - def setup_tests(self, create_log_template, tmp_path_factory, session, testing_dag_bundle): + def setup_tests(self, tmp_path_factory, session, testing_dag_bundle): with conf_vars({("logging", "remote_log_conn_id"): "aws_default"}): self.remote_log_base = "s3://bucket/remote/log/location" self.remote_log_location = "s3://bucket/remote/log/location/1.log" self.remote_log_key = "remote/log/location/1.log" self.local_log_location = str(tmp_path_factory.mktemp("local-s3-log-location")) - create_log_template("{try_number}.log") self.s3_task_handler = S3TaskHandler(self.local_log_location, self.remote_log_base) # Verify the hook now with the config override assert self.s3_task_handler.io.hook is not None diff --git a/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py b/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py index b5e2adb757797..3a345c28146db 100644 --- a/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py +++ b/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py @@ -39,8 +39,7 @@ class TestHdfsTaskHandler: @pytest.fixture(autouse=True) - def ti(self, create_task_instance, create_log_template): - create_log_template("{try_number}.log") + def ti(self, create_task_instance): ti = create_task_instance( dag_id="dag_for_testing_hdfs_task_handler", task_id="task_for_testing_hdfs_log_handler", diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py index c38469bc17230..becb3f23c07a4 100644 --- a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py +++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py @@ -244,8 +244,10 @@ def _render_log_id(self, ti: TaskInstance | TaskInstanceKey, try_number: int) -> if isinstance(ti, TaskInstanceKey): ti = _ensure_ti(ti, session) dag_run = ti.get_dagrun(session=session) - if USE_PER_RUN_LOG_ID: + if USE_PER_RUN_LOG_ID and AIRFLOW_V_3_0_PLUS: log_id_template = dag_run.get_log_template(session=session).elasticsearch_id + else: + log_id_template = conf.get("elasticsearch", "log_id_template") if self.json_format: data_interval_start = self._clean_date(dag_run.data_interval_start) diff --git a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py index ee47d8b6d4798..8d1f679f18b42 100644 --- a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py +++ b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py @@ -47,7 +47,7 @@ from tests_common.test_utils.config import conf_vars from tests_common.test_utils.db import clear_db_dags, clear_db_runs from tests_common.test_utils.paths import AIRFLOW_PROVIDERS_ROOT_PATH -from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS +from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS from unit.elasticsearch.log.elasticmock import elasticmock from unit.elasticsearch.log.elasticmock.utilities import SearchFailedException @@ -77,14 +77,16 @@ class TestElasticsearchTaskHandler: @pytest.fixture def ti(self, create_task_instance, create_log_template): - create_log_template( - self.FILENAME_TEMPLATE, - ( - "{dag_id}-{task_id}-{logical_date}-{try_number}" - if AIRFLOW_V_3_0_PLUS - else "{dag_id}-{task_id}-{execution_date}-{try_number}" - ), - ) + # we remove the LogTemplate in 3.2.0, so we don't need to create it here + if not AIRFLOW_V_3_2_PLUS: + create_log_template( + self.FILENAME_TEMPLATE, + ( + "{dag_id}-{task_id}-{logical_date}-{try_number}" + if AIRFLOW_V_3_0_PLUS + else "{dag_id}-{task_id}-{execution_date}-{try_number}" + ), + ) yield get_ti( dag_id=self.DAG_ID, task_id=self.TASK_ID, diff --git a/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py b/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py index 135d443244559..e777ec70affff 100644 --- a/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py +++ b/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py @@ -56,8 +56,7 @@ def local_log_location(self, tmp_path_factory): return str(tmp_path_factory.mktemp("local-gcs-log-location")) @pytest.fixture(autouse=True) - def gcs_task_handler(self, create_log_template, local_log_location): - create_log_template("{try_number}.log") + def gcs_task_handler(self, local_log_location): self.gcs_task_handler = GCSTaskHandler( base_log_folder=local_log_location, gcs_log_folder="gs://bucket/remote/log/location", diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py b/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py index 4a09b28983375..09497950bab80 100644 --- a/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py +++ b/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py @@ -43,8 +43,7 @@ class TestWasbTaskHandler: @pytest.fixture(autouse=True) - def ti(self, create_task_instance, create_log_template): - create_log_template("{try_number}.log") + def ti(self, create_task_instance): ti = create_task_instance( dag_id="dag_for_testing_wasb_task_handler", task_id="task_for_testing_wasb_log_handler", diff --git a/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py b/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py index bd3966743e033..267240da0febd 100644 --- a/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py +++ b/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py @@ -301,8 +301,10 @@ def _render_log_id(self, ti: TaskInstance | TaskInstanceKey, try_number: int) -> if isinstance(ti, TaskInstanceKey): ti = _ensure_ti(ti, session) dag_run = ti.get_dagrun(session=session) - if USE_PER_RUN_LOG_ID: + if USE_PER_RUN_LOG_ID and AIRFLOW_V_3_0_PLUS: log_id_template = dag_run.get_log_template(session=session).elasticsearch_id + else: + log_id_template = conf.get("opensearch", "log_id_template") if self.json_format: data_interval_start = self._clean_date(dag_run.data_interval_start) diff --git a/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py b/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py index c976182f2c6e9..11ee575605138 100644 --- a/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py +++ b/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py @@ -43,7 +43,7 @@ from tests_common.test_utils.config import conf_vars from tests_common.test_utils.db import clear_db_dags, clear_db_runs -from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS +from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS from unit.opensearch.conftest import MockClient opensearchpy = pytest.importorskip("opensearchpy") @@ -72,12 +72,17 @@ class TestOpensearchTaskHandler: @pytest.fixture def ti(self, create_task_instance, create_log_template): - if AIRFLOW_V_3_0_PLUS: - create_log_template(self.FILENAME_TEMPLATE, "{dag_id}-{task_id}-{logical_date}-{try_number}") + if AIRFLOW_V_3_2_PLUS: + # we remove the LogTemplate in 3.2.0, so we don't need to create it here + pass else: create_log_template( self.FILENAME_TEMPLATE, - "{dag_id}-{task_id}-{execution_date}-{try_number}", + ( + "{dag_id}-{task_id}-{logical_date}-{try_number}" + if AIRFLOW_V_3_0_PLUS + else "{dag_id}-{task_id}-{execution_date}-{try_number}" + ), ) yield get_ti( dag_id=self.DAG_ID,