States of the time field corresponding to ti.state is not clear

[kzosabe]

What do you see as an issue?

There was a problem with the emitting of statistics logs regarding execution time introduced in #30612 .

Problem details:

• Create metrics to track Scheduled->Queued->Running task state transition times #30612 (comment)
• Suppress false warning when TI state is QUEUED and TI doesn't have a start_date #34771 (comment)

When issuing such statistical indicators of execution time, it is important to first clarify what the corresponding time fields are for each state, and to make recognition and implementation consistent.

The time fields corresponding to the transitions to each state of TIs, as I read the code, are as follows:

SCHEDULED: (no time field for this state currently exists)
QUEUED: queued_dttm
RUNNING: start_date
SUCCESS: end_date
FAILED: end_date
SKIPPED: end_date
UPSTREAM_FAILED: end_date
REMOVED: end_date
UP_FOR_RETRY: end_date
UP_FOR_RESCHEDULE: (uncertain)
RESTARTING: (uncertain)
DEFERRED: (uncertain)

So as I understood, the matrix of states and TI dates should look like this (corrections welcome):

---

SCHEDULED         - queued_dttm: no,  start_date: no,  end_date: no
QUEUED            - queued_dttm: yes, start_date: no,  end_date: no
RUNNING           - queued_dttm: yes, start_date: yes, end_date: no
SUCCESS           - queued_dttm: yes, start_date: yes, end_date: yes(1)
FAILED            - queued_dttm: yes, start_date: yes, end_date: yes(1)
SKIPPED           - queued_dttm: yes, start_date: yes, end_date: yes(1)
UPSTREAM_FAILED   - queued_dttm: yes, start_date: yes, end_date: yes(1)
REMOVED           - queued_dttm: yes, start_date: yes, end_date: yes(1)
UP_FOR_RETRY      - queued_dttm: yes, start_date: yes, end_date: yes(1)
UP_FOR_RESCHEDULE - queued_dttm: yes, start_date: yes, end_date: yes?(2)
RESTARTING        - queued_dttm: no?, start_date: no?, end_date: no?(3)
DEFERRED          - queued_dttm: no?, start_date: no?, end_date: no?(4)

(1)

airflow/airflow/models/taskinstance.py

Lines 1894 to 1896 in fce3a58

	if ti.state in State.finished or ti.state == TaskInstanceState.UP_FOR_RETRY:
	ti.end_date = ti.end_date or current_time
	ti.duration = (ti.end_date - ti.start_date).total_seconds()

and

airflow/airflow/utils/state.py

Lines 160 to 168 in 33e5d03

	finished: frozenset[TaskInstanceState] = frozenset(
	[
	TaskInstanceState.SUCCESS,
	TaskInstanceState.FAILED,
	TaskInstanceState.SKIPPED,
	TaskInstanceState.UPSTREAM_FAILED,
	TaskInstanceState.REMOVED,
	]
	)

(2)

airflow/airflow/models/taskinstance.py

Line 2829 in fce3a58

self.end_date = timezone.utcnow()

Is this state being handled this way because the task is interpreted as finished but waiting to be picked up by the scheduler?

(3)
https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/tasks.html#task-instances
One interpretation of the above diagram should look like this.

If RESTARTING and UP_FOR_RETRY are interpreted as pre-scheduled states, then all times should be no.
If we interpret them as finished states like (2), then all times should be yes.
We could also treat these as having indeterminate time field states, but that would probably not be a good idea.

(4)
Undocumented, but commented as Deferrable operator waiting on a trigger on this line.

airflow/airflow/utils/state.py

Line 59 in 33e5d03

DEFERRED = "deferred" # Deferrable operator waiting on a trigger

And these lines imply that this state is a pre-scheduled state

airflow/airflow/models/trigger.py

Lines 191 to 201 in db3181c

	TaskInstance.trigger_id == trigger_id, TaskInstance.state == TaskInstanceState.DEFERRED
	)
	):
	# Add the event's payload into the kwargs for the task
	next_kwargs = task_instance.next_kwargs or {}
	next_kwargs["event"] = event.payload
	task_instance.next_kwargs = next_kwargs
	# Remove ourselves as its trigger
	task_instance.trigger_id = None
	# Finally, mark it as scheduled so it gets re-queued
	task_instance.state = TaskInstanceState.SCHEDULED

---

Solving the problem

So it will be great that

• Fill in the matrix as above by community consensus (and write it into the official document if we can)
• Fix codes that do not follow the matrix above
• Add tests or debug logs that can check for incorrect state and time combinations

Is it worth working on this?

In the first, I tried to tackle the set_state issue mentioned in #34771 (comment) .

#38631

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[uranusjr]

DEFERRED is the easy one, it is basically the same as RUNNING from the scheduler’s perspective, except it is waiting for the trigger to continue. For a ti to entered the DEFERRED, it must first be in the RUNNING state (because deferral happens in the worker), so it should have set start_date, but not end_date.

