Detect and delete stuck pods (#2029)

Author: russgold

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

@@ -109,6 +109,8 @@ public class Main {
   private static String principal;
   private static KubernetesVersion version = null;
   private static SemanticVersion productVersion = null;
+  private static final StuckPodProcessing stuckPodProcessing
+        = new StuckPodProcessing(operatorNamespace, Main::readExistingResources);
 
   static {
     try {
@@ -222,10 +224,15 @@ private static void completeBegin() {
 
       // start periodic retry and recheck
       int recheckInterval = tuningAndConfig.getMainTuning().targetNamespaceRecheckIntervalSeconds;
+      int stuckPodInterval = tuningAndConfig.getMainTuning().stuckPodRecheckSeconds;
       engine
           .getExecutor()
           .scheduleWithFixedDelay(
               recheckDomains(), recheckInterval, recheckInterval, TimeUnit.SECONDS);
+      engine
+          .getExecutor()
+          .scheduleWithFixedDelay(
+              checkStuckPods(), stuckPodInterval, stuckPodInterval, TimeUnit.SECONDS);
 
       // Wait until all other initialization is done before marking ready and
       // starting liveness thread
@@ -336,6 +343,14 @@ static Runnable recheckDomains() {
     };
   }
 
+  static Runnable checkStuckPods() {
+    return () -> getTargetNamespaces().stream().map(Main::checkStuckPodsIn).forEach(Main::runSteps);
+  }
+
+  private  static Step checkStuckPodsIn(String namespace) {
+    return stuckPodProcessing.createStuckPodCheckSteps(namespace);
+  }
+
   static Step readExistingResources(String operatorNamespace, String ns) {
     return Step.chain(
         new ReadExistingResourcesBeforeStep(),

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

@@ -0,0 +1,156 @@
+// 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 java.util.function.BiFunction;
+
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
+import io.kubernetes.client.openapi.models.V1Status;
+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 String operatorNamespace;
+  private final BiFunction<String,String,Step> readExistingNamespaces;
+
+  public StuckPodProcessing(String operatorNamespace, BiFunction<String, String, Step> readExistingNamespaces) {
+    this.operatorNamespace = operatorNamespace;
+    this.readExistingNamespaces = readExistingNamespaces;
+  }
+
+  Step createStuckPodCheckSteps(String namespace) {
+    return new CallBuilder()
+          .withLabelSelectors(LabelConstants.getCreatedbyOperatorSelector())
+          .listPodAsync(namespace, new PodListProcessing(namespace, SystemClock.now()));
+  }
+
+  @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 doNext(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.apply(operatorNamespace, namespace), packet, startDetails);
+      }
+    }
+
+    private Step createForcedDeletePodStep(V1Pod pod) {
+      return new CallBuilder()
+            .withGracePeriodSeconds(0)
+            .deletePodAsync(getName(pod), getNamespace(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<V1Status> {
+
+    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<V1Status> 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 targetNamespaceRecheckIntervalSeconds;
     public final int statusUpdateTimeoutSeconds;
     public final int unchangedCountToDelayStatusRecheck;
+    public final int stuckPodRecheckSeconds;
     public final long initialShortDelay;
     public final long eventualLongDelay;
 
@@ -48,6 +49,7 @@ public static class MainTuning {
      * @param targetNamespaceRecheckIntervalSeconds target 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
      */
@@ -58,6 +60,7 @@ public MainTuning(
         int targetNamespaceRecheckIntervalSeconds,
         int statusUpdateTimeoutSeconds,
         int unchangedCountToDelayStatusRecheck,
+        int stuckPodRecheckSeconds,
         long initialShortDelay,
         long eventualLongDelay) {
       this.domainPresenceFailureRetrySeconds = domainPresenceFailureRetrySeconds;
@@ -66,6 +69,7 @@ public MainTuning(
       this.targetNamespaceRecheckIntervalSeconds = targetNamespaceRecheckIntervalSeconds;
       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("targetNamespaceRecheckIntervalSeconds", 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/helpers/CallBuilder.java

@@ -289,7 +289,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;
 
@@ -499,6 +499,11 @@ public CallBuilder withFieldSelector(String fieldSelector) {
     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;

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

@@ -133,6 +133,7 @@ public class MessageKeys {
   public static final String INTROSPECTOR_JOB_FAILED = "WLSKO-0175";
   public static final String INTROSPECTOR_JOB_FAILED_DETAIL = "WLSKO-0176";
   public static final String INTROSPECTOR_POD_FAILED = "WLSKO-0177";
+  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

@@ -187,6 +187,7 @@ WLSKO-0175=Job {0} in namespace {1} failed with status {2}. Check log messages \
   copied from the introspector pod {3} log for additional information.
 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-0179=Pod {0} in namespace {1} detected as stuck, and force-deleted
 
 # Domain status messages
 

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

@@ -201,7 +201,6 @@ private Map<String, AtomicBoolean> createNamespaceFlags() {
         .collect(Collectors.toMap(identity(), a -> new AtomicBoolean()));
   }
 
-  @SuppressWarnings("unchecked")
   private void invoke_stopNamespace(String namespace, boolean inTargetNamespaceList)
       throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
     if (stopNamespace == null) {
@@ -227,7 +226,7 @@ public static Memento install(int newValue) throws NoSuchFieldException {
 
     @Override
     public MainTuning getMainTuning() {
-      return new MainTuning(2, 2, domainPresenceRecheckIntervalSeconds, 2, 2, 2, 2L, 2L);
+      return new MainTuning(2, 2, domainPresenceRecheckIntervalSeconds, 2, 2, 2, 30, 2L, 2L);
     }
   }