An example of streaming pod events from k8s for a namespace

stream_pod.py

from datetime import datetime
from kubernetes import client, config, watch

config.load_kube_config()
# Create a configuration object
with_ssl_disabled_config = client.Configuration()
with_ssl_disabled_config.verify_ssl = False
api_client = client.ApiClient(with_ssl_disabled_config)

v1 = client.CoreV1Api(api_client)
watcher = watch.Watch()
kwargs = {'label_selector': 'airflow-worker=worker_uuid'}
namespace = 'airflow-worker'

for event in watcher.stream(v1.list_namespaced_pod, namespace, **kwargs):
    task = event['object']
    phase = task.status.phase
    annotations = task.metadata.annotations

    dag_id = annotations['dag_id']
    task_id = annotations['task_id']
    execution_date = annotations['execution_date']
    try_number = annotations['try_number']

    print('[xxx] airflow info', dag_id, task_id, execution_date, try_number)

    print(datetime.now(), event['type'], task.metadata.name, task.status.phase, task.metadata.annotations, task.metadata.resource_version)