Add better debug logging to K8sexec and K8sPodOp (#11502)

airflow/executors/base_executor.py

@@ -112,6 +112,7 @@ class BaseExecutor(LoggingMixin):
             pool=pool,
             pickle_id=pickle_id,
             cfg_path=cfg_path)
+        self.log.debug("created command %s", command_list_to_run)
         self.queue_command(
             task_instance,
             command_list_to_run,

airflow/executors/kubernetes_executor.py

@@ -317,7 +317,7 @@ class AirflowKubernetesScheduler(LoggingMixin):
 
     def _health_check_kube_watcher(self):
         if self.kube_watcher.is_alive():
-            pass
+            self.log.debug("KubeJobWatcher alive, continuing")
         else:
             self.log.error(
                 'Error while health checking kube watcher process. '
@@ -368,6 +368,7 @@ class AirflowKubernetesScheduler(LoggingMixin):
     def delete_pod(self, pod_id: str, namespace: str) -> None:
         """Deletes POD"""
         try:
+            self.log.debug("Deleting pod %s in namespace %s", pod_id, namespace)
             self.kube_client.delete_namespaced_pod(
                 pod_id, namespace, body=client.V1DeleteOptions(**self.kube_config.delete_option_kwargs),
                 **self.kube_config.kube_client_request_args)
@@ -385,11 +386,13 @@ class AirflowKubernetesScheduler(LoggingMixin):
         :return:
 
         """
+        self.log.debug("Syncing KubernetesExecutor")
         self._health_check_kube_watcher()
         while True:
             try:
                 task = self.watcher_queue.get_nowait()
                 try:
+                    self.log.debug("Processing task %s", task)
                     self.process_watcher_task(task)
                 finally:
                     self.watcher_queue.task_done()
@@ -409,6 +412,7 @@ class AirflowKubernetesScheduler(LoggingMixin):
             self.result_queue.put((key, state, pod_id, namespace, resource_version))
 
     def _annotations_to_key(self, annotations: Dict[str, str]) -> Optional[TaskInstanceKey]:
+        self.log.debug("Creating task key for annotations %s", annotations)
         dag_id = annotations['dag_id']
         task_id = annotations['task_id']
         try_number = int(annotations['try_number'])
@@ -519,6 +523,7 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
         proper support
         for State.LAUNCHED
         """
+        self.log.debug("Clearing tasks that have not been launched")
         if not self.kube_client:
             raise AirflowException(NOT_STARTED_MESSAGE)
         queued_tasks = session \
@@ -531,6 +536,7 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
 
         for task in queued_tasks:
             # pylint: disable=protected-access
+            self.log.debug("Checking task %s", task)
             dict_string = (
                 "dag_id={},task_id={},execution_date={},airflow-worker={}".format(
                     pod_generator.make_safe_label_value(task.dag_id),

airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py

@@ -357,7 +357,9 @@ class KubernetesPodOperator(BaseOperator):  # pylint: disable=too-many-instance-
         will supersede all other values.
 
         """
+        self.log.debug("Creating pod for K8sPodOperator task %s", self.task_id)
         if self.pod_template_file:
+            self.log.debug("Pod template file found, will parse for base pod")
             pod_template = pod_generator.PodGenerator.deserialize_model_file(self.pod_template_file)
         else:
             pod_template = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="name"))
@@ -405,8 +407,10 @@ class KubernetesPodOperator(BaseOperator):  # pylint: disable=too-many-instance-
         pod = PodGenerator.reconcile_pods(pod_template, pod)
 
         for secret in self.secrets:
+            self.log.debug("Adding secret to task %s", self.task_id)
             pod = secret.attach_to_pod(pod)
         if self.do_xcom_push:
+            self.log.debug("Adding xcom sidecar to task %s", self.task_id)
             pod = PodGenerator.add_xcom_sidecar(pod)
         return pod
 
@@ -421,6 +425,7 @@ class KubernetesPodOperator(BaseOperator):  # pylint: disable=too-many-instance-
         if not (self.full_pod_spec or self.pod_template_file):
             # Add Airflow Version to the label
             # And a label to identify that pod is launched by KubernetesPodOperator
+            self.log.debug("Adding k8spodoperator labels to pod before launch for task %s", self.task_id)
             self.labels.update(
                 {
                     'airflow_version': airflow_version.replace('+', '-'),
@@ -442,6 +447,7 @@ class KubernetesPodOperator(BaseOperator):  # pylint: disable=too-many-instance-
             raise
         finally:
             if self.is_delete_operator_pod:
+                self.log.debug("Deleting pod for task %s", self.task_id)
                 launcher.delete_pod(self.pod)
         return final_state, self.pod, result