basic event bus

Author: Zsailer

jupyter_server/base/handlers.py

@@ -340,6 +340,10 @@ def kernel_spec_manager(self):
     def config_manager(self):
         return self.settings["config_manager"]
 
+    @property
+    def event_bus(self):
+        return self.settings["event_bus"]
+
     # ---------------------------------------------------------------
     # CORS
     # ---------------------------------------------------------------

jupyter_server/serverapp.py

@@ -120,6 +120,7 @@
     AsyncContentsManager,
     ContentsManager,
 )
+from jupyter_server.services.events.bus import EventBus
 from jupyter_server.services.kernels.kernelmanager import (
     AsyncMappingKernelManager,
     MappingKernelManager,
@@ -164,6 +165,7 @@
     sessions=["jupyter_server.services.sessions.handlers"],
     shutdown=["jupyter_server.services.shutdown"],
     view=["jupyter_server.view.handlers"],
+    events=["jupyter_server.services.events.handlers"],
 )
 
 # Added for backwards compatibility from classic notebook server.
@@ -207,6 +209,7 @@ def __init__(
         session_manager,
         kernel_spec_manager,
         config_manager,
+        event_bus,
         extra_services,
         log,
         base_url,
@@ -242,6 +245,7 @@ def __init__(
             session_manager,
             kernel_spec_manager,
             config_manager,
+            event_bus,
             extra_services,
             log,
             base_url,
@@ -263,6 +267,7 @@ def init_settings(
         session_manager,
         kernel_spec_manager,
         config_manager,
+        event_bus,
         extra_services,
         log,
         base_url,
@@ -354,6 +359,7 @@ def init_settings(
             config_manager=config_manager,
             authorizer=authorizer,
             identity_provider=identity_provider,
+            event_bus=event_bus,
             # handlers
             extra_services=extra_services,
             # Jupyter stuff
@@ -769,6 +775,7 @@ class ServerApp(JupyterApp):
         GatewaySessionManager,
         GatewayClient,
         Authorizer,
+        EventBus,
     ]
 
     subcommands = dict(
@@ -794,6 +801,7 @@ class ServerApp(JupyterApp):
         "sessions",
         "shutdown",
         "view",
+        "events",
     )
 
     _log_formatter_cls = LogFormatter
@@ -1561,6 +1569,11 @@ def _default_kernel_spec_manager_class(self):
         ),
     )
 
+    event_bus = Instance(
+        EventBus,
+        help="An EventBus for emitting structured event data from Jupyter Server and extensions.",
+    )
+
     info_file = Unicode()
 
     @default("info_file")
@@ -1906,6 +1919,10 @@ def init_logging(self):
         logger.parent = self.log
         logger.setLevel(self.log.level)
 
+    def init_eventbus(self):
+        """Initialize the Event Bus."""
+        self.event_bus = EventBus.instance(parent=self)
+
     def init_webapp(self):
         """initialize tornado webapp"""
         self.tornado_settings["allow_origin"] = self.allow_origin
@@ -1970,6 +1987,7 @@ def init_webapp(self):
             self.session_manager,
             self.kernel_spec_manager,
             self.config_manager,
+            self.event_bus,
             self.extra_services,
             self.log,
             self.base_url,
@@ -2436,6 +2454,7 @@ def initialize(
         if find_extensions:
             self.find_server_extensions()
         self.init_logging()
+        self.init_eventbus()
         self.init_server_extensions()
 
         # Special case the starter extension and load

jupyter_server/services/events/__init__.py

@@ -0,0 +1,12 @@
+from jupyter_telemetry.eventlog import EventLog
+from traitlets.config import SingletonConfigurable
+
+
+class EventBus(EventLog, SingletonConfigurable):
+    """A Jupyter EventLog as a Singleton, making it easy to
+    access from anywhere and log events.
+    """
+
+    def record_event(self, *args, **kwargs):
+
+        super().record_event(*args, **kwargs)

jupyter_server/services/events/bus.py

@@ -0,0 +1,46 @@
+import logging
+
+import tornado.web
+import tornado.websocket
+from jupyter_telemetry.eventlog import _skip_message
+from pythonjsonlogger import jsonlogger
+
+from jupyter_server.base.handlers import JupyterHandler
+
+
+class TornadoWebSocketLoggingHandler(logging.Handler):
+    """Logging handler that routes records to a Tornado websocket."""
+
+    def __init__(self, websocket):
+        super().__init__()
+        self.websocket = websocket
+
+    def emit(self, record):
+        """Emit the message across the websocket"""
+        self.websocket.write_message(record.msg)
+
+
+class SubscribeWebsocket(
+    JupyterHandler,
+    tornado.websocket.WebSocketHandler,
+):
+    @property
+    def event_bus(self):
+        return self.settings["event_bus"]
+
+    def open(self):
+        self.logging_handler = TornadoWebSocketLoggingHandler(self)
+        # Add a JSON formatter to the handler.
+        formatter = jsonlogger.JsonFormatter(json_serializer=_skip_message)
+        self.logging_handler.setFormatter(formatter)
+        # To do: add an eventlog.add_handler method to jupyter_telemetry.
+        self.event_bus.log.addHandler(self.logging_handler)
+        self.event_bus.handlers.append(self.logging_handler)
+
+    def on_close(self):
+        self.event_bus.log.removeHandler(self.logging_handler)
+
+
+default_handlers = [
+    (r"/api/events/subscribe", SubscribeWebsocket),
+]

jupyter_server/services/events/handlers.py

@@ -0,0 +1,99 @@
+[metadata]
+name = jupyter_server
+version = attr: jupyter_server.__version__
+description = The backend—i.e. core services, APIs, and REST endpoints—to Jupyter web applications.
+long_description = file: README.md
+long_description_content_type = text/markdown
+license_files = COPYING.md
+author = Jupyter Development Team
+author_email = [email protected]
+url = https://jupyter-server.readthedocs.io
+platforms = Linux, Mac OS X, Windows
+keywords = ipython, jupyter
+classifiers =
+    Development Status :: 5 - Production/Stable
+    Framework :: Jupyter
+    Intended Audience :: Developers
+    Intended Audience :: Science/Research
+    Intended Audience :: System Administrators
+    License :: OSI Approved :: BSD License
+    Programming Language :: Python
+    Programming Language :: Python :: 3
+    Programming Language :: Python :: 3 :: Only
+    Programming Language :: Python :: 3.7
+    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3.10
+project_urls =
+    Documentation = https://jupyter-server.readthedocs.io
+    Funding = https://numfocus.org/donate
+    Source = https://github.com/jupyter-server/jupyter_server
+    Tracker = https://github.com/jupyter-server/jupyter_server/issues
+
+[options]
+zip_safe = False
+include_package_data = True
+packages = find:
+package_dir =
+    "" = "jupyter_server"
+python_requires = >=3.7
+install_requires =
+    anyio>=3.1.0,<4
+    argon2-cffi
+    jinja2
+    jupyter_client>=6.1.12
+    jupyter_core>=4.7.0
+    jupyter_server_terminals
+    nbconvert>=6.4.4
+    nbformat>=5.2.0
+    packaging
+    prometheus_client
+    pywinpty;os_name=='nt'
+    pyzmq>=17
+    Send2Trash
+    terminado>=0.8.3
+    tornado>=6.1.0
+    traitlets>=5.1
+    websocket-client
+    jupyter_telemetry
+
+[options.extras_require]
+test =
+    coverage
+    ipykernel
+    pre-commit
+    pytest-console-scripts
+    pytest-cov
+    pytest-timeout
+    pytest-tornasync
+    pytest>=6.0
+    requests
+
+[options.entry_points]
+console_scripts =
+    jupyter-server = jupyter_server.serverapp:main
+
+[options.packages.find]
+exclude =
+    docs.*
+    examples.*
+    tests
+    tests.*
+
+[flake8]
+ignore = E501, W503, E402
+builtins = c, get_config
+exclude =
+    .cache,
+    .github,
+    docs,
+    setup.py
+enable-extensions = G
+extend-ignore =
+    G001, G002, G004, G200, G201, G202,
+    # black adds spaces around ':'
+    E203,
+per-file-ignores =
+    # B011: Do not call assert False since python -O removes these calls
+    # F841 local variable 'foo' is assigned to but never used
+    tests/*: B011, F841

setup.cfg

@@ -0,0 +1,13 @@
+$id: event.mock.jupyter.com/message
+version: 1
+title: Message
+description: |
+  Emit a message
+type: object
+properties:
+  event_message:
+    title: Event Messages
+    description: |
+      Mock event message to read.
+required:
+  - event_message

tests/services/events/__init__.py

@@ -0,0 +1,6 @@
+# Function that makes these extensions discoverable
+# by the test functions.
+def _jupyter_server_extension_points():
+    return [
+        {"module": "tests.events.mock_extension"},
+    ]

tests/services/events/mock_event.yaml

@@ -0,0 +1,23 @@
+import pathlib
+
+from jupyter_server.base.handlers import JupyterHandler
+from jupyter_server.utils import url_path_join
+
+
+class MockEventHandler(JupyterHandler):
+    def get(self):
+        # Emit an event.
+        self.event_bus.record_event(
+            schema_name="event.mockextension.jupyter.com/message",
+            version=1,
+            event={"message": "Hello world, from mock extension!"},
+        )
+
+
+def _load_jupyter_server_extension(serverapp):
+    # Register a schema with the EventBus
+    schema_file = pathlib.Path(__file__).parent / "mock_event_schema.yaml"
+    serverapp.event_bus.register_schema_file(schema_file)
+    serverapp.web_app.add_handlers(
+        ".*$", [(url_path_join(serverapp.base_url, "/mock/event"), MockEventHandler)]
+    )

tests/services/events/mockextension/__init__.py

@@ -0,0 +1,13 @@
+$id: event.mockextension.jupyter.com/message
+version: 1
+title: Message
+description: |
+  Emit a message
+type: object
+properties:
+  event_message:
+    title: Event Message
+    description: |
+      Mock event message to read.
+required:
+  - event_message

tests/services/events/mockextension/mock_extension.py

@@ -0,0 +1,32 @@
+import json
+import pathlib
+
+import pytest
+
+
+@pytest.fixture
+def event_bus(jp_serverapp):
+    event_bus = jp_serverapp.event_bus
+    # Register the event schema defined in this directory.
+    schema_file = pathlib.Path(__file__).parent / "mock_event.yaml"
+    event_bus.register_schema_file(schema_file)
+    #
+    event_bus.allowed_schemas = ["event.mock.jupyter.com/message"]
+    return event_bus
+
+
+async def test_subscribe_websocket(jp_ws_fetch, event_bus):
+    # Open a websocket connection.
+    ws = await jp_ws_fetch("/api/events/subscribe")
+
+    event_bus.record_event(
+        schema_name="event.mock.jupyter.com/message",
+        version=1,
+        event={"event_message": "Hello, world!"},
+    )
+    message = await ws.read_message()
+    event_data = json.loads(message)
+    # Close websocket
+    ws.close()
+
+    assert event_data.get("event_message") == "Hello, world!"

tests/services/events/mockextension/mock_extension_event.yaml

@@ -0,0 +1,24 @@
+import json
+import pathlib
+
+import pytest
+
+
+@pytest.fixture
+def event_bus(jp_serverapp):
+    event_bus = jp_serverapp.event_bus
+    event_bus.allowed_schemas = ["event.mockextension.jupyter.com/message"]
+    return event_bus
+
+
+async def test_subscribe_websocket(jp_ws_fetch, jp_fetch, event_bus):
+    # Open a websocket connection.
+    ws = await jp_ws_fetch("/api/events/subscribe")
+
+    await jp_fetch("/mock/event")
+    message = await ws.read_message()
+    event_data = json.loads(message)
+    # Close websocket
+    ws.close()
+
+    assert event_data.get("event_message") == "Hello world, from mock extension!"