Enable already existing "task.scheduled_duration" metric

airflow/api_connexion/schemas/task_instance_schema.py

@@ -60,6 +60,7 @@ class Meta:
     priority_weight = auto_field()
     operator = auto_field()
     queued_dttm = auto_field(data_key="queued_when")
+    scheduled_dttm = auto_field(data_key="scheduled_when")
     pid = auto_field()
     executor = auto_field()
     executor_config = auto_field()
@@ -102,6 +103,7 @@ class Meta:
     priority_weight = auto_field()
     operator = auto_field()
     queued_dttm = auto_field(data_key="queued_when")
+    scheduled_dttm = auto_field(data_key="scheduled_when")
     pid = auto_field()
     executor = auto_field()
     executor_config = auto_field()

airflow/api_fastapi/core_api/datamodels/task_instances.py

@@ -64,6 +64,7 @@ class TaskInstanceResponse(BaseModel):
     priority_weight: int | None
     operator: str | None
     queued_dttm: datetime | None = Field(alias="queued_when")
+    scheduled_dttm: datetime | None = Field(alias="scheduled_when")
     pid: int | None
     executor: str | None
     executor_config: Annotated[str, BeforeValidator(str)]
@@ -147,6 +148,7 @@ class TaskInstanceHistoryResponse(BaseModel):
     priority_weight: int | None
     operator: str | None
     queued_dttm: datetime | None = Field(alias="queued_when")
+    scheduled_dttm: datetime | None = Field(alias="scheduled_when")
     pid: int | None
     executor: str | None
     executor_config: Annotated[str, BeforeValidator(str)]

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -9635,6 +9635,12 @@ components:
             format: date-time
           - type: 'null'
           title: Queued When
+        scheduled_when:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Scheduled When
         pid:
           anyOf:
           - type: integer
@@ -9669,6 +9675,7 @@ components:
       - priority_weight
       - operator
       - queued_when
+      - scheduled_when
       - pid
       - executor
       - executor_config
@@ -9762,6 +9769,12 @@ components:
             format: date-time
           - type: 'null'
           title: Queued When
+        scheduled_when:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Scheduled When
         pid:
           anyOf:
           - type: integer
@@ -9820,6 +9833,7 @@ components:
       - priority_weight
       - operator
       - queued_when
+      - scheduled_when
       - pid
       - executor
       - executor_config

airflow/jobs/scheduler_job_runner.py

@@ -775,7 +775,7 @@ def process_executor_events(
                 "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s, map_index=%s, "
                 "run_start_date=%s, run_end_date=%s, "
                 "run_duration=%s, state=%s, executor=%s, executor_state=%s, try_number=%s, max_tries=%s, "
-                "pool=%s, queue=%s, priority_weight=%d, operator=%s, queued_dttm=%s, "
+                "pool=%s, queue=%s, priority_weight=%d, operator=%s, queued_dttm=%s, scheduled_dttm=%s,"
                 "queued_by_job_id=%s, pid=%s"
             )
             cls.logger().info(
@@ -797,6 +797,7 @@ def process_executor_events(
                 ti.priority_weight,
                 ti.operator,
                 ti.queued_dttm,
+                ti.scheduled_dttm,
                 ti.queued_by_job_id,
                 ti.pid,
             )
@@ -1808,6 +1809,7 @@ def _reschedule_stuck_task(self, ti: TaskInstance, session: Session):
             .values(
                 state=TaskInstanceState.SCHEDULED,
                 queued_dttm=None,
+                scheduled_dttm=timezone.utcnow(),
             )
             .execution_options(synchronize_session=False)
         )
@@ -1962,6 +1964,7 @@ def check_trigger_timeouts(
                         state=TaskInstanceState.SCHEDULED,
                         next_method=TRIGGER_FAIL_REPR,
                         next_kwargs={"error": TriggerFailureReason.TRIGGER_TIMEOUT},
+                        scheduled_dttm=timezone.utcnow(),
                         trigger_id=None,
                     )
                 ).rowcount

airflow/migrations/versions/0057_3_0_0_add_new_task_instance_field_scheduled_.py

