[BUG] Failed_node_id is failure node's id instead of the actual failed node's id

[cjidboon94]

Describe the bug

When using a failure node/task https://docs.flyte.org/en/latest/user_guide/development_lifecycle/failure_node.html for cleanup of a workflow for example, the task receives the input parameters of the workflow + a FlyteError object containing the error message and a failed_node_id. Given that the name of the field is failed_node_id and not failure_node_id, I would expect the value to have the id of the failed node/task instead of the (current) failure_node.

Expected behavior

When printing/accessing a FlyteErrors failed_node_id field in a Failure Node task, I expect to get the failed node id. Not the id of the failure node.

A big nice to have would be to actually get the Task ID in addition to the node id as this gives more context than just the node id.

Additional context to reproduce

Based on documentation example

from flytekit import task, workflow
import typing

@task
def t1():
    raise ValueError("Fail!")

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name} due to {err.message} in {err.failed_node_id}") # Expected it to be n0, but is instead fn0


@workflow(on_failure=clean_up)
def wf(name: str):
    t1()

Screenshots

No response

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes