Ability to clear a specific DAG Run's task instances via REST APIs

Discussed in #23220

^{Originally posted by yashk97 April 25, 2022}
Hi,

My use case is in case multiple DAG Runs fail on some task (not the same one in all of them), I want to individually re-trigger each of these DAG Runs. Currently, I have to rely on the Airflow UI (attached screenshots) where I select the failed task and clear its state (along with the downstream tasks) to re-run from that point. While this works, it becomes tedious if the number of failed DAG runs is huge.

I checked the REST API Documentation and came across the clear Task Instances API with the following URL: /api/v1/dags/{dag_id}/clearTaskInstances

However, it filters task instances of the specified DAG in a given date range.

I was wondering if, for a specified DAG Run, we can clear a task along with its downstream tasks irrespective of the states of the tasks or the DAG run through REST API.

This will give us more granular control over re-running DAGs from the point of failure.

I wonder whether we should overload clearTaskInstances, or have a separate endpoint under the DAG run ID prefix instead.

Having a separate endpoint with the DAG run ID will allow maintaining consistency with the flow to clear the task through the Airflow UI.

The parameters can include the following:

Task ID to be cleared (A single task ID)
Whether to clear upstream/downstream tasks too.
Reset DAG run to set its state to running.

Could I take this issue and try fixing it? This is the first time I am working on the REST APIs part.

@Bowrna

I have assigned it to you @Bowrna

I'm sorry, but I received an email about a comment by Rafael Carrasco (have attached the screenshot). I wasn't able to locate the same on the issue page. I think it would be better to clear things up so that two people don't end up working on the same thing.

@yashk97

@yashk97 Please disregard that message, as I deleted it. After reviewing, the endpoint in the other issue I am assigned is for a single task instance and this issue includes a date range. They are not related, unfortunately. Apologies for sending that previous message.

@erdos2n

@erdos2n Thanks for the clarification.

@erdos2n

@erdos2n As mentioned in this issue #23246 the above task is also very similar.
Here the already existing api /api/v1/dags/{dag_id}/clearTaskInstances filters the dag run instances based on the date.

The above issue asks for API endpoint with dag_run_id rather than date filter passed as param in /api/v1/dags/{dag_id}/clearTaskInstances

If I am right the endpoint expected for this issue is :
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/clearTaskInstances with param supported :
task_id
include_upstream
include_downstream
include_future
include_past
reset_dag_runs

cc: @yashk97

I think the code for the other issue that you are working on will work for this case too. @erdos2n except that you pass the task_id as part of the endpoint.

Can anybody tell me if my understanding is correct? Is it still relevant to work on this issue or is it the same as #23246 ?

So clear and and mark state are two distinct actions so I would say #23246 is different. This issue is to replace /clear in the webserver

I believe the URL should then be /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/clear (ie task_id should be in the the url not the params)

Two additonal params will be needed: recursive and only_failed.

Like the existing task instances clear we should include dry_run which will return json of the task instances that will be affected.

#23419 is more related, but will clear the entire dag run

@bbovenzi

This issue is to replace /clear in the webserver

@bbovenzi This API has map_index and execution_date as param. Do I have to include those in params too?

@bbovenzi

@bbovenzi This API has map_index and execution_date as param. Do I have to include those in params too?

Yes, we will need to string together even multiple map_indexs. But we can use dag_run_id instead of execution_date

@bbovenzi

@bbovenzi @Bowrna I'm leaning towards updating the existing clearTaskInstances endpoint to optionally accept dag_run_id. This is a duplication of the existing endpoint (/api/v1/dags/{dag_id}/clearTaskInstances). There's nothing new that we want to introduce with the new endpoint. We could update the existing one to either accept execution_date or dagrun_id. I believe that's what the UI is doing. The current clear task instance endpoint can clear a specific dagrun given the execution date

@ephraimbuddy

@ephraimbuddy @bbovenzi How to pass map_index in this case of using /api/v1/dags/{dag_id}/clearTaskInstances endpoint?
Because in /api/v1/dags/{dag_id}/clearTaskInstances we can get array of task_id. If we get an array of task_id with map_index, how to ensure to which task_id that map_index belong?

