Merge branch 'master' into openshift_visit

Author: shawkins

CHANGELOG.md

@@ -8,12 +8,15 @@
 * Fix #3295: Fix wrong kind getting registered in KubernetesDeserializer in SharedInformerFactory
 * Fix #3318: Informer relist add/update should not always be sync events
 * Fix #3328: Allow for generic watches of known types
+* Fix #3240: MicroTime serialises incorrectly; add custom serializer/deserializer for MicroTime
 * Fix #3329: Moved the parsing of resource(String) to the common base client
 * Fix #3330: Added usage of the openshift specific handlers for resource(String/HasMetadata) to pickup the right Readiness
 
 #### Improvements
+* Fix #3285: adding waiting for list contexts `Informable.informOnCondition`
 * Fix #3284: refined how builders are obtained / used by HasMetadataOperation
 * Fix #3314: Add more detailed information about what is generated by the CRD generator
+* Fix #3307: refined the support for `Readiable.isReady`
 
 #### Dependency Upgrade
 * Update Tekton Pipeline Model to v0.25.0
@@ -22,6 +25,10 @@
 * Fix #3291: Retrying the HTTP operation in case of IOException too
 * Fix #2712: Add support for watching logs in multi-container Controller resources (Deployments, StatefulSets, ReplicaSet etc)
 
+#### _**Note**_: Breaking changes in the API
+##### DSL Changes:
+- #3307 `Readiable.isReady` returns a boolean rather than a Boolean
+
 ### 5.5.0 (2021-06-30)
 
 #### Bugs

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

