diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst index 6419525c8ab0b..a91cb5814a112 100644 --- a/airflow-core/docs/migrations-ref.rst +++ b/airflow-core/docs/migrations-ref.rst @@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | Revision ID | Revises ID | Airflow Version | Description | +=========================+==================+===================+==============================================================+ -| ``436dc127462c`` (head) | ``5a5d3253e946`` | ``3.4.0`` | Drop span_status column. | +| ``4f6723e37686`` (head) | ``436dc127462c`` | ``3.4.0`` | Drop log_template table. | ++-------------------------+------------------+-------------------+--------------------------------------------------------------+ +| ``436dc127462c`` | ``5a5d3253e946`` | ``3.4.0`` | Drop span_status column. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | ``5a5d3253e946`` | ``d2f4e1b3c5a7`` | ``3.4.0`` | Add GIN index on asset_event.extra for PostgreSQL. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ diff --git a/airflow-core/newsfragments/69520.significant.rst b/airflow-core/newsfragments/69520.significant.rst new file mode 100644 index 0000000000000..0a84ffcc3dd88 --- /dev/null +++ b/airflow-core/newsfragments/69520.significant.rst @@ -0,0 +1,54 @@ +Drop the ``LogTemplate`` DB model + +Airflow tracked historical values of ``[logging] log_filename_template`` and +``[elasticsearch] log_id_template`` in a ``log_template`` table so a ``DagRun``'s +logs stayed retrievable after either config value changed. This model has been +removed: the Elasticsearch/OpenSearch task handlers reached into the metadata +database directly to read it, which conflicts with the Airflow 3 execution +model where workers/handlers should not query the metadata DB directly, and it +otherwise only existed to duplicate two strings already available from config. + +**What changed:** + +- The ``LogTemplate`` model, the ``log_template`` table, and ``DagRun.log_template_id`` + are gone. +- ``DagRun.get_log_template()`` has been removed. +- Log filename and log ID templates are now read directly from + ``[logging] log_filename_template`` and ``[elasticsearch] log_id_template`` at + render time, instead of being pinned to the value in effect when the ``DagRun`` + was created. + +**Behaviour changes:** + +- Changing ``log_filename_template`` or ``log_id_template`` now applies + immediately and retroactively to *all* ``DagRun``s, including ones created + before the change. There is no more per-run historical pinning: if you change + either setting, logs written under the previous template are no longer + reachable through the new one. + +**Migration:** + +- A database migration drops the ``log_template`` table and the + ``dag_run.log_template_id`` column. No user action is required. +- Any custom log handler or plugin calling ``DagRun.get_log_template()`` or + importing ``airflow.models.tasklog.LogTemplate`` must be updated to read + ``[logging] log_filename_template`` / ``[elasticsearch] log_id_template`` + from config instead. + +* Types of change + + * [ ] Dag changes + * [x] Config changes + * [ ] API changes + * [ ] CLI changes + * [x] Behaviour changes + * [ ] Plugin changes + * [ ] Dependency changes + * [x] Code interface changes + +* Migration rules needed + + * Remove any usage of ``DagRun.get_log_template()`` or + ``airflow.models.tasklog.LogTemplate`` in custom log handlers/plugins; + read ``[logging] log_filename_template`` / ``[elasticsearch] log_id_template`` + from config instead. diff --git a/airflow-core/src/airflow/migrations/versions/0126_3_4_0_drop_log_template.py b/airflow-core/src/airflow/migrations/versions/0126_3_4_0_drop_log_template.py new file mode 100644 index 0000000000000..851399250a5ad --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0126_3_4_0_drop_log_template.py @@ -0,0 +1,70 @@ +# +# 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. +""" +Drop log_template table. + +Revision ID: 4f6723e37686 +Revises: 436dc127462c +Create Date: 2026-07-05 00:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +from airflow.migrations.utils import disable_sqlite_fkeys +from airflow.utils.sqlalchemy import UtcDateTime + +# revision identifiers, used by Alembic. +revision = "4f6723e37686" +down_revision = "436dc127462c" +branch_labels = None +depends_on = None +airflow_version = "3.4.0" + + +def upgrade(): + """Apply Drop log_template table.""" + with disable_sqlite_fkeys(op): + with op.batch_alter_table("dag_run", schema=None) as batch_op: + batch_op.drop_constraint(batch_op.f("task_instance_log_template_id_fkey"), type_="foreignkey") + batch_op.drop_column("log_template_id") + op.drop_table("log_template") + + +def downgrade(): + """Unapply Drop log_template table.""" + with disable_sqlite_fkeys(op): + op.create_table( + "log_template", + sa.Column("id", sa.Integer(), nullable=False, autoincrement=True), + sa.Column("filename", sa.Text(), nullable=False), + sa.Column("elasticsearch_id", sa.Text(), nullable=False), + sa.Column("created_at", UtcDateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id", name="log_template_pkey"), + ) + with op.batch_alter_table("dag_run", schema=None) as batch_op: + batch_op.add_column(sa.Column("log_template_id", sa.Integer(), nullable=True)) + batch_op.create_foreign_key( + batch_op.f("task_instance_log_template_id_fkey"), + "log_template", + ["log_template_id"], + ["id"], + ondelete="NO ACTION", + ) diff --git a/airflow-core/src/airflow/models/__init__.py b/airflow-core/src/airflow/models/__init__.py index f5bffb4e57173..43191e1202090 100644 --- a/airflow-core/src/airflow/models/__init__.py +++ b/airflow-core/src/airflow/models/__init__.py @@ -75,7 +75,6 @@ def import_all_models(): import airflow.models.serialized_dag import airflow.models.task_state_store 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 edc4d903aa26c..c7e25d3135d28 100644 --- a/airflow-core/src/airflow/models/dagrun.py +++ b/airflow-core/src/airflow/models/dagrun.py @@ -80,7 +80,6 @@ from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel from airflow.models.taskinstance import TaskInstance as TI, clear_task_instances from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH -from airflow.models.tasklog import LogTemplate from airflow.models.taskmap import TaskMap from airflow.serialization.definitions.deadline import SerializedReferenceModels from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set @@ -231,15 +230,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 | None] = 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)), - nullable=True, - ) # This is nullable because it's too costly to migrate dagruns created prior # to this column's addition (Airflow 3.2.0). If you want a reasonable # meaningful non-null value, use ``dr.created_at or dr.run_after``. @@ -2217,25 +2207,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 09216a21834e5..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, mapped_column - -from airflow._shared.timezones import timezone -from airflow.models.base import Base -from airflow.utils.sqlalchemy import UtcDateTime - - -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/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py index abae0f9778572..7d6dfaef2cdf9 100644 --- a/airflow-core/src/airflow/serialization/definitions/dag.py +++ b/airflow-core/src/airflow/serialization/definitions/dag.py @@ -48,7 +48,6 @@ from airflow.models.deadline import Deadline from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel from airflow.models.taskinstancekey import TaskInstanceKey -from airflow.models.tasklog import LogTemplate from airflow.sdk.definitions.deadline import VariableInterval from airflow.serialization.decoders import decode_deadline_alert from airflow.serialization.definitions.deadline import DeadlineAlertFields, SerializedReferenceModels @@ -1425,9 +1424,7 @@ def _create_orm_dagrun( partition_date=partition_date, note=note, ) - # 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 + # Load default into the following field to ensure result can be serialized detached 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 d0105e86efa20..7f007c4142671 100644 --- a/airflow-core/src/airflow/utils/cli.py +++ b/airflow-core/src/airflow/utils/cli.py @@ -105,11 +105,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() # Set server context so validation logic (e.g. DagBundlesManager) runs correctly original_ctx = os.environ.pop("_AIRFLOW_PROCESS_CONTEXT", None) try: diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index ee10c008f9b08..9612107add594 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -48,7 +48,6 @@ func, inspect, literal, - or_, select, text, ) @@ -117,7 +116,7 @@ class MappedClassProtocol(Protocol): "3.1.8": "509b94a1042d", "3.2.0": "1d6611b6ab7c", "3.3.0": "d2f4e1b3c5a7", - "3.4.0": "436dc127462c", + "3.4.0": "4f6723e37686", } # Prefix used to identify tables holding data moved during migration. @@ -864,9 +863,8 @@ def initdb(*, session: Session = NEW_SESSION, use_migration_files: bool = False) _create_db_from_orm(session=session) external_db_manager.initdb(session, use_migration_files=use_migration_files) - # 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(): @@ -1004,82 +1002,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. @@ -1217,7 +1139,6 @@ def _run_upgradedb( external_db_manager.upgradedb(work_session, use_migration_files=use_migration_files) add_default_pool_if_not_exists(session=work_session) - synchronize_log_template(session=work_session) @provide_session 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 d22ce48fa0897..ffa04791e42e4 100644 --- a/airflow-core/src/airflow/utils/log/file_task_handler.py +++ b/airflow-core/src/airflow/utils/log/file_task_handler.py @@ -529,7 +529,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/tests/unit/api_fastapi/common/test_exceptions.py b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py index 118f0b1a5d794..708d438023fb0 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py @@ -328,7 +328,7 @@ def test_handle_multiple_columns_unique_constraint_error_without_stacktrace( 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", "orig_error": "UNIQUE constraint failed: dag_run.dag_id, dag_run.run_id", }, ), @@ -336,7 +336,7 @@ def test_handle_multiple_columns_unique_constraint_error_without_stacktrace( 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 (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, (SELECT max(log_template.id) AS max_1 \nFROM log_template), %s, %s, %s, %s, %s, %s, %s, %s)", + "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 (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)", "orig_error": "(1062, \"Duplicate entry 'test_dag_id-test_run_id' for key 'dag_run.dag_run_dag_id_run_id_key'\")", }, ), @@ -344,7 +344,7 @@ def test_handle_multiple_columns_unique_constraint_error_without_stacktrace( 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 (%(dag_id)s, %(queued_at)s, %(logical_date)s, %(start_date)s, %(end_date)s, %(state)s, %(run_id)s, %(creating_job_id)s, %(run_type)s, %(triggered_by)s, %(triggering_user_name)s, %(conf)s, %(data_interval_start)s, %(data_interval_end)s, %(run_after)s, %(last_scheduling_decision)s, (SELECT max(log_template.id) AS max_1 \nFROM log_template), %(updated_at)s, %(clear_number)s, %(backfill_id)s, %(bundle_version)s, %(scheduled_by_job_id)s, %(context_carrier)s, %(created_dag_version_id)s, %(partition_key)s) RETURNING dag_run.id", + "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 (%(dag_id)s, %(queued_at)s, %(logical_date)s, %(start_date)s, %(end_date)s, %(state)s, %(run_id)s, %(creating_job_id)s, %(run_type)s, %(triggered_by)s, %(triggering_user_name)s, %(conf)s, %(data_interval_start)s, %(data_interval_end)s, %(run_after)s, %(last_scheduling_decision)s, %(updated_at)s, %(clear_number)s, %(backfill_id)s, %(bundle_version)s, %(scheduled_by_job_id)s, %(context_carrier)s, %(created_dag_version_id)s, %(partition_key)s) RETURNING dag_run.id", "orig_error": 'duplicate key value violates unique constraint "dag_run_dag_id_run_id_key"\nDETAIL: Key (dag_id, run_id)=(test_dag_id, test_run_id) already exists.\n', }, ), 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 12fd8fe111ab0..edb09f73ec571 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 2c9e995efabcc..39af0cf51ff45 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 1c65cb1806b14..5af88920a5365 100644 --- a/airflow-core/tests/unit/utils/log/test_log_reader.py +++ b/airflow-core/tests/unit/utils/log/test_log_reader.py @@ -28,7 +28,6 @@ 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.utils.log.log_reader import TaskLogReader from airflow.utils.log.logging_mixin import ExternalLoggingMixin from airflow.utils.state import TaskInstanceState @@ -102,26 +101,21 @@ 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, - task_id=self.TASK_ID, - start_date=self.DEFAULT_DATE, - run_type=DagRunType.SCHEDULED, - logical_date=self.DEFAULT_DATE, - state=TaskInstanceState.RUNNING, - ) - ti.try_number = 3 - ti.hostname = "localhost" - self.ti = ti - yield + with conf_vars({("logging", "log_filename_template"): self.FILENAME_TEMPLATE}): + ti = create_task_instance( + dag_id=self.DAG_ID, + task_id=self.TASK_ID, + start_date=self.DEFAULT_DATE, + run_type=DagRunType.SCHEDULED, + logical_date=self.DEFAULT_DATE, + state=TaskInstanceState.RUNNING, + ) + ti.try_number = 3 + ti.hostname = "localhost" + self.ti = ti + yield clear_db_runs() clear_db_dags() - session.delete(log_template) - session.commit() def test_test_read_log_chunks_should_read_one_try(self): task_log_reader = TaskLogReader() diff --git a/airflow-core/tests/unit/utils/test_db.py b/airflow-core/tests/unit/utils/test_db.py index c81859edbdcfa..2607519c6e098 100644 --- a/airflow-core/tests/unit/utils/test_db.py +++ b/airflow-core/tests/unit/utils/test_db.py @@ -100,7 +100,6 @@ def test_initdb_use_migration_files_uses_alembic_for_empty_db(self, mocker): mock_upgradedb = mocker.patch("airflow.utils.db.upgradedb") mock_create_from_orm = mocker.patch("airflow.utils.db._create_db_from_orm") mocker.patch("airflow.utils.db.add_default_pool_if_not_exists") - mocker.patch("airflow.utils.db.synchronize_log_template") initdb(session=session, use_migration_files=True) diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index 4d66a96b1a358..e5c2ff2e87468 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -623,7 +623,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_log_handlers.py b/airflow-core/tests/unit/utils/test_log_handlers.py index b977ef0d3e0b5..8bc8533221d87 100644 --- a/airflow-core/tests/unit/utils/test_log_handlers.py +++ b/airflow-core/tests/unit/utils/test_log_handlers.py @@ -735,8 +735,8 @@ 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") + @conf_vars({("logging", "log_filename_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", @@ -754,9 +754,9 @@ 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): + @conf_vars({("logging", "log_filename_template"): "{dag_id}/{task_id}/{logical_date}/{try_number}.log"}) + 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", @@ -778,8 +778,15 @@ 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") + @conf_vars( + { + ( + "logging", + "log_filename_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", @@ -797,9 +804,16 @@ 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): + @conf_vars( + { + ( + "logging", + "log_filename_template", + ): "{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log" + } + ) + 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", @@ -821,11 +835,9 @@ 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 - ): + @conf_vars({("logging", "log_filename_template"): "{{ ti.id }}.log"}) + 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 f10a6d10dbcff..2875293c7d148 100644 --- a/devel-common/src/tests_common/pytest_plugin.py +++ b/devel-common/src/tests_common/pytest_plugin.py @@ -1856,13 +1856,27 @@ def _get(dag_id: str): @pytest.fixture -def create_log_template(request): +def create_log_template(request, monkeypatch): + """Configure log templates the way the core under test resolves them. + + Airflow 3.4.0 dropped the ``LogTemplate`` model and reads the filename + template from live config; older cores pin the templates per DagRun via the + ``log_template`` table. Always sets the filename-template config env var, + and additionally seeds the table when the model exists, so the same test + works against any supported core. ``elasticsearch_id`` only feeds the + table row: on 3.4.0+ the ES/OS handlers no longer read a per-run log ID + template, so there is nothing to configure for it. + """ from airflow import settings - from airflow.models.tasklog import LogTemplate - - session = settings.Session() def _create_log_template(filename_template, elasticsearch_id=""): + monkeypatch.setenv("AIRFLOW__LOGGING__LOG_FILENAME_TEMPLATE", filename_template) + try: + from airflow.models.tasklog import LogTemplate # removed in Airflow 3.4.0 + except ImportError: + return + + session = settings.Session() log_template = LogTemplate(filename=filename_template, elasticsearch_id=elasticsearch_id) session.add(log_template) session.commit() diff --git a/devel-common/src/tests_common/test_utils/config.py b/devel-common/src/tests_common/test_utils/config.py index 9a278346d3068..ded63e6043c64 100644 --- a/devel-common/src/tests_common/test_utils/config.py +++ b/devel-common/src/tests_common/test_utils/config.py @@ -88,7 +88,12 @@ def conf_vars(overrides): original_env_vars[env] = os.environ.pop(env) for i, conf in enumerate(configs): - originals[i][(section, key)] = conf.get(section, key) if conf.has_option(section, key) else None + # Snapshot the raw stored form: restoring the interpolated form via + # conf.set() fails for values with a literal % (e.g. Jinja {% %} in + # the log_filename_template default). + originals[i][(section, key)] = ( + conf.get(section, key, raw=True) if conf.has_option(section, key) else None + ) if value is not None: if not conf.has_section(section): conf.add_section(section) diff --git a/generated/known_airflow_exceptions.txt b/generated/known_airflow_exceptions.txt index 1fb461854905d..8d891f851707c 100644 --- a/generated/known_airflow_exceptions.txt +++ b/generated/known_airflow_exceptions.txt @@ -9,7 +9,7 @@ airflow-core/src/airflow/jobs/base_job_runner.py::1 airflow-core/src/airflow/jobs/job.py::1 airflow-core/src/airflow/models/connection.py::4 airflow-core/src/airflow/models/crypto.py::1 -airflow-core/src/airflow/models/dagrun.py::2 +airflow-core/src/airflow/models/dagrun.py::1 airflow-core/src/airflow/models/pool.py::2 airflow-core/src/airflow/secrets/local_filesystem.py::4 airflow-core/src/airflow/serialization/definitions/dag.py::1 diff --git a/providers/elasticsearch/docs/logging/index.rst b/providers/elasticsearch/docs/logging/index.rst index df2a6e6be71f2..e5c8c0aff6c50 100644 --- a/providers/elasticsearch/docs/logging/index.rst +++ b/providers/elasticsearch/docs/logging/index.rst @@ -342,14 +342,8 @@ To enable it, ``airflow.cfg`` must be configured as in the example below. Note t Changes to ``[elasticsearch] log_id_template`` '''''''''''''''''''''''''''''''''''''''''''''' -If you ever need to make changes to ``[elasticsearch] log_id_template``, Airflow 2.3.0+ is able to keep track of -old values so your existing task runs logs can still be fetched. Once you are on Airflow 2.3.0+, in general, you -can just change ``log_id_template`` at will and Airflow will keep track of the changes. - -However, when you are upgrading to 2.3.0+, Airflow may not be able to properly save your previous ``log_id_template``. -If after upgrading you find your task logs are no longer accessible, try adding a row in the ``log_template`` table with ``id=0`` -containing your previous ``log_id_template``. For example, if you used the defaults in 2.2.5: - -.. code-block:: sql - - INSERT INTO log_template (id, filename, elasticsearch_id, created_at) VALUES (0, '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log', '{dag_id}-{task_id}-{execution_date}-{try_number}', NOW()); +``[elasticsearch] log_id_template`` is read directly from the live configuration whenever a log ID +is rendered. Airflow no longer tracks historical values of this setting in the metadata database +(the ``log_template`` table was removed), so changing it applies immediately and retroactively to +how *all* task logs, past and future, are located. If you change ``log_id_template``, older logs +written under a previous template will no longer be reachable through the new one. 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 fa05b2a0b34cf..9cd67d2eab5dc 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 @@ -43,7 +43,6 @@ import airflow.logging_config as alc from airflow.exceptions import AirflowProviderDeprecationWarning -from airflow.models.dagrun import DagRun from airflow.providers.common.compat.sdk import conf from airflow.providers.elasticsearch._compat import apply_compat_with from airflow.providers.elasticsearch.log.es_json_formatter import ElasticsearchJSONFormatter @@ -79,11 +78,6 @@ LOG_LINE_DEFAULTS = {"exc_text": "", "stack_info": ""} # Elasticsearch hosted log type -# Compatibility: Airflow 2.3.3 and up uses this method, which accesses the -# LogTemplate model to record the log ID template used. If this function does -# not exist, the task handler should use the log_id_template attribute instead. -USE_PER_RUN_LOG_ID = hasattr(DagRun, "get_log_template") - TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger", "error_detail", "message", "levelname"] 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 77ed6830e8b1e..9ab7f9d17a9a9 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 @@ -68,6 +68,9 @@ else: from airflow.utils import timezone # type: ignore[attr-defined,no-redef] +# Compatibility: cores before Airflow 3.4.0 pin the log ID template per DagRun +# in the (since removed) LogTemplate model; keep honoring it while such cores +# are supported. On 3.4.0+ the attribute is gone and this is always False. USE_PER_RUN_LOG_ID = hasattr(DagRun, "get_log_template") LOG_LINE_DEFAULTS = {"exc_text": "", "stack_info": ""} TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger", "error_detail", "message", "levelname"] @@ -515,7 +518,9 @@ def _render_log_id(self, ti: TaskInstance | TaskInstanceKey, try_number: int) -> ti = _ensure_ti(ti, session) dag_run = ti.get_dagrun(session=session) if USE_PER_RUN_LOG_ID: - log_id_template = dag_run.get_log_template(session=session).elasticsearch_id + log_id_template = dag_run.get_log_template( # type: ignore[attr-defined] + session=session + ).elasticsearch_id 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 f9d9fa8942603..94772518d52c1 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 @@ -152,6 +152,11 @@ class TestOpensearchTaskHandler: LOGICAL_DATE = datetime(2016, 1, 1) LOG_ID = f"{DAG_ID}-{TASK_ID}-2016-01-01T00:00:00+00:00-1" JSON_LOG_ID = f"{DAG_ID}-{TASK_ID}-{OpensearchTaskHandler._clean_date(LOGICAL_DATE)}-1" + LOG_ID_TEMPLATE = ( + "{dag_id}-{task_id}-{logical_date}-{try_number}" + if AIRFLOW_V_3_0_PLUS + else "{dag_id}-{task_id}-{execution_date}-{try_number}" + ) FILENAME_TEMPLATE = "{try_number}.log" @pytest.fixture(autouse=True) @@ -187,6 +192,9 @@ def _setup_handler(self, tmp_path): json_fields=self.json_fields, host_field=self.host_field, offset_field=self.offset_field, + # On cores without per-run pinning (3.4.0+) the handler falls back + # to this template; older cores read it from the seeded DB row. + log_id_template=self.LOG_ID_TEMPLATE, ) @pytest.fixture