KubernetesPodOperator: exit as success when another task is completed.

[yanxon]

Dear Community,

I need to run a KubernetesPodOperator (task 1) that will create a Pod (pull an image from a private repo). Then, I need to "do something" with the Pod in task 2. Hence, I need to run task 1 and task 2 in parallel, where task 2 will wait until pod in task 1 is up.

Here is the workflow: t_start >> [t1, t2] >> t_end.

The problem rises as task 1 will run in perpetuity since I don't issue cmds command. I can't use the cmds command because I need to use the default docker from the image that I pulled.

My question is how should I mark t1 as complete as soon as t2 is completed?

[hussein-awala]

Currently the KubernetesPodOperator doesn't support this kind of interaction between pods, and IMHO it should not support them.

How can you achieve this?

You need to add 2 more tasks:

• a new sensor operator which use the KubernetesHook to check if the pod 1 is created and running
• a new task executed after the pod 2 to kill/stop pod 1 using KubernetesHook

The new dag will be:

t_start >> pod1 >> t_end
t_start >> sensor_pod1 >> pod2 >> pod1_killer >> t_end

With this dag, pod1 and sensor_pod1 will be running in parallel, once pod1 is running, sensor_pod1 will be pass to success state, and pod2 will be executed. Then when it finishes, pod1_killer will be executed to stop pod1 (you need to be careful to avoid marking the task as failed), the t_end will be executed.

[yanxon]

Thank you for your prompt response, @hussein-awala. I truly appreciate it.

That makes sense.

I have another related question to KubernetesPodOperator/KPO. Since I need the KPO to run my default docker command, cmds argument cannot be set. If the cmds argument is not set, the KPO task will run in perpetuity. Is there a way to run the default docker command + some python argument so that it will trigger as success after running?

[hussein-awala]

I don't think you can achieve that with just an extra command.

However, you can add a signal_handler to your python code (https://stackoverflow.com/a/1112350/9560594), then instead of killing the pod via pod1_killer, you can just send the specific signal to trigger the signal_handler which calls sys.exit(0). You can do the same think for the other languages if your code is not python.

You can use this code as reference to send the signal to your pod:

airflow/airflow/providers/cncf/kubernetes/utils/pod_manager.py

Lines 728 to 747 in b5057e0

	self._exec_pod_command(resp, "kill -s SIGINT 1")
	def _exec_pod_command(self, resp, command: str) -> str | None:
	res = None
	if resp.is_open():
	self.log.info("Running command... %s\n", command)
	resp.write_stdin(command + "\n")
	while resp.is_open():
	resp.update(timeout=1)
	while resp.peek_stdout():
	res = res + resp.read_stdout() if res else resp.read_stdout()
	error_res = None
	while resp.peek_stderr():
	error_res = error_res + resp.read_stderr() if error_res else resp.read_stderr()
	if error_res:
	self.log.info("stderr from command: %s", error_res)
	break
	if res:
	return res
	return res

(Keep in mind that this method is private, so it might change in the future)