Dump Pod as YAML in logs for KubernetesPodOperator (apache#9895)

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

@@ -18,6 +18,7 @@
 import re
 from typing import Dict, List, Optional, Tuple
 
+import yaml
 from kubernetes.client import models as k8s
 
 from airflow.exceptions import AirflowException
@@ -378,7 +379,7 @@ def create_new_pod_for_operator(self, labels, launcher) -> Tuple[State, k8s.V1Po
         )
 
         self.pod = pod
-        self.log.info("Starting pod %s", pod)
+        self.log.debug("Starting pod:\n%s", yaml.safe_dump(pod.to_dict()))
         try:
             launcher.start_pod(
                 pod,

kubernetes_tests/test_kubernetes_pod_operator.py

@@ -16,8 +16,10 @@
 # specific language governing permissions and limitations
 # under the License.
 import json
+import logging
 import os
 import shutil
+import textwrap
 import unittest
 from unittest import mock
 from unittest.mock import ANY
@@ -739,7 +741,20 @@ def test_pod_template_file(
         )
         monitor_mock.return_value = (State.SUCCESS, None)
         context = create_context(k)
-        k.execute(context)
+        with self.assertLogs(k.log, level=logging.DEBUG) as cm:
+            k.execute(context)
+            expected_line = textwrap.dedent("""\
+            DEBUG:airflow.task.operators:Starting pod:
+            api_version: v1
+            kind: Pod
+            metadata:
+              annotations: null
+              cluster_name: null
+              creation_timestamp: null
+              deletion_grace_period_seconds: null\
+            """).strip()
+            self.assertTrue(any(line.startswith(expected_line) for line in cm.output))
+
         actual_pod = self.api_client.sanitize_for_serialization(k.pod)
         self.assertEqual({
             'apiVersion': 'v1',