Change ds, ts, etc. back to use logical date

airflow/models/taskinstance.py

@@ -27,7 +27,7 @@
 from datetime import datetime, timedelta
 from functools import partial
 from tempfile import NamedTemporaryFile
-from typing import IO, TYPE_CHECKING, Any, Dict, Iterable, List, NamedTuple, Optional, Tuple, Union
+from typing import IO, TYPE_CHECKING, Any, Callable, Dict, Iterable, List, NamedTuple, Optional, Tuple, Union
 from urllib.parse import quote
 
 import dill
@@ -1783,17 +1783,16 @@ def get_template_context(self, session: Session = None, ignore_param_exceptions:
         if conf.getboolean('core', 'dag_run_conf_overrides_params'):
             self.overwrite_params_with_dag_run_conf(params=params, dag_run=dag_run)
 
-        interval_start = dag.get_run_data_interval(dag_run).start
-        ds = interval_start.strftime('%Y-%m-%d')
+        logical_date = timezone.coerce_datetime(self.execution_date)
+        ds = logical_date.strftime('%Y-%m-%d')
+        ds_nodash = ds.replace('-', '')
+        ts = logical_date.isoformat()
+        ts_nodash = logical_date.strftime('%Y%m%dT%H%M%S')
+        ts_nodash_with_tz = ts.replace('-', '').replace(':', '')
 
         # Now validates Params and convert them into a simple dict
         task.params = params.validate()
 
-        ds_nodash = ds.replace('-', '')
-        ts = interval_start.isoformat()
-        ts_nodash = interval_start.strftime('%Y%m%dT%H%M%S')
-        ts_nodash_with_tz = ts.replace('-', '').replace(':', '')
-
         @cache  # Prevent multiple database access.
         def _get_previous_dagrun_success() -> Optional["DagRun"]:
             return self.get_previous_dagrun(state=State.SUCCESS, session=session)
@@ -1905,14 +1904,23 @@ def get(
 
         # Create lazy proxies for deprecated stuff.
 
-        def deprecated_proxy(func, *, key, replacement=None) -> lazy_object_proxy.Proxy:
+        def deprecated_proxy(
+            func: Callable[[], Any],
+            *,
+            key: str,
+            replacements: Optional[List[str]] = None,
+        ) -> lazy_object_proxy.Proxy:
             def deprecated_func():
                 message = (
                     f"Accessing {key!r} from the template is deprecated and "
                     f"will be removed in a future version."
                 )
-                if replacement:
-                    message += f" Please use {replacement!r} instead."
+                if replacements:
+                    display_except_last = ", ".join(repr(r) for r in replacements[:-1])
+                    if display_except_last:
+                        message += f" Please use {display_except_last} or {replacements[-1]!r} instead."
+                    else:
+                        message += f" Please use {replacements[-1]!r} instead."
                 warnings.warn(message, DeprecationWarning)
                 return func()
 
@@ -1990,22 +1998,23 @@ def get_prev_ds_nodash() -> Optional[str]:
             'ds': ds,
             'ds_nodash': ds_nodash,
             'execution_date': deprecated_proxy(
-                lambda: timezone.coerce_datetime(self.execution_date),
+                lambda: logical_date,
                 key='execution_date',
-                replacement='data_interval_start',
+                replacements=['logical_date', 'data_interval_start'],
             ),
             'inlets': task.inlets,
+            'logical_date': logical_date,
             'macros': macros,
-            'next_ds': deprecated_proxy(get_next_ds, key="next_ds", replacement="data_interval_end | ds"),
+            'next_ds': deprecated_proxy(get_next_ds, key="next_ds", replacements=["data_interval_end | ds"]),
             'next_ds_nodash': deprecated_proxy(
                 get_next_ds_nodash,
                 key="next_ds_nodash",
-                replacement="data_interval_end | ds_nodash",
+                replacements=["data_interval_end | ds_nodash"],
             ),
             'next_execution_date': deprecated_proxy(
                 get_next_execution_date,
                 key='next_execution_date',
-                replacement='data_interval_end',
+                replacements=['data_interval_end'],
             ),
             'outlets': task.outlets,
             'params': task.params,
@@ -2017,7 +2026,7 @@ def get_prev_ds_nodash() -> Optional[str]:
             'prev_execution_date_success': deprecated_proxy(
                 lambda: self.get_previous_execution_date(state=State.SUCCESS, session=session),
                 key='prev_execution_date_success',
-                replacement='prev_data_interval_start_success',
+                replacements=['prev_data_interval_start_success'],
             ),
             'prev_start_date_success': lazy_object_proxy.Proxy(get_prev_start_date_success),
             'run_id': self.run_id,

docs/apache-airflow/templates-ref.rst

@@ -38,15 +38,14 @@ Variable                                    Description
 ==========================================  ====================================
 ``{{ data_interval_start }}``               Start of the data interval (`pendulum.DateTime`_).
 ``{{ data_interval_end }}``                 End of the data interval (`pendulum.DateTime`_).
-``{{ ds }}``                                Start of the data interval as ``YYYY-MM-DD``.
-                                            Same as ``{{ data_interval_start | ds }}``.
-``{{ ds_nodash }}``                         Start of the data interval as ``YYYYMMDD``.
-                                            Same as ``{{ data_interval_start | ds_nodash }}``.
-``{{ ts }}``                                Same as ``{{ data_interval_start | ts }}``.
+``{{ ds }}``                                The DAG run's logical date as ``YYYY-MM-DD``.
+                                            Same as ``{{ dag_run.logical_date | ds }}``.
+``{{ ds_nodash }}``                         Same as ``{{ dag_run.logical_date | ds_nodash }}``.
+``{{ ts }}``                                Same as ``{{ dag_run.logical_date | ts }}``.
                                             Example: ``2018-01-01T00:00:00+00:00``.
-``{{ ts_nodash_with_tz }}``                 Same as ``{{ data_interval_start | ts_nodash_with_tz }}``.
+``{{ ts_nodash_with_tz }}``                 Same as ``{{ dag_run.logical_date | ts_nodash_with_tz }}``.
                                             Example: ``20180101T000000+0000``.
-``{{ ts_nodash }}``                         Same as ``{{ data_interval_start | ts_nodash }}``.
+``{{ ts_nodash }}``                         Same as ``{{ dag_run.logical_date | ts_nodash }}``.
                                             Example: ``20180101T000000``.
 ``{{ prev_data_interval_start_success }}``  Start of the data interval from prior successful DAG run
                                             (`pendulum.DateTime`_ or ``None``).
@@ -78,13 +77,18 @@ Variable                                    Description
                                             subcommand.
 ==========================================  ====================================
 
+.. note::
+
+    The DAG run's logical date, and values derived from it, such as ``ds`` and
+    ``ts``, **should not** be considered unique in a DAG. Use ``run_id`` instead.
+
 The following variables are deprecated. They are kept for backward compatibility, but you should convert
 existing code to use other variables instead.
 
 =====================================   ====================================
 Deprecated Variable                     Description
 =====================================   ====================================
-``{{ execution_date }}``                the execution date (logical date), same as ``logical_date``
+``{{ execution_date }}``                the execution date (logical date), same as ``dag_run.logical_date``
 ``{{ next_execution_date }}``           the next execution date (if available) (`pendulum.DateTime`_)
                                         if ``{{ execution_date }}`` is ``2018-01-01 00:00:00`` and
                                         ``schedule_interval`` is ``@weekly``, ``{{ next_execution_date }}``

tests/utils/log/test_log_reader.py

@@ -16,16 +16,21 @@
 # under the License.
 
 import copy
+import datetime
 import logging
 import os
 import sys
 import tempfile
 from unittest import mock
 
+import pendulum
 import pytest
 
 from airflow import settings
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
+from airflow.models import DagRun
+from airflow.operators.python import PythonOperator
+from airflow.timetables.base import DataInterval
 from airflow.utils import timezone
 from airflow.utils.log.log_reader import TaskLogReader
 from airflow.utils.log.logging_mixin import ExternalLoggingMixin
@@ -87,7 +92,7 @@ def prepare_log_files(self, log_dir):
                 f.flush()
 
     @pytest.fixture(autouse=True)
-    def prepare_db(self, session, create_task_instance):
+    def prepare_db(self, create_task_instance):
         ti = create_task_instance(
             dag_id=self.DAG_ID,
             task_id=self.TASK_ID,
@@ -238,3 +243,50 @@ 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_interval="@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
+        # execution date, to check if they correctly use different log files.
+        scheduled_dagrun: DagRun = dag_maker.create_dagrun(
+            run_type=DagRunType.SCHEDULED,
+            execution_date=start,
+            data_interval=DataInterval(start, end),
+        )
+        manual_dagrun: DagRun = dag_maker.create_dagrun(
+            run_type=DagRunType.MANUAL,
+            execution_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)