### Description

Every line in the scheduler log pertaining to a particular task instance should be stamped with that task instance's identifying information. In the Kubernetes executor, some lines are stamped only with the pod name instead.

### Use case/motivation

When trying to trace the lifecycle of a task in the kubernetes executor, you currently must search first for the name of the pod created by the task, then search for the pod name in the logs. This means you need to be pretty familiar with the structure of the scheduler logs in order to search effectively for the lifecycle of a task that had a problem.

Some log statements like `Attempting to finish pod` do have the annotations for the pod, which include dag name, task name, and run_id, but others do not. For instance, `Event: podname-a2f2c1ac706 had an event of type DELETED` has no such annotations.



### Related issues

_No response_

### Are you willing to submit a PR?

- [X] 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)

"""Terminates the watcher."""
        self.log.debug("Terminating kube_watchers...")
        for namespace, kube_watcher in self.kube_watchers.items():
            kube_watcher.terminate()
            kube_watcher.join()
            self.log.debug("kube_watcher=%s", kube_watcher)
        self.log.debug("Flushing watcher_queue...")
        self._flush_watcher_queue()
        
        self.watcher_queue.join()
        self.log.debug("Shutting down manager...")
        self._manager.shutdown()
def get_base_pod_from_template(pod_template_file: str | None, kube_config: Any) -> k8s.V1Pod:
    """
    Get base pod from template.
    Reads either the pod_template_file set in the executor_config or the base pod_template_file
    set in the airflow.cfg to craft a "base pod" that will be used by the KubernetesExecutor
    :param pod_template_file: absolute path to a pod_template_file.yaml or None
    :param kube_config: The KubeConfig class generated by airflow that contains all kube metadata
    :return: a V1Pod that can be used as the base pod for k8s tasks
    """
    if pod_template_file:
        return PodGenerator.deserialize_model_file(pod_template_file)
    else:
        return PodGenerator.deserialize_model_file(kube_config.pod_template_file)
class KubernetesExecutor(BaseExecutor):
    """Executor for Kubernetes."""
    supports_ad_hoc_ti_run: bool = True
    def __init__(self):