@@ -23,7 +23,10 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 public interface Informable<T> {
 
@@ -34,6 +37,21 @@ public interface Informable<T> {
    */
   Informable<T> withIndexers(Map<String, Function<T, List<String>>> indexers);
 
+  /**
+   * Similar to a {@link Watch}, but will attempt to handle failures after successfully started.
+   * and provides a store of all the current resources.
+   * <p>This returned informer will not support resync.
+   * <p>This call will be blocking for the initial list and watch.
+   * <p>You are expected to call stop to terminate the underlying Watch.
+   * <p>Additional handlers can be added, but processing of the events will be in the websocket thread, 
+   * so consider non-blocking handler operations for more than one handler.
+   * 
+   * @return a running {@link SharedIndexInformer}
+   */
+  default SharedIndexInformer<T> inform() {
+    return inform(null, 0);
+  }
+  
   /**
    * Similar to a {@link Watch}, but will attempt to handle failures after successfully started.
    * and provides a store of all the current resources.
@@ -56,12 +74,21 @@ default SharedIndexInformer<T> inform(ResourceEventHandler<T> handler) {
    * <p>This call will be blocking for the initial list and watch.
    * <p>You are expected to call stop to terminate the underlying Watch.
    * <p>Additional handlers can be added, but processing of the events will be in the websocket thread, 
-   * so consider non-blocking handleroperations for more than one handler.
+   * so consider non-blocking handler operations for more than one handler.
    * 
    * @param handler to notify
    * @param resync the resync period or 0 for no resync
    * @return a running {@link SharedIndexInformer}
    */
   SharedIndexInformer<T> inform(ResourceEventHandler<T> handler, long resync);
+  
+  /**
+   * Return a {@link Future} when the list at this context satisfies the given {@link Predicate}.
+   * The predicate will be tested against the state of the underlying informer store on every event.
+   * The returned future should be cancelled by the caller if not waiting for completion to close the underlying informer
+   * @param condition the {@link Predicate} to test
+   * @return a {@link CompletableFuture} of the list of items after the condition is met
+   */
+  CompletableFuture<List<T>> informOnCondition(Predicate<List<T>> condition);
 
 }

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

@@ -17,5 +17,9 @@
 
 public interface Readiable {
 
-  Boolean isReady();
+  /**
+   * Check if the resource is ready.  If no readiness check exists, this is just an existence check.
+   * @return true if the resource exists and is ready.
+   */
+  boolean isReady();
 }

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

@@ -25,5 +25,6 @@ public interface VisitFromServerGetWatchDeleteRecreateWaitApplicable<T> extends
   Watchable<Watcher<T>>,
   Waitable<T, T>,
   VisitFromServerWritable<T>,
-  DryRunable<VisitFromServerWritable<T>> {
+  DryRunable<VisitFromServerWritable<T>>,
+  Readiable {
 }

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

@@ -18,8 +18,6 @@
 import io.fabric8.kubernetes.api.model.ObjectReference;
 import io.fabric8.kubernetes.client.dsl.WritableOperation;
 import io.fabric8.kubernetes.client.utils.CreateOrReplaceHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.builder.TypedVisitor;
 import io.fabric8.kubernetes.api.builder.Visitor;
@@ -78,7 +76,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -94,7 +92,6 @@ public class BaseOperation<T extends HasMetadata, L extends KubernetesResourceLi
   MixedOperation<T, L, R>,
   Resource<T> {
 
-  private static final Logger LOG = LoggerFactory.getLogger(BaseOperation.class);
   private static final String READ_ONLY_UPDATE_EXCEPTION_MESSAGE = "Cannot update read-only resources";
   private static final String READ_ONLY_EDIT_EXCEPTION_MESSAGE = "Cannot edit read-only resources";
 
@@ -953,8 +950,12 @@ public Readiness getReadiness() {
   }
 
   @Override
-  public final Boolean isReady() {
-    return getReadiness().isReady(get());
+  public final boolean isReady() {
+    T item = fromServer().get();
+    if (item == null) {
+      return false;
+    }
+    return getReadiness().isReady(item);
   }
 
   @Override
@@ -964,56 +965,70 @@ public T waitUntilReady(long amount, TimeUnit timeUnit) {
 
   @Override
   public T waitUntilCondition(Predicate<T> condition, long amount, TimeUnit timeUnit) {
-    CompletableFuture<T> future = new CompletableFuture<>();
-    // tests the condition, trapping any exceptions
-    Consumer<T> tester = obj -> {
+    CompletableFuture<T> futureCondition = informOnCondition(l -> {
+      if (l.isEmpty()) {
+        return condition.test(null);
+      }
+      return condition.test(l.get(0));
+    }).thenApply(l -> l.isEmpty() ? null : l.get(0));
+    
+    if (!Utils.waitUntilReady(futureCondition, amount, timeUnit)) {
+      futureCondition.cancel(true);
+      T i = getItem();
+      if (i != null) {
+        throw new KubernetesClientTimeoutException(i, amount, timeUnit);
+      }
+      throw new KubernetesClientTimeoutException(getKind(), getName(), getNamespace(), amount, timeUnit);
+    }
+    return futureCondition.getNow(null);
+  }
+  
+  @Override
+  public CompletableFuture<List<T>> informOnCondition(Predicate<List<T>> condition) {
+    CompletableFuture<List<T>> future = new CompletableFuture<>();
+    AtomicReference<Runnable> tester = new AtomicReference<>();
+    
+    // create an informer that supplies the tester with events and empty list handling
+    SharedIndexInformer<T> informer = this.createInformer(0);
+    
+    // prevent unnecessary watches and handle closure
+    future.whenComplete((r, t) -> informer.stop());
+    
+    // use the cache to evaluate the list predicate, trapping any exceptions
+    Runnable test = () -> {
       try {
-        if (condition.test(obj)) {
-          future.complete(obj);
+        // could skip if lastResourceVersion has not changed
+        List<T> list = informer.getStore().list();
+        if (condition.test(list)) {
+          future.complete(list);
         }
       } 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>() {
-
+    tester.set(test);
+    
+    informer.addEventHandler(new ResourceEventHandler<T>() {
       @Override
       public void onAdd(T obj) {
-        tester.accept(obj);
+        test.run();
       }
-
       @Override
-      public void onUpdate(T oldObj, T newObj) {
-        tester.accept(newObj);
+      public void onDelete(T obj, boolean deletedFinalStateUnknown) {
+        test.run();
       }
-
       @Override
-      public void onDelete(T obj, boolean deletedFinalStateUnknown) {
-        tester.accept(null);
+      public void onUpdate(T oldObj, T newObj) {
+        test.run();
       }
-    })) {
-      // prevent unnecessary watches
-      future.whenComplete((r,t) -> informer.stop());
-      informer.run();
-      return future.get(amount, timeUnit);
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-      throw KubernetesClientException.launderThrowable(e.getCause());
-    } catch (ExecutionException e) {
-      throw KubernetesClientException.launderThrowable(e.getCause());
-    } catch (TimeoutException e) {
-      T i = getItem();
-      if (i != null) {
-        throw new KubernetesClientTimeoutException(i, amount, timeUnit);
+      @Override
+      public void onNothing() {
+        test.run();
       }
-      throw new KubernetesClientTimeoutException(getKind(), getName(), getNamespace(), amount, timeUnit);
-    }
-  }
+    });
+    informer.run();
+    return future;
+  }  
 
   public void setType(Class<T> type) {
     this.type = type;
@@ -1041,14 +1056,17 @@ public Informable<T> withIndexers(Map<String, Function<T, List<String>>> indexer
 
   @Override
   public SharedIndexInformer<T> inform(ResourceEventHandler<T> handler, long resync) {
-    DefaultSharedIndexInformer<T, L> result = createInformer(resync, null, handler);
+    DefaultSharedIndexInformer<T, L> result = createInformer(resync);
+    if (handler != null) {
+      result.addEventHandler(handler);
+    }
     // synchronous start list/watch must succeed in the calling thread
     // initial add events will be processed in the calling thread as well
     result.run();
     return result;
   }
 
-  private DefaultSharedIndexInformer<T, L> createInformer(long resync, Consumer<L> onList, ResourceEventHandler<T> handler) {
+  private DefaultSharedIndexInformer<T, L> createInformer(long resync) {
     T i = getItem();
     String name = (Utils.isNotNullOrEmpty(getName()) || i != null) ? checkName(i) : null;
 
@@ -1060,11 +1078,7 @@ public L list(ListOptions params, String namespace, OperationContext context) {
         if (name != null) {
           params.setFieldSelector("metadata.name="+name);
         }
-        L result = BaseOperation.this.list(params);
-        if (onList != null) {
-          onList.accept(result);
-        }
-        return result;
+        return BaseOperation.this.list(params);
       }
 
       @Override
@@ -1075,9 +1089,6 @@ public Watch watch(ListOptions params, String namespace, OperationContext contex
     if (indexers != null) {
       informer.addIndexers(indexers);
     }
-    if (handler != null) {
-      informer.addEventHandler(handler);
-    }
     return informer;
   }
 }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/NamespaceVisitFromServerGetWatchDeleteRecreateWaitApplicableImpl.java

@@ -126,7 +126,7 @@ public HasMetadata apply() {
 
   @Override
   public HasMetadata createOrReplace() {
-    HasMetadata meta = acceptVisitors(item, visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     String namespaceToUse = meta.getMetadata().getNamespace();
     if (Boolean.TRUE.equals(deletingExisting)) {
@@ -143,29 +143,22 @@ public Waitable<HasMetadata, HasMetadata> createOrReplaceAnd() {
   @Override
   public Boolean delete() {
     //First pass check before deleting
-    HasMetadata meta = acceptVisitors(item, visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     return h.delete(client, config, meta.getMetadata().getNamespace(), propagationPolicy, gracePeriodSeconds, meta, dryRun);
   }
 
   @Override
   public HasMetadata get() {
+    HasMetadata meta = acceptVisitors(item, visitors);
     if (fromServer) {
-      HasMetadata meta = acceptVisitors(item, visitors);
       ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
-      HasMetadata reloaded = h.reload(client, config, meta.getMetadata().getNamespace(), meta);
-      if (reloaded != null) {
-        HasMetadata edited = reloaded;
-        //Let's apply any visitor that might have been specified.
-        for (Visitor v : visitors) {
-          edited = h.edit(edited).accept(v).build();
-        }
-        return edited;
+      meta = h.reload(client, config, meta.getMetadata().getNamespace(), meta);
+      if (meta != null) {
+        return acceptVisitors(meta, visitors);
       }
-      return null;
-    } else {
-      return acceptVisitors(item, visitors);
     }
+    return meta;
   }
 
   @Override
@@ -216,21 +209,21 @@ public Waitable<HasMetadata, HasMetadata> withWaitRetryBackoff(long initialBacko
 
   @Override
   public Watch watch(Watcher<HasMetadata> watcher) {
-    HasMetadata meta = acceptVisitors(item, visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     return h.watch(client, config, meta.getMetadata().getNamespace(), meta, watcher);
   }
 
   @Override
   public Watch watch(String resourceVersion, Watcher<HasMetadata> watcher) {
-    HasMetadata meta = acceptVisitors(item, visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     return h.watch(client, config, meta.getMetadata().getNamespace(), meta, resourceVersion, watcher);
   }
 
   @Override
   public Watch watch(ListOptions options, Watcher<HasMetadata> watcher) {
-    HasMetadata meta = acceptVisitors(item, visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     return h.watch(client, config, meta.getMetadata().getNamespace(), meta, options, watcher);
   }
@@ -240,13 +233,17 @@ protected Readiness getReadiness() {
   }
 
   @Override
-  public final Boolean isReady() {
-    return getReadiness().isReady(get());
+  public final boolean isReady() {
+    HasMetadata meta = fromServer().get();
+    if (meta == null) {
+      return false;
+    }
+    return getReadiness().isReady(meta);
   }
 
   @Override
   public HasMetadata waitUntilReady(long amount, TimeUnit timeUnit) {
-    HasMetadata meta = acceptVisitors(get(), visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     return h.waitUntilReady(client, config, meta.getMetadata().getNamespace(), meta, amount, timeUnit);
   }
@@ -259,12 +256,11 @@ public VisitFromServerWritable<HasMetadata> dryRun(boolean isDryRun) {
   @Override
   public HasMetadata waitUntilCondition(Predicate<HasMetadata> condition, long amount,
     TimeUnit timeUnit) {
-    HasMetadata meta = acceptVisitors(get(), visitors);
+    HasMetadata meta = get();
     ResourceHandler<HasMetadata, ?> h = handlerOf(meta);
     return h.waitUntilCondition(client, config, meta.getMetadata().getNamespace(), meta, condition, amount, timeUnit);
   }
 
-
   private static HasMetadata acceptVisitors(HasMetadata item, List<Visitor> visitors) {
     ResourceHandler<HasMetadata, ?> h = handlerOf(item);
     VisitableBuilder<HasMetadata, ?> builder = h.edit(item);