Merge pull request #1393 from minrk/launchmessages

Emit launch events

Author: consideRatio

binderhub/builder.py

@@ -582,17 +582,30 @@ async def launch(self, provider):
                 username = launcher.unique_name_from_repo(self.repo_url)
                 server_name = ''
             try:
+
+                async def handle_progress_event(event):
+                    message = event["message"]
+                    await self.emit(
+                        {
+                            "phase": "launching",
+                            "message": message + "\n",
+                        }
+                    )
+
                 extra_args = {
                     'binder_ref_url': self.ref_url,
                     'binder_launch_host': self.binder_launch_host,
                     'binder_request': self.binder_request,
                     'binder_persistent_request': self.binder_persistent_request,
                 }
-                server_info = await launcher.launch(image=self.image_name,
-                                                    username=username,
-                                                    server_name=server_name,
-                                                    repo_url=self.repo_url,
-                                                    extra_args=extra_args)
+                server_info = await launcher.launch(
+                    image=self.image_name,
+                    username=username,
+                    server_name=server_name,
+                    repo_url=self.repo_url,
+                    extra_args=extra_args,
+                    event_callback=handle_progress_event,
+                )
             except Exception as e:
                 duration = time.perf_counter() - launch_starttime
                 if i + 1 == launcher.retries:

binderhub/launcher.py

@@ -1,6 +1,7 @@
 """
 Launch an image with a temporary user via JupyterHub
 """
+import asyncio
 import base64
 import json
 import random
@@ -9,6 +10,7 @@
 from urllib.parse import urlparse, quote
 import uuid
 import os
+from datetime import timedelta
 
 from tornado.log import app_log
 from tornado import web, gen
@@ -18,6 +20,8 @@
 from jupyterhub.traitlets import Callable
 from jupyterhub.utils import maybe_future
 
+from .utils import url_path_join
+
 # pattern for checking if it's an ssh repo and not a URL
 # used only after verifying that `://` is not present
 _ssh_repo_pat = re.compile(r'.*@.*\:')
@@ -80,6 +84,13 @@ def _default_hub_url_local(self):
         Receives 5 parameters: launcher, image, username, server_name, repo_url
         """
     )
+    launch_timeout = Integer(
+        600,
+        config=True,
+        help="""
+        Wait this many seconds until server is ready, raise TimeoutError otherwise.
+        """,
+    )
 
     async def api_request(self, url, *args, **kwargs):
         """Make an API request to JupyterHub"""
@@ -148,7 +159,15 @@ def unique_name_from_repo(self, repo_url):
         # add a random suffix to avoid collisions for users on the same image
         return '{}-{}'.format(prefix, ''.join(random.choices(SUFFIX_CHARS, k=SUFFIX_LENGTH)))
 
-    async def launch(self, image, username, server_name='', repo_url='', extra_args=None):
+    async def launch(
+        self,
+        image,
+        username,
+        server_name="",
+        repo_url="",
+        extra_args=None,
+        event_callback=None,
+    ):
         """Launch a server for a given image
 
         - creates a temporary user on the Hub if authentication is not enabled
