[Core feature] Flytekit plugin termination hook

[hajapy]

Motivation: Why do you think this is important?

As a flytekit plugin author, I would like to be able to define cleanup actions (in python) that should be run when a flyte task or workflow is aborted (or terminates unexpectedly). For example, this could be used to let the plugin cancel long-running operations in external services if a user aborts a workflow execution.

Goal: What should the final outcome look like, ideally?

The flytekit PythonTask interface could expose a new method (analogous to execute) that would be called upon termination of the task via an external signal.

Describe alternatives you've considered

An effort was made to use the stdlib's signal module to register handlers from the custom plugin module, but this didn't ultimately work (the signal seemed to be intercepted by pyflyte-execute).

Propose: Link/Inline OR Additional context

This suggestion was discussed in a flyte slack thread.

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

• [X] Yes

Have you read the Code of Conduct?

• [X] Yes