1fanwang commented on code in PR #66405:
URL: https://github.com/apache/airflow/pull/66405#discussion_r3678834923
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -329,18 +330,20 @@ def process_status(
# since kube server have received request to delete pod set TI
state failed
if event["type"] == "DELETED" and pod.metadata.deletion_timestamp:
self.log.info(
- "Event: Pod %s deleted before it could complete,
annotations: %s",
+ "Event: Pod %s deleted while running, annotations: %s",
pod_name,
annotations_string,
)
+ # An Airflow-initiated stop sets the TI terminal first, so the
scheduler
+ # excludes that case by state and only a real disruption
reaches here.
self.watcher_queue.put(
KubernetesWatch(
pod_name,
namespace,
TaskInstanceState.FAILED,
annotations,
resource_version,
- None,
+ {"pod_status": "Running", "pod_reason":
POD_DELETED_REASON},
Review Comment:
can we use kwargs, same everywhere in new / updated code path
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -409,6 +412,51 @@ def collect_pod_failure_details(pod: k8s.V1Pod, logger) ->
FailureDetails | None
}
+# A pod taken by the platform while its task ran: node drain, preemption, spot
reclaim,
+# or a force-delete. Distinct from a container that ran and exited on its own.
+POD_DELETED_REASON = "PodDeleted"
Review Comment:
type hint for all vars and methods pls, same everywhere for updated / added
code
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -409,6 +412,51 @@ def collect_pod_failure_details(pod: k8s.V1Pod, logger) ->
FailureDetails | None
}
+# A pod taken by the platform while its task ran: node drain, preemption, spot
reclaim,
+# or a force-delete. Distinct from a container that ran and exited on its own.
+POD_DELETED_REASON = "PodDeleted"
+
+# Pod/node-level reasons meaning the platform ended the pod, as opposed to the
container
+# terminating on its own.
+_INFRA_FAILURE_REASONS = frozenset(
+ {
+ "Evicted",
+ "Preempting",
+ "NodeShutdown",
+ "Shutdown",
+ "NodeLost",
+ "TerminationByKubelet",
+ "DeletionByTaintManager",
+ "DisruptionTarget",
+ POD_DELETED_REASON,
+ }
+)
Review Comment:
is there a python model/constant importable/published from k8s side? if not
can we at least link the permalink for the error code / reference / mapping?
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1467,6 +1468,9 @@ def process_executor_events(
job_id,
)
state, info = event_buffer.pop(buffer_key)
+ # Pop unconditionally, not only on the killed-externally branch
below, so a
+ # self-reporting task can't leak its entry.
+ executor_failure_kind = executor.get_task_failure_info(ti.key)
Review Comment:
initialize with type hint first please, so hard to read otherwise, same for
other new/added code
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