@@ -215,38 +234,100 @@ async def launch(self, image, username, server_name='', repo_url='', extra_args=
 
         # start server
         app_log.info(f"Starting server{_server_name} for user {username} with image {image}")
+        ready_event_future = asyncio.Future()
+
+        def _cancel_ready_event(f=None):
+            if not ready_event_future.done():
+                if f and f.exception():
+                    ready_event_future.set_exception(f.exception())
+                else:
+                    ready_event_future.cancel()
         try:
             resp = await self.api_request(
                 'users/{}/servers/{}'.format(escaped_username, server_name),
                 method='POST',
                 body=json.dumps(data).encode('utf8'),
             )
-            if resp.code == 202:
-                # Server hasn't actually started yet
-                # We wait for it!
-                # NOTE: This ends up being about ten minutes
-                for i in range(64):
-                    user_data = await self.get_user_data(escaped_username)
-                    if user_data['servers'][server_name]['ready']:
-                        break
-                    if not user_data['servers'][server_name]['pending']:
-                        raise web.HTTPError(500, "Image %s for user %s failed to launch" % (image, username))
-                    # FIXME: make this configurable
-                    # FIXME: Measure how long it takes for servers to start
-                    # and tune this appropriately
-                    await gen.sleep(min(1.4 ** i, 10))
-                else:
-                    raise web.HTTPError(500, f"Image {image} for user {username} took too long to launch")
+            # listen for pending spawn (launch) events until server is ready
+            # do this even if previous request finished!
+            buffer_list = []
+
+            async def handle_chunk(chunk):
+                lines = b"".join(buffer_list + [chunk]).split(b"\n\n")
+                # the last item in the list is usually an empty line ('')
+                # but it can be the partial line after the last `\n\n`,
+                # so put it back on the buffer to handle with the next chunk
+                buffer_list[:] = [lines[-1]]
+                for line in lines[:-1]:
+                    if line:
+                        line = line.decode("utf8", "replace")
+                    if line and line.startswith("data:"):
+                        event = json.loads(line.split(":", 1)[1])
+                        if event_callback:
+                            await event_callback(event)
+
+                        # stream ends when server is ready or fails
+                        if event.get("ready", False):
+                            if not ready_event_future.done():
+                                ready_event_future.set_result(event)
+                        elif event.get("failed", False):
+                            if not ready_event_future.done():
+                                ready_event_future.set_exception(
+                                    web.HTTPError(
+                                        500, event.get("message", "unknown error")
+                                    )
+                                )
+
+            url_parts = ["users", username]
+            if server_name:
+                url_parts.extend(["servers", server_name, "progress"])
+            else:
+                url_parts.extend(["server/progress"])
+            progress_api_url = url_path_join(*url_parts)
+            self.log.debug("Requesting progress for {username}: {progress_api_url}")
+            resp_future = self.api_request(
+                progress_api_url,
+                streaming_callback=lambda chunk: asyncio.ensure_future(
+                    handle_chunk(chunk)
+                ),
+                request_timeout=self.launch_timeout,
+            )
+            try:
+                await gen.with_timeout(
+                    timedelta(seconds=self.launch_timeout), resp_future
+                )
+            except (gen.TimeoutError, TimeoutError):
+                _cancel_ready_event()
+                raise web.HTTPError(
+                    500,
+                    f"Image {image} for user {username} took too long to launch",
+                )
 
         except HTTPError as e:
+            _cancel_ready_event()
             if e.response:
                 body = e.response.body
             else:
                 body = ''
 
-            app_log.error("Error starting server{} for user {}: {}\n{}".
-                          format(_server_name, username, e, body))
+            app_log.error(
+                f"Error starting server{_server_name} for user {username}: {e}\n{body}"
+            )
             raise web.HTTPError(500, f"Failed to launch image {image}")
+        except Exception:
+            _cancel_ready_event()
+            raise
+
+        # verify that the server is running!
+        try:
+            # this should already be done, but it's async so wait a finite time
+            ready_event = await gen.with_timeout(
+                timedelta(seconds=5), ready_event_future
+            )
+        except (gen.TimeoutError, TimeoutError):
+            raise web.HTTPError(
+                500, f"Image {image} for user {username} failed to launch"
+            )
 
-        data['url'] = self.hub_url + 'user/%s/%s' % (escaped_username, server_name)
+        data["url"] = url_path_join(self.hub_url, ready_event["url"])
         return data

binderhub/tests/test_build.py

@@ -48,21 +48,29 @@ async def test_build(app, needs_build, needs_launch, always_build, slug, pytestc
     r = await async_requests.get(build_url, stream=True)
     r.raise_for_status()
     events = []
+    launch_events = 0
     async for line in async_requests.iter_lines(r):
         line = line.decode('utf8', 'replace')
         if line.startswith('data:'):
             event = json.loads(line.split(':', 1)[1])
             events.append(event)
             assert 'message' in event
-            sys.stdout.write(event['message'])
+            sys.stdout.write(f"{event.get('phase', '')}: {event['message']}")
             # this is the signal that everything is ready, pod is launched
             # and server is up inside the pod. Break out of the loop now
             # because BinderHub keeps the connection open for many seconds
             # after to avoid "reconnects" from slow clients
             if event.get('phase') == 'ready':
                 r.close()
                 break
-
+            if event.get("phase") == "launching" and not event["message"].startswith(
+                ("Launching server...", "Launch attempt ")
+            ):
+                # skip standard launching events of builder
+                # we are interested in launching events from spawner
+                launch_events += 1
+
+    assert launch_events > 0
     final = events[-1]
     assert 'phase' in final
     assert final['phase'] == 'ready'

helm-chart/binderhub/values.yaml

@@ -211,7 +211,7 @@ jupyterhub:
 
         # launch the notebook server
         os.execvp("jupyter-notebook", sys.argv)
-    events: false
+    events: true
     storage:
       type: none
     memory:

testing/local-binder-local-hub/jupyterhub_config.py

@@ -44,3 +44,9 @@ class LocalContainerSpawner(BinderSpawnerMixin, DockerSpawner):
     "environment": {"JUPYTERHUB_EXTERNAL_URL": os.getenv("JUPYTERHUB_EXTERNAL_URL", "")}
 }]
 c.JupyterHub.default_url = f"/services/{binderhub_service_name}/"
+
+c.JupyterHub.tornado_settings = {
+    "slow_spawn_timeout": 0,
+}
+
+c.KubeSpawner.events_enabled = True