Resiliency for namespace listing and additional liveness checks

Author: rjeberhard

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

@@ -17,9 +17,11 @@
 import java.security.cert.CertificateException;
 import java.security.spec.InvalidKeySpecException;
 import java.time.OffsetDateTime;
+import java.util.Collection;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
@@ -159,14 +161,14 @@ public void run() {
     }
   }
 
-  void markReadyAndStartLivenessThread() {
+  void markReadyAndStartLivenessThread(Collection<ScheduledFuture<?>> futures) {
     try {
       new DeploymentReady(delegate).create();
 
       logStartingLivenessMessage();
       // every five seconds we need to update the last modified time on the liveness file
       wrappedExecutorService.scheduleWithFixedDelay(
-              new DeploymentLiveness(delegate), 5, 5, TimeUnit.SECONDS);
+              new DeploymentLiveness(futures, delegate), 5, 5, TimeUnit.SECONDS);
     } catch (IOException io) {
       LOGGER.severe(MessageKeys.EXCEPTION, io);
     }

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

@@ -1,11 +1,13 @@
-// Copyright (c) 2017, 2022, Oracle and/or its affiliates.
+// Copyright (c) 2017, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Date;
+import java.util.concurrent.ScheduledFuture;
 
 import oracle.kubernetes.common.logging.MessageKeys;
 import oracle.kubernetes.operator.logging.LoggingFacade;
@@ -18,9 +20,11 @@
 public class DeploymentLiveness implements Runnable {
 
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
+  private final Collection<ScheduledFuture<?>> futures;
   private final File livenessFile;
 
-  public DeploymentLiveness(CoreDelegate delegate) {
+  public DeploymentLiveness(Collection<ScheduledFuture<?>> futures, CoreDelegate delegate) {
+    this.futures = futures;
     livenessFile = new File(delegate.getProbesHome(), ".alive");
   }
 
@@ -33,7 +37,8 @@ public void run() {
     } catch (IOException ioe) {
       LOGGER.warning(MessageKeys.COULD_NOT_CREATE_LIVENESS_FILE);
     }
-    if (livenessFile.setLastModified(new Date().getTime())) {
+    if (futures.stream().filter(ScheduledFuture::isDone).findAny().isEmpty()
+          && livenessFile.setLastModified(new Date().getTime())) {
       LOGGER.fine("Liveness file last modified time set");
     }
   }

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020, 2023, Oracle and/or its affiliates.
+// Copyright (c) 2020, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -35,6 +35,8 @@
 import oracle.kubernetes.operator.work.Packet;
 import oracle.kubernetes.operator.work.Step;
 
+import static oracle.kubernetes.operator.KubernetesConstants.HTTP_GONE;
+import static oracle.kubernetes.operator.calls.AsyncRequestStep.FIBER_TIMEOUT;
 import static oracle.kubernetes.operator.helpers.EventHelper.EventItem.NAMESPACE_WATCHING_STARTED;
 import static oracle.kubernetes.operator.helpers.NamespaceHelper.getOperatorNamespace;
 import static oracle.kubernetes.operator.logging.ThreadLoggingContext.setThreadContext;
@@ -161,7 +163,14 @@ private NamespaceListResponseStep() {
     protected NextAction onFailureNoRetry(Packet packet, CallResponse<V1NamespaceList> callResponse) {
       return useBackupStrategy(callResponse)
             ? doNext(createStartNamespacesStep(Namespaces.getConfiguredDomainNamespaces()), packet)
-            : super.onFailureNoRetry(packet, callResponse);
+            : onFailureNoRetryCheckForGone(packet, callResponse);
+    }
+
+    protected NextAction onFailureNoRetryCheckForGone(Packet packet, CallResponse<V1NamespaceList> callResponse) {
+      if (Optional.ofNullable(callResponse).map(CallResponse::getStatusCode).orElse(FIBER_TIMEOUT) == HTTP_GONE) {
+        return doEnd(packet);
+      }
+      return super.onFailureNoRetry(packet, callResponse);
     }
 
     // Returns true if the failure wasn't due to authorization, and we have a list of namespaces to manage.
@@ -172,12 +181,13 @@ private boolean useBackupStrategy(CallResponse<V1NamespaceList> callResponse) {
     @Override
     public NextAction onSuccess(Packet packet, CallResponse<V1NamespaceList> callResponse) {
       final Set<String> namespacesToStart = getNamespacesToStart(callResponse.getResult());
-      Namespaces.getFoundDomainNamespaces(packet).addAll(namespacesToStart);
+      Collection<String> foundDomainNamespaces = Namespaces.getFoundDomainNamespaces(packet);
+      foundDomainNamespaces.addAll(namespacesToStart);
 
-      return doContinueListOrNext(callResponse, packet, createNextSteps(namespacesToStart));
+      return doContinueListOrNext(callResponse, packet, () -> createNextSteps(foundDomainNamespaces));
     }
 
-    private Step createNextSteps(Set<String> namespacesToStartNow) {
+    private Step createNextSteps(Collection<String> namespacesToStartNow) {
       if (!namespacesToStartNow.isEmpty()) {
         List<Step> nextSteps = new ArrayList<>();
         nextSteps.add(createStartNamespacesStep(namespacesToStartNow));
@@ -190,7 +200,7 @@ private Step createNextSteps(Set<String> namespacesToStartNow) {
       return current;
     }
 
-    private Step createNamespaceReviewStep(Set<String> namespacesToStartNow) {
+    private Step createNamespaceReviewStep(Collection<String> namespacesToStartNow) {
       return RunInParallel.perNamespace(namespacesToStartNow, DomainRecheck.this::createNamespaceReview);
     }
 

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018, 2023, Oracle and/or its affiliates.
+// Copyright (c) 2018, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -317,7 +317,7 @@ public NextAction onSuccess(Packet packet, CallResponse<V1PodList> callResponse)
 
         if (jobPod == null) {
           terminationState.remove(packet);
-          return doContinueListOrNext(callResponse, packet, getNext());
+          return doContinueListOrNext(callResponse, packet);
         } else {
           terminationState.setFromPod(jobPod);
           return doNext(getNext(), packet);

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2017, 2023, Oracle and/or its affiliates.
+// Copyright (c) 2017, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -13,9 +13,11 @@
 import java.time.OffsetDateTime;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -301,10 +303,11 @@ void completeBegin() {
       // start periodic retry and recheck
       int recheckInterval = TuningParameters.getInstance().getDomainNamespaceRecheckIntervalSeconds();
       int stuckPodInterval = TuningParameters.getInstance().getStuckPodRecheckSeconds();
-      mainDelegate.scheduleWithFixedDelay(recheckDomains(), recheckInterval, recheckInterval, TimeUnit.SECONDS);
-      mainDelegate.scheduleWithFixedDelay(checkStuckPods(), stuckPodInterval, stuckPodInterval, TimeUnit.SECONDS);
+      Collection<ScheduledFuture<?>> futures = List.of(
+          mainDelegate.scheduleWithFixedDelay(recheckDomains(), recheckInterval, recheckInterval, TimeUnit.SECONDS),
+          mainDelegate.scheduleWithFixedDelay(checkStuckPods(), stuckPodInterval, stuckPodInterval, TimeUnit.SECONDS));
 
-      markReadyAndStartLivenessThread();
+      markReadyAndStartLivenessThread(futures);
 
     } catch (Throwable e) {
       LOGGER.warning(MessageKeys.EXCEPTION, e);

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

@@ -1,11 +1,13 @@
-// Copyright (c) 2022, 2023, Oracle and/or its affiliates.
+// Copyright (c) 2022, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -134,9 +136,10 @@ void completeBegin() {
 
       // start periodic recheck of CRD
       int recheckInterval = TuningParameters.getInstance().getDomainNamespaceRecheckIntervalSeconds();
-      delegate.scheduleWithFixedDelay(recheckCrd(), recheckInterval, recheckInterval, TimeUnit.SECONDS);
+      ScheduledFuture<?> future =
+          delegate.scheduleWithFixedDelay(recheckCrd(), recheckInterval, recheckInterval, TimeUnit.SECONDS);
 
-      markReadyAndStartLivenessThread();
+      markReadyAndStartLivenessThread(List.of(future));
 
     } catch (Exception e) {
       LOGGER.warning(MessageKeys.EXCEPTION, e);

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018, 2023, Oracle and/or its affiliates.
+// Copyright (c) 2018, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator.helpers;
@@ -818,7 +818,7 @@ public NextAction onSuccess(Packet packet, CallResponse<V1PodList> callResponse)
               .orElse(null);
 
         if (jobPod == null) {
-          return doContinueListOrNext(callResponse, packet, processIntrospectorPodLog(getNext()));
+          return doContinueListOrNext(callResponse, packet, () -> processIntrospectorPodLog(getNext()));
         } else if (hasImagePullError(jobPod) || initContainersHaveImagePullError(jobPod)) {
           return doNext(cleanUpAndReintrospect(getNext()), packet);
         } else if (isJobPodTimedOut(jobPod)) {

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018, 2024, Oracle and/or its affiliates.
+// Copyright (c) 2018, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator.helpers;
@@ -7,6 +7,7 @@
 import java.net.SocketTimeoutException;
 import java.util.Collections;
 import java.util.Optional;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 
@@ -140,7 +141,7 @@ private NextAction getPotentialRetryAction(Packet packet) {
    * @return Next action for list continue
    */
   protected final NextAction doContinueListOrNext(CallResponse<T> callResponse, Packet packet) {
-    return doContinueListOrNext(callResponse, packet, getNext());
+    return doContinueListOrNext(callResponse, packet, this::getNext);
   }
 
   /**
@@ -149,10 +150,10 @@ protected final NextAction doContinueListOrNext(CallResponse<T> callResponse, Pa
    *
    * @param callResponse Call response
    * @param packet Packet
-   * @param next Next step, if no continuation
+   * @param next Supplier of next step, if no continuation
    * @return Next action for list continue
    */
-  protected final NextAction doContinueListOrNext(CallResponse<T> callResponse, Packet packet, Step next) {
+  protected final NextAction doContinueListOrNext(CallResponse<T> callResponse, Packet packet, Supplier<Step> next) {
     String cont = accessContinue(callResponse.getResult());
     if (cont != null) {
       packet.put(CONTINUE, cont);
@@ -161,9 +162,9 @@ protected final NextAction doContinueListOrNext(CallResponse<T> callResponse, Pa
       return resetRetryStrategyAndReinvokeRequest(packet);
     }
     if (callResponse.getResult() instanceof KubernetesListObject kubernetesListObject) {
-      return doNext(next, packet).withDebugComment(kubernetesListObject, this::toComment);
+      return doNext(next.get(), packet).withDebugComment(kubernetesListObject, this::toComment);
     } else {
-      return doNext(next, packet);
+      return doNext(next.get(), packet);
     }
   }
 

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2022, Oracle and/or its affiliates.
+// Copyright (c) 2022, 2025, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -8,6 +8,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.logging.Level;
@@ -56,7 +57,7 @@ public void tearDown() throws Exception {
 
   @Test
   void whenNoExistingLivenessFile_fileCreated() {
-    DeploymentLiveness deploymentLiveness = new DeploymentLiveness(coreDelegate);
+    DeploymentLiveness deploymentLiveness = new DeploymentLiveness(Collections.emptyList(), coreDelegate);
     deploymentLiveness.run();
 
     File aliveFile = new File(coreDelegate.probesHome, ".alive");
@@ -72,7 +73,7 @@ void whenExistingLivenessFile_onlyLogLastModifiedUpdated() throws IOException {
     File aliveFile = new File(coreDelegate.probesHome, ".alive");
     assertTrue(aliveFile.createNewFile());
 
-    DeploymentLiveness deploymentLiveness = new DeploymentLiveness(coreDelegate);
+    DeploymentLiveness deploymentLiveness = new DeploymentLiveness(Collections.emptyList(), coreDelegate);
     deploymentLiveness.run();
 
     assertThat(coreDelegate.probesHome, anExistingDirectory());
@@ -86,7 +87,7 @@ void whenExistingLivenessFile_onlyLogLastModifiedUpdated() throws IOException {
   void whenCantCreateLivenessFile_logWarning() throws IOException {
     assertTrue(coreDelegate.probesHome.setWritable(false, false));
 
-    DeploymentLiveness deploymentLiveness = new DeploymentLiveness(coreDelegate);
+    DeploymentLiveness deploymentLiveness = new DeploymentLiveness(Collections.emptyList(), coreDelegate);
     deploymentLiveness.run();
 
     assertThat(coreDelegate.probesHome, anExistingDirectory());