List continuation and watch bookmark support (#1881)

* Allow watch bookmarks

* Work in progress

* Work in progress

* Work in progress

* Work in progress

* AsyncRequestStep tests

* Fix tests

* Bookmark tests

* Clarify names

* Work in progress

* Better support for namespace lists spanning REST calls

* Revert changes to charts

* Bug fixes

* Disassociate CRD creation from namespace startup

* Make unit-test more generic

* Clarify continue pattern

* Save continue for reinvoke of async request

* Correct method name

Author: rjeberhard

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

@@ -916,7 +916,8 @@ public NextAction onSuccess(Packet packet, CallResponse<V1PodList> callResponse)
           }
         }
       }
-      return doNext(packet);
+
+      return doContinueListOrNext(callResponse, packet);
     }
   }
 
@@ -972,7 +973,7 @@ public NextAction onSuccess(Packet packet, CallResponse<V1ServiceList> callRespo
         }
       }
 
-      return doNext(packet);
+      return doContinueListOrNext(callResponse, packet);
     }
   }
 

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

@@ -62,6 +62,7 @@
 import oracle.kubernetes.operator.logging.MessageKeys;
 import oracle.kubernetes.operator.rest.RestConfigImpl;
 import oracle.kubernetes.operator.rest.RestServer;
+import oracle.kubernetes.operator.steps.ActionResponseStep;
 import oracle.kubernetes.operator.steps.ConfigMapAfterStep;
 import oracle.kubernetes.operator.work.Component;
 import oracle.kubernetes.operator.work.Container;
@@ -207,13 +208,12 @@ private static void begin() {
 
       Step strategy = Step.chain(
           new InitializeNamespacesSecurityStep(configuredDomainNamespaces),
-          new NamespaceRulesReviewStep());
+          new NamespaceRulesReviewStep(),
+          CrdHelper.createDomainCrdStep(version, productVersion));
       if (!DomainNamespaceSelectionStrategy.Dedicated.equals(selectionStrategy)) {
         strategy = Step.chain(strategy, readExistingNamespaces(selectionStrategy, configuredDomainNamespaces, false));
       } else {
-        strategy = Step.chain(strategy, CrdHelper.createDomainCrdStep(
-                version, productVersion,
-                new StartNamespacesStep(configuredDomainNamespaces, false)));
+        strategy = Step.chain(strategy, new StartNamespacesStep(configuredDomainNamespaces, false));
       }
       runSteps(
           strategy,
@@ -322,14 +322,13 @@ static Step createDomainRecheckSteps(DateTime now) {
 
     Step strategy = Step.chain(
         new InitializeNamespacesSecurityStep(configuredDomainNamespaces),
-        new NamespaceRulesReviewStep());
+        new NamespaceRulesReviewStep(),
+        CrdHelper.createDomainCrdStep(version, productVersion));
     if (!DomainNamespaceSelectionStrategy.Dedicated.equals(selectionStrategy)) {
       strategy = Step.chain(strategy, readExistingNamespaces(
               selectionStrategy, configuredDomainNamespaces, isFullRecheck));
     } else {
-      strategy = Step.chain(strategy, CrdHelper.createDomainCrdStep(
-          version, productVersion,
-          new StartNamespacesStep(configuredDomainNamespaces, isFullRecheck)));
+      strategy = Step.chain(strategy, new StartNamespacesStep(configuredDomainNamespaces, isFullRecheck));
     }
     return strategy;
   }
