如何进行KubernetesClientException资源版本太旧的分析

如何进行KubernetesClientException资源版本太旧的分析，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。

背景

公司目前在基于k8s做调度（基于io.fabric8:kubernetes-client:4.2.0），在运行的过程中，遇到了如下问题：

DEBUG io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager - WebSocket close received. code: 1000, reason:DEBUG io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager - Submitting reconnect task to the executor[scheduleReconnect|Executor for Watch 1880052106] DEBUG io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager - Scheduling reconnect task[reconnectAttempt|Executor for Watch 1880052106] DEBUG io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager - Connecting websocket ... io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager@700f518a199 - 2020-11-17T06:39:13.874Z -[merlion-k8s-backend]-[merlion-k8s-backend-6b4cc44855-s6wnq]: 06:39:13.873 [OkHttp https://10.25.61.82:6443/...] DEBUG io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager - WebSocket successfully opened  WARN PodsWatchSnapshotSource: Kubernetes client has been closed (this is expected if the application is shutting down.)io.fabric8.kubernetes.client.KubernetesClientException: too old resource version: 135562761 (135563127)at io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager$1.onMessage(WatchConnectionManager.java:254)[kubernetes-client-4.2.2.jar:?]at okhttp3.internal.ws.RealWebSocket.onReadMessage(RealWebSocket.java:323) [okhttp-3.12.0.jar:?]at okhttp3.internal.ws.WebSocketReader.readMessageFrame(WebSocketReader.java:219) [okhttp-3.12.0.jar:?]at okhttp3.internal.ws.WebSocketReader.processNextFrame(WebSocketReader.java:105) [okhttp-3.12.0.jar:?]at okhttp3.internal.ws.RealWebSocket.loopReader(RealWebSocket.java:274) [okhttp-3.12.0.jar:?]at okhttp3.internal.ws.RealWebSocket$2.onResponse(RealWebSocket.java:214) [okhttp-3.12.0.jar:?]at okhttp3.RealCall$AsyncCall.execute(RealCall.java:206) [okhttp-3.12.0.jar:?]at okhttp3.internal.NamedRunnable.run(NamedRunnable.java:32) [okhttp-3.12.0.jar:?]at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_191]at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_191]at java.lang.Thread.run(Thread.java:748) [?:1.8.0_191]

单凭这个问题其实没什么，但是代码中是：

  watchConnection = kubernetesClient.pods()         .withLabel(MERLION_TASK_LABEL, applicationId)         //      .withResourceVersion(resourceVersion)         .watch(new TaskPodsWatcher())

因为我们已经注释掉了withResourceVersion(resourceVersion),（如果没有注释掉，说明我们的代码中设置的resourceVersion太小）但是还会报too old resource version

分析

直接跳转到WatchConnectionManager onClosed 如下：

     @Override      public void onClosed(WebSocket webSocket, int code, String reason) {        logger.debug("WebSocket close received. code: {}, reason: {}", code, reason);        if (forceClosed.get()) {          logger.debug("Ignoring onClose for already closed/closing websocket");          return;        }        if (currentReconnectAttempt.get() >= reconnectLimit && reconnectLimit >= 0) {          closeEvent(new KubernetesClientException("Connection unexpectedly closed"));          return;        }        scheduleReconnect();      }

对于onclosed的解释是

 /**   * Invoked when both peers have indicated that no more messages will be transmitted and the   * connection has been successfully released. No further calls to this listener will be made.   */  public void onClosed(WebSocket webSocket, int code, String reason) {  }

说明由于长时间没有event的传输，导致该connect被释放了，从而导致WebSocket 被关闭了（这种在任务不是很多的情况下发生的概率很大），从而进行了重联操作scheduleReconnect，而该方法调用了runWatch():

executor.schedule(new NamedRunnable("reconnectAttempt") {              @Override              public void execute() {                try {                  runWatch();                  reconnectPending.set(false);                } catch (Exception e) {                  // An unexpected error occurred and we didn't even get an onFailure callback.                  logger.error("Exception in reconnect", e);                  webSocketRef.set(null);                  closeEvent(new KubernetesClientException("Unhandled exception in reconnect attempt", e));                  close();                }              }            }, nextReconnectInterval(), TimeUnit.MILLISECONDS);          }

而在runWatch()方法中，我们又调用了

 if (this.resourceVersion.get() != null) {      httpUrlBuilder.addQueryParameter("resourceVersion", this.resourceVersion.get());    }

而this.resourceVersion 值的设置在 public void onMessage(WebSocket webSocket, String message) 方法中：

WatchEvent event = readWatchEvent(message);          Object object = event.getObject();          if (object instanceof HasMetadata) {            @SuppressWarnings("unchecked")            T obj = (T) object;            // Dirty cast - should always be valid though            resourceVersion.set(((HasMetadata) obj).getMetadata().getResourceVersion());            Watcher.Action action = Watcher.Action.valueOf(event.getType());            watcher.eventReceived(action, obj);          } else if (object instanceof KubernetesResourceList) {            @SuppressWarnings("unchecked")            KubernetesResourceList list = (KubernetesResourceList) object;            // Dirty cast - should always be valid though            resourceVersion.set(list.getMetadata().getResourceVersion());            Watcher.Action action = Watcher.Action.valueOf(event.getType());            List items = list.getItems();            if (items != null) {              for (HasMetadata item : items) {                watcher.eventReceived(action, (T) item);              }            }

也就是说，假如说如果重联的时候距离上次设置resourceVersion超过了etc保留的最小resourceVersion的话，就会报too old resource version错误:

解决

通过网上查询kubernetes-too-old-resource-version,该Kubernetes Client team memeber 提到了：

Fabric8 does not handle it with plain watch. But it is handling it in SharedInformer API, see ReflectorWatcher. I would recommend using informer API when writing operators since it's better than plain list and watch

也就是说,我们可以用SharedInformer api来实现，而watch机制处理不了这种情况，所以我们可以用SharedInformer实现，截止到2020年11月16日，我们获取到kubernetes-client最新版本，kubernetes-client:4.13.0,编码实现:

val sharedInformerFactory = kubernetesClient.informers()    val podInformer = sharedInformerFactory      .sharedIndexInformerFor(classOf[Pod], classOf[PodList],        new OperationContext().withNamespace("test"), 30 * 1000L)    podInformer.addEventHandler(new ResourceEventHandler[Pod] {      override def onAdd(obj: Pod): Unit = {        eventReceived(obj, "ADD")      }      override def onDelete(obj: Pod, deletedFinalStateUnknown: Boolean): Unit = {        eventReceived(obj, "DELETE")      }      override def onUpdate(oldObj: Pod, newObj: Pod): Unit = {        eventReceived(newObj, "UPDATE")      }      private def idShouldUpdate(pod: Pod): Boolean = {        pod.getMetadata.getLabels.getOrDefault(MERLION_TASK_LABEL, "") == applicationId      }      private def eventReceived(pod: Pod, action: String): Unit = {        if (idShouldUpdate(pod)) {          val podName = pod.getMetadata.getName          logger.info(s"Received job pod update for pod named $podName, action ${action}")          snapshotsStore.updatePod(pod)        }      }    })    sharedInformerFactory.startAllRegisteredInformers() }

其中SharedInformerFactory的机制和k8s Informer机制一样的，能够保证消息的可靠性, 其中最主要的是ReflectorWatcher和Reflector和DefaultSharedIndexInformer,我们简单的分析一下：

public DefaultSharedIndexInformer(Class apiTypeClass, ListerWatcher listerWatcher, long resyncPeriod, OperationContext context, ConcurrentLinkedQueue eventListeners) {    this.resyncCheckPeriodMillis = resyncPeriod;    this.defaultEventHandlerResyncPeriod = resyncPeriod;    this.processor = new SharedProcessor<>();    this.indexer = new Cache();    DeltaFIFO fifo = new DeltaFIFO<>(Cache::metaNamespaceKeyFunc, this.indexer);    this.controller = new Controller<>(apiTypeClass, fifo, listerWatcher, this::handleDeltas, processor::shouldResync, resyncCheckPeriodMillis, context, eventListeners);    controllerThread = new Thread(controller::run, "informer-controller-" + apiTypeClass.getSimpleName());  }

DefaultSharedIndexInformer 中，用DeltaFIFO作为event的存储，而this::handleDeltas的调用是在Controller作为this.queue.pop 的参数processFunc函数被调用，也就是说这个函数来消费fifo里面的event,如下：

private void processLoop() throws Exception {    while (true) {      try {        this.queue.pop(this.processFunc);      } catch (InterruptedException t) {        log.error("DefaultController#processLoop got interrupted {}", t.getMessage(), t);        return;      } catch (Exception e) {        log.error("DefaultController#processLoop recovered from crashing {} ", e.getMessage(), e);        throw e;      }    }

而queue也是DeltaFIFO的形参传进来的，也就是说queue就是fifo，而fifo里面的数据从哪里来呢？在controller::run函数中:

 if (fullResyncPeriod > 0) {          reflector = new Reflector<>(apiTypeClass, listerWatcher, queue, operationContext, fullResyncPeriod);        } else {          reflector = new Reflector<>(apiTypeClass, listerWatcher, queue, operationContext, DEFAULT_PERIOD);        }reflector.listAndWatch()

将会调用reflector.listAndWatch()方法，该方法进行类似k8s的list-watch机制，如下：

public void listAndWatch() throws Exception {    try {      log.info("Started ReflectorRunnable watch for {}", apiTypeClass);      reListAndSync();      resyncExecutor.scheduleWithFixedDelay(this::reListAndSync, 0L, resyncPeriodMillis, TimeUnit.MILLISECONDS);      startWatcher();    } catch (Exception exception) {      store.isPopulated(false);      throw new RejectedExecutionException("Error while starting ReflectorRunnable watch", exception);    }  }

reListAndSync进行全量event数据的拉取，startWatcher进行watch，获取增量event数据，那这个watch是什么呢？如下：

 watch.set(        listerWatcher.watch(new ListOptionsBuilder()          .withWatch(Boolean.TRUE).withResourceVersion(lastSyncResourceVersion.get()).withTimeoutSeconds(null).build(),        operationContext.getNamespace(), operationContext, watcher)      )

这里的watcher在reflector的构造函数中初始化

watcher = new ReflectorWatcher<>(store, lastSyncResourceVersion, this::startWatcher, this::reListAndSync);

而ReflectorWatcher是继承自Watcher，所以也有对应的eventReceived方法和onClose方法，如下:

@Override  public void eventReceived(Action action, T resource) {    if (action == null) {      final String errorMessage = String.format("Unrecognized event %s", resource.getMetadata().getName());      log.error(errorMessage);      throw new KubernetesClientException(errorMessage);    }    log.info("Event received {}", action.name());    switch (action) {      case ERROR:        final String errorMessage = String.format("ERROR event for %s", resource.getMetadata().getName());        log.error(errorMessage);        throw new KubernetesClientException(errorMessage);      case ADDED:        store.add(resource);        break;      case MODIFIED:        store.update(resource);        break;      case DELETED:        store.delete(resource);        break;    }    lastSyncResourceVersion.set(resource.getMetadata().getResourceVersion());    log.debug("{}#Receiving resourceVersion {}", resource.getKind(), lastSyncResourceVersion.get());  }  @Override  public void onClose(KubernetesClientException exception) {    log.error("Watch closing");    Optional.ofNullable(exception)      .map(e -> {        log.debug("Exception received during watch", e);        return exception;      })      .map(KubernetesClientException::getStatus)      .map(Status::getCode)      .filter(c -> c.equals(HttpURLConnection.HTTP_GONE))      .ifPresent(c -> onHttpGone.run());    onClose.run();  }

在eventReceived方法中，所有消息的都会添加到store中也就是fifo的queue中,在onClose方法中，我们看到如果HTTP_GONE，也就是too old resource version的话，会进行onHttpGone.run(),也会进行onClose.run(）,而 onHttpGone就是Reflector的reListAndSync函数，onClose是Reflector的startWatcher函数,也就是说一旦该watcher被关闭，就会重新进行watch。

注意

在kubernetes-client:4.6.2中,WatchConnectionManager onMessage 对于HTTP_GONE的处理是不一样的,如下：

if (status.getCode() == HTTP_GONE) {                logger.info("The resource version {} no longer exists. Scheduling a reconnect.", resourceVersion.get());                resourceVersion.set(null);                scheduleReconnect();              } else {                logger.error("Error received: {}", status.toString());              }

一旦发生了HTTP_GONE,,会把resourceVersion设置为null,也就是获取最新的event，而且会立即重联，而在4.13.0版本和4.2.0版本，是不会立即重联，而是让用户去处理的。

看完上述内容，你们掌握如何进行KubernetesClientException资源版本太旧的分析的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注行业资讯频道，感谢各位的阅读！