Detect and shut down stuck server pods

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

@@ -73,8 +73,10 @@ public class Main {
   private static final AtomicReference<DateTime> lastFullRecheck =
       new AtomicReference<>(DateTime.now());
   private static final Semaphore shutdownSignal = new Semaphore(0);
+  private static final int DEFAULT_STUCK_POD_RECHECK_SECONDS = 30;
 
   private final MainDelegate delegate;
+  private final StuckPodProcessing stuckPodProcessing;
   private NamespaceWatcher namespaceWatcher;
   private boolean warnedOfCrdAbsence;
 
@@ -291,6 +293,7 @@ DomainNamespaces getDomainNamespaces() {
 
   Main(MainDelegate delegate) {
     this.delegate = delegate;
+    stuckPodProcessing = new StuckPodProcessing(delegate);
   }
 
   void startOperator(Runnable completionAction) {
@@ -336,10 +339,9 @@ private void completeBegin() {
 
       // start periodic retry and recheck
       int recheckInterval = TuningParameters.getInstance().getMainTuning().domainNamespaceRecheckIntervalSeconds;
-      delegate.getEngine()
-          .getExecutor()
-          .scheduleWithFixedDelay(
-              recheckDomains(), recheckInterval, recheckInterval, TimeUnit.SECONDS);
+      int stuckPodInterval = getStuckPodInterval();
+      delegate.scheduleWithFixedDelay(recheckDomains(), recheckInterval, recheckInterval, TimeUnit.SECONDS);
+      delegate.scheduleWithFixedDelay(checkStuckPods(), stuckPodInterval, stuckPodInterval, TimeUnit.SECONDS);
 
       markReadyAndStartLivenessThread();
 
@@ -348,6 +350,13 @@ private void completeBegin() {
     }
   }
 
+  private int getStuckPodInterval() {
+    return Optional.ofNullable(TuningParameters.getInstance())
+          .map(TuningParameters::getMainTuning)
+          .map(t -> t.stuckPodRecheckSeconds)
+          .orElse(DEFAULT_STUCK_POD_RECHECK_SECONDS);
+  }
+
   NamespaceWatcher getNamespaceWatcher() {
     return namespaceWatcher;
   }
@@ -360,6 +369,11 @@ Runnable recheckDomains() {
     return () -> delegate.runSteps(createDomainRecheckSteps());
   }
 
+  Runnable checkStuckPods() {
+    return () -> getDomainNamespaces().getNamespaces().forEach(stuckPodProcessing::checkStuckPods);
+  }
+
+
   Step createDomainRecheckSteps() {
     return createDomainRecheckSteps(DateTime.now());
   }

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

@@ -3,6 +3,9 @@
 
 package oracle.kubernetes.operator;
 
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
 import oracle.kubernetes.operator.helpers.KubernetesVersion;
 import oracle.kubernetes.operator.helpers.SemanticVersion;
 import oracle.kubernetes.operator.logging.LoggingFacade;
@@ -36,4 +39,6 @@ default void runSteps(Step firstStep) {
   DomainNamespaces getDomainNamespaces();
 
   KubernetesVersion getKubernetesVersion();
+
+  ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
 }

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

@@ -0,0 +1,159 @@
+// Copyright (c) 2020, Oracle Corporation 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.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import javax.annotation.Nonnull;
+
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
+import oracle.kubernetes.operator.calls.CallResponse;
+import oracle.kubernetes.operator.helpers.CallBuilder;
+import oracle.kubernetes.operator.helpers.PodHelper;
+import oracle.kubernetes.operator.logging.LoggingFacade;
+import oracle.kubernetes.operator.logging.LoggingFactory;
+import oracle.kubernetes.operator.steps.DefaultResponseStep;
+import oracle.kubernetes.operator.work.NextAction;
+import oracle.kubernetes.operator.work.Packet;
+import oracle.kubernetes.operator.work.Step;
+import oracle.kubernetes.utils.SystemClock;
+import org.joda.time.DateTime;
+
+import static oracle.kubernetes.operator.logging.MessageKeys.POD_FORCE_DELETED;
+
+/**
+ * Under certain circumstances, when a Kubernetes node goes down, it may mark its pods as terminating, but never
+ * actually remove them. This code detects such cases, deletes the pods and triggers the necessary make-right flows.
+ */
+public class StuckPodProcessing {
+  private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
+
+  private final MainDelegate mainDelegate;
+
+  public StuckPodProcessing(MainDelegate mainDelegate) {
+    this.mainDelegate = mainDelegate;
+  }
+
+  void checkStuckPods(String namespace) {
+    Step step = new CallBuilder()
+          .withLabelSelectors(LabelConstants.getCreatedbyOperatorSelector())
+          .listPodAsync(namespace, new PodListProcessing(namespace, SystemClock.now()));
+    mainDelegate.runSteps(step);
+  }
+
+  @SuppressWarnings("unchecked")
+  private List<V1Pod> getStuckPodList(Packet packet) {
+    return (List<V1Pod>) packet.computeIfAbsent("STUCK_PODS", k -> new ArrayList<>());
+  }
+
+  class PodListProcessing extends DefaultResponseStep<V1PodList> {
+
+    private final DateTime now;
+
+    public PodListProcessing(String namespace, DateTime dateTime) {
+      super(new PodActionsStep(namespace));
+      now = dateTime;
+    }
+
+    @Override
+    public NextAction onSuccess(Packet packet, CallResponse<V1PodList> callResponse) {
+      callResponse.getResult().getItems().stream()
+            .filter(pod -> isStuck(pod, now))
+            .forEach(pod -> addStuckPodToPacket(packet, pod));
+
+      return doContinueListOrNext(callResponse, packet);
+    }
+
+    private boolean isStuck(V1Pod pod, DateTime now)  {
+      return getExpectedDeleteTime(pod).isBefore(now);
+    }
+
+    private DateTime getExpectedDeleteTime(V1Pod pod) {
+      return getDeletionTimeStamp(pod).plusSeconds((int) getDeletionGracePeriodSeconds(pod));
+    }
+
+    private long getDeletionGracePeriodSeconds(V1Pod pod) {
+      return Optional.of(pod).map(V1Pod::getMetadata).map(V1ObjectMeta::getDeletionGracePeriodSeconds).orElse(1L);
+    }
+
+    private DateTime getDeletionTimeStamp(V1Pod pod) {
+      return Optional.of(pod).map(V1Pod::getMetadata).map(V1ObjectMeta::getDeletionTimestamp).orElse(SystemClock.now());
+    }
+
+    private void addStuckPodToPacket(Packet packet, V1Pod stuckPod) {
+      getStuckPodList(packet).add(stuckPod);
+    }
+  }
+
+  class PodActionsStep extends Step {
+
+    private final String namespace;
+
+    public PodActionsStep(String namespace) {
+      this.namespace = namespace;
+    }
+
+    @Override
+    public NextAction apply(Packet packet) {
+      final List<V1Pod> stuckPodList = getStuckPodList(packet);
+      if (stuckPodList.isEmpty()) {
+        return doNext(packet);
+      } else {
+        Collection<StepAndPacket> startDetails = new ArrayList<>();
+
+        for (V1Pod pod : stuckPodList) {
+          startDetails.add(new StepAndPacket(createForcedDeletePodStep(pod), packet.clone()));
+        }
+        return doForkJoin(readExistingNamespaces(), packet, startDetails);
+      }
+    }
+
+    @Nonnull
+    private Step readExistingNamespaces() {
+      return mainDelegate.getDomainNamespaces().readExistingResources(namespace, mainDelegate.getDomainProcessor());
+    }
+
+    private Step createForcedDeletePodStep(V1Pod pod) {
+      return new CallBuilder()
+            .withGracePeriodSeconds(0)
+            .deletePodAsync(getName(pod), getNamespace(pod), getDomainUid(pod), null,
+                  new ForcedDeleteResponseStep(getName(pod), getNamespace(pod)));
+    }
+
+    private String getName(V1Pod pod) {
+      return Objects.requireNonNull(pod.getMetadata()).getName();
+    }
+
+    private String getNamespace(V1Pod pod) {
+      return Objects.requireNonNull(pod.getMetadata()).getNamespace();
+    }
+
+    private String getDomainUid(V1Pod pod) {
+      return PodHelper.getPodDomainUid(pod);
+    }
+  }
+
+  static class ForcedDeleteResponseStep extends DefaultResponseStep<V1Pod> {
+
+    private final String name;
+    private final String namespace;
+
+    public ForcedDeleteResponseStep(String name, String namespace) {
+      this.name = name;
+      this.namespace = namespace;
+    }
+
+    @Override
+    public NextAction onSuccess(Packet packet, CallResponse<V1Pod> callResponse) {
+      LOGGER.info(POD_FORCE_DELETED, name, namespace);
+      return super.onSuccess(packet, callResponse);
+    }
+  }
+
+}

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

@@ -37,6 +37,7 @@ public static class MainTuning {
     public final int domainNamespaceRecheckIntervalSeconds;
     public final int statusUpdateTimeoutSeconds;
     public final int unchangedCountToDelayStatusRecheck;
+    public final int stuckPodRecheckSeconds;
     public final long initialShortDelay;
     public final long eventualLongDelay;
 
@@ -48,24 +49,27 @@ public static class MainTuning {
      * @param domainNamespaceRecheckIntervalSeconds domain namespace recheck interval
      * @param statusUpdateTimeoutSeconds status update timeout
      * @param unchangedCountToDelayStatusRecheck unchanged count to delay status recheck
+     * @param stuckPodRecheckSeconds time between checks for stuck pods
      * @param initialShortDelay initial short delay
      * @param eventualLongDelay eventual long delay
      */
     public MainTuning(
-        int domainPresenceFailureRetrySeconds,
-        int domainPresenceFailureRetryMaxCount,
-        int domainPresenceRecheckIntervalSeconds,
-        int domainNamespaceRecheckIntervalSeconds,
-        int statusUpdateTimeoutSeconds,
-        int unchangedCountToDelayStatusRecheck,
-        long initialShortDelay,
-        long eventualLongDelay) {
+          int domainPresenceFailureRetrySeconds,
+          int domainPresenceFailureRetryMaxCount,
+          int domainPresenceRecheckIntervalSeconds,
+          int domainNamespaceRecheckIntervalSeconds,
+          int statusUpdateTimeoutSeconds,
+          int unchangedCountToDelayStatusRecheck,
+          int stuckPodRecheckSeconds,
+          long initialShortDelay,
+          long eventualLongDelay) {
       this.domainPresenceFailureRetrySeconds = domainPresenceFailureRetrySeconds;
       this.domainPresenceFailureRetryMaxCount = domainPresenceFailureRetryMaxCount;
       this.domainPresenceRecheckIntervalSeconds = domainPresenceRecheckIntervalSeconds;
       this.domainNamespaceRecheckIntervalSeconds = domainNamespaceRecheckIntervalSeconds;
       this.statusUpdateTimeoutSeconds = statusUpdateTimeoutSeconds;
       this.unchangedCountToDelayStatusRecheck = unchangedCountToDelayStatusRecheck;
+      this.stuckPodRecheckSeconds = stuckPodRecheckSeconds;
       this.initialShortDelay = initialShortDelay;
       this.eventualLongDelay = eventualLongDelay;
     }

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

@@ -54,6 +54,7 @@ private void update() {
             (int) readTuningParameter("domainNamespaceRecheckIntervalSeconds", 3),
             (int) readTuningParameter("statusUpdateTimeoutSeconds", 10),
             (int) readTuningParameter("statusUpdateUnchangedCountToDelayStatusRecheck", 10),
+            (int) readTuningParameter("stuckPodRecheckSeconds", 30),
             readTuningParameter("statusUpdateInitialShortDelay", 5),
             readTuningParameter("statusUpdateEventualLongDelay", 30));
 

operator/src/main/java/oracle/kubernetes/operator/calls/AsyncRequestStep.java

@@ -83,7 +83,7 @@ public AsyncRequestStep(
       String labelSelector,
       String resourceVersion) {
     this(next, requestParams, factory, null, helper, timeoutSeconds, maxRetryCount,
-            fieldSelector, labelSelector, resourceVersion);
+            null, fieldSelector, labelSelector, resourceVersion);
   }
 
   /**
@@ -108,6 +108,7 @@ public AsyncRequestStep(
           ClientPool helper,
           int timeoutSeconds,
           int maxRetryCount,
+          Integer gracePeriodSeconds,
           String fieldSelector,
           String labelSelector,
           String resourceVersion) {

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

@@ -17,6 +17,7 @@ <T> Step createRequestAsync(
       ClientPool helper,
       int timeoutSeconds,
       int maxRetryCount,
+      Integer gracePeriodSeconds,
       String fieldSelector,
       String labelSelector,
       String resourceVersion);

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

@@ -303,7 +303,7 @@ public <T> T execute(
   private final CallFactory<V1Secret> readSecret =
       (requestParams, usage, cont, callback) ->
           wrap(readSecretAsync(usage, requestParams.name, requestParams.namespace, callback));
-  private final Integer gracePeriodSeconds = null;
+  private Integer gracePeriodSeconds = null;
   private final Boolean orphanDependents = null;
   private final String propagationPolicy = null;
 
@@ -540,6 +540,11 @@ public CallBuilder withTimeoutSeconds(int timeoutSeconds) {
     return this;
   }
 
+  public CallBuilder withGracePeriodSeconds(int gracePeriodSeconds) {
+    this.gracePeriodSeconds = gracePeriodSeconds;
+    return this;
+  }
+
   private void tuning(int limit, int timeoutSeconds, int maxRetryCount) {
     this.limit = limit;
     this.timeoutSeconds = timeoutSeconds;
@@ -1909,6 +1914,7 @@ private <T> Step createRequestAsync(
         helper,
         timeoutSeconds,
         maxRetryCount,
+        gracePeriodSeconds,
         fieldSelector,
         labelSelector,
         resourceVersion);
@@ -1924,6 +1930,7 @@ private <T> Step createRequestAsync(
             helper,
             timeoutSeconds,
             maxRetryCount,
+            gracePeriodSeconds,
             fieldSelector,
             labelSelector,
             resourceVersion);
@@ -1939,6 +1946,7 @@ private <T> Step createRequestAsync(
             helper,
             timeoutSeconds,
             maxRetryCount,
+            gracePeriodSeconds,
             fieldSelector,
             labelSelector,
             resourceVersion);

operator/src/main/java/oracle/kubernetes/operator/logging/MessageKeys.java

@@ -134,6 +134,7 @@ public class MessageKeys {
   public static final String INTROSPECTOR_JOB_FAILED_DETAIL = "WLSKO-0176";
   public static final String INTROSPECTOR_POD_FAILED = "WLSKO-0177";
   public static final String CRD_NOT_INSTALLED = "WLSKO-0178";
+  public static final String POD_FORCE_DELETED = "WLSKO-0179";
 
   // domain status messages
   public static final String DUPLICATE_SERVER_NAME_FOUND = "WLSDO-0001";

operator/src/main/java/oracle/kubernetes/utils/SystemClock.java

@@ -8,8 +8,9 @@
 /** A wrapper for the system clock that facilitates unit testing of time. */
 public abstract class SystemClock {
 
-  private static SystemClock DELEGATE =
-      new SystemClock() {
+  // Leave as non-final; unit tests may replace this value
+  @SuppressWarnings("FieldMayBeFinal")
+  private static SystemClock DELEGATE = new SystemClock() {
         @Override
         public DateTime getCurrentTime() {
           return DateTime.now();

operator/src/main/resources/Operator.properties

@@ -188,6 +188,7 @@ WLSKO-0175=Job {0} in namespace {1} failed with status {2}. Check log messages \
 WLSKO-0176=Job {1} in namespace {0} failed, job details are {2}
 WLSKO-0177=Pod {0} in namespace {1} failed, the pod status is {2}
 WLSKO-0178=Operator cannot proceed, as the Custom Resource Definition for ''domains.weblogic.oracle'' is not installed.
+WLSKO-0179=Pod {0} in namespace {1} detected as stuck, and force-deleted
 
 # Domain status messages