enforcing read-timeout for informer-factory processor via property "kubernetes.informer.clientReadTimeout"

Author: yue9944882

spring/src/main/java/io/kubernetes/client/spring/extended/controller/KubernetesInformerFactoryProcessor.java

@@ -17,12 +17,18 @@
 import io.kubernetes.client.informer.SharedInformer;
 import io.kubernetes.client.informer.SharedInformerFactory;
 import io.kubernetes.client.informer.cache.Lister;
+import io.kubernetes.client.informer.cache.ReflectorRunnable;
 import io.kubernetes.client.openapi.ApiClient;
 import io.kubernetes.client.spring.extended.controller.annotation.KubernetesInformer;
 import io.kubernetes.client.spring.extended.controller.annotation.KubernetesInformers;
+import io.kubernetes.client.spring.extended.controller.config.KubernetesInformerProperties;
+import io.kubernetes.client.spring.extended.controller.factory.KubernetesControllerFactory;
 import io.kubernetes.client.util.generic.GenericKubernetesApi;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
@@ -45,6 +51,10 @@ public class KubernetesInformerFactoryProcessor
 
   public static final int ORDER = 0;
 
+  private static final Logger log = LoggerFactory.getLogger(KubernetesControllerFactory.class);
+
+  @Autowired private KubernetesInformerProperties informerProperties;
+
   private ConfigurableListableBeanFactory beanFactory;
 
   @Override
@@ -122,6 +132,20 @@ private <T extends KubernetesObject> Lister<T> lister(Class<T> type) {
   private <T extends KubernetesObject> SharedInformer<T> informer(
       Class<T> type, KubernetesInformer kubernetesInformer) {
     ApiClient apiClient = this.beanFactory.getBean(ApiClient.class);
+
+    if (apiClient.getHttpClient().readTimeoutMillis() > 0) {
+      log.warn(
+          "Enforcing read-timeout of the ApiClient {} to {} so that the watch connection won't abort from client-side",
+          apiClient,
+          ReflectorRunnable.REFLECTOR_WATCH_CLIENTSIDE_TIMEOUT);
+      apiClient.setHttpClient(
+          apiClient
+              .getHttpClient()
+              .newBuilder()
+              .readTimeout(informerProperties.getClientReadTimeout())
+              .build());
+    }
+
     SharedInformerFactory sharedInformerFactory =
         this.beanFactory.getBean(SharedInformerFactory.class);
     final GenericKubernetesApi api =

spring/src/main/java/io/kubernetes/client/spring/extended/controller/config/KubernetesInformerAutoConfiguration.java

@@ -18,11 +18,15 @@
 import io.kubernetes.client.util.ClientBuilder;
 import java.io.IOException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 @Configuration(proxyBeanMethods = false)
 @ConditionalOnKubernetesInformerEnabled
+@EnableConfigurationProperties({
+  KubernetesInformerProperties.class,
+})
 public class KubernetesInformerAutoConfiguration {
 
   @Bean

spring/src/main/java/io/kubernetes/client/spring/extended/controller/config/KubernetesInformerProperties.java

@@ -0,0 +1,32 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package io.kubernetes.client.spring.extended.controller.config;
+
+import io.kubernetes.client.informer.cache.ReflectorRunnable;
+import java.time.Duration;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties("kubernetes.informer")
+public class KubernetesInformerProperties {
+
+  private Duration clientReadTimeout = ReflectorRunnable.REFLECTOR_WATCH_CLIENTSIDE_TIMEOUT;
+
+  public Duration getClientReadTimeout() {
+    return clientReadTimeout;
+  }
+
+  public KubernetesInformerProperties setClientReadTimeout(Duration clientReadTimeout) {
+    this.clientReadTimeout = clientReadTimeout;
+    return this;
+  }
+}

util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java

@@ -37,6 +37,8 @@ public class ReflectorRunnable<
         ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
     implements Runnable {
 
+  public static Duration REFLECTOR_WATCH_CLIENTSIDE_TIMEOUT = Duration.ofMinutes(5);
+
   private static final Logger log = LoggerFactory.getLogger(ReflectorRunnable.class);
 
   private String lastSyncResourceVersion;
@@ -113,7 +115,7 @@ public void run() {
                   new CallGeneratorParams(
                       Boolean.TRUE,
                       lastSyncResourceVersion,
-                      Long.valueOf(Duration.ofMinutes(5).getSeconds()).intValue()));
+                      Long.valueOf(REFLECTOR_WATCH_CLIENTSIDE_TIMEOUT.getSeconds()).intValue()));
 
           synchronized (this) {
             if (!isActive.get()) {