switching to an informer based implementation

also fully removing the additional operation/context fields

Author: shawkins

CHANGELOG.md

@@ -15,6 +15,7 @@
 * Fix #3216: made the mock server aware of apiVersions
 * Fix #3225: Pod metric does not have corresponding label selector variant
 * Fix #3243: pipes provided to exec command are no longer closed on connection close, so that client can fully read the buffer after the command finishes.
+* Fix #3271: waitUntilReady and waitUntilCondition should handle resource too old
 
 #### Improvements
 * Fix #3078: adding javadocs to further clarify patch, edit, replace, etc. and note the possibility of items being modified.
@@ -54,6 +55,7 @@
 ##### DSL Changes:
 - #3127 `StatusUpdatable.updateStatus` deprecated, please use patchStatus, editStatus, or replaceStatus
 - #3239 deprecated methods on SharedInformerFactory directly dealing with the OperationContext, withName, and withNamespace - the Informable interface should be used instead.
+- #3271 waitUntilReady and waitUntilCondition with throw a KubernetesClientTimeoutException instead of an IllegalArgumentException on timeout
 
 ##### Util Changes:
 - #3197 `Utils.waitUntilReady` now accepts a Future, rather than a BlockingQueue

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/KubernetesClientTimeoutException.java

@@ -30,6 +30,11 @@ public class KubernetesClientTimeoutException extends KubernetesClientException
   private static final String KNOWS_RESOURCES_FORMAT = "Timed out waiting for [%d] milliseconds for multiple resources. %s";
 
   private final List<HasMetadata> resourcesNotReady;
