feat: add user alerts (#1121)

Enables Renku administrators and alerting systems to send alerts to users, which are then displayed in their Renku sessions:

- Users can retrieve their active alerts for specific sessions
- Admins can manually create and resolve alerts via REST API
- Support for creating and resolving alerts from Prometheus Alertmanager webhooks, secured with OAuth2 client credentials flow

Author: wesjdj

Makefile

@@ -52,7 +52,8 @@ API_SPECS := \
     components/renku_data_services/notebooks/apispec.py \
     components/renku_data_services/platform/apispec.py \
     components/renku_data_services/data_connectors/apispec.py \
-    components/renku_data_services/search/apispec.py
+    components/renku_data_services/search/apispec.py \
+    components/renku_data_services/notifications/apispec.py
 
 schemas: ${API_SPECS}  ## Generate pydantic classes from apispec yaml files
 	@echo "generated classes based on ApiSpec"

bases/renku_data_services/data_api/app.py

@@ -26,6 +26,7 @@
 from renku_data_services.data_connectors.blueprints import DataConnectorsBP
 from renku_data_services.namespace.blueprints import GroupsBP
 from renku_data_services.notebooks.blueprints import NotebooksBP, NotebooksNewBP
+from renku_data_services.notifications.blueprints import NotificationsBP
 from renku_data_services.platform.blueprints import PlatformConfigBP, PlatformUrlRedirectBP
 from renku_data_services.project.blueprints import ProjectsBP, ProjectSessionSecretBP
 from renku_data_services.repositories.blueprints import RepositoriesBP
@@ -261,6 +262,13 @@ def register_all_handlers(app: Sanic, dm: DependencyManager) -> Sanic:
         authenticator=dm.authenticator,
         metrics=dm.metrics,
     )
+    notifications = NotificationsBP(
+        name="notifications",
+        url_prefix=url_prefix,
+        notifications_repo=dm.notifications_repo,
+        authenticator=dm.authenticator,
+        alertmanager_webhook_role=dm.config.alertmanager_webhook_role,
+    )
     app.blueprint(
         [
             resource_pools.blueprint(),
@@ -289,6 +297,7 @@ def register_all_handlers(app: Sanic, dm: DependencyManager) -> Sanic:
             search.blueprint(),
             data_connectors.blueprint(),
             platform_redirects.blueprint(),
+            notifications.blueprint(),
         ]
     )
     if builds is not None:

bases/renku_data_services/data_api/config.py

@@ -40,6 +40,7 @@ class Config:
     gitlab_url: str | None
     log_cfg: LoggingConfig
     version: str
+    alertmanager_webhook_role: str
 
     @classmethod
     def from_env(cls, db: DBConfig | None = None) -> Self:
@@ -82,4 +83,5 @@ def from_env(cls, db: DBConfig | None = None) -> Self:
             server_options=ServerOptionsConfig.from_env(),
             gitlab_url=gitlab_url,
             log_cfg=LoggingConfig.from_env(),
+            alertmanager_webhook_role=os.environ.get("ALERTMANAGER_WEBHOOK_ROLE", "alertmanager-webhook"),
         )

bases/renku_data_services/data_api/dependencies.py

@@ -14,6 +14,7 @@
 import renku_data_services.connected_services
 import renku_data_services.crc
 import renku_data_services.data_connectors
+import renku_data_services.notifications
 import renku_data_services.platform
 import renku_data_services.repositories
 import renku_data_services.search
@@ -56,6 +57,7 @@
 from renku_data_services.notebooks.config import GitProviderHelperProto, get_clusters
 from renku_data_services.notebooks.constants import AMALTHEA_SESSION_GVK, JUPYTER_SESSION_GVK
 from renku_data_services.notebooks.image_check import ImageCheckRepository
+from renku_data_services.notifications.db import NotificationsRepository
 from renku_data_services.platform.db import PlatformRepository, UrlRedirectRepository
 from renku_data_services.project.db import (
     ProjectMemberRepository,
@@ -147,6 +149,7 @@ class DependencyManager:
     shipwright_client: ShipwrightClient | None
     url_redirect_repo: UrlRedirectRepository
     git_provider_helper: GitProviderHelperProto
+    notifications_repo: NotificationsRepository
     oauth_http_client_factory: OAuthHttpClientFactory
 
     spec: dict[str, Any] = field(init=False, repr=False, default_factory=dict)
@@ -175,6 +178,7 @@ def load_apispec() -> dict[str, Any]:
             renku_data_services.platform.__file__,
             renku_data_services.data_connectors.__file__,
             renku_data_services.search.__file__,
+            renku_data_services.notifications.__file__,
         ]
 
         api_specs = []
@@ -394,6 +398,10 @@ def from_env(cls) -> DependencyManager:
             project_repo=project_repo,
             data_connector_repo=data_connector_repo,
         )
+        notifications_repo = NotificationsRepository(
+            session_maker=config.db.async_session_maker,
+            alertmanager_webhook_role=config.alertmanager_webhook_role,
+        )
         return cls(
             config,
             authenticator=authenticator,
@@ -431,5 +439,6 @@ def from_env(cls) -> DependencyManager:
             low_level_user_secrets_repo=low_level_user_secrets_repo,
             url_redirect_repo=url_redirect_repo,
             git_provider_helper=git_provider_helper,
+            notifications_repo=notifications_repo,
             oauth_http_client_factory=oauth_http_client_factory,
         )

components/renku_data_services/authn/dummy.py

@@ -78,4 +78,5 @@ async def authenticate(self, access_token: str, request: Request) -> base_models
             email=user_props.get("email", "john.doe@gmail.com") if is_set else None,
             full_name=user_props.get("full_name", "John Doe") if is_set else None,
             refresh_token=request.headers.get("Renku-Auth-Refresh-Token"),
