[ZEPPELIN-6575] Reclaim idle interpreters on the server with a per-setting timeout

conf/zeppelin-site.xml.template

@@ -576,7 +576,9 @@
   <name>zeppelin.interpreter.lifecyclemanager.class</name>
   <value>org.apache.zeppelin.interpreter.lifecycle.TimeoutLifecycleManager</value>
   <description>LifecycleManager class for managing the lifecycle of interpreters, by default interpreter will
-  be closed after timeout</description>
+  be closed after timeout. With TimeoutLifecycleManager, Zeppelin server tracks the last use of each
+  interpreter group and closes the idle ones itself, so the threshold below can be overridden per
+  interpreter setting</description>
 </property>
 
 <property>
@@ -588,7 +590,9 @@
 <property>
   <name>zeppelin.interpreter.lifecyclemanager.timeout.threshold</name>
   <value>1h</value>
-  <description>Interpreter timeout threshold, by default it is 1 hour</description>
+  <description>Interpreter timeout threshold, by default it is 1 hour. Set the same property on an
+  individual interpreter setting to override it for that interpreter only, or set it to 0 there to
+  keep that interpreter from ever being reclaimed</description>
 </property>
 -->
 

docs/usage/interpreter/overview.md

@@ -115,6 +115,29 @@ Before 0.8.0, Zeppelin doesn't have lifecycle management for interpreters. Users
 `NullLifecycleManager` will do nothing, i.e., the user needs to control the lifecycle of interpreter by themselves as before. `TimeoutLifecycleManager` will shut down interpreters after an interpreter remains idle for a while. By default, the idle threshold is 1 hour.
 Users can change this threshold via the `zeppelin.interpreter.lifecyclemanager.timeout.threshold` setting. `NullLifecycleManager` is the default lifecycle manager, and users can change it via `zeppelin.interpreter.lifecyclemanager.class`.
 
+### Per interpreter idle threshold
+
+One global threshold is not always enough: a Spark interpreter holding tens of gigabytes of cluster
+memory is worth reclaiming quickly, while a JDBC interpreter that only keeps a few connections open
+is usually worth keeping. With `TimeoutLifecycleManager` configured, Zeppelin server keeps track of
+when each interpreter group was last used and closes the idle ones itself, which lets an individual
+interpreter setting override the global value.
+
+To do so, add `zeppelin.interpreter.lifecyclemanager.timeout.threshold` as a property of that
+interpreter on the interpreter setting page, or in `interpreter.json`. The value accepts the same
+formats as the global one, i.e. a plain number of milliseconds or a unit suffix such as `10m`:
+
+| Value on an interpreter setting | Effect on that interpreter |
+|---|---|
+| not set | the global `zeppelin.interpreter.lifecyclemanager.timeout.threshold` applies |
+| `10m` | it is shut down after 10 minutes of idle time, whatever the global value is |
+| `0` | it is never shut down for being idle |
+
+A paragraph that is still running keeps its interpreter alive regardless of the threshold. The check
+runs every `zeppelin.interpreter.lifecyclemanager.timeout.checkinterval`, which is shared with the
+global behaviour, so an interpreter can be shut down up to one interval after its threshold has
+passed.
+
 
 ## Inline Generic Configuration
 

zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java

@@ -780,6 +780,38 @@ public String getLifecycleManagerClass() {
     return getString(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS);
   }
 
+  /**
+   * Shared with {@code TimeoutLifecycleManager} so that both ways of reclaiming an idle
+   * interpreter check at the same cadence.
+   *
+   * @return interval in milliseconds between two idle checks
+   */
+  public long getInterpreterIdleCheckInterval() {
+    return getTimeMillis(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL);
+  }
+
+  /**
+   * Global idle threshold, which an interpreter setting can override with its own
+   * {@code zeppelin.interpreter.lifecyclemanager.timeout.threshold} property.
+   *
+   * @return threshold in milliseconds
+   */
+  public long getInterpreterIdleTimeoutThreshold() {
+    return getTimeMillis(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD);
+  }
+
+  /**
+   * 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.
+   */
+  private long getTimeMillis(ConfVars c) {
+    String value = getString(c);
+    if (StringUtils.isBlank(value)) {
+      return c.getLongValue();
+    }
+    return parseTimeMillis(value);
+  }
+
   public boolean getZeppelinImpersonateSparkProxyUser() {
       return getBoolean(ConfVars.ZEPPELIN_IMPERSONATE_SPARK_PROXY_USER);
   }
@@ -1239,4 +1271,20 @@ public static long timeUnitToMill(String timeStrWithUnit) {
     return Duration.parse("PT" + timeStrWithUnit).toMillis();
   }
 
+  /**
+   * Parses a time value that is either a plain millisecond number or carries a unit suffix,
+   * e.g. {@code 600000}, {@code 10m} or {@code 500ms}.
+   *
+   * @throws NumberFormatException if the value carries no unit and is not a number
+   * @throws java.time.format.DateTimeParseException if the unit suffix is not understood
+   */
+  public static long parseTimeMillis(String timeStr) {
+    String trimmed = timeStr.trim();
+    try {
+      return Long.parseLong(trimmed);
+    } catch (NumberFormatException e) {
+      return timeUnitToMill(trimmed);
+    }
+  }
+
 }

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java

