Fix http throwable responses (#2251)

* handle throwable response plus cleanup

Author: doxiao

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

@@ -4,6 +4,7 @@
 package oracle.kubernetes.operator.helpers;
 
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Function;
 
@@ -36,22 +37,22 @@ static void annotateForPrometheus(V1ObjectMeta meta, String applicationContext,
     meta.putAnnotationsItem("prometheus.io/scrape", "true");
   }
 
-  public static V1Pod withSha256Hash(V1Pod pod) {
+  static V1Pod withSha256Hash(V1Pod pod) {
     return DEBUG ? addHashAndDebug(pod) : addHash(pod);
   }
 
   public static <K extends KubernetesObject> K withSha256Hash(K kubernetesObject) {
     return withSha256Hash(kubernetesObject, kubernetesObject);
   }
 
-  public static <K extends KubernetesObject> K withSha256Hash(K kubernetesObject, Object objectToHash) {
+  static <K extends KubernetesObject> K withSha256Hash(K kubernetesObject, Object objectToHash) {
     return addHash(kubernetesObject, objectToHash);
   }
 
   private static V1Pod addHashAndDebug(V1Pod pod) {
     String dump = Yaml.dump(pod);
     addHash(pod);
-    pod.getMetadata().putAnnotationsItem(HASHED_STRING, dump);
+    Objects.requireNonNull(pod.getMetadata()).putAnnotationsItem(HASHED_STRING, dump);
     return pod;
   }
 

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

@@ -126,7 +126,7 @@ private Step getConflictStep() {
 
   abstract Map<String, String> getPodAnnotations();
 
-  String getNamespace() {
+  private String getNamespace() {
     return info.getNamespace();
   }
 
@@ -156,7 +156,7 @@ Domain getDomain() {
     return info.getDomain();
   }
 
-  String getDomainName() {
+  private String getDomainName() {
     return domainTopology.getName();
   }
 
@@ -172,7 +172,7 @@ String getAsName() {
     return domainTopology.getAdminServerName();
   }
 
-  Integer getAsPort() {
+  private Integer getAsPort() {
     return domainTopology
         .getServerConfig(domainTopology.getAdminServerName())
         .getLocalAdminProtocolChannelPort();
@@ -375,7 +375,7 @@ private Step patchCurrentPod(V1Pod currentPod, Step next) {
     return createProgressingStep(patchPod(currentPod, next));
   }
 
-  protected Step patchPod(V1Pod currentPod, Step next) {
+  private Step patchPod(V1Pod currentPod, Step next) {
     JsonPatchBuilder patchBuilder = Json.createPatchBuilder();
     KubernetesUtils.addPatches(
         patchBuilder, "/metadata/labels/", getLabels(currentPod), getNonHashedPodLabels());

operator/src/main/java/oracle/kubernetes/operator/http/HttpAsyncRequestStep.java

@@ -8,7 +8,6 @@
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -53,7 +52,7 @@ private HttpAsyncRequestStep(HttpRequest request, HttpResponseStep responseStep)
    * @param responseStep the step to handle the response
    * @return a new step to run as part of a fiber, linked to the response step
    */
-  public static HttpAsyncRequestStep createGetRequest(String url, HttpResponseStep responseStep) {
+  static HttpAsyncRequestStep createGetRequest(String url, HttpResponseStep responseStep) {
     HttpRequest request = HttpRequest.newBuilder(URI.create(url)).GET().build();
     return create(request, responseStep);
   }
@@ -107,11 +106,14 @@ private void checkTimeout(AsyncFiber fiber) {
     }
 
     private void resume(AsyncFiber fiber, HttpResponse<String> response, Throwable throwable) {
-      if (throwable != null) {
-        LOGGER.fine(MessageKeys.HTTP_REQUEST_TIMED_OUT, request.method(), request.uri(), throwable);
+      if (throwable instanceof HttpTimeoutException) {
+        LOGGER.fine(MessageKeys.HTTP_REQUEST_TIMED_OUT, throwable.getMessage());
+      } else if (response != null) {
+        recordResponse(response);
+      } else if (throwable != null) {
+        recordThrowableResponse(throwable);
       }
-      
-      Optional.ofNullable(response).ifPresent(this::recordResponse);
+
       fiber.resume(packet);
     }
 
@@ -121,8 +123,12 @@ private void recordResponse(HttpResponse<String> response) {
       }
       HttpResponseStep.addToPacket(packet, response);
     }
-  }
 
+    private void recordThrowableResponse(Throwable throwable) {
+      LOGGER.warning(MessageKeys.HTTP_REQUEST_GOT_THROWABLE, request.method(), request.uri(), throwable.getMessage());
+      HttpResponseStep.addToPacket(packet, throwable);
+    }
+  }
 
   private static CompletableFuture<HttpResponse<String>> createFuture(HttpRequest request) {
     return httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString());
@@ -132,7 +138,7 @@ static class HttpTimeoutException extends RuntimeException {
     private final String method;
     private final URI uri;
 
-    public HttpTimeoutException(String method, URI uri) {
+    HttpTimeoutException(String method, URI uri) {
       this.method = method;
       this.uri = uri;
     }

operator/src/main/java/oracle/kubernetes/operator/http/HttpResponseStep.java

@@ -21,7 +21,19 @@ public HttpResponseStep(Step next) {
 
   @Override
   public NextAction apply(Packet packet) {
-    return Optional.ofNullable(getResponse(packet)).map(r -> doApply(packet, r)).orElse(doNext(packet));
+    return Optional.ofNullable(getResponse(packet))
+        .map(r -> doApply(packet, r))
+        .orElse(handlePossibleThrowableOrContinue(packet));
+  }
+
+  private NextAction handlePossibleThrowableOrContinue(Packet packet) {
+    return Optional.ofNullable(getThrowableResponse(packet))
+        .map(t -> onFailure(packet, null))
+        .orElse(doNext(packet));
+  }
+
+  private Throwable getThrowableResponse(Packet packet) {
+    return packet.getSpi(Throwable.class);
   }
 
   private NextAction doApply(Packet packet, HttpResponse<String> response) {
@@ -46,6 +58,15 @@ static void addToPacket(Packet packet, HttpResponse<String> response) {
     packet.getComponents().put(RESPONSE, Component.createFor(HttpResponse.class, response));
   }
 
+  /**
+   * Adds the specified throwable to a packet so that this step can access it via {@link Packet#getSpi(Class)} call.
+   * @param packet the packet to which the response should be added
+   * @param throwable the throwable from the server
+   */
+  static void addToPacket(Packet packet, Throwable throwable) {
+    packet.getComponents().put(RESPONSE, Component.createFor(Throwable.class, throwable));
+  }
+
   /**
    * Removes any current response from the packet.
    * @param packet the packet from which the response should be removed

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

@@ -139,6 +139,7 @@ public class MessageKeys {
   public static final String BEGIN_MANAGING_NAMESPACE = "WLSKO-0186";
   public static final String END_MANAGING_NAMESPACE = "WLSKO-0187";
   public static final String MII_DOMAIN_DYNAMICALLY_UPDATED = "WLSKO-0188";
+  public static final String HTTP_REQUEST_GOT_THROWABLE = "WLSKO-0189";
 
   // domain status messages
   public static final String DUPLICATE_SERVER_NAME_FOUND = "WLSDO-0001";

operator/src/main/java/oracle/kubernetes/operator/steps/ManagedServerUpIteratorStep.java

@@ -49,11 +49,11 @@ public class ManagedServerUpIteratorStep extends Step {
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
 
   /** The interval in msec that the operator will wait to ensure that started pods have been scheduled on a node. */
-  public static final int SCHEDULING_DETECTION_DELAY = 100;
+  static final int SCHEDULING_DETECTION_DELAY = 100;
 
   private final Collection<ServerStartupInfo> startupInfos;
 
-  public ManagedServerUpIteratorStep(Collection<ServerStartupInfo> startupInfos, Step next) {
+  ManagedServerUpIteratorStep(Collection<ServerStartupInfo> startupInfos, Step next) {
     super(next);
     this.startupInfos = startupInfos;
   }
@@ -133,7 +133,7 @@ private StepAndPacket createManagedServerUpWaiters(Packet packet, ServerStartupI
             createPacketForServer(packet, ssi));
   }
 
-  String getPodName(DomainPresenceInfo info, String serverName) {
+  private String getPodName(DomainPresenceInfo info, String serverName) {
     return LegalNames.toPodName(info.getDomainUid(), serverName);
   }
 
@@ -234,7 +234,7 @@ private static class StartClusteredServersStepFactory {
       this.maxConcurrency = maxConcurrency;
     }
 
-    public int getMaxConcurrency() {
+    int getMaxConcurrency() {
       return this.maxConcurrency;
     }
 

operator/src/main/java/oracle/kubernetes/operator/steps/ReadHealthStep.java

@@ -53,8 +53,8 @@
 
 public class ReadHealthStep extends Step {
 
-  public static final String OVERALL_HEALTH_NOT_AVAILABLE = "Not available";
-  public static final String OVERALL_HEALTH_FOR_SERVER_OVERLOADED =
+  static final String OVERALL_HEALTH_NOT_AVAILABLE = "Not available";
+  static final String OVERALL_HEALTH_FOR_SERVER_OVERLOADED =
       OVERALL_HEALTH_NOT_AVAILABLE + " (possibly overloaded)";
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
 
@@ -305,7 +305,10 @@ private Map<String, String> getServerStateMap() {
       }
 
       private boolean isServerOverloaded() {
-        return isServerOverloaded(getResponse().statusCode());
+        return Optional.ofNullable(getResponse())
+            .map(HttpResponse::statusCode)
+            .map(this::isServerOverloaded)
+            .orElse(false);
       }
 
       private boolean isServerOverloaded(int statusCode) {

operator/src/main/resources/Operator.properties

@@ -118,7 +118,7 @@ WLSKO-0166=for {0}
 WLSKO-0167=with {0}
 WLSKO-0168={0}: {1}
 WLSKO-0169=Job {0} is created at {1}
-WLSKO-0170=HTTP timeout: exception {0}.
+WLSKO-0170=HTTP timeout: {0}.
 WLSKO-0171=Namespace {0} is in operator's domain namespaces list but doesn't exist
 WLSKO-0173=Replace custom resource definition failed: {0}
 WLSKO-0174=Create custom resource definition failed: {0}
@@ -137,6 +137,7 @@ WLSKO-0185=Patching Pod Disruption Budget for WebLogic domain with UID: {0}. Clu
 WLSKO-0186=Start managing namespace {0}
 WLSKO-0187=Stop managing namespace {0}
 WLSKO-0188=Model in Image domain with DomainUID ''{0}'' and server name ''{1}'' online updated successfully. No restart is necessary
+WLSKO-0189=HTTP request method {0} to {1} failed with exception {2}.
 
 # Domain status messages
 

operator/src/test/java/oracle/kubernetes/operator/http/HttpAsyncRequestStepTest.java

@@ -29,8 +29,10 @@
 
 import static com.meterware.simplestub.Stub.createStub;
 import static oracle.kubernetes.operator.logging.MessageKeys.HTTP_METHOD_FAILED;
+import static oracle.kubernetes.operator.logging.MessageKeys.HTTP_REQUEST_GOT_THROWABLE;
 import static oracle.kubernetes.operator.logging.MessageKeys.HTTP_REQUEST_TIMED_OUT;
 import static oracle.kubernetes.utils.LogMatcher.containsFine;
+import static oracle.kubernetes.utils.LogMatcher.containsWarning;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -109,6 +111,12 @@ private void receiveResponseBeforeTimeout(NextAction nextAction, HttpResponse<St
     FiberTestSupport.doOnExit(nextAction, fiber);
   }
 
+  private void completeWithThrowableBeforeTimeout(NextAction nextAction, Throwable throwable) {
+    responseFuture.completeExceptionally(throwable);
+    FiberTestSupport.doOnExit(nextAction, fiber);
+  }
+
+
   @Test
   public void whenErrorResponseReceived_logMessage() {
     final NextAction nextAction = requestStep.apply(packet);
@@ -118,6 +126,17 @@ public void whenErrorResponseReceived_logMessage() {
     assertThat(logRecords, containsFine(HTTP_METHOD_FAILED));
   }
 
+  @Test
+  public void whenThrowableResponseReceived_logMessage() {
+    consoleMemento
+        .collectLogMessages(logRecords, HTTP_REQUEST_GOT_THROWABLE)
+        .withLogLevel(Level.WARNING);
+    final NextAction nextAction = requestStep.apply(packet);
+
+    completeWithThrowableBeforeTimeout(nextAction, new Throwable("Test"));
+
+    assertThat(logRecords, containsWarning(HTTP_REQUEST_GOT_THROWABLE));
+  }
 
   @Test
   public void whenResponseReceived_populatePacket() {