Backstop wait for ready with periodic rechecks

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

@@ -175,23 +175,33 @@ public boolean equals(Object o) {
   public static class WatchTuning {
     public final int watchLifetime;
     public final int watchMinimumDelay;
+    public final int watchBackstopRecheckDelay;
 
-    public WatchTuning(int watchLifetime, int watchMinimumDelay) {
+    /**
+     * Create watch tuning.
+     * @param watchLifetime Watch lifetime
+     * @param watchMinimumDelay Minimum delay before accepting new events to prevent hot loops
+     * @param watchBackstopRecheckDelay Recheck delay for get while waiting for a status to backstop missed watch events
+     */
+    public WatchTuning(int watchLifetime, int watchMinimumDelay, int watchBackstopRecheckDelay) {
       this.watchLifetime = watchLifetime;
       this.watchMinimumDelay = watchMinimumDelay;
+      this.watchBackstopRecheckDelay = watchBackstopRecheckDelay;
     }
 
     @Override
     public String toString() {
       return new ToStringBuilder(this)
           .append("watchLifetime", watchLifetime)
           .append("watchMinimumDelay", watchMinimumDelay)
+          .append("watchBackstopRecheckDelay", watchBackstopRecheckDelay)
           .toString();
     }
 
     @Override
     public int hashCode() {
-      return new HashCodeBuilder().append(watchLifetime).append(watchMinimumDelay).toHashCode();
+      return new HashCodeBuilder()
+              .append(watchLifetime).append(watchMinimumDelay).append(watchBackstopRecheckDelay).toHashCode();
     }
 
     @Override
@@ -206,6 +216,7 @@ public boolean equals(Object o) {
       return new EqualsBuilder()
           .append(watchLifetime, wt.watchLifetime)
           .append(watchMinimumDelay, wt.watchMinimumDelay)
+          .append(watchBackstopRecheckDelay, wt.watchBackstopRecheckDelay)
           .isEquals();
     }
   }

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

@@ -66,7 +66,8 @@ private void update() {
     WatchTuning watch =
         new WatchTuning(
             (int) readTuningParameter("watchLifetime", 300),
-            (int) readTuningParameter("watchMinimumDelay", 5));
+            (int) readTuningParameter("watchMinimumDelay", 5),
+            (int) readTuningParameter("watchBackstopRecheckDelaySeconds", 5));
 
     PodTuning pod =
         new PodTuning(

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

@@ -3,6 +3,8 @@
 
 package oracle.kubernetes.operator;
 
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
@@ -21,6 +23,14 @@
  * @param <T> the type of resource handled by this step
  */
 abstract class WaitForReadyStep<T> extends Step {
+  private static final int DEFAULT_RECHECK_SECONDS = 5;
+
+  static int getWatchBackstopRecheckDelaySeconds() {
+    return Optional.ofNullable(TuningParameters.getInstance())
+            .map(parameters -> parameters.getWatchTuning().watchBackstopRecheckDelay)
+            .orElse(DEFAULT_RECHECK_SECONDS);
+  }
+
   private final T initialResource;
 
   /**
@@ -146,11 +156,15 @@ private void checkUpdatedResource(Packet packet, AsyncFiber fiber, Callback call
     fiber
         .createChildFiber()
         .start(
-            createReadAsyncStep(getName(), getNamespace(), resumeIfReady(callback)),
+            createReadAndIfReadyCheckStep(callback),
             packet.clone(),
             null);
   }
 
+  private Step createReadAndIfReadyCheckStep(Callback callback) {
+    return createReadAsyncStep(getName(), getNamespace(), resumeIfReady(callback));
+  }
+
   private String getNamespace() {
     return getMetadata(initialResource).getNamespace();
   }
@@ -165,8 +179,10 @@ private DefaultResponseStep<T> resumeIfReady(Callback callback) {
       public NextAction onSuccess(Packet packet, CallResponse<T> callResponse) {
         if (isReady(callResponse.getResult())) {
           callback.proceedFromWait(callResponse.getResult());
+          return doNext(packet);
         }
-        return doNext(packet);
+        return doDelay(createReadAndIfReadyCheckStep(callback), packet,
+                getWatchBackstopRecheckDelaySeconds(), TimeUnit.SECONDS);
       }
     };
   }

operator/src/main/java/oracle/kubernetes/operator/helpers/KubernetesVersion.java

@@ -8,7 +8,7 @@
 /** Major and minor version of Kubernetes API Server. */
 public class KubernetesVersion extends SemanticVersion {
   public static final KubernetesVersion TEST_VERSION = new KubernetesVersion(1, 10);
-  private static final String[] MINIMUM_K8S_VERSIONS = {"1.13.5", "1.14.8", "1.15.7"};
+  private static final String[] MINIMUM_K8S_VERSIONS = {"1.14.8", "1.15.7", "1.16.0", "1.17.0", "1.18.0"};
   static final KubernetesVersion UNREADABLE = new KubernetesVersion(0, 0);
   private final String version;
 

operator/src/test/java/oracle/kubernetes/operator/NamespaceTest.java

@@ -47,7 +47,8 @@ public class NamespaceTest {
   public static final String NAMESPACE_STOPPING_MAP = "namespaceStoppingMap";
 
   private Domain domain = DomainProcessorTestSetup.createTestDomain();
-  private final TuningParameters.WatchTuning tuning = new TuningParameters.WatchTuning(30, 0);
+  private final TuningParameters.WatchTuning tuning
+          = new TuningParameters.WatchTuning(30, 0, 5);
   private List<Memento> mementos = new ArrayList<>();
   private Set<String> currentNamespaces = new HashSet<>();
   private Map<String,String> helmValues = new HashMap<>();

operator/src/test/java/oracle/kubernetes/operator/PodWatcherTest.java

@@ -5,6 +5,7 @@
 
 import java.math.BigInteger;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
@@ -44,6 +45,7 @@ public class PodWatcherTest extends WatcherTestBase implements WatchListener<V1P
   private static final BigInteger INITIAL_RESOURCE_VERSION = new BigInteger("234");
   private static final String NS = "ns";
   private static final String NAME = "test";
+  private static final int RECHECK_SECONDS = 10;
   private KubernetesTestSupport testSupport = new KubernetesTestSupport();
   private final TerminalStep terminalStep = new TerminalStep();
   private java.util.List<com.meterware.simplestub.Memento> mementos = new java.util.ArrayList<>();
@@ -247,6 +249,15 @@ public void whenPodNotReadyLaterAndThenReady_runNextStep() {
     assertThat(terminalStep.wasRun(), is(true));
   }
 
+  @Test
+  public void whenPodNotReadyLaterAndThenReadyButNoWatchEvent_runNextStep() {
+    makeModifiedPodReadyWithNoWatchEvent(this::markPodReady);
+
+    testSupport.setTime(RECHECK_SECONDS, TimeUnit.SECONDS);
+
+    assertThat(terminalStep.wasRun(), is(true));
+  }
+
   @Test
   public void whenIntrospectPodNotReadyWithTerminatedReason_logPodStatus() {
     sendIntrospectorPodModifiedWatchAfterWaitForReady(this::addContainerStateTerminatedReason);
@@ -280,6 +291,24 @@ private void sendPodModifiedWatchAfterWaitForReady(Function<V1Pod,V1Pod>... modi
     }
   }
 
+  // Simulates a pod that is ready but where Kubernetes has failed to send the watch event
+  @SafeVarargs
+  private void makeModifiedPodReadyWithNoWatchEvent(Function<V1Pod,V1Pod>... modifiers) {
+    AtomicBoolean stopping = new AtomicBoolean(false);
+    PodWatcher watcher = createWatcher(stopping);
+    V1Pod pod = createPod();
+    testSupport.defineResources(pod);
+
+    try {
+      testSupport.runSteps(watcher.waitForReady(createPod(), terminalStep));
+      for (Function<V1Pod,V1Pod> modifier : modifiers) {
+        modifier.apply(pod);
+      }
+    } finally {
+      stopping.set(true);
+    }
+  }
+
   // Starts the waitForReady step with an incomplete pod and sends a watch indicating that the pod has changed
   @SafeVarargs
   private void sendIntrospectorPodModifiedWatchAfterWaitForReady(Function<V1Pod,V1Pod>... modifiers) {

operator/src/test/java/oracle/kubernetes/operator/WatcherTestBase.java

@@ -35,7 +35,7 @@ public abstract class WatcherTestBase extends ThreadFactoryTestBase implements A
   private static final BigInteger INITIAL_RESOURCE_VERSION = new BigInteger("214748364700");
   private static final String NAMESPACE = "testspace";
   private final RuntimeException hasNextException = new RuntimeException(Watcher.HAS_NEXT_EXCEPTION_MESSAGE);
-  final WatchTuning tuning = new WatchTuning(30, 0);
+  final WatchTuning tuning = new WatchTuning(30, 0, 5);
   private List<Memento> mementos = new ArrayList<>();
   private List<Watch.Response<?>> callBacks = new ArrayList<>();
   private BigInteger resourceVersion = INITIAL_RESOURCE_VERSION;

operator/src/test/java/oracle/kubernetes/operator/helpers/VersionCheckTest.java

@@ -89,11 +89,13 @@ public static Collection<Object[]> data() {
           {LOG_MSG_TEST, "1", "10", "1+cor.0", containsWarning(K8S_VERSION_TOO_LOW), noIgnores()},
           {LOG_MSG_TEST, "1", "10", "11", containsWarning(K8S_VERSION_TOO_LOW), noIgnores()},
           {LOG_MSG_TEST, "1", "11", "4", containsWarning(K8S_VERSION_TOO_LOW), noIgnores()},
-          {LOG_MSG_TEST, "1", "13", "5", containsInfo(K8S_VERSION_CHECK), noIgnores()},
+          {LOG_MSG_TEST, "1", "13", "5", containsWarning(K8S_VERSION_TOO_LOW), noIgnores()},
           {LOG_MSG_TEST, "1", "12", "2", containsWarning(K8S_VERSION_TOO_LOW), noIgnores()},
           {LOG_MSG_TEST, "1", "14", "8", containsInfo(K8S_VERSION_CHECK), noIgnores()},
           {VERSION_TEST, "1", "15", "7", returnsVersion(1, 15), ignoring(K8S_VERSION_CHECK)},
-          {LOG_MSG_TEST, "1", "13", "6", containsInfo(K8S_VERSION_CHECK), noIgnores()},
+          {VERSION_TEST, "1", "16", "1", returnsVersion(1, 16), ignoring(K8S_VERSION_CHECK)},
+          {VERSION_TEST, "1", "17", "2", returnsVersion(1, 17), ignoring(K8S_VERSION_CHECK)},
+          {VERSION_TEST, "1", "18", "0", returnsVersion(1, 18), ignoring(K8S_VERSION_CHECK)},
           {VERSION_TEST, "2", "7", "", returnsVersion(2, 7), ignoring(K8S_VERSION_CHECK)},
           {LOG_MSG_TEST, "2", "", "", containsInfo(K8S_VERSION_CHECK), noIgnores()},
         });