You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jordi-crespo (via GitHub)" <gi...@apache.org> on 2024/09/10 20:57:45 UTC
[I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
jordi-crespo opened a new issue, #42144:
URL: https://github.com/apache/airflow/issues/42144
### Apache Airflow Provider(s)
cncf-kubernetes
### Versions of Apache Airflow Providers
apache-airflow-providers-cncf-kubernetes==8.3.4
### Apache Airflow version
2.9.2
### Operating System
Docker, creating the image from python:3.10-slim-buster
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
_No response_
### What happened
When using `KubernetesJobOperator` in deferred mode, the job executes correctly, but logs from the associated pod are not retrieved.
The issue arises in the [`execute_complete`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/job.py#L226) method, where the pod_name and pod_namespace passed from the trigger are None, which causes the following code to fail:
```
if self.get_logs:
pod_name = event["pod_name"]
pod_namespace = event["pod_namespace"]
self.pod = self.hook.get_pod(pod_name, pod_namespace)
if not self.pod:
raise PodNotFoundException("Could not find pod after resuming from deferral")
self._write_logs(self.pod)
```
First, I create a `V1Pod` object that I pass to the `kubernetesJobOperator` using the parameter `full_pod_spec`. This works fine as well in the `kubernetesPodOperator`. The `kubernetesJobOperator` creates the job it creates a `V1Pod` from my [`full_pod_spec`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/job.py#L307). Later in this [line of code](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/job.py#L169) it creates the `self.pod` object.
Because I am executing the opeartor as deferred the `self.execute_deferrable()` is executed. This method uses the `KubernetesJobTrigger` class and [passes the `pod_name` and `pod_namespace`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/job.py#L212). Because I don't get an error, I assume that the self.pod object exist, because it can access the following pod attributes:
````
pod_name=self.pod.metadata.name, # type: ignore[union-attr]
pod_namespace=self.pod.metadata.namespace, # type: ignore[union-attr]
````
[`KubernetesJobTrigger`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/triggers/job.py#L32), yields [`TriggerEvent`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/triggers/job.py#L122) object, passing the `pod_name`and `pod_namespace` like this:
````
yield TriggerEvent(
{
"pod_name": pod.metadata.name if self.get_logs else None,
"pod_namespace": pod.metadata.namespace if self.get_logs else None,
}
)
````
This pod object comes from this [piece of code](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/triggers/job.py#L105):
````
if self.get_logs or self.do_xcom_push:
pod = await self.hook.get_pod(name=self.pod_name, namespace=self.pod_namespace)
````
Next, the method [`execute_complete`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/job.py#L226) is executed is getting the `pod_name`and `pod_namespace` from the event.
But I found that there these values are `None`. I believe that the `TriggerEvent` is yielding None values.
To resolve this, I had to manually extract the pod name from the job itself, by inheriting the `kubernetesJobOperator` and creating a new `execute_complete` method.
```
pod_list_response = self.hook.get_namespaced_pod_list(
label_selector=f"job-name={job_name}", namespace=job_namespace
)
pod_list = json.loads(pod_list_response.read())
pods = pod_list.get("items", [])
if not pods:
raise PodNotFoundException(f"No pods found for job: {job_name} in namespace: {job_namespace}")
pod = pods[0]
pod_name = pod["metadata"]["name"]
pod_namespace = pod["metadata"]["namespace"]
```
After retrieving the pod details, I encountered another issue with the client used to fetch logs. By default, [`self.client`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/job.py#L470) in `KubernetesJobOperator` points to the `BatchV1Api`:
```
@cached_property
def client(self) -> BatchV1Api:
return self.hook.batch_v1_client
```
This is happening in the following piece of code:
````
if self.get_logs:
pod_name = event["pod_name"]
pod_namespace = event["pod_namespace"]
self.pod = self.hook.get_pod(pod_name, pod_namespace)
if not self.pod:
raise PodNotFoundException("Could not find pod after resuming from deferral")
self._write_logs(self.pod)
````
[`self._write_logs(self.pod)`](https://github.com/apache/airflow/blob/main/airflow/providers/cncf/kubernetes/operators/pod.py#L832) is created in the `kubernetesPodOperator` that is creating the `self.client using use `CoreV1Api`. I resolved this by switching the client before calling the `_write_logs` function:
```
self.client = self.hook.core_v1_client
self._write_logs(self.pod)
```
### What you think should happen instead
We should be able to send the pod_name and pod_namespace from the `KubernetesJobTrigger` and that they are not None values.
Also, a raise error is executed if the job is in failed state before being able to print the logs:
````
def execute_complete(self, context: Context, event: dict, **kwargs):
ti = context["ti"]
ti.xcom_push(key="job", value=event["job"])
if event["status"] == "error":
raise AirflowException(event["message"])
if self.get_logs:
pod_name = event["pod_name"]
pod_namespace = event["pod_namespace"]
self.pod = self.hook.get_pod(pod_name, pod_namespace)
if not self.pod:
raise PodNotFoundException("Could not find pod after resuming from deferral")
self._write_logs(self.pod)
````
But even if the job fails, we would still want to see the logs:
````
def execute_complete(self, context: Context, event: dict, **kwargs):
ti = context["ti"]
ti.xcom_push(key="job", value=event["job"])
if self.get_logs:
pod_name = event["pod_name"]
pod_namespace = event["pod_namespace"]
self.pod = self.hook.get_pod(pod_name, pod_namespace)
if not self.pod:
raise PodNotFoundException("Could not find pod after resuming from deferral")
self._write_logs(self.pod)
if event["status"] == "error":
raise AirflowException(event["message"])
````
### How to reproduce
Write a DAG with the following operator:
````
from kubernetes.client import V1Pod, V1ObjectMeta, V1PodSpec, V1Container
from airflow.providers.cncf.kubernetes.operators.job import KubernetesJobOperator
# Define the pod
pod = V1Pod(
metadata=V1ObjectMeta(name="example-pod"),
spec=V1PodSpec(
containers=[V1Container(
name="example-container",
image="busybox",
command=["/bin/sh", "-c", "echo Hello, Kubernetes! && sleep 60"]
)],
restart_policy="Never"
)
)
# Define the KubernetesJobOperator
job = KubernetesJobOperator(
task_id="example_k8s_job",
full_pod_spec=pod,
get_logs=True,
in_cluster=True
)
````
### Anything else
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-2357170155
Not been able to reproduce, in my case its working fine without issues , could you please share bit more details or can you check on the latest version of provider?
<img width="1345" alt="image" src="https://github.com/user-attachments/assets/1840e367-f035-4d4c-9be7-82c6f0f442da">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "jordi-crespo (via GitHub)" <gi...@apache.org>.
jordi-crespo commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-2357198992
Hello, the Pod created by the job will be scheduled in a nodepool that needs to be autoescaled.
Because of this, the pod will be in a pending state until the node is available to execute pods.
It is just a hunch that this can be the cause of the problem. I am not entertained sure how the trigger function works.
But I think that using the pod name and pod namespace from the class attribute will solve the problem.
This is how works in the kubernetesPodOperator
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "jordi-ydteam (via GitHub)" <gi...@apache.org>.
jordi-ydteam commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-2357198265
Hello, the Pod created by the job will be scheduled in a nodepool that needs to be autoescaled.
Because of this, the pod will be in a pending state until the node is available to execute pods.
It is just a hunch that this can be the cause of the problem. I am not entertained sure how the trigger function works.
But I think that using the pod name and pod namespace from the class attribute will solve the problem.
This is how works in the kubernetesPodOperator
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-3425037807
This issue has been closed because it has not received response from the issue author.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-3305702979
This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-2357180910
Are you referring to pod waiting states? meaning its still not available to do further steps?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on issue #42144:
URL: https://github.com/apache/airflow/issues/42144#issuecomment-2357216935
ok, do you have debug logs ? there are logs which tells what response got from when the pod creation api called.
if you could look logs this pattern: `Pod Creation Request:` `Pod Creation Response:` and possible share? that will help us to understand whats happening really there, when these creation call made.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed issue #42144: Issue with Pod Log Retrieval in KubernetesJobOperator When Using Deferred Mode
URL: https://github.com/apache/airflow/issues/42144
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org