@@ -47,6 +47,7 @@
 import org.apache.zeppelin.display.AngularObjectRegistryListener;
 import org.apache.zeppelin.helium.ApplicationEventListener;
 import org.apache.zeppelin.interpreter.Interpreter.RegisteredInterpreter;
+import org.apache.zeppelin.interpreter.lifecycle.IdleInterpreterReclaimer;
 import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
 import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
@@ -146,6 +147,7 @@ public class InterpreterSettingManager implements NoteEventListener {
   private Map<String, String> jupyterKernelLanguageMap = new HashMap<>();
   private List<String> includesInterpreters;
   private List<String> excludesInterpreters;
+  private final IdleInterpreterReclaimer idleInterpreterReclaimer;
 
   @Inject
   public InterpreterSettingManager(ZeppelinConfiguration zConf,
@@ -206,6 +208,14 @@ public InterpreterSettingManager(ZeppelinConfiguration zConf,
 
     this.configStorage = configStorage;
     init();
+
+    this.idleInterpreterReclaimer = new IdleInterpreterReclaimer(zConf, this);
+    this.idleInterpreterReclaimer.start();
+  }
+
+  @VisibleForTesting
+  public IdleInterpreterReclaimer getIdleInterpreterReclaimer() {
+    return idleInterpreterReclaimer;
   }
 
   public RemoteInterpreterEventServer getInterpreterEventServer() {
@@ -1111,6 +1121,7 @@ public void close(String settingId) {
   }
 
   public void close() {
+    idleInterpreterReclaimer.stop();
     List<Thread> closeThreads = interpreterSettings.values().stream()
             .map(intpSetting-> new Thread(intpSetting::close, intpSetting.getId() + "-close"))
             .peek(t -> t.setUncaughtExceptionHandler((th, e) ->

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java

@@ -19,6 +19,7 @@
 package org.apache.zeppelin.interpreter;
 
 import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.lifecycle.IdleInterpreterReclaimer;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
 import org.apache.zeppelin.scheduler.Job;
 import org.apache.zeppelin.scheduler.Scheduler;
@@ -43,6 +44,8 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
   private RemoteInterpreterProcess remoteInterpreterProcess; // attached remote interpreter process
   private Object interpreterProcessCreationLock = new Object();
   private final ZeppelinConfiguration zConf;
+  private volatile long lastUsedTimeInMillis = System.currentTimeMillis();
+  private volatile boolean launchingInterpreterProcess;
 
   /**
    * Create InterpreterGroup with given id and interpreterSetting, used in ZeppelinServer
@@ -64,19 +67,54 @@ public RemoteInterpreterProcess getOrCreateInterpreterProcess(String userName,
                                                                 Properties properties)
       throws IOException {
     synchronized (interpreterProcessCreationLock) {
-      if (remoteInterpreterProcess == null) {
-        LOGGER.info("Create InterpreterProcess for InterpreterGroup: {}", getId());
-        remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, userName,
-                properties);
-        remoteInterpreterProcess.start(userName);
-        remoteInterpreterProcess.init(zConf);
-        getInterpreterSetting().getRecoveryStorage()
-                .onInterpreterClientStart(remoteInterpreterProcess);
+      try {
+        if (remoteInterpreterProcess == null) {
+          LOGGER.info("Create InterpreterProcess for InterpreterGroup: {}", getId());
+          launchingInterpreterProcess = true;
+          remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, userName,
+                  properties);
+          remoteInterpreterProcess.start(userName);
+          remoteInterpreterProcess.init(zConf,
+                  IdleInterpreterReclaimer.processConfigurationOverrides(zConf,
+                          interpreterSetting));
+          getInterpreterSetting().getRecoveryStorage()
+                  .onInterpreterClientStart(remoteInterpreterProcess);
+        }
+        return remoteInterpreterProcess;
+      } finally {
+        // Reset the idle clock before dropping the flag, so that this group is never momentarily
+        // visible as idle with a timestamp from before the launch.
+        onInterpreterUse();
+        launchingInterpreterProcess = false;
       }
-      return remoteInterpreterProcess;
     }
   }
 
+  /**
+   * A launch takes a while - minutes for Spark on YARN - and counts as activity rather than as
+   * idle time.
+   *
+   * @return whether a process is currently being launched for this group
+   */
+  public boolean isLaunchingInterpreterProcess() {
+    return launchingInterpreterProcess;
+  }
+
+  /**
+   * Records that this group has just been used, so that server side idle reclaim does not consider
+   * it idle. A single volatile write on purpose: while a paragraph runs this is called on every
+   * status poll of that paragraph.
+   *
+   * @see org.apache.zeppelin.interpreter.lifecycle.IdleInterpreterReclaimer
+   */
+  public void onInterpreterUse() {
+    lastUsedTimeInMillis = System.currentTimeMillis();
+  }
+
+  public long getLastUsedTimeInMillis() {
+    return lastUsedTimeInMillis;
+  }
+
   public RemoteInterpreterProcess getInterpreterProcess() {
     return remoteInterpreterProcess;
   }