使用 kubernetes python 客户端重试手表时如何避免"资源太旧"？

Question

下面的代码将在第二个watch.stream()上抛出一个ApiException 410 resource too old:

# python3 -m venv venv
# source venv/bin/activate
# pip install 'kubernetes==23.3.0'
from kubernetes import client,config,watch
config.load_kube_config(context='my-eks-context')
v1 = client.CoreV1Api()
watcher = watch.Watch()

namespace = 'kube-system'
last_resource_version=0
# this watch will timeout in 5s to have a fast way to simulate a watch that need to be retried 
for i in  watcher.stream(v1.list_namespaced_pod, namespace, resource_version=last_resource_version, timeout_seconds=5):
print(i['object'].metadata.resource_version)
last_resource_version = i['object'].metadata.resource_version

# we retry the watch starting from the last resource version known
# but this will raise a kubernetes.client.exceptions.ApiException: (410)
# Reason: Expired: too old resource version: 379140622 (380367990)
for i in  watcher.stream(v1.list_namespaced_pod, namespace, resource_version=last_resource_version, timeout_seconds=5):
print('second loop', i['object'].metadata.resource_version)
last_resource_version = i['object'].metadata.resource_version

kubernetes文档说明:

如果一个客户端手表被断开连接，那么该客户端可以从最后返回的资源版本开始一个新的手表

，这是我在上面的代码中想要的，它总是给出以下异常:

Traceback (most recent call last):
File "main.py", line 24, in <module>
File "/Users/rubelagu/git/python-kubernetes-client/venv/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 379164133 (380432814)

我做错了什么?

Answer 1

似乎在对手表(来自EKS集群1.21)的初始响应中，事件可以以任何顺序返回。

我在随后的两个表中间隔两秒，它们包含相同的30个事件，但顺序完全不同。

因此不能保证您看到的最后一个资源版本实际上是实际的最后一个，也不能保证您可以从resourceVersion/resource_version恢复。此外，你不允许通过resourceVersion对这些事件进行排序/整理，因为kubernetes的资源版本语义文档明确地说:

资源版本必须被视为不透明[…你不能假设资源版本为数字或可排序.

您必须通过捕获resource too old exception并在不指定资源版本的情况下重试来解释这个问题，参见下面的示例:

from kubernetes import client,config,watch
from kubernetes.client.exceptions import ApiException
config.load_kube_config(context='eks-prod')
v1 = client.CoreV1Api()
# v1 = config.new_client_from_config(context="eks-prod").CoreV1Api()
watcher = watch.Watch()

namespace = 'my-namespace'

def list_pods(resource_version=None):
print('start watch from resource version: ', str(resource_version))
try: 
for i in  watcher.stream(v1.list_namespaced_pod, namespace, resource_version=resource_version, timeout_seconds=2):
print(i['object'].metadata.resource_version)
last_resource_version = i['object'].metadata.resource_version
except ApiException as e: 
if e.status == 410: # Resource too old
return list_pods(resource_version=None)
else:
raise
return last_resource_version
last_resource_version = list_pods()
print('last_resource_version', last_resource_version)
list_pods(last_resource_version)