+            roles=user_props.get("roles", []),
         )

components/renku_data_services/authn/keycloak.py

@@ -116,14 +116,26 @@ async def authenticate(
         with suppress(errors.UnauthorizedError, jwt.InvalidTokenError):
             token = str(header_value).removeprefix("Bearer ").removeprefix("bearer ")
             parsed = self._validate(token)
-            is_admin = self.admin_role in parsed.get("realm_access", {}).get("roles", [])
+            roles = parsed.get("realm_access", {}).get("roles", [])
+            is_admin = self.admin_role in roles
             exp = parsed.get("exp")
             id = parsed.get("sub")
             email = parsed.get("email")
-            if id is None or email is None:
+
+            if email is None:
+                client_id = parsed.get("azp")
+                if client_id:
+                    email = f"service-account-{client_id}@renku.local"
+                else:
+                    raise errors.UnauthorizedError(
+                        message="Your credentials are invalid or expired, please log in again."
+                    ) from None
+
+            if id is None:
                 raise errors.UnauthorizedError(
                     message="Your credentials are invalid or expired, please log in again."
                 ) from None
+
             user = base_models.AuthenticatedAPIUser(
                 is_admin=is_admin,
                 id=id,
@@ -134,6 +146,7 @@ async def authenticate(
                 email=email,
                 refresh_token=str(refresh_token) if refresh_token else None,
                 access_token_expires_at=datetime.fromtimestamp(exp) if exp is not None else None,
+                roles=roles,
             )
         if user is not None:
             return user

components/renku_data_services/base_api/auth.py

@@ -156,3 +156,36 @@ async def decorated_function(*args: _P.args, **kwargs: _P.kwargs) -> _T:
         return response
 
     return decorated_function
+
+
+def require_role(
+    role: str,
+) -> Callable[
+    [Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]]],
+    Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]],
+]:
+    """Decorator for a Sanic handler that errors out if the user does not have the specified role.
+
+    Args:
+        role: The role name to check for (e.g., "alertmanager-webhook")
+    """
+
+    def decorator(
+        f: Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]],
+    ) -> Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]]:
+        @wraps(f)
+        async def decorated_function(request: Request, user: APIUser, *args: _P.args, **kwargs: _P.kwargs) -> _T:
+            if user is None or user.access_token is None:
+                raise errors.UnauthorizedError(
+                    message="Please provide valid access credentials in the Authorization header."
+                )
+            if role not in user.roles and not user.is_admin:
+                raise errors.ForbiddenError(message=f"You do not have the required role '{role}' for this operation.")
+
+            # the user is authenticated and has the required role
+            response = await f(request, user, *args, **kwargs)
+            return response
+
+        return decorated_function
+
+    return decorator

components/renku_data_services/base_models/core.py

@@ -33,6 +33,7 @@ class APIUser:
     email: str | None = None
     access_token_expires_at: datetime | None = None
     is_admin: bool = False
+    roles: list[str] = field(default_factory=list)
 
     @property
     def is_authenticated(self) -> bool:

components/renku_data_services/migrations/env.py

@@ -9,6 +9,7 @@
 from renku_data_services.metrics.orm import BaseORM as metrics
 from renku_data_services.migrations.utils import run_migrations
 from renku_data_services.namespace.orm import BaseORM as namespaces
+from renku_data_services.notifications.orm import BaseORM as notifications
 from renku_data_services.platform.orm import BaseORM as platform
 from renku_data_services.project.orm import BaseORM as project
 from renku_data_services.search.orm import BaseORM as search
@@ -19,20 +20,21 @@
 
 all_metadata = [
     authz.metadata,
-    crc.metadata,
     connected_services.metadata,
+    crc.metadata,
     data_connectors.metadata,
     events.metadata,
     k8s_cache.metadata,
     metrics.metadata,
     namespaces.metadata,
+    notifications.metadata,
     platform.metadata,
     project.metadata,
+    search.metadata,
     secrets.metadata,
     sessions.metadata,
     storage.metadata,
     users.metadata,
-    search.metadata,
 ]
 
 run_migrations(all_metadata)

components/renku_data_services/migrations/versions/5ec28ea89e0a_add_alerts_table.py

@@ -0,0 +1,51 @@
+"""add alerts table
+
+Revision ID: 5ec28ea89e0a
+Revises: 328803606473
+Create Date: 2025-11-05 13:41:25.972261
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+from renku_data_services.utils.sqlalchemy import ULIDType
+
+# revision identifiers, used by Alembic.
+revision = "5ec28ea89e0a"
+down_revision = "328803606473"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "alerts",
+        sa.Column("id", ULIDType(), server_default=sa.text("generate_ulid()"), nullable=False),
+        sa.Column("title", sa.String(), nullable=False),
+        sa.Column("message", sa.String(), nullable=False),
+        sa.Column("event_type", sa.String(), nullable=False),
+        sa.Column("user_id", sa.String(), nullable=False),
+        sa.Column("session_name", sa.String(), nullable=True),
+        sa.Column("creation_date", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("resolved_date", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["user_id"], ["users.users.keycloak_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        schema="notifications",
+    )
+    op.create_index(
+        op.f("ix_notifications_alerts_event_type"), "alerts", ["event_type"], unique=False, schema="notifications"
+    )
+    op.create_index(
+        op.f("ix_notifications_alerts_user_id"), "alerts", ["user_id"], unique=False, schema="notifications"
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f("ix_notifications_alerts_user_id"), table_name="alerts", schema="notifications")
+    op.drop_index(op.f("ix_notifications_alerts_event_type"), table_name="alerts", schema="notifications")
+    op.drop_table("alerts", schema="notifications")
+    # ### end Alembic commands ###