UP_FOR_RESCHEDULE is similar, except it’s done by a (synchronous) sensor in reschedule mode. It is also basically the same as RUNNING except waiting for the scheduler to pick it up again. Since signalling a reschedule is done in the worker, the ti should have set start_date. The situation with end_state is… complicated (this uncertainly is a part of the reason why sensors are disfavoured). Personally I would prefer if it does not set end_date (nor re-set start_date when the scheduler restarts the ti), but it is what it is and we can’t change that without breaking compatibility (and when we can break compatibility in Airflow 3, we probably want to remove synchronous sensors altogether anyway). I think UP_FOR_RESCHEDULE should be considered a special state that is also finished and pre-scheduling and keep its current behaviour.

RESTARTING is also similar (it is actually relative new) in that it’s set after the ti have started, so start_date should have been set. It should keep its start_date until the scheduler picks the ti back up, and end_date should remain null. The same applies for UP_FOR_RETRY. (I think your reference on UP_FOR_RETRY’s end_date is wrong? The code it links does not handle UP_FOR_RETRY, from what I can tell.)

[kzosabe]

@uranusjr Thanks for the explanation!

Just to confirm, does a DEFFERED task really transition directly to RUNNING?
I may be missing something, but I could not find a code path that directly transitions from DEFFERED to RUNNING.
As far as I have investigated, it seems that DEFFERED tasks are first rolled back to SCHEDULED, as in the following test code.

airflow/tests/models/test_trigger.py

Lines 93 to 118 in db3181c

	def test_submit_event(session, create_task_instance):
	"""
	Tests that events submitted to a trigger re-wake their dependent
	task instances.
	"""
	# Make a trigger
	trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
	trigger.id = 1
	session.add(trigger)
	session.commit()
	# Make a TaskInstance that's deferred and waiting on it
	task_instance = create_task_instance(
	session=session, execution_date=timezone.utcnow(), state=State.DEFERRED
	)
	task_instance.trigger_id = trigger.id
	task_instance.next_kwargs = {"cheesecake": True}
	session.commit()
	# Call submit_event
	Trigger.submit_event(trigger.id, TriggerEvent(42), session=session)
	# commit changes made by submit event and expire all cache to read from db.
	session.flush()
	session.expunge_all()
	# Check that the task instance is now scheduled
	updated_task_instance = session.query(TaskInstance).one()
	assert updated_task_instance.state == State.SCHEDULED
	assert updated_task_instance.next_kwargs == {"event": 42, "cheesecake": True}

However, whether there is a direct transition or not, your explanation seems to be correct.

As I was checking your comment, I noticed that, at least the presence or absence of start_date is not be determined by the value of ti.state. So the form of matrix I wrote above may be nonsense.

From reading your comment, in my interpretation, the current desired behavior of start_date is:

• The datetime when the task is first touched by worker is stored.
• And is normally retained no matter how the state changes thereafter (including changes to pre-running state).
• If there is a action that can be regarded as restart , it will be overwritten with the datetime of it.
  • Perhaps we don't want to consider a run from DEFERRED as a restart, but we want to consider a run from UP_FOR_RESCHEDULE as a restart?

Is this understanding correct?

If this is correct, then at least the changes I've made in #38631 should be reasonable in terms of conforming to this interpretation.
Because in the current implementation, the change from QUEUED to SCHEDULED also sets the end_date, which is also clearly contrary to the above.

However, it seems that changes such as putting None in start_date when reverting to a pre-running state, as described in #38631 (comment) , are wrong and should not be done.

I think your reference on UP_FOR_RETRY’s end_date is wrong?

According to my referenced code, ti.set_state(TaskInstanceState.UP_FOR_RETRY) actually sets current time as ti.end_date (because of or ti.state == TaskInstanceState.UP_FOR_RETRY).
However, from reading your commentary, this may also be the wrong behavior.

[uranusjr]

Just to confirm, does a DEFFERED task really transition directly to RUNNING?

I didn’t say that; I only said a task transitions from RUNNING to DEFERRED 😛 A deferred task is indeed set to SCHEDULED by the triggerer, when the trigger is fulfilled, to be handed back to the scheduler; in some cases the handed back ti isn’t even transitioned back to RUNNING at all, but to a finished state directly (when things time out etc.)

the current desired behavior of start_date is [snip]

Yeah I think that’s about right. UP_FOR_RESCHEDULE is technically a restart because a ti set to it will completely go through the worker cycle again. But again it is special (in a bad way) and let’s try to just not change it.

According to my referenced code, ti.set_state(TaskInstanceState.UP_FOR_RETRY) actually sets current time as ti.end_date (because of or ti.state == TaskInstanceState.UP_FOR_RETRY). However, from reading your commentary, this may also be the wrong behavior.

Ah no, my comment on this was wrong, actually. I was confusing UP_FOR_RETRY and UP_FOR_RESCHEDULE. And although we can technically argue UP_FOR_RETRY maybe should not set end_date, it needs to do so because retry in Airflow depends on a task instance to have an end_date.

Also note that there’s currently a discussion on keeping task instance history across tries, so instead of overwriting those dates on retry, we will have a separate row in the database, each having its start and end dates, thus changing the situation a bit.

---

By the way, I want to specifically point out the work you put in this. It is a mess where Airflow sets ti states in various places, and it’s evident you must have traced a lot of code to reach the understanding on the issue. Thanks for putting in the effort.

[shahar1]

@kzosabe Would you still like to work on this issue?

[kzosabe]

No, we can close this. Thanks!