Mark Success but remote task continue running

[dotrungkien3210]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.9.0

What happened?

I had manage a airflow HA celery cluster and manage nearly 150 DAGs, almost working correctly. But only one DAG met trouble. When I mark success, the remote executer trigger by ssh operator continue running. The problem is just one DAG happend, another DAG work correctly although they have same logic

What you think should happen instead?

I think some trouble from code stuck and waiting

How to reproduce

hope you can make muntiple SSHook and process in diffirent server in one DAG

Operating System

Ubuntu 20.04

Versions of Apache Airflow Providers

2.9.0

Deployment

Other

Deployment details

airflow HA celery cluster in 2 node, with 2 scheduler, 2 consumer, 2 worker

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[nathadfield]

@dotrungkien3210 I'm afraid you are going to have to provide more information and context for anyone to be able to assist you in this. If this is a bug then you'll need to demonstrate it in a way that can be reproduced. If this is more of a support/implementation problem then you will be much better served by speaking to the community in Slack.

[dotrungkien3210]

Hi @nathadfield this is full DAG code
command_build = 'bash bin/airflow_build.sh '
checkout_branch = 'master'
build_command = 'bash /home/airflow/platform/check_git/checkgit.sh ' + path_project + ' ' + project_name + ' "' + command_build + '" ' + checkout_branch
build_ope = SSHOperator(
ssh_hook=sshHook,
task_id='crawler-system_build',
command=build_command,
on_failure_callback=create.task_on_failure_callback,
get_pty=True,
do_xcom_push=False,
dag=dag)
command = 'cd ' + path_project + project_name + ' && bash bin/run_migrate_story_book.sh '
run_ope = SSHOperator(
ssh_hook=sshHook,
task_id=dag_name,
do_xcom_push=False,
on_failure_callback=create.task_on_failure_callback,
command=command,
dag=dag)

config = create_dag.getConfig(path="/airflow/crawler-system")

for key, value in config.items():
getConf = SSHOperator(
ssh_hook=sshHook,
task_id='get_config_' + key,
command='echo "' + value + '" > ' + path_project + project_name + '/' + 'properties/' + key,
on_failure_callback=create.task_on_failure_callback,
get_pty=True,
do_xcom_push=False,
dag=dag
)
getConf >> build_ope
pass

server_bs = create_dag.select_server('', '')
sshHook_bs = SSHHook(server_bs)
command_build_bs = 'bash bin/schedule/build_airflow.sh '
build_command_bs = 'bash /home/airflow/platform/check_git/checkgit.sh ' + path_project_bs + ' ' + project_name_bs + ' "' + command_build_bs + '" ' + checkout_branch
build_ope_bs = SSHOperator(
ssh_hook=sshHook_bs,
task_id='brandSafety_build',
command=build_command_bs,
on_failure_callback=create.task_on_failure_callback,
get_pty=True,
do_xcom_push=False,
dag=dag)

command_bs = 'cd ' + path_project_bs + project_name_bs + ' && bash bin/schedule/run_check_coverage_storybook.sh '
run_ope_bs = SSHOperator(
ssh_hook=sshHook_bs,
task_id='check_coverage_storybook',
do_xcom_push=False,
on_failure_callback=create.task_on_failure_callback,
command=command_bs,
dag=dag)

config_bs = create_dag.getConfig(path="/airflow/bs-core")
for key, value in config_bs.items():
getConf = SSHOperator(
ssh_hook=sshHook_bs,
task_id='get_config_bs_' + key,
command='echo "' + value + '" > ' + path_project_bs + project_name_bs + '/' + 'properties/' + key,
on_failure_callback=create.task_on_failure_callback,
get_pty=True,
do_xcom_push=False,
dag=dag
)
getConf >> build_ope_bs
pass

command_build_repush = 'bash bin/airflow_build.sh '
build_command_repush = 'bash /home/airflow/platform/check_git/checkgit.sh ' + path_project + ' ' + project_name + ' "' + command_build_repush + '" ' + checkout_branch
build_ope_repush = SSHOperator(
ssh_hook=sshHook_bs,
task_id='crawler-system_build_repush',
command=build_command_repush,
on_failure_callback=create.task_on_failure_callback,
get_pty=True,
do_xcom_push=False,
dag=dag)

command_repush = 'cd ' + path_project + project_name + ' && bash bin/run_repush_storybook.sh '
run_ope_repush = SSHOperator(
ssh_hook=sshHook_bs,
task_id='repush_story_book',
do_xcom_push=False,
on_failure_callback=create.task_on_failure_callback,
command=command_repush,
dag=dag)

config_repush = create_dag.getConfig(path="/airflow/bs-core")
for key, value in config_bs.items():
getConf = SSHOperator(
ssh_hook=sshHook_bs,
task_id='get_config_repush_' + key,
command='echo "' + value + '" > ' + path_project + project_name + '/' + 'properties/' + key,
on_failure_callback=create.task_on_failure_callback,
get_pty=True,
do_xcom_push=False,
dag=dag
)
getConf >> build_ope_repush
pass

[build_ope, build_ope_bs, build_ope_repush] >> run_ope >> run_ope_bs >> run_ope_repush

[nathadfield]

Thanks but this does not provide us with the ability to reproduce the problem, if indeed there even is a bug behind the problem you are facing and it not be something to do with your code or your implementation. You will have to try and pinpoint the problem a bit more on your side before anyone can really assess whether there truly is a bug or not.

[potiuk]

If you are using SSHOperator and it gets killed, the SSH connection is closed (by the nature of SSH conneciton) - but whether the action originated by SSH continues running depends solely on what command you send it. For example if your command will execute nohup run_something - that run_something will continue runing because your remote process will ignore HUP signal that is generated by SSH server when connection is broken.

Another thing with SSH is that it might not realise the connection is broken if the command on the other might not realize that the connection has been broken - which is the nature of TCP connection that gets broken - so you should look in your code.

Also I see that SSHOperator does not have on_kill implemented that would gently and deliberately close the ssh connection - that might help in some cases. I will open a separate task for it.

[potiuk]

Created feature and marked it as good first issue #40343 - that might help in some cases so feel free to implement the task, but in general case, it might simply be that your command is not reacting to being closed by the SSH server (ignoring the signals). Reviewing your commands and implementing on_kill to gently close the client should likely help in all cases.