@@ -369,14 +368,31 @@ private static Step readExistingPods(String ns) {
   }
 
   static Step readExistingNamespaces(DomainNamespaceSelectionStrategy selectionStrategy,
-                                             Collection<String> domainNamespaces,
-                                             boolean isFullRecheck) {
+                                     Collection<String> domainNamespaces,
+                                     boolean isFullRecheck) {
     CallBuilder builder = new CallBuilder();
     String selector = selectionStrategy.getLabelSelector();
     if (selector != null) {
       builder.withLabelSelectors(selector);
     }
-    return builder.listNamespaceAsync(new NamespaceListStep(selectionStrategy, domainNamespaces, isFullRecheck));
+    return builder.listNamespaceAsync(
+      new ActionResponseStep<V1NamespaceList>(new NamespaceListAfterStep(selectionStrategy)) {
+        private Step startNamespaces(Collection<String> namespacesToStart, boolean isFullRecheck) {
+          return new StartNamespacesStep(namespacesToStart, isFullRecheck);
+        }
+
+        @Override
+        protected NextAction onFailureNoRetry(Packet packet, CallResponse<V1NamespaceList> callResponse) {
+          return !selectionStrategy.isRequireList() && isNotAuthorizedOrForbidden(callResponse)
+                  ? doNext(startNamespaces(domainNamespaces, isFullRecheck), packet) :
+                  super.onFailureNoRetry(packet, callResponse);
+        }
+
+        @Override
+        public Step createSuccessStep(V1NamespaceList result, Step next) {
+          return new NamespaceListStep(result, selectionStrategy, domainNamespaces, isFullRecheck, next);
+        }
+      });
   }
 
   private static ConfigMapAfterStep createConfigMapStep(String ns) {
@@ -682,7 +698,7 @@ private abstract static class ForEachNamespaceStep extends Step {
 
     @Override
     protected String getDetail() {
-      return String.join(",", domainNamespaces);
+      return Optional.ofNullable(domainNamespaces).map(d -> String.join(",", d)).orElse(null);
     }
 
     protected abstract Step action(String ns);
@@ -873,7 +889,8 @@ public NextAction onSuccess(Packet packet, CallResponse<DomainList> callResponse
         domainWatchers.put(
             ns, createDomainWatcher(ns, getResourceVersion(callResponse.getResult())));
       }
-      return doNext(packet);
+
+      return doContinueListOrNext(callResponse, packet);
     }
 
     String getResourceVersion(DomainList result) {
@@ -916,7 +933,8 @@ public NextAction onSuccess(Packet packet, CallResponse<V1ServiceList> callRespo
       if (!serviceWatchers.containsKey(ns)) {
         serviceWatchers.put(ns, createServiceWatcher(ns, getInitialResourceVersion(result)));
       }
-      return doNext(packet);
+
+      return doContinueListOrNext(callResponse, packet);
     }
 
     private String getInitialResourceVersion(V1ServiceList result) {
@@ -946,7 +964,8 @@ public NextAction onSuccess(Packet packet, CallResponse<V1EventList> callRespons
       if (!eventWatchers.containsKey(ns)) {
         eventWatchers.put(ns, createEventWatcher(ns, getInitialResourceVersion(result)));
       }
-      return doNext(packet);
+
+      return doContinueListOrNext(callResponse, packet);
     }
 
     private String getInitialResourceVersion(V1EventList result) {
@@ -990,103 +1009,86 @@ public NextAction onSuccess(Packet packet, CallResponse<V1PodList> callResponse)
       if (!podWatchers.containsKey(ns)) {
         podWatchers.put(ns, createPodWatcher(ns, getInitialResourceVersion(result)));
       }
-      return doNext(packet);
+
+      return doContinueListOrNext(callResponse, packet);
     }
 
     private String getInitialResourceVersion(V1PodList result) {
       return result != null ? result.getMetadata().getResourceVersion() : "";
     }
   }
 
-  private static class NamespaceListStep extends ResponseStep<V1NamespaceList> {
+  private static final String ALL_DOMAIN_NAMESPACES = "ALL_DOMAIN_NAMESPACES";
+
+  private static class NamespaceListStep extends Step {
+    private final V1NamespaceList list;
     private final DomainNamespaceSelectionStrategy selectionStrategy;
     private final Collection<String> configuredDomainNamespaces;
     private final boolean isFullRecheck;
 
-    NamespaceListStep(DomainNamespaceSelectionStrategy selectionStrategy,
-                      Collection<String> configuredDomainNamespaces,
-                      boolean isFullRecheck) {
+    NamespaceListStep(V1NamespaceList list,
+            DomainNamespaceSelectionStrategy selectionStrategy,
+            Collection<String> configuredDomainNamespaces,
+            boolean isFullRecheck,
+            Step next) {
+      super(next);
+      this.list = list;
       this.selectionStrategy = selectionStrategy;
       this.configuredDomainNamespaces = configuredDomainNamespaces;
       this.isFullRecheck = isFullRecheck;
     }
 
     @Override
-    public NextAction onFailure(Packet packet, CallResponse<V1NamespaceList> callResponse) {
-      return callResponse.getStatusCode() == CallBuilder.NOT_FOUND
-          ? onSuccess(packet, callResponse)
-          : super.onFailure(packet, callResponse);
-    }
-
-    @Override
-    protected NextAction onFailureNoRetry(Packet packet, CallResponse<V1NamespaceList> callResponse) {
-      return !selectionStrategy.isRequireList() && isNotAuthorizedOrForbidden(callResponse)
-          ? doNext(createDomainCrdAndStartNamespaces(configuredDomainNamespaces, isFullRecheck), packet) :
-            super.onFailureNoRetry(packet, callResponse);
-    }
-
-    @Override
-    public NextAction onSuccess(Packet packet, CallResponse<V1NamespaceList> callResponse) {
-      V1NamespaceList result = callResponse.getResult();
+    public NextAction apply(Packet packet) {
       // don't bother processing pre-existing events
-      String intialResourceVersion = getInitialResourceVersion(result);
-      List<String> nsList = getExistingNamespaces(result);
+      String intialResourceVersion = getInitialResourceVersion(list);
+      List<String> nsPossiblyPartialList = getExistingNamespaces(list);
 
-      Set<String> namespacesToStart;
+      Set<String> namespacesToStartNow;
       if (selectionStrategy.isRequireList()) {
-        namespacesToStart = new TreeSet<>(nsList);
+        namespacesToStartNow = new TreeSet<>(nsPossiblyPartialList);
         String regexp = selectionStrategy.getRegExp();
         if (regexp != null) {
           try {
-            namespacesToStart = namespacesToStart.stream().filter(
+            namespacesToStartNow = namespacesToStartNow.stream().filter(
                     Pattern.compile(regexp).asPredicate()).collect(Collectors.toSet());
           } catch (PatternSyntaxException pse) {
             LOGGER.severe(MessageKeys.EXCEPTION, pse);
           }
         }
       } else {
-        namespacesToStart = new TreeSet<>(configuredDomainNamespaces);
-        for (String ns : configuredDomainNamespaces) {
-          if (!nsList.contains(ns)) {
-            try (LoggingContext stack = LoggingContext.setThreadContext().namespace(ns)) {
-              LOGGER.warning(MessageKeys.NAMESPACE_IS_MISSING, ns);
-            }
-            namespacesToStart.remove(ns);
-          }
-        }
+        namespacesToStartNow = new TreeSet<>(configuredDomainNamespaces);
+        namespacesToStartNow.retainAll(nsPossiblyPartialList);
       }
+
       Step strategy = null;
-      if (!namespacesToStart.isEmpty()) {
-        strategy = Step.chain(createDomainCrdAndStartNamespaces(namespacesToStart, isFullRecheck),
-          new CreateNamespaceWatcherStep(selectionStrategy, intialResourceVersion));
+      if (!namespacesToStartNow.isEmpty()) {
+        strategy = Step.chain(
+          startNamespaces(namespacesToStartNow, isFullRecheck),
+          new CreateNamespaceWatcherStep(selectionStrategy, intialResourceVersion),
+          getNext());
 
         if (configuredDomainNamespaces == null) {
-          strategy = new InitializeNamespacesSecurityStep(namespacesToStart, strategy);
+          strategy = new InitializeNamespacesSecurityStep(namespacesToStartNow, strategy);
         }
       } else {
-        strategy = CrdHelper.createDomainCrdStep(
-          version, productVersion,
-            new CreateNamespaceWatcherStep(selectionStrategy, intialResourceVersion));
+        strategy = Step.chain(
+          new CreateNamespaceWatcherStep(selectionStrategy, intialResourceVersion),
+          getNext());
       }
 
-      // Check for namespaces that are removed from the operator's
-      // domainNamespaces list, or that are deleted from the Kubernetes cluster.
-      Set<String> namespacesToStop = new TreeSet<>(namespaceStoppingMap.keySet());
-      for (String ns : namespacesToStart) {
-        // the active namespaces are the ones that will not be stopped
-        if (delegate.isNamespaceRunning(ns)) {
-          namespacesToStop.remove(ns);
-        }
+      Collection<String> allDomainNamespaces = (Collection<String>) packet.get(ALL_DOMAIN_NAMESPACES);
+      if (allDomainNamespaces == null) {
+        allDomainNamespaces = new HashSet<>();
+        packet.put(ALL_DOMAIN_NAMESPACES, allDomainNamespaces);
       }
-      stopNamespaces(namespacesToStart, namespacesToStop);
+      allDomainNamespaces.addAll(namespacesToStartNow);
 
       return doNext(strategy, packet);
     }
-    
-    private Step createDomainCrdAndStartNamespaces(Collection<String> namespacesToStart, boolean isFullRecheck) {
-      return CrdHelper.createDomainCrdStep(
-          version, productVersion,
-            new StartNamespacesStep(namespacesToStart, isFullRecheck));
+
+    private Step startNamespaces(Collection<String> namespacesToStart, boolean isFullRecheck) {
+      return new StartNamespacesStep(namespacesToStart, isFullRecheck);
     }
 
     private String getInitialResourceVersion(V1NamespaceList result) {
@@ -1103,7 +1105,48 @@ private List<String> getExistingNamespaces(V1NamespaceList result) {
       return namespaces;
     }
   }
-  
+
+  private static class NamespaceListAfterStep extends Step {
+    private final DomainNamespaceSelectionStrategy selectionStrategy;
+
+    public NamespaceListAfterStep(DomainNamespaceSelectionStrategy selectionStrategy) {
+      this.selectionStrategy = selectionStrategy;
+    }
+
+    @Override
+    public NextAction apply(Packet packet) {
+      Collection<String> allDomainNamespaces = (Collection<String>) packet.get(ALL_DOMAIN_NAMESPACES);
+      if (allDomainNamespaces == null) {
+        allDomainNamespaces = new HashSet<>();
+      }
+
+      Collection<String> configuredDomainNamespaces = selectionStrategy.getConfiguredList();
+      if (configuredDomainNamespaces != null) {
+        for (String ns : configuredDomainNamespaces) {
+          if (!allDomainNamespaces.contains(ns)) {
+            try (LoggingContext stack = LoggingContext.setThreadContext().namespace(ns)) {
+              LOGGER.warning(MessageKeys.NAMESPACE_IS_MISSING, ns);
+            }
+          }
+        }
+      }
+
+      // Check for namespaces that are removed from the operator's
+      // domainNamespaces list, or that are deleted from the Kubernetes cluster.
+      Set<String> namespacesToStop = new TreeSet<>(namespaceStoppingMap.keySet());
+      for (String ns : allDomainNamespaces) {
+        // the active namespaces are the ones that will not be stopped
+        if (delegate.isNamespaceRunning(ns)) {
+          namespacesToStop.remove(ns);
+        }
+      }
+
+      stopNamespaces(allDomainNamespaces, namespacesToStop);
+
+      return doNext(packet);
+    }
+  }
+
   private static class CreateNamespaceWatcherStep extends Step {
     private final DomainNamespaceSelectionStrategy selectionStrategy;
     private final String initialResourceVersion;

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

@@ -59,7 +59,7 @@ private void update() {
 
     CallBuilderTuning callBuilder =
         new CallBuilderTuning(
-            (int) readTuningParameter("callRequestLimit", 500),
+            (int) readTuningParameter("callRequestLimit", 50),
             (int) readTuningParameter("callMaxRetryCount", 5),
             (int) readTuningParameter("callTimeoutSeconds", 10));
 

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

@@ -93,6 +93,11 @@ void waitForExit() {
     }
   }
 
+  // for test
+  String getResourceVersion() {
+    return resourceVersion;
+  }
+
   /**
    * Sets the listener for watch events.
    *

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

@@ -4,8 +4,8 @@
 package oracle.kubernetes.operator.builders;
 
 /** An object which encapsulates common parameters for Kubernetes API calls. */
-class CallParamsImpl implements CallParams {
-  private static final int DEFAULT_LIMIT = 500;
+public class CallParamsImpl implements CallParams {
+  private static final int DEFAULT_LIMIT = 50;
   private static final int DEFAULT_TIMEOUT = 30;
 
   private Integer limit = CallParamsImpl.DEFAULT_LIMIT;
@@ -20,7 +20,7 @@ public Integer getLimit() {
     return limit;
   }
 
-  void setLimit(Integer limit) {
+  public void setLimit(Integer limit) {
     this.limit = limit;
   }
 
@@ -29,7 +29,7 @@ public Integer getTimeoutSeconds() {
     return timeoutSeconds;
   }
 
-  void setTimeoutSeconds(Integer timeoutSeconds) {
+  public void setTimeoutSeconds(Integer timeoutSeconds) {
     this.timeoutSeconds = timeoutSeconds;
   }
 

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

@@ -29,7 +29,7 @@ public class WatchBuilder {
   /** Ignored for watches. */
   private static final String START_LIST = null;
 
-  private static final Boolean ALLOW_BOOKMARKS = false;
+  private static final Boolean ALLOW_BOOKMARKS = true;
 
   @SuppressWarnings("FieldMayBeFinal") // Leave non-final for unit test
   private static WatchFactory FACTORY = new WatchFactoryImpl();