如何通过k8s Python API读取Jupyter notebook的token？

核心逻辑 #

官方minimal-notebook镜像启动时，会自动生成随机登录token，存放在两个位置：

1. 容器启动的标准输出日志中，格式为http://<hostname>:8888/lab?token=<xxx>的行
2. 容器内部路径/home/jovyan/.local/share/jupyter/runtime/下的nbserver-*.json文件的token字段
   我们可以通过K8s Python API调用对应能力读取上述任意位置的内容提取token。

前置准备 #

首先安装依赖包：

pip install kubernetes

确保你的运行环境已经完成K8s集群身份认证（本地开发默认读取~/.kube/config，集群内运行默认读取挂载的ServiceAccount凭证），且使用的身份有对应命名空间下的pods/list、pods/log、pods/exec权限。 #

完整实现代码 #

from kubernetes import client, config
import json
import re

# 加载集群配置
config.load_kube_config() # 集群内运行替换为 config.load_incluster_config()
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()

# 配置参数
NAMESPACE = "你的命名空间"
DEPLOYMENT_NAME = "你的minimal-notebook的Deployment名称"

# 步骤1：获取Deployment的标签选择器，过滤对应Pod
deploy = apps_v1.read_namespaced_deployment(name=DEPLOYMENT_NAME, namespace=NAMESPACE)
label_selector = ",".join([f"{k}={v}" for k,v in deploy.spec.selector.match_labels.items()])
pods = v1.list_namespaced_pod(namespace=NAMESPACE, label_selector=label_selector)

if not pods.items:
    raise Exception("未找到对应Pod，请确认Deployment已完成调度")

# 取第一个Running状态的Pod（假设Deployment副本数为1，多副本自行遍历）
target_pod = None
for pod in pods.items:
    if pod.status.phase == "Running":
        target_pod = pod.metadata.name
        break
if not target_pod:
    raise Exception("未找到运行状态的Pod")

# ---------------------------
# 方式1：从Pod日志中提取Token（实现最简单，适合刚启动的Pod）
# ---------------------------
log = v1.read_namespaced_pod_log(name=target_pod, namespace=NAMESPACE)
token_match = re.search(r"token=([a-zA-Z0-9]+)", log)
if token_match:
    jupyter_token = token_match.group(1)
    print(f"从日志提取到的Token：{jupyter_token}")
else:
    print("日志中未找到Token，建议使用方式2")

# ---------------------------
# 方式2：Exec进入容器读取runtime配置文件（更稳定，不受日志滚动影响）
# ---------------------------
exec_command = [
    "/bin/sh",
    "-c",
    "cat /home/jovyan/.local/share/jupyter/runtime/nbserver-*.json"
]

resp = v1.connect_get_namespaced_pod_exec(
    name=target_pod,
    namespace=NAMESPACE,
    command=exec_command,
    stderr=True,
    stdin=False,
    stdout=True,
    tty=False
)

nb_config = json.loads(resp)
jupyter_token = nb_config.get("token")
print(f"从配置文件读取到的Token：{jupyter_token}")

权限说明 #

如果你是在集群内通过ServiceAccount运行该代码，需要配置如下RBAC权限：

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: <你的命名空间>
  name: jupyter-token-reader
rules:
- apiGroups: ["apps"]
  resources: ["deployments"]
  verbs: ["get"]
- apiGroups: [""]
  resources: ["pods", "pods/log"]
  verbs: ["get", "list"]
- apiGroups: [""]
  resources: ["pods/exec"]
  verbs: ["create"]

注意事项 #

• 如果你在启动Jupyter时手动指定了JUPYTER_TOKEN环境变量，也可以直接通过读取Pod的环境变量获取token，不需要读日志或exec
• 若Deployment配置了多副本，需要自行遍历所有Pod获取对应每个实例的token

内容的提问来源于stack exchange，提问作者ktm5124