DAG - On error abandon currently running tasks

# Summary

It would be extremely useful if a DAG could give up if argo already knows the end result will be failure. Currently the failFast option will stop scheduling dependencies if there has been a failure but if there are concurrently running tasks and one of them fails the others will be left to finish running before the workflow will finish.

# Use Cases

I have a DAG workflow which uses RabbitMQ to pass work items between tasks. This works well with the large number of items I need to process. However if the message consumer part dies or is evicted (for example due to an OOM event) then the workflow will stall and will never finish without manual intervention.

While it would be possible to invent a way to detect this kind of failure in my message consumers and have them give up, perhaps involving a a regular heartbeat message between the message producer and consumer, Argo already knows all hope is lost so it would be much cleaner if it could just kill everything off and call it a day.

I discussed this with @sarabala1979 on one of your office hours zoom calls a couple of weeks ago.

Example workflow:

This is an extremely simple and contrived example of the kind of hang I'm hoping to avoid. The message producer pretends to have been evicted due to an out of memory event but it equally well might have crashed due to a bug. The end result is that the message consumer ends up hanging around forever waiting for messages which will never arrive.

```
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: hang-
spec:
entrypoint: hang
templates:
- name: rabbitmq
daemon: true
container:
image: rabbitmq:3
startupProbe:
exec:
command: ['su', 'rabbitmq', '-c', '/opt/rabbitmq/sbin/rabbitmq-diagnostics status']

- name: generate-messages
inputs:
parameters:
- name: amqp-url
script:
image: ubuntu:20.04
command: [bash]
source: |
echo Preparing to publish some messages
apt-get update
apt-get install -y amqp-tools
amqp-declare-queue --url={{inputs.parameters.amqp-url}} -q workitems -d
for i in $(seq 1 10); do amqp-publish --url={{inputs.parameters.amqp-url}} -r workitems -p -b "Work item: $i"; done
# --== Pretend we got evicted ==--
echo Oh no! We caused an OOM
exit 137
# --== End ==--
amqp-publish --url={{inputs.parameters.amqp-url}} -r workitems -p -b "FINISHED"

- name: consume-messages
inputs:
parameters:
- name: amqp-url
script:
image: ubuntu:20.04
command: [bash]
source: |
echo Preparing to consume some messages
apt-get update
apt-get install -y amqp-tools
amqp-declare-queue --url={{inputs.parameters.amqp-url}} -q workitems -d
echo Foo
echo Bar
while : ; do
msg="$(amqp-consume --url={{inputs.parameters.amqp-url}} -q workitems -c 1 cat)"
[ "$msg" == "FINISHED" ] && break
echo "Pretending to work on: $msg"
sleep 1
done
echo FINISHED!

- name: hang
dag:
tasks:
- name: mq
template: rabbitmq
- name: message-producer
dependencies: [mq]
arguments:
parameters:
- name: "amqp-url"
value: "amqp://guest:guest@{{tasks.mq.ip}}:5672"
template: generate-messages
- name: message-consumer
dependencies: [mq]
arguments:
parameters:
- name: "amqp-url"
value: "amqp://guest:guest@{{tasks.mq.ip}}:5672"
template: consume-messages

```

What I'd like to see happen in this case is for Argo to kill the message consumer task once it's detected the failure of the message producer task. ie, once Argo knows that the dag has failed, give up. I doubt everyone wants this behaviour so perhaps it can be requested via an extension to the failFast flag.

If there are steps which have a retryStrategy set, then obviously things should keep running until the retry limit has been reached.

---

**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.

No source files or tests are named. Start by locating the Go workflow-controller entry points for DAG failFast handling and task termination, then trace how retryStrategy affects failure propagation. Done means a configured DAG can stop concurrently running tasks after an unrecoverable failure while allowing retries to reach their limit.

Written by the indexing model from the issue text.