[ZEPPELIN-6576] Add REST API for a bounded on-demand interpreter health check

conf/zeppelin-site.xml.template

@@ -433,6 +433,12 @@
   <description>Interpreter process connect timeout. Default time unit is msec.</description>
 </property>
 
+<property>
+  <name>zeppelin.interpreter.healthcheck.timeout</name>
+  <value>3s</value>
+  <description>How long one interpreter health check request waits, shared by all interpreter processes of the setting. Default time unit is msec.</description>
+</property>
+
 <property>
   <name>zeppelin.interpreter.output.limit</name>
   <value>102400</value>

docs/setup/operation/configuration.md

@@ -376,6 +376,12 @@ Sources descending by priority:
     <td>600s</td>
     <td>Interpreter process connect timeout. Default time unit is msec</td>
   </tr>
+  <tr>
+    <td><h6 class="properties">ZEPPELIN_INTERPRETER_HEALTHCHECK_TIMEOUT</h6></td>
+    <td><h6 class="properties">zeppelin.interpreter.healthcheck.timeout</h6></td>
+    <td>3s</td>
+    <td>How long one interpreter health check request waits, shared by all interpreter processes of the setting. Default time unit is msec</td>
+  </tr>
   <tr>
     <td><h6 class="properties">ZEPPELIN_DEP_LOCALREPO</h6></td>
     <td><h6 class="properties">zeppelin.dep.localrepo</h6></td>

docs/usage/rest_api/interpreter.md

@@ -520,6 +520,154 @@ The role of registered interpreters, settings and interpreters group are describ
     </tr>
   </table>
 
+<br/>
+### Health check an interpreter
+
+  <table class="table-configuration">
+    <col width="200">
+    <tr>
+      <td>Description</td>
+      <td>This ```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.<br/><br/>
+        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.
+      </td>
+    </tr>
+    <tr>
+      <td>URL</td>
+      <td>```http://[zeppelin-server]:[zeppelin-port]/api/interpreter/setting/[interpreter ID]/healthcheck```</td>
+    </tr>
+    <tr>
+      <td>Success code</td>
+      <td>200</td>
+    </tr>
+    <tr>
+      <td>Fail code</td>
+      <td> 403 if the note of the request may not be acted on, 404 if there is no such interpreter setting </td>
+    </tr>
+    <tr>
+      <td>Sample JSON input (Optional)</td>
+      <td>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"
+}
+```
+</td>
+    </tr>
+    <tr>
+      <td>Sample JSON response</td>
+      <td>```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
+      }
+    ]
+  }
+}
+```
+</td>
+    </tr>
+    <tr>
+      <td>Reasons</td>
+      <td>Only ```OK``` is healthy. The other reasons are not equally bad and are worth telling
+        apart.
+        <ul>
+          <li>```OK```: the launcher reports the interpreter as running.</li>
+          <li>```NOT_REACHABLE```: a process exists, but the launcher reports it as not running,
+            or the probe failed.</li>
+          <li>```PROBE_TIMEOUT```: the probe started but did not finish within the budget.</li>
+          <li>```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.</li>
+          <li>```LAUNCHING```: the process is still being launched and was left unprobed.</li>
+          <li>```NOT_RUNNING```: no process is running, the normal state right after a restart.</li>
+        </ul>
+      </td>
+    </tr>
+    <tr>
+      <td>Check methods</td>
+      <td>What ```checkMethod``` says the launcher looked at. None of them is a reply from the
+        interpreter.
+        <ul>
+          <li>```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.</li>
+          <li>```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.</li>
+          <li>```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.</li>
+          <li>```KUBERNETES_POD_PHASE``` (Kubernetes launcher): the Pod phase is ```Running```
+            and the interpreter registered with the server once.</li>
+          <li>```UNKNOWN```: a launcher that does not describe its check.</li>
+        </ul>
+      </td>
+    </tr>
+    <tr>
+      <td>Sample JSON response of an interpreter that is not running</td>
+      <td>
+
+```json
+{
+  "status": "OK",
+  "message": "",
+  "body": {
+    "settingId": "2CH2VMY7B",
+    "settingName": "spark",
+    "reason": "NOT_RUNNING",
+    "groups": []
+  }
+}
+```
+</td>
+    </tr>
+  </table>
+
 <br/>
 ### Add a new repository for dependency resolving
 

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;

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();

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;
+  }
 }

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),