我有一个用TypeScript编写的前端，后端是Python 3.11，用来把Docker镜像部署到Kubernetes集群（使用kubernetes 29.3.0）。在这个过程中，它会调用一个API，在镜像启动时获取状态，以便一旦服务启动就把用户重定向到相应的服务。

当前的代码采用一个相对简单、粗糙的解决方案来查看部署的状态：

• 如果部署不存在，就没有状态
• 如果部署存在且只有一个可用副本（这是我们所规定的），它就是 running
• 否则就是 pending

并且UI实际上把这个“状态”呈现在界面上供用户查看。

95% 的时间，镜像在几秒内就启动完成，且没有问题。

我希望在UI中看到更细粒度/更微妙的提示，尤其是在一切顺利时。例如，极少数情况下镜像尚未在节点上，需要拉取镜像——这会花费时间，或者所选节点资源不足以启动该镜像。最好能真正把发生了什么信息通知给用户。

我需要一段代码来获取一个Pod的最后一个事件：

from kubernetes import watch
from kubernetes.client import CoreV1Api

def get_client:
    pseudo_code: return client.CoreV1Api() with config

# return the _last_ event for a pod
def watch_pod_events(name, namespace):
    v1 = get_client()
    w = watch.Watch()
    message = None
    for event in w.stream(v1.list_namespaced_event, namespace=namespace):
        involved_obj = event['object'].involved_object
        if involved_obj.name == name:
            message = event['object'].message
    return message

如果我在循环中打印出消息，会得到实际数据——但它并没有向上传递到上层。

我怀疑是因为 w.stream() 是一个可迭代对象（iterable），而不是一个实际的列表，但我不确定。

我也尝试过：

my_events = list(filter(lambda e: e['object'].involved_object.name == name, w.stream(v1.list_namespaced_event, namespace=namespace)))

my_events =  [e for e in w.stream(v1.list_namespaced_event, namespace=namespace) if e['object'].involved_object.name == name]

my_events = v1.list_namespaced_event(namespace=namespace, field_selector=f"involvedObject.name={name}").items

但都不起作用。

解决方案

所以这就是我最终的做法……它先获取Pod启动时的事件，然后读取Pod应用的日志，筛选出任何 Info、Warning或 Error行。

Pod事件在经过之后就会消失，应用日志会继续追加。

import re
import logging

from django.conf import settings
from kubernetes.client import exceptions as k8s_exceptions
from typing import Optional

from MyLibs.kubernetes import get_client

logger = logging.getLogger(__name__)

def get_notebook_server_status(pod_name: str, namespace: str = settings.K8S_NAMESPACE) -> Optional[str]:
    v1 = get_client()

    message = ""
    if pod_name is None:
        return None

    field_selector = (
        f"involvedObject.kind=Pod," f"involvedObject.namespace={namespace}," f"involvedObject.name={pod_name}"
    )

    try:
        pod_events_list = v1.list_namespaced_event(namespace=namespace, field_selector=field_selector)
        items = pod_events_list.items
        for e in items:
            message = f"{e.reason}: {e.message}"
    except KeyboardInterrupt:
        return
    except k8s_exceptions.ApiException as exc:
        # If the watch RV is too old, clear it and relist/watch from "now".
        if exc.status == 410:
            return None

    # If pod has started, get messages from the pod logs
    p = re.compile("PodReadyToStartContainers: True")
    if p.search(message) or message == "Started: Started container notebook":
        try:
            # Get the logs for the pod, split into a list, and filter.
            # If we get a "Use Control-C to stop this server" message, we know the pod is running and can exit
            pod_logs = v1.read_namespaced_pod_log(
                name=pod_name, namespace=namespace, insecure_skip_tls_verify_backend=True, timestamps=False
            )
            pod_logs_list = pod_logs.splitlines()

            # Now filter the log_list to only include lines that start '\[[IWE] '
            filtered_logs = ["Waiting for jupyter to start...."]
            for line in pod_logs_list:
                if filtered_logs == ["Running"]:
                  break
                if re.match(r"\[[IWE] ", line):
                  filtered_logs.append(line)
            message = "\n".join(filtered_logs)
        except KeyboardInterrupt:
            return
        except k8s_exceptions.ApiException as exc:
            # If the watch RV is too old, clear it and relist/watch from "now".
            if exc.status == 410:
                return None
        except Exception:
            return
    else:
        logger.warning(
            f"get_notebook_server_status: Pod {pod_name} has not started yet, current message: {message}"
        )
    return message