@@ -0,0 +1,64 @@
+#
+# 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.
+
+"""
+add new task_instance field scheduled_dttm.
+
+Revision ID: 33b04e4bfa19
+Revises: 8ea135928435
+Create Date: 2025-01-22 11:22:01.272681
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.utils.sqlalchemy import UtcDateTime
+
+# revision identifiers, used by Alembic.
+revision = "33b04e4bfa19"
+down_revision = "8ea135928435"
+branch_labels = None
+depends_on = None
+airflow_version = "3.0.0"
+
+
+def upgrade():
+    """Apply add new task_instance field scheduled_dttm."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("task_instance", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("scheduled_dttm", UtcDateTime(timezone=True), nullable=True))
+
+    with op.batch_alter_table("task_instance_history", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("scheduled_dttm", UtcDateTime(timezone=True), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    """Unapply add new task_instance field scheduled_dttm."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("task_instance_history", schema=None) as batch_op:
+        batch_op.drop_column("scheduled_dttm")
+
+    with op.batch_alter_table("task_instance", schema=None) as batch_op:
+        batch_op.drop_column("scheduled_dttm")
+
+    # ### end Alembic commands ###

airflow/models/dag.py

@@ -1666,6 +1666,7 @@ def add_logger_if_needed(ti: TaskInstance):
                     if s.state != TaskInstanceState.UP_FOR_RESCHEDULE:
                         s.try_number += 1
                     s.state = TaskInstanceState.SCHEDULED
+                    s.scheduled_dttm = timezone.utcnow()
                 session.commit()
                 # triggerer may mark tasks scheduled so we read from DB
                 all_tis = set(dr.get_task_instances(session=session))

airflow/models/dagrun.py

@@ -1662,6 +1662,7 @@ def schedule_tis(
                     )
                     .values(
                         state=TaskInstanceState.SCHEDULED,
+                        scheduled_dttm=timezone.utcnow(),
                         try_number=case(
                             (
                                 or_(TI.state.is_(None), TI.state != TaskInstanceState.UP_FOR_RESCHEDULE),

airflow/models/taskinstance.py

@@ -824,6 +824,7 @@ def _set_ti_attrs(target, source, include_dag_run=False):
     target.operator = source.operator
     target.custom_operator_name = source.custom_operator_name
     target.queued_dttm = source.queued_dttm
+    target.scheduled_dttm = source.scheduled_dttm
     target.queued_by_job_id = source.queued_by_job_id
     target.last_heartbeat_at = source.last_heartbeat_at
     target.pid = source.pid
@@ -1712,6 +1713,7 @@ class TaskInstance(Base, LoggingMixin):
     operator = Column(String(1000))
     custom_operator_name = Column(String(1000))
     queued_dttm = Column(UtcDateTime)
+    scheduled_dttm = Column(UtcDateTime)
     queued_by_job_id = Column(Integer)
 
     last_heartbeat_at = Column(UtcDateTime)
@@ -2705,23 +2707,24 @@ def emit_state_change_metric(self, new_state: TaskInstanceState) -> None:
             timing = timezone.utcnow() - self.queued_dttm
         elif new_state == TaskInstanceState.QUEUED:
             metric_name = "scheduled_duration"
-            if self.start_date is None:
-                # This check does not work correctly before fields like `scheduled_dttm` are implemented.
-                # TODO: Change the level to WARNING once it's viable.
-                # see #30612 #34493 and #34771 for more details
-                self.log.debug(
+            if self.scheduled_dttm is None:
+                self.log.warning(
                     "cannot record %s for task %s because previous state change time has not been saved",
                     metric_name,
                     self.task_id,
                 )
                 return
-            timing = timezone.utcnow() - self.start_date
+            timing = timezone.utcnow() - self.scheduled_dttm
         else:
             raise NotImplementedError("no metric emission setup for state %s", new_state)
 
         # send metric twice, once (legacy) with tags in the name and once with tags as tags
         Stats.timing(f"dag.{self.dag_id}.{self.task_id}.{metric_name}", timing)
-        Stats.timing(f"task.{metric_name}", timing, tags={"task_id": self.task_id, "dag_id": self.dag_id})
+        Stats.timing(
+            f"task.{metric_name}",
+            timing,
+            tags={"task_id": self.task_id, "dag_id": self.dag_id, "queue": self.queue},
+        )
 
     def clear_next_method_args(self) -> None:
         """Ensure we unset next_method and next_kwargs to ensure that any retries don't reuse them."""

airflow/models/taskinstancehistory.py

@@ -77,6 +77,7 @@ class TaskInstanceHistory(Base):
     operator = Column(String(1000))
     custom_operator_name = Column(String(1000))
     queued_dttm = Column(UtcDateTime)
+    scheduled_dttm = Column(UtcDateTime)
     queued_by_job_id = Column(Integer)
     pid = Column(Integer)
     executor = Column(String(1000))

airflow/models/trigger.py

@@ -281,6 +281,7 @@ def submit_failure(cls, trigger_id, exc=None, session: Session = NEW_SESSION) ->
             task_instance.trigger_id = None
             # Finally, mark it as scheduled so it gets re-queued
             task_instance.state = TaskInstanceState.SCHEDULED
+            task_instance.scheduled_dttm = timezone.utcnow()
 
     @classmethod
     @provide_session

airflow/triggers/base.py

@@ -25,6 +25,7 @@
 
 from airflow.callbacks.callback_requests import TaskCallbackRequest
 from airflow.callbacks.database_callback_sink import DatabaseCallbackSink
+from airflow.utils import timezone
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.state import TaskInstanceState
@@ -172,6 +173,7 @@ def handle_submit(self, *, task_instance: TaskInstance, session: Session = NEW_S
 
         # Set the state of the task instance to scheduled
         task_instance.state = TaskInstanceState.SCHEDULED
+        task_instance.scheduled_dttm = timezone.utcnow()
 
 
 class BaseTaskEndEvent(TriggerEvent):

airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -4667,6 +4667,18 @@ export const $TaskInstanceHistoryResponse = {
       ],
       title: "Queued When",
     },
+    scheduled_when: {
+      anyOf: [
+        {
+          type: "string",
+          format: "date-time",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Scheduled When",
+    },
     pid: {
       anyOf: [
         {
@@ -4715,6 +4727,7 @@ export const $TaskInstanceHistoryResponse = {
     "priority_weight",
     "operator",
     "queued_when",
+    "scheduled_when",
     "pid",
     "executor",
     "executor_config",
@@ -4882,6 +4895,18 @@ export const $TaskInstanceResponse = {
       ],
       title: "Queued When",
     },
+    scheduled_when: {
+      anyOf: [
+        {
+          type: "string",
+          format: "date-time",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Scheduled When",
+    },
     pid: {
       anyOf: [
         {
@@ -4979,6 +5004,7 @@ export const $TaskInstanceResponse = {
     "priority_weight",
     "operator",
     "queued_when",
+    "scheduled_when",
     "pid",
     "executor",
     "executor_config",

airflow/ui/openapi-gen/requests/types.gen.ts

@@ -1223,6 +1223,7 @@ export type TaskInstanceHistoryResponse = {
   priority_weight: number | null;
   operator: string | null;
   queued_when: string | null;
+  scheduled_when: string | null;
   pid: number | null;
   executor: string | null;
   executor_config: string;
@@ -1253,6 +1254,7 @@ export type TaskInstanceResponse = {
   priority_weight: number | null;
   operator: string | null;
   queued_when: string | null;
+  scheduled_when: string | null;
   pid: number | null;
   executor: string | null;
   executor_config: string;

airflow/utils/db.py

@@ -94,7 +94,7 @@ class MappedClassProtocol(Protocol):
     "2.9.2": "686269002441",
     "2.10.0": "22ed7efa9da2",
     "2.10.3": "5f2621c13b39",
-    "3.0.0": "8ea135928435",
+    "3.0.0": "33b04e4bfa19",
 }
 
 

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-ff7265e5bc09d6b46d8e95f0c247b3dc5b1262451ab128c711888ffafa21c9db
+829be35e333798f7c33c5fe0130ed12fad481c92145abc398ae23b815dd7b6ed