+  
+  public KubernetesClientTimeoutException(String kind, String name, String namespace, long amount, TimeUnit timeUnit) {
+    super(String.format(RESOURCE_FORMAT, timeUnit.toMillis(amount), kind, name, namespace));
+    this.resourcesNotReady = Collections.emptyList();
+  }
 
   public KubernetesClientTimeoutException(HasMetadata resource, long amount, TimeUnit timeUnit) {
     super(String.format(RESOURCE_FORMAT, timeUnit.toMillis(amount), resource.getKind(), resource.getMetadata().getName(), resource.getMetadata().getNamespace()));

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/Waitable.java

@@ -20,7 +20,9 @@
 
 public interface Waitable<T, P> {
 
+  @Deprecated
   long DEFAULT_INITIAL_BACKOFF_MILLIS = 5L;
+  @Deprecated
   double DEFAULT_BACKOFF_MULTIPLIER = 2d;
 
   T waitUntilReady(long amount, TimeUnit timeUnit) throws InterruptedException;

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/base/BaseOperation.java

@@ -16,7 +16,6 @@
 package io.fabric8.kubernetes.client.dsl.base;
 
 import io.fabric8.kubernetes.api.model.ObjectReference;
-import io.fabric8.kubernetes.client.WatcherException;
 import io.fabric8.kubernetes.client.dsl.WritableOperation;
 import io.fabric8.kubernetes.client.utils.CreateOrReplaceHelper;
 import org.slf4j.Logger;
@@ -38,6 +37,7 @@
 import io.fabric8.kubernetes.client.Config;
 import io.fabric8.kubernetes.client.ConfigBuilder;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.KubernetesClientTimeoutException;
 import io.fabric8.kubernetes.client.OperationInfo;
 import io.fabric8.kubernetes.client.ResourceNotFoundException;
 import io.fabric8.kubernetes.client.Watch;
@@ -59,7 +59,6 @@
 import io.fabric8.kubernetes.client.informers.impl.DefaultSharedIndexInformer;
 import io.fabric8.kubernetes.client.internal.readiness.Readiness;
 import io.fabric8.kubernetes.client.utils.HttpClientUtils;
-import io.fabric8.kubernetes.client.utils.KubernetesResourceUtil;
 import io.fabric8.kubernetes.client.utils.URLUtils;
 import io.fabric8.kubernetes.client.utils.Utils;
 import io.fabric8.kubernetes.client.utils.WatcherToggle;
@@ -73,11 +72,11 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -89,8 +88,6 @@
 import okhttp3.HttpUrl;
 import okhttp3.Request;
 
-import static java.util.concurrent.TimeUnit.NANOSECONDS;
-
 public class BaseOperation<T extends HasMetadata, L extends KubernetesResourceList<T>, R extends Resource<T>>
   extends OperationSupport
   implements
@@ -1123,34 +1120,53 @@ public T waitUntilReady(long amount, TimeUnit timeUnit) throws InterruptedExcept
   @Override
   public T waitUntilCondition(Predicate<T> condition, long amount, TimeUnit timeUnit)
     throws InterruptedException {
+    
+    CompletableFuture<T> future = new CompletableFuture<>();
+    // tests the condition, trapping any exceptions
+    Consumer<T> tester = obj -> {
+      try {
+        if (condition.test(obj)) {
+          future.complete(obj);
+        }
+      } catch (Exception e) {
+        future.completeExceptionally(e);
+      }
+    };
+    // start an informer that supplies the tester with events and empty list handling
+    try (SharedIndexInformer<T> informer = this.createInformer(0, l -> {
+      if (l.getItems().isEmpty()) {
+        tester.accept(null);
+      }
+    }, new ResourceEventHandler<T>() {
+      
+      @Override
+      public void onAdd(T obj) {
+        tester.accept(obj);
+      }
 
-    long remainingNanosToWait = timeUnit.toNanos(amount);
-    while (remainingNanosToWait > 0) {
-
-      T item = fromServer().get();
-      if (condition.test(item)) {
-        return item;
+      @Override
+      public void onUpdate(T oldObj, T newObj) {
+        tester.accept(newObj);
       }
 
-      final WaitForConditionWatcher<T> watcher = new WaitForConditionWatcher<>(condition);
-      final long startTime = System.nanoTime();
-      try (Watch ignored = watch(KubernetesResourceUtil.getResourceVersion(item), watcher)) {
-        return watcher.getFuture().get(remainingNanosToWait, NANOSECONDS);
-      } catch (ExecutionException e) {
-        Throwable cause = e.getCause();
-        if (cause instanceof WatcherException && ((WatcherException)cause).isHttpGone()) {
-          LOG.debug("Restarting the watch due to http gone");
-          remainingNanosToWait -= (System.nanoTime() - startTime);
-          continue;
-        }
-        throw KubernetesClientException.launderThrowable(cause);
-      } catch (TimeoutException e) {
-        break;
+      @Override
+      public void onDelete(T obj, boolean deletedFinalStateUnknown) {
+        tester.accept(null);
       }
+    })) {
+      // prevent unnecessary watches
+      future.whenComplete((r,t) -> informer.stop());
+      informer.run();
+      return future.get(amount, timeUnit);
+    } catch (ExecutionException e) {
+      throw KubernetesClientException.launderThrowable(e.getCause());
+    } catch (TimeoutException e) {
+      T item = getItem();
+      if (item != null) {
+        throw new KubernetesClientTimeoutException(item, amount, timeUnit);
+      }
+      throw new KubernetesClientTimeoutException(getKind(), getName(), getNamespace(), amount, timeUnit);
     }
-
-    LOG.debug("ran out of time waiting for watcher, wait condition not met");
-    throw new IllegalArgumentException(type.getSimpleName() + " with name:[" + name + "] in namespace:[" + namespace + "] matching condition not found!");
   }
 
   public void setType(Class<T> type) {
@@ -1179,30 +1195,41 @@ public Informable<T> withIndexers(Map<String, Function<T, List<String>>> indexer
 
   @Override
   public SharedIndexInformer<T> inform(ResourceEventHandler<T> handler, long resync) {
-    // convert the name into something listable
-    FilterWatchListDeletable<T, L> baseOperation =
-        getName() == null ? this : withFields(Collections.singletonMap("metadata.name", getName()));
+    DefaultSharedIndexInformer<T, L> result = createInformer(resync, null, handler);
+    result.run();
+    return result;
+  }
 
+  private DefaultSharedIndexInformer<T, L> createInformer(long resync, Consumer<L> onList, ResourceEventHandler<T> handler) {
+    T item = getItem();
+    String name = (Utils.isNotNullOrEmpty(getName()) || item != null) ? checkName(item) : null;
+    
     // use the local context / namespace
     DefaultSharedIndexInformer<T, L> informer = new DefaultSharedIndexInformer<>(getType(), new ListerWatcher<T, L>() {
       @Override
       public L list(ListOptions params, String namespace, OperationContext context) {
-        return baseOperation.list(params);
+        // convert the name into something listable
+        if (name != null) {
+          params.setFieldSelector("metadata.name="+name);
+        }
+        L result = BaseOperation.this.list(params);
+        if (onList != null) {
+          onList.accept(result);
+        }
+        return result;
       }
 
       @Override
       public Watch watch(ListOptions params, String namespace, OperationContext context, Watcher<T> watcher) {
-        return baseOperation.watch(params, watcher);
+        return BaseOperation.this.watch(params, watcher);
       }
     }, resync, context, Runnable::run); // just run the event notification in the websocket thread
-    if (handler != null) {
-      informer.addEventHandler(handler);
-    }
     if (indexers != null) {
       informer.addIndexers(indexers);
     }
-    // synchronous start list/watch must succeed in the calling thread
-    informer.run();
+    if (handler != null) {
+      informer.addEventHandler(handler);
+    }
     return informer;
   }
 }