[grid] Improving Node health check execution

diemol · diemol · commit 127fe0fe6aae · 2022-01-21T18:37:46.000-05:00
This commit does a couple of things... First, it removes the Regularly used to run all health checks and now uses a ScheduledExecutorService. This was done because this replacement has shown better CPU performance (checked with VisualVM(. Second, it runs the health check when the Node is registered to set the availability to UP. This was already being done by Regularly. But if the check was failing for any reason, no retry was attempted, and the next check execution was when the Regularly schedule interval kicks in. Instead of that, we use now Failsafe to run a retry when the initial execution of the health check fails. This is useful in environments like Kubernetes, since the pod comes up, but it might take a few seconds until the Distributor can reach the IP assigned to the pod. #9847
diff --git a/java/src/org/openqa/selenium/HealthCheckFailedException.java b/java/src/org/openqa/selenium/HealthCheckFailedException.java
@@ -0,0 +1,28 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.openqa.selenium;
+
+/**
+ * Indicates that a Node health check failed.
+ */
+public class HealthCheckFailedException extends WebDriverException {
+
+  public HealthCheckFailedException(String msg, Throwable cause) {
+    super(msg, cause);
+  }
+}
diff --git a/java/src/org/openqa/selenium/grid/distributor/local/BUILD.bazel b/java/src/org/openqa/selenium/grid/distributor/local/BUILD.bazel
@@ -30,5 +30,6 @@ java_library(
         "//java/src/org/openqa/selenium/json",
         "//java/src/org/openqa/selenium/remote",
         artifact("com.google.guava:guava"),
+        artifact("net.jodah:failsafe"),
     ],
 )
diff --git a/java/src/org/openqa/selenium/grid/distributor/local/LocalDistributor.java b/java/src/org/openqa/selenium/grid/distributor/local/LocalDistributor.java
@@ -28,15 +28,19 @@
 import static org.openqa.selenium.remote.tracing.AttributeKey.SESSION_URI;
 import static org.openqa.selenium.remote.tracing.Tags.EXCEPTION;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.RetryPolicy;
+
 import org.openqa.selenium.Beta;
 import org.openqa.selenium.Capabilities;
+import org.openqa.selenium.HealthCheckFailedException;
 import org.openqa.selenium.ImmutableCapabilities;
 import org.openqa.selenium.RetrySessionRequestException;
 import org.openqa.selenium.SessionNotCreatedException;
 import org.openqa.selenium.WebDriverException;
-import org.openqa.selenium.concurrent.Regularly;
 import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.config.Config;
 import org.openqa.selenium.grid.data.CreateSessionRequest;
@@ -116,7 +120,6 @@ public class LocalDistributor extends Distributor implements Closeable {
   private final SessionMap sessions;
   private final SlotSelector slotSelector;
   private final Secret registrationSecret;
-  private final Regularly hostChecker = new Regularly("distributor host checker");
   private final Map<NodeId, Runnable> allChecks = new HashMap<>();
   private final Duration healthcheckInterval;
 
@@ -142,6 +145,15 @@ public class LocalDistributor extends Distributor implements Closeable {
         return thread;
       });
 
+  private final ScheduledExecutorService nodeHealthCheckService =
+    Executors.newSingleThreadScheduledExecutor(
+      r -> {
+        Thread thread = new Thread(r);
+        thread.setDaemon(true);
+        thread.setName("Local Distributor - Node Health Check");
+        return thread;
+      });
+
   private final Executor sessionCreatorExecutor = Executors.newFixedThreadPool(
     Runtime.getRuntime().availableProcessors(),
     r -> {
@@ -197,6 +209,12 @@ public LocalDistributor(
     Runnable purgeDeadNodes = model::purgeDeadNodes;
     purgeDeadNodesService.scheduleAtFixedRate(purgeDeadNodes, 30, 30, TimeUnit.SECONDS);
 
+    nodeHealthCheckService.scheduleAtFixedRate(
+      runNodeHealthChecks(),
+      this.healthcheckInterval.toMillis(),
+      this.healthcheckInterval.toMillis(),
+      TimeUnit.MILLISECONDS);
+
     // if sessionRequestRetryInterval is 0, we will schedule session creation every 10 millis
     long period = sessionRequestRetryInterval.isZero() ?
                   10 : sessionRequestRetryInterval.toMillis();
@@ -281,14 +299,26 @@ public LocalDistributor add(Node node) {
     try {
       model.add(node.getStatus());
       nodes.put(node.getId(), node);
-    } catch (Exception e){
+    } catch (Exception e) {
       return this;
     }
 
     // Extract the health check
     Runnable runnableHealthCheck = asRunnableHealthCheck(node);
     allChecks.put(node.getId(), runnableHealthCheck);
-    hostChecker.submit(runnableHealthCheck, healthcheckInterval, Duration.ofSeconds(30));
+
+    // Running the health check right after the Node registers itself. We retry the
+    // execution because the Node might on a complex network topology. For example,
+    // Kubernetes pods with IPs that take a while before they are reachable.
+    RetryPolicy<Object> initialHealthCheckPolicy = new RetryPolicy<>()
+      .withMaxAttempts(-1)
+      .withMaxDuration(Duration.ofSeconds(90))
+      .withDelay(Duration.ofSeconds(15))
+      .abortIf(result -> true);
+
+    LOG.log(getDebugLogLevel(), "Running initial health check for Node " + node.getId());
+    Executors.newSingleThreadExecutor().submit(
+      () -> Failsafe.with(initialHealthCheckPolicy).run(runnableHealthCheck::run));
 
     LOG.info(String.format(
       "Added node %s at %s. Health check every %ss",
@@ -301,10 +331,21 @@ public LocalDistributor add(Node node) {
     return this;
   }
 
+  private Runnable runNodeHealthChecks() {
+    return () -> {
+      ImmutableMap<NodeId, Runnable> nodeHealthChecks = ImmutableMap.copyOf(allChecks);
+      for (Runnable nodeHealthCheck : nodeHealthChecks.values()) {
+        nodeHealthCheck.run();
+      }
+    };
+  }
+
   private Runnable asRunnableHealthCheck(Node node) {
     HealthCheck healthCheck = node.getHealthCheck();
     NodeId id = node.getId();
     return () -> {
+      boolean checkFailed = false;
+      Exception failedCheckException = null;
       LOG.log(getDebugLogLevel(), "Running health check for " + node.getId());
 
       HealthCheck.Result result;
@@ -313,19 +354,27 @@ private Runnable asRunnableHealthCheck(Node node) {
       } catch (Exception e) {
         LOG.log(Level.WARNING, "Unable to process node " + id, e);
         result = new HealthCheck.Result(DOWN, "Unable to run healthcheck. Assuming down");
+        checkFailed = true;
+        failedCheckException = e;
       }
 
       Lock writeLock = lock.writeLock();
       writeLock.lock();
       try {
         LOG.log(
           getDebugLogLevel(),
-          String.format("Health check result for %s was %s", node.getId(), result.getAvailability()));
+          String.format(
+            "Health check result for %s was %s",
+            node.getId(),
+            result.getAvailability()));
         model.setAvailability(id, result.getAvailability());
         model.updateHealthCheckCount(id, result.getAvailability());
       } finally {
         writeLock.unlock();
       }
+      if (checkFailed) {
+        throw new HealthCheckFailedException("Node " + id, failedCheckException);
+      }
     };
   }
 
@@ -355,10 +404,7 @@ public void remove(NodeId nodeId) {
     try {
       nodes.remove(nodeId);
       model.remove(nodeId);
-      Runnable runnable = allChecks.remove(nodeId);
-      if (runnable != null) {
-        hostChecker.remove(runnable);
-      }
+      allChecks.remove(nodeId);
     } finally {
       writeLock.unlock();
     }
@@ -599,7 +645,7 @@ private boolean reserve(SlotId id) {
   public void close() {
     LOG.info("Shutting down Distributor executor service");
     purgeDeadNodesService.shutdown();
-    hostChecker.shutdown();
+    nodeHealthCheckService.shutdown();
     newSessionService.shutdown();
   }
 

Original file line number	Diff line number	Diff line change
`@@ -30,5 +30,6 @@ java_library(`
`30`	`30`	`"//java/src/org/openqa/selenium/json",`
`31`	`31`	`"//java/src/org/openqa/selenium/remote",`
`32`	`32`	`artifact("com.google.guava:guava"),`
	`33`	`+ artifact("net.jodah:failsafe"),`
`33`	`34`	`],`
`34`	`35`	`)`