diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index d5e54b91f16..d04aee833e6 100755 --- a/conf/zeppelin-site.xml.template +++ b/conf/zeppelin-site.xml.template @@ -576,7 +576,9 @@ zeppelin.interpreter.lifecyclemanager.class org.apache.zeppelin.interpreter.lifecycle.TimeoutLifecycleManager LifecycleManager class for managing the lifecycle of interpreters, by default interpreter will - be closed after timeout + 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 @@ -588,7 +590,9 @@ zeppelin.interpreter.lifecyclemanager.timeout.threshold 1h - Interpreter timeout threshold, by default it is 1 hour + 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 --> diff --git a/docs/usage/interpreter/overview.md b/docs/usage/interpreter/overview.md index fe5cf3bd0b9..e7869357708 100644 --- a/docs/usage/interpreter/overview.md +++ b/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 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 179dcce6e85..b2e15160b52 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 @@ -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); + } + } + } 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 e2959382206..f7b3f675d68 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 @@ -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 jupyterKernelLanguageMap = new HashMap<>(); private List includesInterpreters; private List 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 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/ManagedInterpreterGroup.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java index 8f2c16c0743..71f61c9e86a 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java +++ b/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; } diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimer.java new file mode 100644 index 00000000000..ccb48366b45 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimer.java @@ -0,0 +1,204 @@ +/* + * 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.lifecycle; + +import com.google.common.annotations.VisibleForTesting; + +import org.apache.commons.lang3.StringUtils; +import org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; +import org.apache.zeppelin.interpreter.InterpreterSetting; +import org.apache.zeppelin.interpreter.InterpreterSettingManager; +import org.apache.zeppelin.interpreter.ManagedInterpreterGroup; +import org.apache.zeppelin.scheduler.ExecutorFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +/** + * Closes interpreter groups that have been idle for longer than a threshold, driven by Zeppelin + * server rather than by the interpreter process itself. + * + *

{@link TimeoutLifecycleManager} does the same thing from inside the interpreter process, where + * the threshold can only arrive through the configuration map pushed over Thrift at startup. That + * map holds {@link ConfVars} entries only, so an interpreter setting property never reaches it and + * every process gets the same global threshold. Deciding here means the threshold of the owning + * interpreter setting can just be read. + * + *

Follows {@code zeppelin.interpreter.lifecyclemanager.class}, which already says whether idle + * reclaim is wanted: its {@link NullLifecycleManager} default leaves a deployment untouched, and + * {@link TimeoutLifecycleManager} enables this. Any other implementation is left alone. The + * in-process manager stays as a fallback for a server that went away and is given the same resolved + * threshold by {@link #processConfigurationOverrides}. + */ +public class IdleInterpreterReclaimer { + + private static final Logger LOGGER = LoggerFactory.getLogger(IdleInterpreterReclaimer.class); + + private static final String SCHEDULER_NAME = "IdleInterpreterReclaimer"; + + /** + * Threshold property. On an interpreter setting, {@code 0} or below means never reclaimed. + */ + public static final String IDLE_TIMEOUT_THRESHOLD_PROPERTY = + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(); + + private final ZeppelinConfiguration zConf; + private final InterpreterSettingManager interpreterSettingManager; + + private ScheduledExecutorService checkScheduler; + + public IdleInterpreterReclaimer(ZeppelinConfiguration zConf, + InterpreterSettingManager interpreterSettingManager) { + this.zConf = zConf; + this.interpreterSettingManager = interpreterSettingManager; + } + + private static boolean isEnabled(ZeppelinConfiguration zConf) { + return TimeoutLifecycleManager.class.getName().equals(zConf.getLifecycleManagerClass()); + } + + public void start() { + if (!isEnabled(zConf)) { + LOGGER.debug("Server driven idle interpreter reclaim is off, {} is {}", + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(), + zConf.getLifecycleManagerClass()); + return; + } + long checkInterval = zConf.getInterpreterIdleCheckInterval(); + if (checkInterval <= 0) { + LOGGER.warn("Not starting idle interpreter reclaim: {} must be positive but is {}", + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL.getVarName(), + checkInterval); + return; + } + checkScheduler = ExecutorFactory.singleton().createOrGetScheduled(SCHEDULER_NAME, 1); + // Fixed delay rather than fixed rate, so that a slow close does not queue up further checks. + checkScheduler.scheduleWithFixedDelay(this::reclaimIdleInterpreterGroups, + checkInterval, checkInterval, MILLISECONDS); + LOGGER.info("Server driven idle interpreter reclaim started with checkInterval: {}ms, " + + "default threshold: {}ms", checkInterval, zConf.getInterpreterIdleTimeoutThreshold()); + } + + public void stop() { + if (checkScheduler != null) { + ExecutorFactory.singleton().shutdown(SCHEDULER_NAME); + checkScheduler = null; + LOGGER.info("Server driven idle interpreter reclaim stopped"); + } + } + + /** + * Closes every interpreter group idle for longer than the threshold of its interpreter setting. + * Uses in-memory state only: {@code isAlive()} and {@code isRunning()} cost a socket connect for + * docker and a kube-apiserver round trip for k8s, and this walks every group on a timer. + */ + @VisibleForTesting + void reclaimIdleInterpreterGroups() { + long now = System.currentTimeMillis(); + for (ManagedInterpreterGroup interpreterGroup : + interpreterSettingManager.getAllInterpreterGroup()) { + try { + reclaimIfIdle(interpreterGroup, now); + } catch (Exception e) { + LOGGER.error("Fail to reclaim idle interpreter group: {}", interpreterGroup.getId(), e); + } + } + } + + private void reclaimIfIdle(ManagedInterpreterGroup interpreterGroup, long now) { + if (interpreterGroup.isLaunchingInterpreterProcess()) { + // The handle is published before the process is ready, and a launch can outlast the + // threshold, so this would close a process that is starting rather than an idle one. + return; + } + if (interpreterGroup.getInterpreterProcess() == null) { + // Like TimeoutLifecycleManager, only manage a group once its process has started. + return; + } + if (interpreterGroup.isEmpty()) { + // No session left: the group is already on its way out through close(). + return; + } + + InterpreterSetting interpreterSetting = interpreterGroup.getInterpreterSetting(); + long threshold = getIdleTimeoutThreshold(zConf, interpreterSetting); + if (threshold <= 0) { + LOGGER.debug("Interpreter group {} is never reclaimed, its threshold is {}ms", + interpreterGroup.getId(), threshold); + return; + } + + long idleTimeInMillis = now - interpreterGroup.getLastUsedTimeInMillis(); + if (idleTimeInMillis <= threshold) { + return; + } + + LOGGER.info("Reclaiming interpreter group {} of interpreter setting {}: idle for {}ms which " + + "exceeds its threshold of {}ms", interpreterGroup.getId(), + interpreterSetting == null ? "?" : interpreterSetting.getName(), + idleTimeInMillis, threshold); + interpreterGroup.close(); + } + + /** + * @return idle threshold in milliseconds for the given interpreter setting, taking its own + * {@link #IDLE_TIMEOUT_THRESHOLD_PROPERTY} property over the global configuration + */ + @VisibleForTesting + static long getIdleTimeoutThreshold(ZeppelinConfiguration zConf, + InterpreterSetting interpreterSetting) { + if (interpreterSetting != null) { + String override = + interpreterSetting.getJavaProperties().getProperty(IDLE_TIMEOUT_THRESHOLD_PROPERTY); + if (StringUtils.isNotBlank(override)) { + try { + return ZeppelinConfiguration.parseTimeMillis(override); + } catch (RuntimeException e) { + LOGGER.warn("Ignoring unparsable {} of interpreter setting {}: {}", + IDLE_TIMEOUT_THRESHOLD_PROPERTY, interpreterSetting.getName(), override, e); + } + } + } + return zConf.getInterpreterIdleTimeoutThreshold(); + } + + /** + * Gives the in-process {@link TimeoutLifecycleManager} fallback the threshold resolved here + * instead of the global one. A setting that opted out gets {@link Long#MAX_VALUE} rather than its + * own {@code 0}, which {@link TimeoutLifecycleManager} would read as "shut down at the next + * check" since it has no way to express "never". + * + * @return entries to put on top of {@link ZeppelinConfiguration#getCompleteConfiguration()}, + * empty when server driven reclaim is off + */ + public static Map processConfigurationOverrides( + ZeppelinConfiguration zConf, InterpreterSetting interpreterSetting) { + if (!isEnabled(zConf)) { + return Collections.emptyMap(); + } + long threshold = getIdleTimeoutThreshold(zConf, interpreterSetting); + return Collections.singletonMap(IDLE_TIMEOUT_THRESHOLD_PROPERTY, + String.valueOf(threshold <= 0 ? Long.MAX_VALUE : threshold)); + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java index 6bd2e202325..efa9ea99a02 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java @@ -111,6 +111,19 @@ public ManagedInterpreterGroup getInterpreterGroup() { return (ManagedInterpreterGroup) super.getInterpreterGroup(); } + /** + * Mirrors the {@code onInterpreterUse} hooks that {@code RemoteInterpreterServer} calls inside + * the interpreter process, so that server driven idle reclaim sees the same activity signal. + * Needed explicitly because {@link #getOrCreateInterpreterProcess()} returns the cached handle + * without going through the interpreter group once the process exists. + */ + private void markInterpreterGroupUsed() { + ManagedInterpreterGroup intpGroup = getInterpreterGroup(); + if (intpGroup != null) { + intpGroup.onInterpreterUse(); + } + } + @Override public void open() throws InterpreterException { synchronized (this) { @@ -194,6 +207,7 @@ public InterpreterResult interpret(final String st, final InterpreterContext con if (LOGGER.isDebugEnabled()) { LOGGER.debug("st:\n{}", st); } + markInterpreterGroupUsed(); final FormType form = getFormType(); RemoteInterpreterProcess interpreterProcess = null; @@ -292,6 +306,7 @@ public int getProgress(final InterpreterContext context) throws InterpreterExcep LOGGER.warn("getProgress is called when RemoterInterpreter is not opened for {}", className); return 0; } + markInterpreterGroupUsed(); RemoteInterpreterProcess interpreterProcess = null; try { interpreterProcess = getOrCreateInterpreterProcess(); @@ -325,6 +340,7 @@ public String getStatus(final String jobId) { LOGGER.warn("getStatus is called when RemoteInterpreter is not opened for {}", className); return Job.Status.UNKNOWN.name(); } + markInterpreterGroupUsed(); RemoteInterpreterProcess interpreterProcess = null; try { interpreterProcess = getOrCreateInterpreterProcess(); 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 95802a64fe7..e994439c890 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 @@ -29,7 +29,10 @@ import java.io.IOException; import java.text.SimpleDateFormat; +import java.util.Collections; import java.util.Date; +import java.util.HashMap; +import java.util.Map; /** * Abstract class for interpreter process @@ -101,8 +104,20 @@ public R callRemoteFunction(PooledRemoteClient.RemoteFunction fun } public void init(ZeppelinConfiguration zConf) { + init(zConf, Collections.emptyMap()); + } + + /** + * Pushes the server configuration into the interpreter process. + * + * @param overrides entries to put on top of the global configuration, for settings that are + * resolved per interpreter setting rather than globally + */ + public void init(ZeppelinConfiguration zConf, Map overrides) { + Map properties = new HashMap<>(zConf.getCompleteConfiguration()); + properties.putAll(overrides); callRemoteFunction(client -> { - client.init(zConf.getCompleteConfiguration()); + client.init(properties); return null; }); } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimerTest.java new file mode 100644 index 00000000000..877b834b04c --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/lifecycle/IdleInterpreterReclaimerTest.java @@ -0,0 +1,337 @@ +/* + * 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.lifecycle; + +import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; +import org.apache.zeppelin.interpreter.AbstractInterpreterTest; +import org.apache.zeppelin.interpreter.ExecutionContext; +import org.apache.zeppelin.interpreter.InterpreterSetting; +import org.apache.zeppelin.interpreter.InterpreterSettingManager; +import org.apache.zeppelin.interpreter.ManagedInterpreterGroup; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreter; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess; +import org.apache.zeppelin.scheduler.Job; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.Map; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +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 server driven idle reclaim, above all that an interpreter setting can override the global + * threshold in either direction. That override is what the interpreter process side + * {@link TimeoutLifecycleManager} cannot offer, because its threshold only reaches the process + * through the global configuration map. + */ +class IdleInterpreterReclaimerTest extends AbstractInterpreterTest { + + private static final String THRESHOLD_PROPERTY = + IdleInterpreterReclaimer.IDLE_TIMEOUT_THRESHOLD_PROPERTY; + + @Override + @BeforeEach + public void setUp() throws Exception { + super.setUp(); + zConf.setProperty(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(), + TimeoutLifecycleManager.class.getName()); + zConf.setProperty( + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL.getVarName(), + "1000"); + // The reclaimer picks these up when it starts, and that already happened while + // super.setUp() built the InterpreterSettingManager, so restart it. + interpreterSettingManager.getIdleInterpreterReclaimer().stop(); + interpreterSettingManager.getIdleInterpreterReclaimer().start(); + } + + /** + * A setting may ask to be reclaimed sooner than the global threshold allows. The global + * threshold stays at its 1h default here, so only the per setting value of 10s can close it. + */ + @Test + void perSettingThresholdReclaimsEarlierThanTheGlobalOne() throws Exception { + InterpreterSetting interpreterSetting = + interpreterSettingManager.getInterpreterSettingByName("test"); + interpreterSetting.setProperty(THRESHOLD_PROPERTY, "10s"); + + startEchoInterpreter(); + assertEquals(1, interpreterSetting.getAllInterpreterGroups().size()); + + waitForInterpreterGroups(interpreterSetting, 0, 40); + assertEquals(0, interpreterSetting.getAllInterpreterGroups().size(), + "the group should be reclaimed after the per setting threshold of 10s"); + } + + /** + * The other direction: a non positive per setting threshold means keep it, whatever the short + * global threshold says. + */ + @Test + void perSettingThresholdCanOptOutOfAShortGlobalThreshold() throws Exception { + zConf.setProperty( + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "5s"); + + InterpreterSetting interpreterSetting = + interpreterSettingManager.getInterpreterSettingByName("test"); + interpreterSetting.setProperty(THRESHOLD_PROPERTY, "0"); + + startEchoInterpreter(); + assertEquals(1, interpreterSetting.getAllInterpreterGroups().size()); + + Thread.sleep(20 * 1000); + assertEquals(1, interpreterSetting.getAllInterpreterGroups().size(), + "the setting opted out of reclaim, so the short global threshold must not apply"); + } + + @Test + void globalThresholdAppliesWhenTheSettingDoesNotOverrideIt() throws Exception { + zConf.setProperty( + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "10s"); + + InterpreterSetting interpreterSetting = + interpreterSettingManager.getInterpreterSettingByName("test"); + + startEchoInterpreter(); + assertEquals(1, interpreterSetting.getAllInterpreterGroups().size()); + + waitForInterpreterGroups(interpreterSetting, 0, 40); + assertEquals(0, interpreterSetting.getAllInterpreterGroups().size()); + } + + /** + * A paragraph running for longer than the threshold must not have its interpreter pulled out + * from under it. While a job runs the server polls its status, which counts as use. + */ + @Test + void aRunningParagraphKeepsItsInterpreterAlive() throws Exception { + zConf.setProperty( + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "5s"); + + InterpreterSetting interpreterSetting = + interpreterSettingManager.getInterpreterSettingByName("test"); + final RemoteInterpreter sleepInterpreter = + (RemoteInterpreter) interpreterFactory.getInterpreter("test.sleep", + new ExecutionContext("user1", "note1", "test")); + + // Submit through the scheduler the way Zeppelin submits a paragraph, so that the job status + // poller runs. + sleepInterpreter.getScheduler().submit(new Job("test-job", null) { + @Override + public Object getReturn() { + return null; + } + + @Override + public int progress() { + return 0; + } + + @Override + public Map info() { + return null; + } + + @Override + protected Object jobRun() throws Throwable { + return sleepInterpreter.interpret("30000", createDummyInterpreterContext()); + } + + @Override + protected boolean jobAbort() { + return false; + } + + @Override + public void setResult(Object results) { + } + }); + + long deadline = System.currentTimeMillis() + 30 * 1000; + while (!sleepInterpreter.isOpened() && System.currentTimeMillis() < deadline) { + Thread.sleep(500); + } + assertTrue(sleepInterpreter.isOpened(), "interpreter did not start"); + assertEquals(1, interpreterSetting.getAllInterpreterGroups().size()); + + Thread.sleep(20 * 1000); + assertEquals(1, interpreterSetting.getAllInterpreterGroups().size(), + "a running paragraph must keep its interpreter group alive"); + } + + /** + * A probe is cheap for the local launcher but not for docker or k8s, and this scan walks every + * group on a timer. + */ + @Test + void scanNeverProbesTheInterpreterProcess() { + RemoteInterpreterProcess process = mock(RemoteInterpreterProcess.class); + + InterpreterSetting interpreterSetting = mock(InterpreterSetting.class); + when(interpreterSetting.getName()).thenReturn("probe-guard"); + when(interpreterSetting.getJavaProperties()).thenReturn(new Properties()); + + ManagedInterpreterGroup interpreterGroup = mock(ManagedInterpreterGroup.class); + when(interpreterGroup.getId()).thenReturn("probe-guard-shared_process"); + when(interpreterGroup.getInterpreterProcess()).thenReturn(process); + when(interpreterGroup.getInterpreterSetting()).thenReturn(interpreterSetting); + when(interpreterGroup.isEmpty()).thenReturn(false); + // Idle since the epoch, so it is well past any threshold and does get closed. + when(interpreterGroup.getLastUsedTimeInMillis()).thenReturn(0L); + + InterpreterSettingManager settingManager = mock(InterpreterSettingManager.class); + when(settingManager.getAllInterpreterGroup()) + .thenReturn(Collections.singletonList(interpreterGroup)); + + new IdleInterpreterReclaimer(zConf, settingManager).reclaimIdleInterpreterGroups(); + + verify(interpreterGroup).close(); + verify(process, never()).isAlive(); + verify(process, never()).isRunning(); + } + + /** + * The handle is published before the process is ready and the group has been idle since it was + * created, so without the launching check the scan closes a process that is starting up. + */ + @Test + void aGroupBeingLaunchedIsNotReclaimed() { + ManagedInterpreterGroup interpreterGroup = mock(ManagedInterpreterGroup.class); + when(interpreterGroup.getId()).thenReturn("launching-shared_process"); + when(interpreterGroup.isLaunchingInterpreterProcess()).thenReturn(true); + when(interpreterGroup.getInterpreterProcess()) + .thenReturn(mock(RemoteInterpreterProcess.class)); + when(interpreterGroup.isEmpty()).thenReturn(false); + when(interpreterGroup.getLastUsedTimeInMillis()).thenReturn(0L); + + InterpreterSettingManager settingManager = mock(InterpreterSettingManager.class); + when(settingManager.getAllInterpreterGroup()) + .thenReturn(Collections.singletonList(interpreterGroup)); + + new IdleInterpreterReclaimer(zConf, settingManager).reclaimIdleInterpreterGroups(); + + verify(interpreterGroup, never()).close(); + } + + @Test + void thresholdResolutionPrefersTheSettingAndFallsBackOnGarbage() { + zConf.setProperty( + ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "1h"); + + assertEquals(3600000L, IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, null), + "no setting at all means the global threshold"); + + InterpreterSetting interpreterSetting = mock(InterpreterSetting.class); + when(interpreterSetting.getName()).thenReturn("threshold-resolution"); + + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties(null)); + assertEquals(3600000L, + IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting), + "no override means the global threshold"); + + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("10s")); + assertEquals(10000L, + IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting)); + + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("600000")); + assertEquals(600000L, + IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting), + "a plain number is milliseconds"); + + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("0")); + assertEquals(0L, + IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting), + "zero opts the setting out of reclaim"); + + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("not-a-duration")); + assertEquals(3600000L, + IdleInterpreterReclaimer.getIdleTimeoutThreshold(zConf, interpreterSetting), + "an unparsable override must fall back to the global threshold"); + } + + /** + * A setting that opted out must not be shut down by the in-process fallback either. Its own + * {@code 0} would mean "shut down at the next check" there, so it never reaches the process. + */ + @Test + void optingOutDisablesTheInProcessFallbackToo() { + InterpreterSetting interpreterSetting = mock(InterpreterSetting.class); + when(interpreterSetting.getName()).thenReturn("opt-out"); + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("0")); + + Map overrides = + IdleInterpreterReclaimer.processConfigurationOverrides(zConf, interpreterSetting); + assertEquals(String.valueOf(Long.MAX_VALUE), overrides.get(THRESHOLD_PROPERTY)); + assertNull(overrides.get(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName()), + "the lifecycle manager the operator configured must be left alone"); + + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("10s")); + overrides = IdleInterpreterReclaimer.processConfigurationOverrides(zConf, interpreterSetting); + assertEquals("10000", overrides.get(THRESHOLD_PROPERTY), + "the process gets the resolved threshold, not the global one"); + } + + /** + * With the default lifecycle manager nothing is reclaimed and nothing is overridden, so an + * existing deployment is untouched. + */ + @Test + void defaultLifecycleManagerLeavesEverythingAlone() { + zConf.setProperty(ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(), + NullLifecycleManager.class.getName()); + + InterpreterSetting interpreterSetting = mock(InterpreterSetting.class); + when(interpreterSetting.getJavaProperties()).thenReturn(thresholdProperties("10s")); + + assertTrue(IdleInterpreterReclaimer.processConfigurationOverrides(zConf, interpreterSetting) + .isEmpty()); + } + + private Properties thresholdProperties(String threshold) { + Properties properties = new Properties(); + if (threshold != null) { + properties.setProperty(THRESHOLD_PROPERTY, threshold); + } + return properties; + } + + private void startEchoInterpreter() throws Exception { + RemoteInterpreter echoInterpreter = + (RemoteInterpreter) interpreterFactory.getInterpreter("test.echo", + new ExecutionContext("user1", "note1", "test")); + echoInterpreter.interpret("hello", createDummyInterpreterContext()); + assertTrue(echoInterpreter.isOpened()); + } + + private void waitForInterpreterGroups(InterpreterSetting interpreterSetting, + int expectedSize, + int maxSeconds) throws Exception { + long deadline = System.currentTimeMillis() + maxSeconds * 1000L; + while (interpreterSetting.getAllInterpreterGroups().size() != expectedSize + && System.currentTimeMillis() < deadline) { + Thread.sleep(1000); + } + } +}