Any update about this?

Hmm yes, thats tricky. maybe we still do want an endpoint just for a single task instance where we can specify map_indices while /api/v1/dags/{dag_id}/clearTaskInstances will just assume all map_indexes of that task_id and run_id

We can use the method we used here:

airflow/airflow/models/dag.py

Line 1381 in 325444e

task_ids: Optional[Collection[Union[str, Tuple[str, int]]]],

Get a tuple of task_ids with the corresponding map_index instead of repeating ourselves on a new endpoint?

We can use the method we used here:

airflow/airflow/models/dag.py

Line 1381 in 325444e

task_ids: Optional[Collection[Union[str, Tuple[str, int]]]],

Get a tuple of task_ids with the corresponding map_index instead of repeating ourselves on a new endpoint?

I will think about this more

@ephraimbuddy

@ephraimbuddy @bbovenzi Could I proceed with getting a tuple of task_ids with the corresponding map_index?

@ephraimbuddy

@ephraimbuddy @bbovenzi Could I proceed with getting a tuple of task_ids with the corresponding map_index?

The other PR has been merged: #23451

@ephraimbuddy

@ephraimbuddy the PR merged #23451 doesn't have task_id as part of the API. Am I missing to understand any point? I thought this particular issue is about clearing DAG with task_ids and corresponding map_index and dag_run_id as one of the params. Can you please clarify?

@ephraimbuddy

@ephraimbuddy that PR was to clear an entire DAG run. This issue is to clear specific task instances. Maybe we need a more complex request body:

{
  instances: [
    { task_id: 'task', map_index: 3 },
    { task_id: 'other_task' } // default to map_index = -1
  ]
}

@bbovenzi

@bbovenzi Currently the below task_ids format is acceptable as suggested by @ephraimbuddy .

{
  task_ids: [
    ('task', 3),
     'other_task' ,// default to map_index = -1
  ]
}

airflow/airflow/models/dag.py

Line 1381 in 325444e

task_ids: Optional[Collection[Union[str, Tuple[str, int]]]],

Do we have to explicitly get a request body like below with task_id and map_index as key name in list of dict ?

{
  instances: [
    { task_id: 'task', map_index: 3 },
    { task_id: 'other_task' } // default to map_index = -1
  ]

@Bowrna

@Bowrna oh yes sorry, the task_ids format is good

@bbovenzi

@bbovenzi Does this API can accept more than one task instance for a specific dag run to clear?

@bbovenzi

@bbovenzi Does this API can accept more than one task instance for a specific dag run to clear?

Yes, because we are modifying the existing /clearTaskInstances endpoint. Since that is already plural, we need to stay consistent.

Tell me, is there a deadline for the task? relevant for me too

Navigation Menu

Repository navigation

Ability to clear a specific DAG Run's task instances via REST APIs #23227

Discussed in #23220

5 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Ability to clear a specific DAG Run's task instances via REST APIs #23227

Description

Discussed in #23220

Activity

uranusjr commented on Apr 26, 2022

yashk97 commented on Apr 27, 2022

Bowrna commented on May 2, 2022

ephraimbuddy commented on May 2, 2022

yashk97 commented on May 2, 2022

erdos2n commented on May 2, 2022

yashk97 commented on May 2, 2022

Bowrna commented on May 4, 2022

bbovenzi commented on May 4, 2022

Bowrna commented on May 5, 2022

bbovenzi commented on May 5, 2022

ephraimbuddy commented on May 9, 2022

5 remaining items

Bowrna commented on May 10, 2022

Bowrna commented on May 15, 2022

bbovenzi commented on May 16, 2022

ephraimbuddy commented on May 17, 2022

ephraimbuddy commented on May 17, 2022

Bowrna commented on May 30, 2022

ephraimbuddy commented on May 30, 2022

Bowrna commented on May 30, 2022

bbovenzi commented on May 30, 2022

Bowrna commented on Jun 1, 2022

bbovenzi commented on Jun 1, 2022

Bowrna commented on Jun 1, 2022

bbovenzi commented on Jun 3, 2022

zakaryanla commented on Jul 7, 2022

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions