diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index d04aee833e6..56b13f5b7d1 100755 --- a/conf/zeppelin-site.xml.template +++ b/conf/zeppelin-site.xml.template @@ -433,6 +433,12 @@ Interpreter process connect timeout. Default time unit is msec. + + zeppelin.interpreter.healthcheck.timeout + 3s + How long one interpreter health check request waits, shared by all interpreter processes of the setting. Default time unit is msec. + + zeppelin.interpreter.output.limit 102400 diff --git a/docs/setup/operation/configuration.md b/docs/setup/operation/configuration.md index 0a53f5179ee..80dbb1bd3c8 100644 --- a/docs/setup/operation/configuration.md +++ b/docs/setup/operation/configuration.md @@ -376,6 +376,12 @@ Sources descending by priority: 600s Interpreter process connect timeout. Default time unit is msec + +
ZEPPELIN_INTERPRETER_HEALTHCHECK_TIMEOUT
+
zeppelin.interpreter.healthcheck.timeout
+ 3s + How long one interpreter health check request waits, shared by all interpreter processes of the setting. Default time unit is msec +
ZEPPELIN_DEP_LOCALREPO
zeppelin.dep.localrepo
diff --git a/docs/usage/rest_api/interpreter.md b/docs/usage/rest_api/interpreter.md index 9d81dd60e06..2ed22c14e7d 100644 --- a/docs/usage/rest_api/interpreter.md +++ b/docs/usage/rest_api/interpreter.md @@ -520,6 +520,154 @@ The role of registered interpreters, settings and interpreters group are describ +
+### Health check an interpreter + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
DescriptionThis ```POST``` method probes the interpreter processes of the given interpreter setting + and reports what their launchers say about them. Useful right after a restart, which only + closes the processes and leaves the next use to start them again. The check never starts + an interpreter: a setting with nothing running is reported with the reason + ```NOT_RUNNING```. A setting can own one process per user or note, so the response holds + an entry per interpreter group. The server bounds how long it waits, set by + ```zeppelin.interpreter.healthcheck.timeout``` (3 seconds by default), and a group whose + probe did not finish in time is reported with the reason ```PROBE_TIMEOUT``` rather than + holding the request.

+ The check does not ask the interpreter itself to reply. Each launcher looks at its own + state, reported per group as ```checkMethod``` (see below), so ```healthy``` means the + launcher reports the interpreter as running, not that the interpreter is known to answer. +
URL```http://[zeppelin-server]:[zeppelin-port]/api/interpreter/setting/[interpreter ID]/healthcheck```
Success code200
Fail code 403 if the note of the request may not be acted on, 404 if there is no such interpreter setting
Sample JSON input (Optional)The note to authorize the caller against, needed only when the caller is not allowed to + reach the interpreter endpoints on their own. Running a paragraph of that note is enough. + +```json +{ + "noteId": "2AVQJVC8N" +} +``` +
Sample JSON response```healthy``` follows what the launcher reported. ```alive```, ```running``` and + ```probeTookMs``` are left out when no probe was made, so that an interpreter found dead + stays distinguishable from one that was never asked. + +```json +{ + "status": "OK", + "message": "", + "body": { + "settingId": "2CH2VMY7B", + "settingName": "spark", + "groups": [ + { + "groupId": "2CH2VMY7B-shared_process", + "healthy": true, + "reason": "OK", + "checkMethod": "KUBERNETES_POD_PHASE", + "alive": true, + "running": true, + "probeTookMs": 3 + }, + { + "groupId": "2CH2VMY7B-user2", + "healthy": false, + "reason": "NOT_REACHABLE", + "checkMethod": "KUBERNETES_POD_PHASE", + "alive": true, + "running": false, + "probeTookMs": 1004 + }, + { + "groupId": "2CH2VMY7B-user3", + "healthy": false, + "reason": "PROBE_TIMEOUT", + "checkMethod": "KUBERNETES_POD_PHASE", + "probeTookMs": 3000 + } + ] + } +} +``` +
ReasonsOnly ```OK``` is healthy. The other reasons are not equally bad and are worth telling + apart. +
    +
  • ```OK```: the launcher reports the interpreter as running.
    • +
  • ```NOT_REACHABLE```: a process exists, but the launcher reports it as not running, + or the probe failed.
    • +
  • ```PROBE_TIMEOUT```: the probe started but did not finish within the budget.
    • +
  • ```PROBE_UNAVAILABLE```: the probe never started, because every probe thread was + busy, typically with earlier probes of unreachable hosts. Nothing was checked, so this + says nothing about the interpreter.
    • +
  • ```LAUNCHING```: the process is still being launched and was left unprobed.
    • +
  • ```NOT_RUNNING```: no process is running, the normal state right after a restart.
    • +
+
Check methodsWhat ```checkMethod``` says the launcher looked at. None of them is a reply from the + interpreter. +
    +
  • ```LOCAL_PROCESS_STATE``` (standard, Spark and Flink launchers, when they start the + interpreter process themselves): the state of the process as last recorded by the + server. An interpreter that stopped answering but whose + process is still up is reported as running.
    • +
  • ```YARN_APPLICATION_STATE``` (YARN launcher): whether the YARN application reached + RUNNING once. It is not cleared afterwards, so a failed application is still reported + as running.
    • +
  • ```TCP_CONNECT``` (Docker launcher, an interpreter set to connect to an existing + process, and processes recovered after a server restart): a TCP connect to the + interpreter port succeeded.
    • +
  • ```KUBERNETES_POD_PHASE``` (Kubernetes launcher): the Pod phase is ```Running``` + and the interpreter registered with the server once.
    • +
  • ```UNKNOWN```: a launcher that does not describe its check.
    • +
+
Sample JSON response of an interpreter that is not running + +```json +{ + "status": "OK", + "message": "", + "body": { + "settingId": "2CH2VMY7B", + "settingName": "spark", + "reason": "NOT_RUNNING", + "groups": [] + } +} +``` +
+
### Add a new repository for dependency resolving diff --git a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java index b7ba89a4a5b..2a397a78597 100644 --- a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java +++ b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java @@ -491,6 +491,11 @@ public boolean isRunning() { return false; } + @Override + public CheckMethod getCheckMethod() { + return CheckMethod.TCP_CONNECT; + } + @Override public String getErrorMessage() { DockerClient client = docker; diff --git a/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sRemoteInterpreterProcess.java b/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sRemoteInterpreterProcess.java index 62fcc38df01..1f74bf259ca 100644 --- a/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sRemoteInterpreterProcess.java +++ b/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sRemoteInterpreterProcess.java @@ -239,6 +239,11 @@ public boolean isRunning() { return "Running".equalsIgnoreCase(getPodPhase()) && started.get(); } + @Override + public CheckMethod getCheckMethod() { + return CheckMethod.KUBERNETES_POD_PHASE; + } + public String getPodPhase() { try { Pod pod = client.pods().inNamespace(interpreterNamespace).withName(podName).get(); diff --git a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java index f47eaf68d76..515f26ae564 100644 --- a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java +++ b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java @@ -644,4 +644,9 @@ public boolean isAlive() { public boolean isRunning() { return isYarnAppRunning.get(); } + + @Override + public CheckMethod getCheckMethod() { + return CheckMethod.YARN_APPLICATION_STATE; + } } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java index b2e15160b52..14e1569650e 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java @@ -800,6 +800,16 @@ public long getInterpreterIdleTimeoutThreshold() { return getTimeMillis(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD); } + /** + * Budget for one on demand interpreter health check, shared by all interpreter processes of the + * setting being checked. + * + * @return budget in milliseconds + */ + public long getInterpreterHealthCheckTimeout() { + return getTimeMillis(ConfVars.ZEPPELIN_INTERPRETER_HEALTHCHECK_TIMEOUT); + } + /** * Reads a time valued property. {@link #getString(ConfVars)} returns null for a ConfVars * declared with a numeric default, so the declared default is used when nothing is configured. @@ -1010,6 +1020,7 @@ public enum ConfVars { ZEPPELIN_INTERPRETER_DEP_MVNREPO("zeppelin.interpreter.dep.mvnRepo", "https://repo1.maven.org/maven2/"), ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT("zeppelin.interpreter.connect.timeout", 600000L), + ZEPPELIN_INTERPRETER_HEALTHCHECK_TIMEOUT("zeppelin.interpreter.healthcheck.timeout", 3000L), ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE("zeppelin.interpreter.connection.poolsize", 100), ZEPPELIN_INTERPRETER_GROUP_DEFAULT("zeppelin.interpreter.group.default", "spark"), ZEPPELIN_INTERPRETER_OUTPUT_LIMIT("zeppelin.interpreter.output.limit", 1024 * 100), diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterHealthCheck.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterHealthCheck.java new file mode 100644 index 00000000000..99ac1bde022 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterHealthCheck.java @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF 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.apache.zeppelin.interpreter; + +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess.CheckMethod; + +import java.util.Collections; +import java.util.List; + +/** + * Result of a user triggered health check of a single interpreter setting. + * + *

A setting owns zero or more {@link ManagedInterpreterGroup}s, one per interpreter process, so + * the result reports an entry per group instead of one verdict for the setting. + * + *

{@code healthy} follows {@code InterpreterClient#isRunning()}. That call is documented as "the + * interpreter can communicate with server", but no launcher asks the interpreter: each one looks at + * its own state, from a process launcher flag to a Kubernetes Pod phase. {@code healthy} therefore + * means the launcher reports the interpreter as running, not that the interpreter replied, and + * {@code checkMethod} says what was looked at. + * + *

{@code alive}, {@code running} and {@code probeTookMs} are {@code null} whenever no probe was + * performed, so that "probed and found dead" does not read the same as "never probed". + */ +public class InterpreterHealthCheck { + + /** + * Why a group is, or is not, healthy. Only {@link #OK} means healthy; the remaining reasons are + * not equally bad, and a client is expected to tell them apart rather than to treat every one of + * them as a failure. + */ + public enum Reason { + /** The launcher reports the interpreter as running. */ + OK, + /** A process handle exists, but the launcher reports it as not running or could not tell. */ + NOT_REACHABLE, + /** The probe started but did not finish within the budget of this health check. */ + PROBE_TIMEOUT, + /** + * The probe never started, because every probe thread was busy and the queue was full or the + * budget ran out first. Nothing was checked, so this says nothing about the interpreter. + */ + PROBE_UNAVAILABLE, + /** A process is currently being launched for this group, which is not a failure. */ + LAUNCHING, + /** No process is running, which is the normal state right after a restart. */ + NOT_RUNNING + } + + private final String settingId; + private final String settingName; + private final Reason reason; + private final List groups; + + private InterpreterHealthCheck(InterpreterSetting setting, Reason reason, + List groups) { + this.settingId = setting.getId(); + this.settingName = setting.getName(); + this.reason = reason; + this.groups = groups; + } + + /** + * A setting that holds no interpreter group at all. Restarting a setting only closes its + * processes and the next use lazily creates them again, so this is the expected state right after + * a restart rather than something being wrong. + */ + public static InterpreterHealthCheck notRunning(InterpreterSetting setting) { + return new InterpreterHealthCheck(setting, Reason.NOT_RUNNING, Collections.emptyList()); + } + + /** + * A setting with groups to report. The reason lives on each group, so no setting wide reason is + * set here. + */ + public static InterpreterHealthCheck of(InterpreterSetting setting, List groups) { + return new InterpreterHealthCheck(setting, null, groups); + } + + public String getSettingId() { + return settingId; + } + + public String getSettingName() { + return settingName; + } + + public Reason getReason() { + return reason; + } + + public List getGroups() { + return groups; + } + + /** + * Health of one interpreter group, which is to say of one interpreter process. + * + *

Instances are created through the factory methods so that {@code healthy} is always derived + * from the same probe result the reason was derived from, and the two can never disagree. + */ + public static class GroupHealth { + + private final String groupId; + private final boolean healthy; + private final Reason reason; + private final CheckMethod checkMethod; + private final Boolean alive; + private final Boolean running; + private final Long probeTookMs; + + private GroupHealth(String groupId, Reason reason, CheckMethod checkMethod, Boolean alive, + Boolean running, Long probeTookMs) { + this.groupId = groupId; + this.reason = reason; + this.checkMethod = checkMethod; + this.alive = alive; + this.running = running; + this.probeTookMs = probeTookMs; + this.healthy = Boolean.TRUE.equals(running); + } + + /** The probe came back, and what the launcher reported decides the reason. */ + public static GroupHealth probed(String groupId, CheckMethod checkMethod, boolean alive, + boolean running, long probeTookMs) { + return new GroupHealth(groupId, running ? Reason.OK : Reason.NOT_REACHABLE, checkMethod, + alive, running, probeTookMs); + } + + /** + * The probe did not come back within the budget. Nothing is known about the process, hence no + * {@code alive} or {@code running}. + */ + public static GroupHealth timedOut(String groupId, CheckMethod checkMethod, + long probeTookMs) { + return new GroupHealth(groupId, Reason.PROBE_TIMEOUT, checkMethod, null, null, + probeTookMs); + } + + /** + * The probe itself failed, which says the interpreter could not be reached but leaves the state + * of the process unknown, hence no {@code alive} or {@code running}. + */ + public static GroupHealth probeFailed(String groupId, CheckMethod checkMethod, + long probeTookMs) { + return new GroupHealth(groupId, Reason.NOT_REACHABLE, checkMethod, null, null, + probeTookMs); + } + + /** + * The probe never got a thread to run on, so nothing was checked and no time was spent on the + * process; reporting a timeout here would blame an interpreter that was never asked. + */ + public static GroupHealth unavailable(String groupId, CheckMethod checkMethod) { + return new GroupHealth(groupId, Reason.PROBE_UNAVAILABLE, checkMethod, null, null, null); + } + + /** A process is being launched for this group, so it is deliberately left unprobed. */ + public static GroupHealth launching(String groupId) { + return new GroupHealth(groupId, Reason.LAUNCHING, null, null, null, null); + } + + /** The group carries no process handle, so there is nothing to probe. */ + public static GroupHealth notRunning(String groupId) { + return new GroupHealth(groupId, Reason.NOT_RUNNING, null, null, null, null); + } + + public String getGroupId() { + return groupId; + } + + public boolean isHealthy() { + return healthy; + } + + public Reason getReason() { + return reason; + } + + public CheckMethod getCheckMethod() { + return checkMethod; + } + + public Boolean getAlive() { + return alive; + } + + public Boolean getRunning() { + return running; + } + + public Long getProbeTookMs() { + return probeTookMs; + } + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterHealthChecker.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterHealthChecker.java new file mode 100644 index 00000000000..95c2a601fc6 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterHealthChecker.java @@ -0,0 +1,273 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF 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.apache.zeppelin.interpreter; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; + +import org.apache.zeppelin.interpreter.InterpreterHealthCheck.GroupHealth; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess.CheckMethod; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Probes the interpreter processes of one interpreter setting and reports what their launchers say + * about them, within a deadline that holds for the whole request. + * + *

Unlike the interpreter status snapshot, which is built from server memory alone, this does + * call into the launcher. That is only acceptable because a probe happens when a user explicitly + * asks for one, against a single setting, and under a deadline: {@code isRunning()} costs a socket + * connect for docker and a kube-apiserver round trip for k8s, so the same call would be an accident + * if a listing made it on a timer. What each launcher looks at differs, and none of them asks the + * interpreter itself, so every result carries the {@link CheckMethod} it was based on. + * + *

The deadline can only be imposed from the outside. {@code isRunning()} takes no timeout + * argument, the socket connect behind the docker implementation hardcodes one second, and the + * kubernetes client is created with the defaults of its library. So the probe runs on another + * thread and this class stops waiting for it, which bounds the response but not the probe: + * {@code Future#cancel(boolean)} interrupts, and neither a blocking socket connect nor an HTTP + * client call ends on an interrupt. A probe thread can therefore stay busy after its result was + * given up on, which is why the pool is small and its threads are daemons. + * + *

Once every probe thread is held that way, later probes cannot run until one comes back. They + * are then reported as {@link InterpreterHealthCheck.Reason#PROBE_UNAVAILABLE} rather than as timed + * out, since the interpreter was never asked, and the queue they wait in is bounded and cleared of + * the probes that were given up on, so that pending work cannot pile up while the threads are held. + */ +public class InterpreterHealthChecker { + + private static final Logger LOGGER = LoggerFactory.getLogger(InterpreterHealthChecker.class); + + /** + * A probe that was given up on keeps its thread, so this bounds how many can pile up while an + * interpreter host is unreachable. + */ + private static final int MAX_CONCURRENT_PROBES = 4; + + /** + * Room for probes waiting for a thread. An isolated setting submits one probe per group in a + * single request, so this has to fit a setting with many users or notes; the bound is there so + * that held threads cannot let waiting work grow without limit. Probes beyond it are reported as + * unavailable at once instead of waiting for the deadline. + */ + private static final int MAX_PENDING_PROBES = 64; + + private final long probeTimeoutInMillis; + private final long probeTimeoutInNanos; + private final ThreadPoolExecutor probeExecutor; + + /** + * @param probeTimeoutInMillis budget for one health check request, shared by all groups of the + * setting rather than granted to each of them, so that an isolated + * setting cannot turn into a per group multiple of it + */ + public InterpreterHealthChecker(long probeTimeoutInMillis) { + this(probeTimeoutInMillis, MAX_CONCURRENT_PROBES, MAX_PENDING_PROBES); + } + + @VisibleForTesting + InterpreterHealthChecker(long probeTimeoutInMillis, int maxConcurrentProbes, + int maxPendingProbes) { + Preconditions.checkArgument(probeTimeoutInMillis > 0, + "Health check timeout must be positive, but was %sms", probeTimeoutInMillis); + this.probeTimeoutInMillis = probeTimeoutInMillis; + // Saturates instead of overflowing, so a huge configured value means "wait very long". + this.probeTimeoutInNanos = TimeUnit.MILLISECONDS.toNanos(probeTimeoutInMillis); + this.probeExecutor = newProbeExecutor(maxConcurrentProbes, maxPendingProbes); + } + + private static ThreadPoolExecutor newProbeExecutor(int maxConcurrentProbes, + int maxPendingProbes) { + ThreadPoolExecutor executor = new ThreadPoolExecutor( + maxConcurrentProbes, maxConcurrentProbes, 60L, TimeUnit.SECONDS, + new ArrayBlockingQueue<>(maxPendingProbes), + runnable -> { + Thread thread = new Thread(runnable, "InterpreterHealthChecker-probe"); + thread.setDaemon(true); + return thread; + }); + executor.allowCoreThreadTimeOut(true); + return executor; + } + + /** + * Probes every interpreter group of the given setting. + * + *

A setting owns one group per interpreter process, so an isolated setting is probed once per + * user or note rather than once in total. The probes are submitted together and then collected + * against one deadline, so a group that does not answer costs the request its budget once instead + * of delaying the groups that would have answered right away. + */ + public InterpreterHealthCheck check(InterpreterSetting interpreterSetting) { + List interpreterGroups = interpreterSetting.getAllInterpreterGroups(); + if (interpreterGroups.isEmpty()) { + return InterpreterHealthCheck.notRunning(interpreterSetting); + } + + // The monotonic clock, so that a wall clock adjustment cannot stretch or cut the budget. + long startTimeInNanos = System.nanoTime(); + // Keeps the report in the order of the groups, whether an entry needed a probe or not. + Map healthByGroupId = new LinkedHashMap<>(); + Map probes = new LinkedHashMap<>(); + + for (ManagedInterpreterGroup interpreterGroup : interpreterGroups) { + String groupId = interpreterGroup.getId(); + if (interpreterGroup.isLaunchingInterpreterProcess()) { + // The handle is published before the process is ready and a launch can take minutes for + // Spark on YARN, so probing here would report a starting interpreter as broken. + healthByGroupId.put(groupId, GroupHealth.launching(groupId)); + continue; + } + // Read the handle once: it is set to null as soon as the last session of the group closes, so + // checking it and then reading it again would risk a NullPointerException in between. + RemoteInterpreterProcess process = interpreterGroup.getInterpreterProcess(); + if (process == null) { + healthByGroupId.put(groupId, GroupHealth.notRunning(groupId)); + continue; + } + CheckMethod checkMethod = process.getCheckMethod(); + Probe pendingProbe = new Probe(checkMethod); + try { + healthByGroupId.put(groupId, null); + pendingProbe.future = probeExecutor.submit(() -> { + if (!pendingProbe.claimed.compareAndSet(false, true)) { + // The request already gave up on this probe and reported it as never started. + return null; + } + return probe(groupId, checkMethod, process); + }); + probes.put(groupId, pendingProbe); + } catch (RejectedExecutionException e) { + // Every thread is busy and the queue is full, or stop() ran. + LOGGER.warn("Not probing interpreter group {}, no probe thread is available", groupId); + healthByGroupId.put(groupId, GroupHealth.unavailable(groupId, checkMethod)); + } + } + + boolean gaveUpOnAProbe = false; + for (Map.Entry probe : probes.entrySet()) { + GroupHealth groupHealth = awaitProbe(probe.getKey(), probe.getValue(), startTimeInNanos); + gaveUpOnAProbe |= probe.getValue().future.isCancelled(); + healthByGroupId.put(probe.getKey(), groupHealth); + } + if (gaveUpOnAProbe) { + // A cancelled probe that never started would otherwise keep its place in the queue until a + // thread frees up, which may be never while the threads are held by unreachable hosts. + probeExecutor.purge(); + } + return InterpreterHealthCheck.of(interpreterSetting, + new ArrayList<>(healthByGroupId.values())); + } + + /** + * Stops accepting probes. Probes already in flight are interrupted, which they may well ignore, + * but their threads are daemons and do not hold up a shutdown. + */ + public void stop() { + probeExecutor.shutdownNow(); + } + + @VisibleForTesting + int getPendingProbeCount() { + return probeExecutor.getQueue().size(); + } + + private GroupHealth probe(String groupId, CheckMethod checkMethod, + RemoteInterpreterProcess process) { + long startTimeInNanos = System.nanoTime(); + try { + boolean running = process.isRunning(); + // An interpreter that answers necessarily has a process, so isAlive() is only worth its own + // remote call when the interpreter did not answer - which is where telling a process that is + // gone from one that is up but mute actually helps. + boolean alive = running || process.isAlive(); + return GroupHealth.probed(groupId, checkMethod, alive, running, + millisSince(startTimeInNanos)); + } catch (Exception e) { + // The probe failing is itself an answer about reachability, so it is reported rather than + // propagated: one unreachable group must not hide the results of the others. + LOGGER.warn("Fail to probe interpreter group: {}", groupId, e); + return GroupHealth.probeFailed(groupId, checkMethod, millisSince(startTimeInNanos)); + } + } + + private GroupHealth awaitProbe(String groupId, Probe probe, long startTimeInNanos) { + // Subtracting elapsed time rather than adding the budget to the start cannot overflow. + long remainingInNanos = + Math.max(probeTimeoutInNanos - (System.nanoTime() - startTimeInNanos), 0); + try { + return probe.future.get(remainingInNanos, TimeUnit.NANOSECONDS); + } catch (TimeoutException e) { + LOGGER.info("Probe of interpreter group {} did not finish within {}ms", + groupId, probeTimeoutInMillis); + return giveUp(groupId, probe, startTimeInNanos); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return giveUp(groupId, probe, startTimeInNanos); + } catch (ExecutionException e) { + // probe() already reports a failing probe, so reaching here means something else broke. + LOGGER.warn("Fail to probe interpreter group: {}", groupId, e.getCause()); + return GroupHealth.probeFailed(groupId, probe.checkMethod, millisSince(startTimeInNanos)); + } + } + + private GroupHealth giveUp(String groupId, Probe probe, long startTimeInNanos) { + // Claiming the probe settles in one step whether it ever started: if this succeeds, its thread + // will find it claimed and not run it, so saying it never started is true for good. A timeout + // here would blame an interpreter that was never asked. + boolean neverStarted = probe.claimed.compareAndSet(false, true); + probe.future.cancel(true); + if (neverStarted) { + return GroupHealth.unavailable(groupId, probe.checkMethod); + } + return GroupHealth.timedOut(groupId, probe.checkMethod, millisSince(startTimeInNanos)); + } + + private static long millisSince(long startTimeInNanos) { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeInNanos); + } + + /** A submitted probe, together with whether it has been claimed to run or to be given up on. */ + private static final class Probe { + private final CheckMethod checkMethod; + /** + * Set by whichever comes first: the probe thread starting the probe, or the request giving up + * on it before it started. + */ + private final AtomicBoolean claimed = new AtomicBoolean(); + private Future future; + + private Probe(CheckMethod checkMethod) { + this.checkMethod = checkMethod; + } + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java index 8b4c2fe55be..c1b6c60d610 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java @@ -148,6 +148,7 @@ public class InterpreterSettingManager implements NoteEventListener { private List includesInterpreters; private List excludesInterpreters; private final IdleInterpreterReclaimer idleInterpreterReclaimer; + private final InterpreterHealthChecker interpreterHealthChecker; @Inject public InterpreterSettingManager(ZeppelinConfiguration zConf, @@ -175,6 +176,9 @@ public InterpreterSettingManager(ZeppelinConfiguration zConf, PluginManager pluginManager) throws IOException { this.zConf = zConf; + // First, so that an invalid timeout fails before anything below starts a thread. + this.interpreterHealthChecker = + new InterpreterHealthChecker(zConf.getInterpreterHealthCheckTimeout()); this.pluginManager = pluginManager; this.defaultOption = defaultOption; this.interpreterDirPath = Paths.get(zConf.getInterpreterDir()); @@ -218,6 +222,16 @@ public IdleInterpreterReclaimer getIdleInterpreterReclaimer() { return idleInterpreterReclaimer; } + /** + * Probes the interpreter processes of the given setting and reports what their launchers say + * about them. + * + * @see InterpreterHealthChecker + */ + public InterpreterHealthCheck healthCheck(InterpreterSetting interpreterSetting) { + return interpreterHealthChecker.check(interpreterSetting); + } + public RemoteInterpreterEventServer getInterpreterEventServer() { return interpreterEventServer; } @@ -1132,6 +1146,7 @@ public void close(String settingId) { public void close() { idleInterpreterReclaimer.stop(); + interpreterHealthChecker.stop(); List closeThreads = interpreterSettings.values().stream() .map(intpSetting-> new Thread(intpSetting::close, intpSetting.getId() + "-close")) .peek(t -> t.setUncaughtExceptionHandler((th, e) -> diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ExecRemoteInterpreterProcess.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ExecRemoteInterpreterProcess.java index 6dd2793adf8..4b9db5dcba6 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ExecRemoteInterpreterProcess.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/ExecRemoteInterpreterProcess.java @@ -156,6 +156,11 @@ public boolean isRunning() { return interpreterProcessLauncher != null && interpreterProcessLauncher.isRunning(); } + @Override + public CheckMethod getCheckMethod() { + return CheckMethod.LOCAL_PROCESS_STATE; + } + @Override public String getErrorMessage() { return this.interpreterProcessLauncher != null diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java index e994439c890..94bf748db0f 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java @@ -143,4 +143,32 @@ public boolean recover() { public abstract void processStarted(int port, String host); public abstract String getErrorMessage(); + + /** + * Says what {@link #isRunning()} of this process actually observes. Despite its documentation, + * no launcher asks the interpreter itself: each one looks at a different piece of state, and a + * caller reporting the answer needs to know which one so as not to promise more than was checked. + * + * @return {@link CheckMethod#UNKNOWN} unless a subclass describes its own check + */ + public CheckMethod getCheckMethod() { + return CheckMethod.UNKNOWN; + } + + /** + * What {@link #isRunning()} looks at to decide its answer. None of these is a reply from the + * interpreter, since the interpreter service has no call to ask for one. + */ + public enum CheckMethod { + /** The state of a process this server launched, as last recorded by its launcher. */ + LOCAL_PROCESS_STATE, + /** Whether the YARN application reached RUNNING once; it is not cleared afterwards. */ + YARN_APPLICATION_STATE, + /** A TCP connect to the interpreter port, which shows the port accepts connections. */ + TCP_CONNECT, + /** The Pod phase reported by the Kubernetes API server. */ + KUBERNETES_POD_PHASE, + /** A launcher whose check is not described here. */ + UNKNOWN + } } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java index 71c3a69b3e9..16707c4f981 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java @@ -107,6 +107,11 @@ public boolean isRunning() { return RemoteInterpreterUtils.checkIfRemoteEndpointAccessible(getHost(), getPort()); } + @Override + public CheckMethod getCheckMethod() { + return CheckMethod.TCP_CONNECT; + } + @Override public void processStarted(int port, String host) { // assume process is externally managed. nothing to do diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java index 3b9d754e919..4d765f71c73 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java @@ -30,6 +30,7 @@ import org.apache.zeppelin.notebook.AuthorizationService; import org.apache.zeppelin.common.Message; import org.apache.zeppelin.common.Message.OP; +import org.apache.zeppelin.rest.message.HealthCheckInterpreterRequest; import org.apache.zeppelin.rest.message.InterpreterInstallationRequest; import org.apache.zeppelin.rest.message.NewInterpreterSettingRequest; import org.apache.zeppelin.rest.message.RestartInterpreterRequest; @@ -206,12 +207,7 @@ public Response restartSetting(String message, @PathParam("settingId") String se if (null == noteId) { interpreterSettingManager.close(settingId); } else { - Set entities = new HashSet<>(); - entities.add(authenticationService.getPrincipal()); - entities.addAll(authenticationService.getAssociatedRoles()); - if (authorizationService.hasRunPermission(entities, noteId) || - authorizationService.hasWritePermission(entities, noteId) || - authorizationService.isOwner(entities, noteId)) { + if (hasPermissionOnNote(noteId)) { interpreterSettingManager.restart(settingId, authenticationService.getPrincipal(), noteId); } else { return new JsonResponse<>(Status.FORBIDDEN, "No privilege to restart interpreter") @@ -229,6 +225,54 @@ public Response restartSetting(String message, @PathParam("settingId") String se return new JsonResponse<>(Status.OK, "", setting).build(); } + /** + * Health check of an interpreter setting: probes its interpreter processes and reports what their + * launchers say about them. + * + *

This is the one interpreter read path that calls into the launchers, which may mean a socket + * connect or a Kubernetes API request. That is acceptable because a user asks for it explicitly, + * it names a single setting, and the server bounds the wait. It never starts an interpreter: a + * setting with nothing running is reported as such, since that is the normal state right after a + * restart. + * + * @param message HealthCheckInterpreterRequest, optional + */ + @POST + @Path("setting/{settingId}/healthcheck") + @ZeppelinApi + public Response healthCheckSetting(String message, @PathParam("settingId") String settingId) { + InterpreterSetting setting = interpreterSettingManager.get(settingId); + if (setting == null) { + return new JsonResponse<>(Status.NOT_FOUND, "", settingId).build(); + } + + HealthCheckInterpreterRequest request = + GSON.fromJson(message, HealthCheckInterpreterRequest.class); + String noteId = request == null ? null : request.getNoteId(); + if (null != noteId && !hasPermissionOnNote(noteId)) { + return new JsonResponse<>(Status.FORBIDDEN, "No privilege to health check interpreter") + .build(); + } + + LOGGER.info("Health check interpreterSetting {}, user={}", settingId, + authenticationService.getPrincipal()); + return new JsonResponse<>(Status.OK, "", interpreterSettingManager.healthCheck(setting)) + .build(); + } + + /** + * @return whether the current user may act on the interpreters of the given note, which running a + * paragraph of it already implies + */ + private boolean hasPermissionOnNote(String noteId) { + Set entities = new HashSet<>(); + entities.add(authenticationService.getPrincipal()); + entities.addAll(authenticationService.getAssociatedRoles()); + return authorizationService.hasRunPermission(entities, noteId) + || authorizationService.hasWritePermission(entities, noteId) + || authorizationService.isOwner(entities, noteId); + } + /** * List all available interpreters by group. */ diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/message/HealthCheckInterpreterRequest.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/message/HealthCheckInterpreterRequest.java new file mode 100644 index 00000000000..89ebdebe64b --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/message/HealthCheckInterpreterRequest.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF 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.apache.zeppelin.rest.message; + +/** + * HealthCheckInterpreter rest api request message. + * + *

The note is what a caller who is not an administrator is authorized against, the same way a + * restart from a note page is. A request without one is only served to whoever the deployment lets + * reach the interpreter endpoints at all. + */ +public class HealthCheckInterpreterRequest { + + private final String noteId; + + public HealthCheckInterpreterRequest(String noteId) { + this.noteId = noteId; + } + + public String getNoteId() { + return noteId; + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/InterpreterHealthCheckerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/InterpreterHealthCheckerTest.java new file mode 100644 index 00000000000..ac98fe0c7ce --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/InterpreterHealthCheckerTest.java @@ -0,0 +1,298 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF 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.apache.zeppelin.interpreter; + +import org.apache.zeppelin.interpreter.InterpreterHealthCheck.GroupHealth; +import org.apache.zeppelin.interpreter.InterpreterHealthCheck.Reason; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess.CheckMethod; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.mockito.stubbing.Answer; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Tests the on demand health check, above all that it stays within its deadline and that it never + * starts an interpreter to answer. + * + *

The deadline is what makes probing acceptable here at all: {@code isRunning()} has no timeout + * of its own, so without it a single unreachable interpreter would hold the request for as long as + * the launcher underneath takes to give up. + */ +class InterpreterHealthCheckerTest { + + /** Short enough to keep the tests quick, long enough that a local probe finishes well inside. */ + private static final long PROBE_TIMEOUT_IN_MILLIS = 500; + + private InterpreterHealthChecker healthChecker; + + /** Lets the probes that ignore interruption return once a test is over. */ + private final CountDownLatch releaseHeldProbes = new CountDownLatch(1); + + @AfterEach + void tearDown() { + releaseHeldProbes.countDown(); + if (healthChecker != null) { + healthChecker.stop(); + } + } + + /** + * A setting whose interpreters are not running is the normal state right after a restart, so it + * is reported as such - and answering must not be what starts them. + */ + @Test + void reportsNotRunningWithoutStartingAnInterpreter() throws Exception { + ManagedInterpreterGroup interpreterGroup = interpreterGroup("group-1", null); + InterpreterSetting interpreterSetting = + interpreterSetting(Collections.singletonList(interpreterGroup)); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + InterpreterHealthCheck healthCheck = healthChecker.check(interpreterSetting); + + assertEquals(1, healthCheck.getGroups().size()); + GroupHealth groupHealth = healthCheck.getGroups().get(0); + assertEquals(Reason.NOT_RUNNING, groupHealth.getReason()); + assertFalse(groupHealth.isHealthy(), "a group without a process cannot be healthy"); + assertNull(groupHealth.getAlive(), "nothing was probed, so nothing is known about the process"); + assertNull(groupHealth.getRunning()); + assertNull(groupHealth.getProbeTookMs()); + verify(interpreterGroup, never()).getOrCreateInterpreterProcess(anyString(), any()); + } + + /** A setting without any group at all reports on the setting rather than on groups. */ + @Test + void reportsNotRunningWhenThereIsNoGroup() { + InterpreterSetting interpreterSetting = interpreterSetting(Collections.emptyList()); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + InterpreterHealthCheck healthCheck = healthChecker.check(interpreterSetting); + + assertEquals(Reason.NOT_RUNNING, healthCheck.getReason()); + assertTrue(healthCheck.getGroups().isEmpty()); + } + + /** + * A reachable interpreter is healthy, and saying so takes a single remote call. The result names + * what the launcher looked at, since that is all healthy can promise. + */ + @Test + void reportsHealthyOnASingleRemoteCall() { + RemoteInterpreterProcess process = mock(RemoteInterpreterProcess.class); + when(process.isRunning()).thenReturn(true); + when(process.getCheckMethod()).thenReturn(CheckMethod.KUBERNETES_POD_PHASE); + InterpreterSetting interpreterSetting = interpreterSetting( + Collections.singletonList(interpreterGroup("group-1", process))); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + GroupHealth groupHealth = healthChecker.check(interpreterSetting).getGroups().get(0); + + assertEquals(Reason.OK, groupHealth.getReason()); + assertTrue(groupHealth.isHealthy()); + assertEquals(CheckMethod.KUBERNETES_POD_PHASE, groupHealth.getCheckMethod()); + assertEquals(Boolean.TRUE, groupHealth.getAlive()); + // Asking whether the process exists would be another remote call for no added answer. + verify(process, never()).isAlive(); + } + + /** + * The point of the whole exercise: a probe that does not come back must not hold the request. The + * interpreter here never answers, which is what an unreachable host looks like. + */ + @Test + void stopsWaitingForAProbeThatDoesNotComeBack() { + InterpreterSetting interpreterSetting = interpreterSetting( + Collections.singletonList(interpreterGroup("group-1", ignoringInterruption()))); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + long startTimeInMillis = System.currentTimeMillis(); + GroupHealth groupHealth = healthChecker.check(interpreterSetting).getGroups().get(0); + long tookInMillis = System.currentTimeMillis() - startTimeInMillis; + + assertEquals(Reason.PROBE_TIMEOUT, groupHealth.getReason()); + assertNull(groupHealth.getAlive(), "the probe never answered, so nothing is known"); + assertTrue(tookInMillis < PROBE_TIMEOUT_IN_MILLIS * 10, + "the request should end on the deadline rather than with the probe, but took " + + tookInMillis + "ms"); + } + + /** + * The deadline belongs to the request, so the groups are probed together: one interpreter that + * does not answer must not cost the others their result. + */ + @Test + void oneUnreachableGroupDoesNotHideTheOthers() { + RemoteInterpreterProcess reachable = mock(RemoteInterpreterProcess.class); + when(reachable.isRunning()).thenReturn(true); + InterpreterSetting interpreterSetting = interpreterSetting(Arrays.asList( + interpreterGroup("group-unreachable", ignoringInterruption()), + interpreterGroup("group-reachable", reachable))); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + List groupHealths = healthChecker.check(interpreterSetting).getGroups(); + + assertEquals(2, groupHealths.size()); + // Reported in the order of the groups, whether an entry had to wait for the deadline or not. + assertEquals("group-unreachable", groupHealths.get(0).getGroupId()); + assertEquals(Reason.PROBE_TIMEOUT, groupHealths.get(0).getReason()); + assertEquals("group-reachable", groupHealths.get(1).getGroupId()); + assertEquals(Reason.OK, groupHealths.get(1).getReason()); + } + + /** A group whose process is still being launched is not broken, and is not probed either. */ + @Test + void reportsLaunchingWithoutProbing() { + RemoteInterpreterProcess process = mock(RemoteInterpreterProcess.class); + ManagedInterpreterGroup interpreterGroup = interpreterGroup("group-1", process); + when(interpreterGroup.isLaunchingInterpreterProcess()).thenReturn(true); + InterpreterSetting interpreterSetting = + interpreterSetting(Collections.singletonList(interpreterGroup)); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + GroupHealth groupHealth = healthChecker.check(interpreterSetting).getGroups().get(0); + + assertEquals(Reason.LAUNCHING, groupHealth.getReason()); + verify(process, never()).isRunning(); + } + + /** A probe that throws says the interpreter could not be reached, not that the request failed. */ + @Test + void reportsAFailingProbeAsUnreachable() { + RemoteInterpreterProcess process = mock(RemoteInterpreterProcess.class); + when(process.isRunning()).thenThrow(new RuntimeException("boom")); + InterpreterSetting interpreterSetting = interpreterSetting( + Collections.singletonList(interpreterGroup("group-1", process))); + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + + GroupHealth groupHealth = healthChecker.check(interpreterSetting).getGroups().get(0); + + assertEquals(Reason.NOT_REACHABLE, groupHealth.getReason()); + assertNull(groupHealth.getRunning(), "the probe failed, so the process state is unknown"); + } + + /** + * A probe that ignores interruption, the way a blocking socket connect or HTTP call does, keeps + * its thread after it was given up on. Probes that come after must still not be reported as timed + * out, since they were never run, and must not stay queued behind the held threads. + */ + @Test + void doesNotBlameAnInterpreterItCouldNotProbe() { + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS); + List unreachableGroups = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + unreachableGroups.add(interpreterGroup("group-unreachable-" + i, ignoringInterruption())); + } + for (GroupHealth groupHealth : + healthChecker.check(interpreterSetting(unreachableGroups)).getGroups()) { + assertEquals(Reason.PROBE_TIMEOUT, groupHealth.getReason()); + } + + RemoteInterpreterProcess reachable = mock(RemoteInterpreterProcess.class); + when(reachable.isRunning()).thenReturn(true); + InterpreterSetting reachableSetting = interpreterSetting( + Collections.singletonList(interpreterGroup("group-reachable", reachable))); + for (int i = 0; i < 3; i++) { + GroupHealth groupHealth = healthChecker.check(reachableSetting).getGroups().get(0); + + assertEquals(Reason.PROBE_UNAVAILABLE, groupHealth.getReason(), + "every probe thread is held, so the interpreter was never asked"); + assertNull(groupHealth.getProbeTookMs()); + } + assertEquals(0, healthChecker.getPendingProbeCount(), + "probes that were given up on must not keep their place in the queue"); + verify(reachable, never()).isRunning(); + } + + /** Probes beyond the queue are turned away at once rather than at the deadline. */ + @Test + void reportsProbesBeyondTheQueueAsUnavailable() { + healthChecker = new InterpreterHealthChecker(PROBE_TIMEOUT_IN_MILLIS, 1, 1); + RemoteInterpreterProcess queued = ignoringInterruption(); + RemoteInterpreterProcess rejected = ignoringInterruption(); + InterpreterSetting interpreterSetting = interpreterSetting(Arrays.asList( + interpreterGroup("group-running", ignoringInterruption()), + interpreterGroup("group-queued", queued), + interpreterGroup("group-rejected", rejected))); + + List groupHealths = healthChecker.check(interpreterSetting).getGroups(); + + assertEquals(Reason.PROBE_TIMEOUT, groupHealths.get(0).getReason()); + assertEquals(Reason.PROBE_UNAVAILABLE, groupHealths.get(1).getReason()); + assertEquals(Reason.PROBE_UNAVAILABLE, groupHealths.get(2).getReason()); + assertEquals(0, healthChecker.getPendingProbeCount()); + verify(queued, never()).isRunning(); + verify(rejected, never()).isRunning(); + } + + /** A budget of zero or less would report every group as timed out without asking any. */ + @Test + void rejectsANonPositiveTimeout() { + assertThrows(IllegalArgumentException.class, () -> new InterpreterHealthChecker(0)); + assertThrows(IllegalArgumentException.class, () -> new InterpreterHealthChecker(-1)); + } + + /** A process whose isRunning() blocks until the test ends, whatever interrupts it. */ + private RemoteInterpreterProcess ignoringInterruption() { + RemoteInterpreterProcess process = mock(RemoteInterpreterProcess.class); + Answer blocks = invocation -> { + while (true) { + try { + releaseHeldProbes.await(); + return false; + } catch (InterruptedException e) { + // Swallowed on purpose: a blocking socket connect does not end on an interrupt either. + } + } + }; + when(process.isRunning()).thenAnswer(blocks); + return process; + } + + private static InterpreterSetting interpreterSetting(List groups) { + InterpreterSetting interpreterSetting = mock(InterpreterSetting.class); + when(interpreterSetting.getId()).thenReturn("setting-1"); + when(interpreterSetting.getName()).thenReturn("test"); + when(interpreterSetting.getAllInterpreterGroups()).thenReturn(groups); + return interpreterSetting; + } + + private static ManagedInterpreterGroup interpreterGroup(String groupId, + RemoteInterpreterProcess process) { + ManagedInterpreterGroup interpreterGroup = mock(ManagedInterpreterGroup.class); + when(interpreterGroup.getId()).thenReturn(groupId); + when(interpreterGroup.getInterpreterProcess()).thenReturn(process); + return interpreterGroup; + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java index 19435b32112..b39c3738ba8 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java @@ -496,6 +496,38 @@ void testAddDeleteRepository() throws IOException { delete.close(); } + /** + * A setting that has never been used has no interpreter process, which is also its state right + * after a restart, so the health check reports that instead of starting one. + */ + @Test + void testHealthCheckOfAnInterpreterThatIsNotRunning() throws IOException { + InterpreterSetting mdIntpSetting = notebook.getInterpreterSettingManager() + .getInterpreterSettingByName("md"); + + CloseableHttpResponse post = + httpPost("/interpreter/setting/" + mdIntpSetting.getId() + "/healthcheck", ""); + JsonObject body = getBodyFieldFromResponse( + EntityUtils.toString(post.getEntity(), StandardCharsets.UTF_8)); + + assertThat("health check of an unused interpreter:", post, isAllowed()); + assertEquals(mdIntpSetting.getId(), body.get("settingId").getAsString()); + assertEquals("NOT_RUNNING", body.get("reason").getAsString()); + assertEquals(0, body.getAsJsonArray("groups").size()); + assertEquals(0, mdIntpSetting.getAllInterpreterGroups().size(), + "the health check must not have started an interpreter"); + post.close(); + } + + @Test + void testHealthCheckOfNonExistInterpreterSetting() throws IOException { + CloseableHttpResponse post = + httpPost("/interpreter/setting/no_such_setting/healthcheck", ""); + + assertThat("health check of an unknown setting:", post, isNotFound()); + post.close(); + } + private JsonObject getBodyFieldFromResponse(String rawResponse) { JsonObject response = gson.fromJson(rawResponse, JsonElement.class).getAsJsonObject(); return response.getAsJsonObject("body");