Drop LogTemplate DB Model

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
 

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

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)

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

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

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()

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(

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

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,
                         },

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

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")