Add option to clear only failed tasks in a graph view

[tirkarthi]

This PR adds a new option to clear only failed tasks in a dag run. The backend API will accept a new parameter only_failed which by default is False for backwards compatibility.

closes: #38031
related: #38031

[tirkarthi]

Nit: I have added "clear only failed tasks" but can make it "clear failed tasks" as per original issue since when the failed task is cleared the downstream tasks that have upstream_failed state are also cleared.

[dirrao]

Nice Feature. LGTM.

[eladkal]

failed tasks are just failed or also upstream_failed?
I am not sure if the request in the issue is for status failed or actually to tasks that are in terminal error state.
I think it's odd to clear only failed without upstream_failed but if people find it useful then we should have it

[bbovenzi]

This is the equivalent of "Clear task" with the parameters upstream+downstream+only_failed set to true, right?

[tirkarthi]

It would clear the failed tasks for the selected run (wit option to select downstream or not).

@eladkal I overlooked this part in the original issue. First I thought passing failed only clears failed tasks but it also cleared upstream failed. I found it useful since after the failed tasks is cleared and it succeeds having the downstream task might result in an extra action if not for the behaviour. Then I found the code as below where the logic is present. The request was also for dags with large number of tasks where downstream=True requirement means this will only clear upstream_failed downstream tasks and not success downstream tasks. Adding downstream or not option will change this into a modal similar to the clear task button on a dagrun level.

@Pad71 let me know if I missed something.

airflow/airflow/models/dag.py

Lines 2333 to 2334 in 3840ec6

	if only_failed:
	state += [TaskInstanceState.FAILED, TaskInstanceState.UPSTREAM_FAILED]

[tirkarthi]

This is the equivalent of "Clear task" with the parameters upstream+downstream+only_failed set to true, right?

Yes, as per my understanding of the code.

[Pad71]

It would clear the failed tasks for the selected run (wit option to select downstream or not).

@eladkal I overlooked this part in the original issue. First I thought passing failed only clears failed tasks but it also cleared upstream failed. I found it useful since after the failed tasks is cleared and it succeeds having the downstream task might result in an extra action if not for the behaviour. Then I found the code as below where the logic is present. The request was also for dags with large number of tasks where downstream=True requirement means this will only clear upstream_failed downstream tasks and not success downstream tasks. Adding downstream or not option will change this into a modal similar to the clear task button on a dagrun level.

@Pad71 let me know if I missed something.

airflow/airflow/models/dag.py

Lines 2333 to 2334 in 3840ec6

	if only_failed:
	state += [TaskInstanceState.FAILED, TaskInstanceState.UPSTREAM_FAILED]

@tirkarthi Yeah, the requirement is - clear all failed tasks in selected dag run and clear their downstream tasks in upstream_failed state.

[tirkarthi]

Thanks @Pad71 , the PR implements the requirement.

[Pad71]

Thanks @Pad71 , the PR implements the requirement.

Thank you @tirkarthi and all involved! Good job.

if you guys have a time to check this discussion, it would be great.
#38181

[jscheffl]

Looks good, LGTM!