Add ability to cull terminals and track last activity

This adds functionality to track a terminal's last activity and
optionally cull terminals that have been inactive for some specified
duration.

Author: kevin-bates

jupyter_server/serverapp.py

@@ -83,6 +83,7 @@
 from .auth.login import LoginHandler
 from .auth.logout import LogoutHandler
 from .base.handlers import FileFindHandler
+from .terminal import TerminalManager
 
 from traitlets.config import Config
 from traitlets.config.application import catch_config_error, boolean_flag
@@ -587,7 +588,7 @@ class ServerApp(JupyterApp):
     classes = [
             KernelManager, Session, MappingKernelManager, KernelSpecManager, AsyncMappingKernelManager,
             ContentsManager, FileContentsManager, AsyncContentsManager, AsyncFileContentsManager, NotebookNotary,
-            GatewayKernelManager, GatewayKernelSpecManager, GatewaySessionManager, GatewayClient
+            TerminalManager, GatewayKernelSpecManager, GatewaySessionManager, GatewayClient
         ]
 
     subcommands = dict(
@@ -1546,7 +1547,7 @@ def init_terminals(self):
 
         try:
             from .terminal import initialize
-            initialize(self.web_app, self.root_dir, self.connection_url, self.terminado_settings)
+            initialize(self.web_app, self.root_dir, self.connection_url, self.terminado_settings, self)
             self.web_app.settings['terminals_available'] = True
         except ImportError as e:
             self.log.warning(_i18n("Terminals not available (error was %s)"), e)

jupyter_server/services/api/api.yaml

@@ -568,7 +568,7 @@ paths:
           schema:
             type: array
             items:
-              $ref: '#/definitions/Terminal_ID'
+              $ref: '#/definitions/Terminal'
         403:
           description: Forbidden to access
         404:
@@ -582,7 +582,7 @@ paths:
         200:
           description: Succesfully created a new terminal
           schema:
-            $ref: '#/definitions/Terminal_ID'
+            $ref: '#/definitions/Terminal'
         403:
           description: Forbidden to access
         404:
@@ -599,7 +599,7 @@ paths:
         200:
           description: Terminal session with given id
           schema:
-            $ref: '#/definitions/Terminal_ID'
+            $ref: '#/definitions/Terminal'
         403:
           description: Forbidden to access
         404:
@@ -845,12 +845,18 @@ definitions:
         type: string
         description: Last modified timestamp
         format: dateTime
-  Terminal_ID:
-    description: A Terminal_ID object
+  Terminal:
+    description: A Terminal object
     type: object
     required:
       - name
     properties:
       name:
         type: string
-        description: name of terminal ID
+        description: name of terminal
+      last_activity:
+        type: string
+        description: |
+          ISO 8601 timestamp for the last-seen activity on this terminal.  Use
+          this to identify which terminals have been inactive since a given time.
+          Timestamps will be UTC, indicated 'Z' suffix.

jupyter_server/terminal/__init__.py

@@ -8,14 +8,14 @@
     raise ImportError("terminado >= 0.8.3 required, found %s" % terminado.__version__)
 
 from ipython_genutils.py3compat import which
-from terminado import NamedTermManager
 from tornado.log import app_log
 from jupyter_server.utils import url_path_join as ujoin
 from . import api_handlers
-from .handlers import TermSocket
+from .handlers import TerminalHandler, TermSocket
+from .terminalmanager import TerminalManager
 
 
-def initialize(webapp, root_dir, connection_url, settings):
+def initialize(webapp, root_dir, connection_url, settings, parent):
     if os.name == 'nt':
         default_shell = 'powershell.exe'
     else:
@@ -33,11 +33,12 @@ def initialize(webapp, root_dir, connection_url, settings):
     # the user has specifically set a preferred shell command.
     if os.name != 'nt' and shell_override is None and not sys.stdout.isatty():
         shell.append('-l')
-    terminal_manager = webapp.settings['terminal_manager'] = NamedTermManager(
+    terminal_manager = webapp.settings['terminal_manager'] = TerminalManager(
         shell_command=shell,
         extra_env={'JUPYTER_SERVER_ROOT': root_dir,
                    'JUPYTER_SERVER_URL': connection_url,
                    },
+        parent=parent,
     )
     terminal_manager.log = app_log
     base_url = webapp.settings['base_url']

jupyter_server/terminal/api_handlers.py

@@ -1,57 +1,35 @@
 import json
 from tornado import web
 from ..base.handlers import APIHandler
-from ..prometheus.metrics import TERMINAL_CURRENTLY_RUNNING_TOTAL
 
 
 
 class TerminalRootHandler(APIHandler):
 
     @web.authenticated
     def get(self):
-        tm = self.terminal_manager
-        terms = [{'name': name} for name in tm.terminals]
-        self.finish(json.dumps(terms))
-
-        # Update the metric below to the length of the list 'terms'
-        TERMINAL_CURRENTLY_RUNNING_TOTAL.set(
-            len(terms)
-        )
+        models = self.terminal_manager.list()
+        self.finish(json.dumps(models))
 
     @web.authenticated
     def post(self):
         """POST /terminals creates a new terminal and redirects to it"""
         data = self.get_json_body() or {}
 
-        name, _ = self.terminal_manager.new_named_terminal(**data)
-        self.finish(json.dumps({'name': name}))
-
-        # Increase the metric by one because a new terminal was created
-        TERMINAL_CURRENTLY_RUNNING_TOTAL.inc()
+        model = self.terminal_manager.create(**data)
+        self.finish(json.dumps(model))
 
 
 class TerminalHandler(APIHandler):
     SUPPORTED_METHODS = ('GET', 'DELETE')
 
     @web.authenticated
     def get(self, name):
-        tm = self.terminal_manager
-        if name in tm.terminals:
-            self.finish(json.dumps({'name': name}))
-        else:
-            raise web.HTTPError(404, "Terminal not found: %r" % name)
+        model = self.terminal_manager.get(name)
+        self.finish(json.dumps(model))
 
     @web.authenticated
     async def delete(self, name):
-        tm = self.terminal_manager
-        if name in tm.terminals:
-            await tm.terminate(name, force=True)
-            self.set_status(204)
-            self.finish()
-
-            # Decrease the metric below by one
-            # because a terminal has been shutdown
-            TERMINAL_CURRENTLY_RUNNING_TOTAL.dec()
-
-        else:
-            raise web.HTTPError(404, "Terminal not found: %r" % name)
+        await self.terminal_manager.terminate(name, force=True)
+        self.set_status(204)
+        self.finish()

jupyter_server/terminal/handlers.py

@@ -11,6 +11,14 @@
 from ..base.zmqhandlers import WebSocketMixin
 
 
+class TerminalHandler(JupyterHandler):
+    """Render the terminal interface."""
+    @web.authenticated
+    def get(self, term_name):
+        self.write(self.render_template('terminal.html',
+                   ws_path="terminals/websocket/%s" % term_name))
+
+
 class TermSocket(WebSocketMixin, JupyterHandler, terminado.TermSocket):
 
     def origin_check(self):
@@ -26,8 +34,14 @@ def get(self, *args, **kwargs):
 
     def on_message(self, message):
         super(TermSocket, self).on_message(message)
-        self.application.settings['terminal_last_activity'] = utcnow()
+        self._update_activity()
 
     def write_message(self, message, binary=False):
         super(TermSocket, self).write_message(message, binary=binary)
-        self.application.settings['terminal_last_activity'] = utcnow()
+        self._update_activity()
+
+    def _update_activity(self):
+        self.application.settings['terminal_last_activity'] = utcnow()
+        # terminal may not be around on deletion/cull
+        if self.term_name in self.terminal_manager.terminals:
+            self.terminal_manager.terminals[self.term_name].last_activity = utcnow()

jupyter_server/terminal/terminalmanager.py

@@ -0,0 +1,140 @@
+"""A MultiTerminalManager for use in the notebook webserver
+- raises HTTPErrors
+- creates REST API models
+"""
+
+# Copyright (c) Jupyter Development Team.
+# Distributed under the terms of the Modified BSD License.
+
+
+from datetime import timedelta
+from notebook._tz import utcnow, isoformat
+from terminado import NamedTermManager
+from tornado import web
+from tornado.ioloop import IOLoop, PeriodicCallback
+from traitlets import Integer
+from traitlets.config import Configurable
+from ..prometheus.metrics import TERMINAL_CURRENTLY_RUNNING_TOTAL
+
+
+class TerminalManager(Configurable, NamedTermManager):
+    """  """
+
+    _culler_callback = None
+
+    _initialized_culler = False
+
+    cull_inactive_timeout = Integer(0, config=True,
+                                    help="""Timeout (in seconds) in which a terminal has been inactive and ready to be culled.
+        Values of 0 or lower disable culling."""
+                                    )
+
+    cull_interval_default = 300  # 5 minutes
+    cull_interval = Integer(cull_interval_default, config=True,
+        help="""The interval (in seconds) on which to check for terminals exceeding the inactive timeout value."""
+                            )
+
+    # -------------------------------------------------------------------------
+    # Methods for managing terminals
+    # -------------------------------------------------------------------------
+    def __init__(self, *args, **kwargs):
+        super(TerminalManager, self).__init__(*args, **kwargs)
+
+    def create(self):
+        name, term = self.new_named_terminal()
+        # Monkey-patch last-activity, similar to kernels.  Should we need
+        # more functionality per terminal, we can look into possible sub-
+        # classing or containment then.
+        term.last_activity = utcnow()
+        model = self.terminal_model(name)
+        # Increase the metric by one because a new terminal was created
+        TERMINAL_CURRENTLY_RUNNING_TOTAL.inc()
+        # Ensure culler is initialized
+        self.initialize_culler()
+        return model
+
+    def get(self, name):
+        model = self.terminal_model(name)
+        return model
+
+    def list(self):
+        models = [self.terminal_model(name) for name in self.terminals]
+
+        # Update the metric below to the length of the list 'terms'
+        TERMINAL_CURRENTLY_RUNNING_TOTAL.set(
+            len(models)
+        )
+        return models
+
+    async def terminate(self, name, force=False):
+        self._check_terminal(name)
+        await super(TerminalManager, self).terminate(name, force=force)
+
+        # Decrease the metric below by one
+        # because a terminal has been shutdown
+        TERMINAL_CURRENTLY_RUNNING_TOTAL.dec()
+
+    def terminal_model(self, name):
+        """Return a JSON-safe dict representing a terminal.
+        For use in representing terminals in the JSON APIs.
+        """
+        self._check_terminal(name)
+        term = self.terminals[name]
+        model = {
+            "name": name,
+            "last_activity": isoformat(term.last_activity),
+        }
+        return model
+
+    def _check_terminal(self, name):
+        """Check a that terminal 'name' exists and raise 404 if not."""
+        if name not in self.terminals:
+            raise web.HTTPError(404, u'Terminal not found: %s' % name)
+
+    def initialize_culler(self):
+        """Start culler if 'cull_inactive_timeout' is greater than zero.
+        Regardless of that value, set flag that we've been here.
+        """
+        if not self._initialized_culler and self.cull_inactive_timeout > 0:
+            if self._culler_callback is None:
+                loop = IOLoop.current()
+                if self.cull_interval <= 0:  # handle case where user set invalid value
+                    self.log.warning("Invalid value for 'cull_interval' detected (%s) - using default value (%s).",
+                        self.cull_interval, self.cull_interval_default)
+                    self.cull_interval = self.cull_interval_default
+                self._culler_callback = PeriodicCallback(
+                    self.cull_terminals, 1000*self.cull_interval)
+                self.log.info("Culling terminals with inactivity > %s seconds at %s second intervals ...",
+                              self.cull_inactive_timeout, self.cull_interval)
+                self._culler_callback.start()
+
+        self._initialized_culler = True
+
+    async def cull_terminals(self):
+        self.log.debug("Polling every %s seconds for terminals inactive for > %s seconds...",
+                       self.cull_interval, self.cull_inactive_timeout)
+        # Create a separate list of terminals to avoid conflicting updates while iterating
+        for name in list(self.terminals):
+            try:
+                await self.cull_inactive_terminal(name)
+            except Exception as e:
+                self.log.exception("The following exception was encountered while checking the "
+                                   "activity of terminal {}: {}".format(name, e))
+
+    async def cull_inactive_terminal(self, name):
+        try:
+            term = self.terminals[name]
+        except KeyError:
+            return  # KeyErrors are somewhat expected since the terminal can be terminated as the culling check is made.
+
+        self.log.debug("name=%s, last_activity=%s", name, term.last_activity)
+        if hasattr(term, 'last_activity'):
+            dt_now = utcnow()
+            dt_inactive = dt_now - term.last_activity
+            # Compute idle properties
+            is_time = dt_inactive > timedelta(seconds=self.cull_inactive_timeout)
+            # Cull the kernel if all three criteria are met
+            if (is_time):
+                inactivity = int(dt_inactive.total_seconds())
+                self.log.warning("Culling terminal '%s' due to %s seconds of inactivity.", name, inactivity)
+                await self.terminate(name, force=True)