Merge pull request #106 from lsst-sqre/tickets/DM-49148/app-metrics

DM-49148: First app metric

Author: fajpunk

.github/workflows/docs.yaml

@@ -5,7 +5,7 @@ env:
   # reason to support multiple Python versions, so all actions are run with
   # this version. Quote the version to avoid interpretation as a floating
   # point number.
-  PYTHON_VERSION: "3.12"
+  PYTHON_VERSION: "3.13"
 
 "on":
   push:

.github/workflows/periodic-ci.yaml

@@ -10,7 +10,7 @@ env:
   # reason to support multiple Python versions, so all actions are run with
   # this version. Quote the version to avoid interpretation as a floating
   # point number.
-  PYTHON_VERSION: "3.12"
+  PYTHON_VERSION: "3.13"
 
 "on":
   schedule:

changelog.d/20250225_155032_danfuchs_HEAD.md

@@ -0,0 +1,4 @@
+<!-- Delete the sections that don't apply -->
+### New features
+
+- Added [Application Metrics](https://safir.lsst.io/user-guide/metrics/index.html) scaffolding, and a single pair of metrics for counting the number of notebook execution tasks that are enqueued.

src/noteburst/config.py

@@ -11,6 +11,7 @@
 from pydantic_settings import BaseSettings
 from safir.arq import ArqMode
 from safir.logging import LogLevel, Profile
+from safir.metrics import MetricsConfiguration, metrics_configuration_factory
 
 __all__ = [
     "Config",
@@ -78,6 +79,14 @@ class Config(BaseSettings):
         ),
     ] = "/noteburst"
 
+    metrics: Annotated[
+        MetricsConfiguration,
+        Field(
+            default_factory=metrics_configuration_factory,
+            title="Metrics configuration",
+        ),
+    ]
+
     environment_url: Annotated[
         HttpUrl,
         Field(

src/noteburst/events.py

@@ -0,0 +1,43 @@
+"""App metrics events."""
+
+from typing import override
+
+from safir.dependencies.metrics import EventDependency, EventMaker
+from safir.metrics import EventManager, EventPayload
+
+__all__ = [
+    "EnqueueNotebookExecutionFailure",
+    "EnqueueNotebookExecutionSuccess",
+    "Events",
+    "events_dependency",
+]
+
+
+class EnqueueNotebookExecutionSuccess(EventPayload):
+    """An nbexec task was successfully enqueued."""
+
+    username: str
+
+
+class EnqueueNotebookExecutionFailure(EventPayload):
+    """An nbexec task failed to be enqueued."""
+
+    username: str
+
+
+class Events(EventMaker):
+    """A container for app metrics event publishers."""
+
+    @override
+    async def initialize(self, manager: EventManager) -> None:
+        """Create event publishers."""
+        self.enqueue_nbexec_success = await manager.create_publisher(
+            "enqueue_nbexec_success", EnqueueNotebookExecutionSuccess
+        )
+        self.enqueue_nbexec_failure = await manager.create_publisher(
+            "enqueue_nbexec_failure", EnqueueNotebookExecutionFailure
+        )
+
+
+events_dependency = EventDependency(Events())
+"""Provides an container that holds event publishers."""

src/noteburst/handlers/v1/handlers.py

@@ -16,6 +16,12 @@
 
 from noteburst.exceptions import JobNotFoundError, NoteburstJobError
 
+from ...events import (
+    EnqueueNotebookExecutionFailure,
+    EnqueueNotebookExecutionSuccess,
+    Events,
+    events_dependency,
+)
 from .models import NotebookResponse, PostNotebookRequest
 
 v1_router = APIRouter(tags=["v1"], route_class=SlackRouteErrorHandler)
@@ -33,8 +39,10 @@ async def post_nbexec(
     *,
     request: Request,
     response: Response,
+    user: Annotated[str, Depends(auth_dependency)],
     logger: Annotated[structlog.BoundLogger, Depends(auth_logger_dependency)],
     arq_queue: Annotated[ArqQueue, Depends(arq_dependency)],
+    events: Annotated[Events, Depends(events_dependency)],
 ) -> NotebookResponse:
     """Submits a notebook for execution. The notebook is executed
     asynchronously via a pool of JupyterLab (Nublado) instances.
@@ -61,14 +69,24 @@ async def post_nbexec(
     and (if available) the notebook (`ipynb`) result. See
     `GET /v1/notebooks/{job_id}` for more information.
     """
-    logger.debug("Enqueing a nbexec task")
-    job_metadata = await arq_queue.enqueue(
-        "nbexec",
-        ipynb=request_data.get_ipynb_as_str(),
-        kernel_name=request_data.kernel_name,
-        enable_retry=request_data.enable_retry,
-        timeout=request_data.timeout,
-    )
+    try:
+        logger.debug("Enqueing a nbexec task")
+        job_metadata = await arq_queue.enqueue(
+            "nbexec",
+            ipynb=request_data.get_ipynb_as_str(),
+            kernel_name=request_data.kernel_name,
+            enable_retry=request_data.enable_retry,
+            timeout=request_data.timeout,
+        )
+        await events.enqueue_nbexec_success.publish(
+            EnqueueNotebookExecutionSuccess(username=user)
+        )
+    except:
+        await events.enqueue_nbexec_failure.publish(
+            EnqueueNotebookExecutionFailure(username=user)
+        )
+        raise
+
     logger.info("Finished enqueing an nbexec task", job_id=job_metadata.id)
     response_data = await NotebookResponse.from_job_metadata(
         job=job_metadata, request=request

src/noteburst/main.py

@@ -24,6 +24,7 @@
 from safir.slack.webhook import SlackRouteErrorHandler
 
 from .config import config
+from .events import events_dependency
 from .handlers.external import external_router
 from .handlers.internal import internal_router
 from .handlers.v1 import v1_router
@@ -48,6 +49,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         mode=config.arq_mode, redis_settings=config.arq_redis_settings
     )
 
+    event_manager = config.metrics.make_manager()
+    await event_manager.initialize()
+    await events_dependency.initialize(event_manager)
+
     yield
 
     # Shut down event

tests/conftest.py

@@ -30,7 +30,7 @@
 
 
 @pytest_asyncio.fixture
-async def app() -> AsyncIterator[FastAPI]:
+async def app(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[FastAPI]:
     """Return a configured test application.
 
     Wraps the application in a lifespan manager so that startup and shutdown

tests/handlers/v1_test.py

@@ -4,11 +4,15 @@
 
 import json
 from pathlib import Path
+from typing import cast
 
 import pytest
 from httpx import AsyncClient
 from safir.arq import MockArqQueue
 from safir.dependencies.arq import arq_dependency
+from safir.metrics import MockEventPublisher
+
+from noteburst.events import events_dependency
 
 
 @pytest.fixture
@@ -28,6 +32,8 @@ async def test_post_nbexec(
     client: AsyncClient, sample_ipynb: str, sample_ipynb_executed: str
 ) -> None:
     """Test ``POST /v1/``, sending a notebook to execute."""
+    events = await events_dependency()
+
     arq_queue = await arq_dependency()
     assert isinstance(arq_queue, MockArqQueue)
 
@@ -45,6 +51,9 @@ async def test_post_nbexec(
     assert job_url == response.headers["Location"]
     job_id = data["job_id"]
 
+    pub = cast(MockEventPublisher, events.enqueue_nbexec_success).published
+    pub.assert_published_all([{"username": "user"}])
+
     response = await client.get(job_url)
     assert response.status_code == 200
     data2 = response.json()

tox.ini

@@ -12,6 +12,9 @@ setenv =
     NOTEBURST_REDIS_URL = redis://localhost:6379
     NOTEBURST_ARQ_MODE = test
     NOTEBURST_WORKER_IDENTITIES_PATH = tests/identities.test.yaml
+    METRICS_APPLICATION = "noteburst"
+    METRICS_ENABLED = false
+    METRICS_MOCK = true
 deps =
     -r{toxinidir}/requirements/main.txt
     -r{toxinidir}/requirements/dev.txt