discard client on watch failure; extend socket timeout

Author: russgold

operator/src/main/java/oracle/kubernetes/operator/Watcher.java

@@ -126,7 +126,13 @@ private void watchForEvents() {
                 .withResourceVersion(resourceVersion.toString())
                 .withTimeoutSeconds(tuning.watchLifetime))) {
       while (watch.hasNext()) {
-        Watch.Response<T> item = watch.next();
+        Watch.Response<T> item;
+        try {
+          item = watch.next();
+        } catch (Throwable e) {
+          watch.discardClient();
+          throw e;
+        }
 
         if (isStopping()) setIsDraining(true);
         if (isDraining()) continue;

operator/src/main/java/oracle/kubernetes/operator/builders/WatchBuilder.java

@@ -7,8 +7,6 @@
 import com.squareup.okhttp.Call;
 import io.kubernetes.client.ApiClient;
 import io.kubernetes.client.ApiException;
-import io.kubernetes.client.ProgressRequestBody;
-import io.kubernetes.client.ProgressResponseBody;
 import io.kubernetes.client.apis.BatchV1Api;
 import io.kubernetes.client.apis.CoreV1Api;
 import io.kubernetes.client.models.V1ConfigMap;
@@ -33,6 +31,8 @@ public class WatchBuilder {
   /** Ignored for watches. */
   private static final String START_LIST = null;
 
+  private static final int ADDITIONAL_TIMEOUT_FOR_SOCKET = 60;
+
   private static WatchFactory FACTORY = new WatchFactoryImpl();
 
   private CallParamsImpl callParams = new CallParamsImpl();
@@ -92,7 +92,7 @@ private class ListNamespacedServiceCall implements BiFunction<ApiClient, CallPar
     @Override
     public Call apply(ApiClient client, CallParams callParams) {
       // Ensure that client doesn't time out before call or watch
-      client.getHttpClient().setReadTimeout(callParams.getTimeoutSeconds(), TimeUnit.SECONDS);
+      client.getHttpClient().setReadTimeout(getSocketTimeout(callParams), TimeUnit.SECONDS);
 
       try {
         return new CoreV1Api(client)
@@ -137,7 +137,7 @@ private class ListPodCall implements BiFunction<ApiClient, CallParams, Call> {
     @Override
     public Call apply(ApiClient client, CallParams callParams) {
       // Ensure that client doesn't time out before call or watch
-      client.getHttpClient().setReadTimeout(callParams.getTimeoutSeconds(), TimeUnit.SECONDS);
+      client.getHttpClient().setReadTimeout(getSocketTimeout(callParams), TimeUnit.SECONDS);
 
       try {
         return new CoreV1Api(client)
@@ -182,7 +182,7 @@ private class ListJobCall implements BiFunction<ApiClient, CallParams, Call> {
     @Override
     public Call apply(ApiClient client, CallParams callParams) {
       // Ensure that client doesn't time out before call or watch
-      client.getHttpClient().setReadTimeout(callParams.getTimeoutSeconds(), TimeUnit.SECONDS);
+      client.getHttpClient().setReadTimeout(getSocketTimeout(callParams), TimeUnit.SECONDS);
 
       try {
         return new BatchV1Api(client)
@@ -227,7 +227,7 @@ private class ListEventCall implements BiFunction<ApiClient, CallParams, Call> {
     @Override
     public Call apply(ApiClient client, CallParams callParams) {
       // Ensure that client doesn't time out before call or watch
-      client.getHttpClient().setReadTimeout(callParams.getTimeoutSeconds(), TimeUnit.SECONDS);
+      client.getHttpClient().setReadTimeout(getSocketTimeout(callParams), TimeUnit.SECONDS);
 
       try {
         return new CoreV1Api(client)
@@ -272,7 +272,7 @@ private class ListDomainsCall implements BiFunction<ApiClient, CallParams, Call>
     @Override
     public Call apply(ApiClient client, CallParams callParams) {
       // Ensure that client doesn't time out before call or watch
-      client.getHttpClient().setReadTimeout(callParams.getTimeoutSeconds(), TimeUnit.SECONDS);
+      client.getHttpClient().setReadTimeout(getSocketTimeout(callParams), TimeUnit.SECONDS);
 
       try {
         return new WeblogicApi(client)
@@ -320,7 +320,7 @@ private class ListNamespacedConfigMapCall implements BiFunction<ApiClient, CallP
     @Override
     public Call apply(ApiClient client, CallParams callParams) {
       // Ensure that client doesn't time out before call or watch
-      client.getHttpClient().setReadTimeout(callParams.getTimeoutSeconds(), TimeUnit.SECONDS);
+      client.getHttpClient().setReadTimeout(getSocketTimeout(callParams), TimeUnit.SECONDS);
 
       try {
         return new CoreV1Api(client)
@@ -343,6 +343,10 @@ public Call apply(ApiClient client, CallParams callParams) {
     }
   }
 
+  private Integer getSocketTimeout(CallParams callParams) {
+    return callParams.getTimeoutSeconds() + ADDITIONAL_TIMEOUT_FOR_SOCKET;
+  }
+
   /**
    * Sets a value for the fieldSelector parameter for the call that will set up this watch. Defaults
    * to null.
@@ -355,7 +359,8 @@ public WatchBuilder withFieldSelector(String fieldSelector) {
     return this;
   }
 
-  public WatchBuilder withIncludeUninitialized(Boolean includeUninitialized) {
+  @SuppressWarnings("SameParameterValue")
+  WatchBuilder withIncludeUninitialized(Boolean includeUninitialized) {
     callParams.setIncludeUninitialized(includeUninitialized);
     return this;
   }
@@ -370,16 +375,12 @@ public WatchBuilder withLabelSelectors(String... labelSelectors) {
     return this;
   }
 
-  public WatchBuilder withLimit(Integer limit) {
+  @SuppressWarnings("SameParameterValue")
+  WatchBuilder withLimit(Integer limit) {
     callParams.setLimit(limit);
     return this;
   }
 
-  public WatchBuilder withPrettyPrinting() {
-    callParams.setPretty("true");
-    return this;
-  }
-
   public WatchBuilder withResourceVersion(String resourceVersion) {
     callParams.setResourceVersion(resourceVersion);
     return this;
@@ -390,17 +391,6 @@ public WatchBuilder withTimeoutSeconds(Integer timeoutSeconds) {
     return this;
   }
 
-  public WatchBuilder withProgressListener(ProgressResponseBody.ProgressListener progressListener) {
-    callParams.setProgressListener(progressListener);
-    return this;
-  }
-
-  public WatchBuilder withProgressRequestListener(
-      ProgressRequestBody.ProgressRequestListener progressRequestListener) {
-    callParams.setProgressRequestListener(progressRequestListener);
-    return this;
-  }
-
   static class WatchFactoryImpl implements WatchFactory {
     @Override
     public <T> WatchI<T> createWatch(
@@ -411,7 +401,7 @@ public <T> WatchI<T> createWatch(
         throws ApiException {
       ApiClient client = pool.take();
       try {
-        return new WatchImpl<T>(
+        return new WatchImpl<>(
             pool,
             client,
             Watch.createWatch(

operator/src/main/java/oracle/kubernetes/operator/builders/WatchI.java

@@ -14,4 +14,6 @@
  * @param <T> the generic object type
  */
 public interface WatchI<T>
-    extends Iterable<Watch.Response<T>>, Iterator<Watch.Response<T>>, java.io.Closeable {}
+    extends Iterable<Watch.Response<T>>, Iterator<Watch.Response<T>>, java.io.Closeable {
+  default void discardClient() {}
+}

operator/src/main/java/oracle/kubernetes/operator/builders/WatchImpl.java

@@ -8,14 +8,15 @@
 import io.kubernetes.client.util.Watch;
 import java.io.IOException;
 import java.util.Iterator;
+import javax.annotation.Nonnull;
 import oracle.kubernetes.operator.helpers.Pool;
 
 /**
  * A pass-through implementation of the Kubernetes Watch class which implements a facade interface.
  */
 public class WatchImpl<T> implements WatchI<T> {
   private final Pool<ApiClient> pool;
-  private final ApiClient client;
+  private ApiClient client;
   private Watch<T> impl;
 
   WatchImpl(Pool<ApiClient> pool, ApiClient client, Watch<T> impl) {
@@ -31,6 +32,12 @@ public void close() throws IOException {
   }
 
   @Override
+  public void discardClient() {
+    client = pool.take();
+  }
+
+  @Override
+  @Nonnull
   public Iterator<Watch.Response<T>> iterator() {
     return impl.iterator();
   }