### 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)

from kubernetes import client, watch
from kubernetes.client import Configuration, models as k8s
from kubernetes.client.rest import ApiException
from sqlalchemy.orm import Session
from urllib3.exceptions import ReadTimeoutError
from airflow.configuration import conf
from airflow.exceptions import AirflowException, PodMutationHookException, PodReconciliationError
from airflow.executors.base_executor import BaseExecutor
from airflow.kubernetes import pod_generator
from airflow.kubernetes.kube_client import get_kube_client
from airflow.kubernetes.kube_config import KubeConfig
from airflow.kubernetes.kubernetes_helper_functions import annotations_to_key, create_pod_id
from airflow.kubernetes.pod_generator import PodGenerator
from airflow.utils.event_scheduler import EventScheduler
from airflow.utils.log.logging_mixin import LoggingMixin, remove_escape_codes
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.state import State, TaskInstanceState
if TYPE_CHECKING:
    from airflow.executors.base_executor import CommandType
    from airflow.models.taskinstance import TaskInstance
    from airflow.models.taskinstancekey import TaskInstanceKey
    
    KubernetesJobType = Tuple[TaskInstanceKey, CommandType, Any, Optional[str]]
    
    KubernetesResultsType = Tuple[TaskInstanceKey, Optional[str], str, str, str]
    
    KubernetesWatchType = Tuple[str, str, Optional[str], Dict[str, str], str]
ALL_NAMESPACES = "ALL_NAMESPACES"
POD_EXECUTOR_DONE_KEY = "airflow_executor_done"
class ResourceVersion: