HsiuChuanHsu commented on code in PR #54115:
URL: https://github.com/apache/airflow/pull/54115#discussion_r2289512429
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -254,28 +275,51 @@ def process_status(
and
container_status_state["waiting"]["message"] == "pull QPS exceeded"
):
continue
- self.log.error(
- "Event: %s has container %s with fatal reason
%s",
+ key = annotations_to_key(annotations=annotations)
+ task_key_str = (
+ f"{key.dag_id}.{key.task_id}.{key.try_number}"
if key else "unknown"
+ )
+ self.log.warning(
+ "Event: %s has container %s with fatal reason
%s, task: %s",
pod_name,
container_status["name"],
container_status_state["waiting"]["reason"],
+ task_key_str,
)
self.watcher_queue.put(
- (pod_name, namespace,
TaskInstanceState.FAILED, annotations, resource_version)
+ (
+ pod_name,
+ namespace,
+ TaskInstanceState.FAILED,
+ annotations,
+ resource_version,
+ failure_details,
Review Comment:
Fixed!
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -210,22 +211,42 @@ def process_status(
if POD_REVOKED_KEY in pod.metadata.labels.keys():
return
+ # Collect failure details for failed pods using the new function
+ failure_details = None
+ if status == "Failed":
+ try:
+ failure_details = collect_pod_failure_details(pod)
+ except Exception as e:
+ self.log.warning(
+ "Failed to collect pod failure details for %s/%s: %s",
namespace, pod_name, e
+ )
+
annotations_string = annotations_for_logging_task_metadata(annotations)
if event["type"] == "DELETED" and not pod.metadata.deletion_timestamp:
# This will happen only when the task pods are adopted by another
executor.
# So, there is no change in the pod state.
# However, need to free the executor slot from the current
executor.
self.log.info("Event: pod %s adopted, annotations: %s", pod_name,
annotations_string)
- self.watcher_queue.put((pod_name, namespace, ADOPTED, annotations,
resource_version))
+ self.watcher_queue.put((pod_name, namespace, ADOPTED, annotations,
resource_version, None))
elif hasattr(pod.status, "reason") and pod.status.reason ==
"ProviderFailed":
# Most likely this happens due to Kubernetes setup (virtual
kubelet, virtual nodes, etc.)
- self.log.error(
- "Event: %s failed to start with reason ProviderFailed,
annotations: %s",
+ key = annotations_to_key(annotations=annotations)
+ task_key_str = f"{key.dag_id}.{key.task_id}.{key.try_number}" if
key else "unknown"
+ self.log.warning(
+ "Event: %s failed to start with reason ProviderFailed, task:
%s, annotations: %s",
pod_name,
+ task_key_str,
annotations_string,
)
self.watcher_queue.put(
- (pod_name, namespace, TaskInstanceState.FAILED, annotations,
resource_version)
+ (
+ pod_name,
+ namespace,
+ TaskInstanceState.FAILED,
+ annotations,
+ resource_version,
+ failure_details,
Review Comment:
Fixed!
--
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-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
