feat: support for search reprovisioning (#392)

Author: mohammad-alisafaee

Makefile

@@ -2,7 +2,7 @@
 
 AMALTHEA_JS_VERSION ?= 0.11.0
 AMALTHEA_SESSIONS_VERSION ?= 0.0.1-new-operator-chart
-codegen_params = --input-file-type openapi --output-model-type pydantic_v2.BaseModel --use-double-quotes --target-python-version 3.12 --collapse-root-models --field-constraints --strict-nullable --set-default-enum-member --openapi-scopes schemas paths parameters --set-default-enum-member --use-one-literal-as-default --use-default
+codegen_params = --input-file-type openapi --output-model-type pydantic_v2.BaseModel --use-double-quotes --target-python-version 3.12 --collapse-root-models --field-constraints --strict-nullable --openapi-scopes schemas paths parameters --set-default-enum-member --use-one-literal-as-default --use-default
 
 define test_apispec_up_to_date
 	$(eval $@_NAME=$(1))
@@ -38,10 +38,12 @@ components/renku_data_services/notebooks/apispec.py: components/renku_data_servi
 	poetry run datamodel-codegen --input components/renku_data_services/notebooks/api.spec.yaml --output components/renku_data_services/notebooks/apispec.py --base-class renku_data_services.notebooks.apispec_base.BaseAPISpec $(codegen_params)
 components/renku_data_services/platform/apispec.py: components/renku_data_services/platform/api.spec.yaml
 	poetry run datamodel-codegen --input components/renku_data_services/platform/api.spec.yaml --output components/renku_data_services/platform/apispec.py --base-class renku_data_services.platform.apispec_base.BaseAPISpec $(codegen_params)
+components/renku_data_services/message_queue/apispec.py: components/renku_data_services/message_queue/api.spec.yaml
+	poetry run datamodel-codegen --input components/renku_data_services/message_queue/api.spec.yaml --output components/renku_data_services/message_queue/apispec.py --base-class renku_data_services.message_queue.apispec_base.BaseAPISpec $(codegen_params)
 
 ##@ Apispec
 
-schemas: components/renku_data_services/crc/apispec.py components/renku_data_services/storage/apispec.py components/renku_data_services/users/apispec.py components/renku_data_services/project/apispec.py components/renku_data_services/namespace/apispec.py components/renku_data_services/secrets/apispec.py components/renku_data_services/connected_services/apispec.py components/renku_data_services/repositories/apispec.py components/renku_data_services/notebooks/apispec.py components/renku_data_services/platform/apispec.py  ## Generate pydantic classes from apispec yaml files
+schemas: components/renku_data_services/crc/apispec.py components/renku_data_services/storage/apispec.py components/renku_data_services/users/apispec.py components/renku_data_services/project/apispec.py components/renku_data_services/namespace/apispec.py components/renku_data_services/secrets/apispec.py components/renku_data_services/connected_services/apispec.py components/renku_data_services/repositories/apispec.py components/renku_data_services/notebooks/apispec.py components/renku_data_services/platform/apispec.py components/renku_data_services/message_queue/apispec.py  ## Generate pydantic classes from apispec yaml files
 	@echo "generated classes based on ApiSpec"
 
 ##@ Avro schemas
@@ -84,6 +86,8 @@ style_checks:  ## Run linting and style checks
 	@$(call test_apispec_up_to_date,"notebooks")
 	@echo "checking platform apispec is up to date"
 	@$(call test_apispec_up_to_date,"platform")
+	@echo "checking message_queue apispec is up to date"
+	@$(call test_apispec_up_to_date,"message_queue")
 	poetry run mypy
 	poetry run ruff format --check
 	poetry run ruff check .

bases/renku_data_services/data_api/app.py

@@ -15,6 +15,7 @@
     ResourcePoolUsersBP,
     UserResourcePoolsBP,
 )
+from renku_data_services.message_queue.blueprints import SearchBP
 from renku_data_services.namespace.blueprints import GroupsBP
 from renku_data_services.platform.blueprints import PlatformConfigBP
 from renku_data_services.project.blueprints import ProjectsBP
@@ -140,6 +141,18 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
         platform_repo=config.platform_repo,
         authenticator=config.authenticator,
     )
+    search = SearchBP(
+        name="search",
+        url_prefix=url_prefix,
+        authenticator=config.authenticator,
+        session_maker=config.db.async_session_maker,
+        reprovisioning_repo=config.reprovisioning_repo,
+        event_repo=config.event_repo,
+        user_repo=config.kc_user_repo,
+        group_repo=config.group_repo,
+        project_repo=config.project_repo,
+        authz=config.authz,
+    )
     app.blueprint(
         [
             resource_pools.blueprint(),
@@ -162,6 +175,7 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
             oauth2_connections.blueprint(),
             repositories.blueprint(),
             platform_config.blueprint(),
+            search.blueprint(),
         ]
     )
 

components/renku_data_services/app_config/config.py

@@ -48,7 +48,7 @@
 from renku_data_services.k8s.clients import DummyCoreClient, DummySchedulingClient, K8sCoreClient, K8sSchedulingClient
 from renku_data_services.k8s.quota import QuotaRepository
 from renku_data_services.message_queue.config import RedisConfig
-from renku_data_services.message_queue.db import EventRepository
+from renku_data_services.message_queue.db import EventRepository, ReprovisioningRepository
 from renku_data_services.message_queue.interface import IMessageQueue
 from renku_data_services.message_queue.redis_queue import RedisQueue
 from renku_data_services.namespace.db import GroupRepository
@@ -166,6 +166,7 @@ class Config:
     _project_repo: ProjectRepository | None = field(default=None, repr=False, init=False)
     _group_repo: GroupRepository | None = field(default=None, repr=False, init=False)
     _event_repo: EventRepository | None = field(default=None, repr=False, init=False)
+    _reprovisioning_repo: ReprovisioningRepository | None = field(default=None, repr=False, init=False)
     _session_repo: SessionRepository | None = field(default=None, repr=False, init=False)
     _user_preferences_repo: UserPreferencesRepository | None = field(default=None, repr=False, init=False)
     _kc_user_repo: KcUserRepo | None = field(default=None, repr=False, init=False)
@@ -176,6 +177,7 @@ class Config:
     _platform_repo: PlatformRepository | None = field(default=None, repr=False, init=False)
 
     def __post_init__(self) -> None:
+        # NOTE: Read spec files required for Swagger
         spec_file = Path(renku_data_services.crc.__file__).resolve().parent / "api.spec.yaml"
         with open(spec_file) as f:
             crc_spec = safe_load(f)
@@ -212,6 +214,10 @@ def __post_init__(self) -> None:
         with open(spec_file) as f:
             platform = safe_load(f)
 
+        spec_file = Path(renku_data_services.message_queue.__file__).resolve().parent / "api.spec.yaml"
+        with open(spec_file) as f:
+            search = safe_load(f)
+
         self.spec = merge_api_specs(
             crc_spec,
             storage_spec,
@@ -222,6 +228,7 @@ def __post_init__(self) -> None:
             connected_services,
             repositories,
             platform,
+            search,
         )
 
         if self.default_resource_pool_file is not None:
@@ -287,6 +294,13 @@ def event_repo(self) -> EventRepository:
             )
         return self._event_repo
 
+    @property
+    def reprovisioning_repo(self) -> ReprovisioningRepository:
+        """The DB adapter for reprovisioning."""
+        if not self._reprovisioning_repo:
+            self._reprovisioning_repo = ReprovisioningRepository(session_maker=self.db.async_session_maker)
+        return self._reprovisioning_repo
+
     @property
     def project_repo(self) -> ProjectRepository:
         """The DB adapter for Renku native projects."""

components/renku_data_services/authz/authz.py

@@ -1,7 +1,7 @@
 """Projects authorization adapter."""
 
 import asyncio
-from collections.abc import AsyncIterable, Awaitable, Callable
+from collections.abc import AsyncGenerator, AsyncIterable, Awaitable, Callable
 from dataclasses import dataclass, field
 from enum import StrEnum
 from functools import wraps
@@ -385,6 +385,15 @@ async def users_with_permission(
                 ids.append(response.subject.subject_object_id)
         return ids
 
+    async def get_all_members(
+        self, resource_type: ResourceType, *, zed_token: ZedToken | None = None
+    ) -> AsyncGenerator[Member, None]:
+        """Get all users that are members of a specific resource."""
+        members = self._get_members_helper(resource_type, resource_id=None, zed_token=zed_token)
+        async for member in members:
+            if member.user_id and member.user_id != "*":
+                yield member
+
     @_is_allowed(Scope.READ)
     async def members(
         self,
@@ -395,41 +404,63 @@ async def members(
         *,
         zed_token: ZedToken | None = None,
     ) -> list[Member]:
+        """Get all users that are members of a specific resource type, if role is None then all roles are retrieved."""
+        members = self._get_members_helper(resource_type, str(resource_id), role, zed_token=zed_token)
+        return [m async for m in members]
+
+    async def _get_members_helper(
+        self,
+        resource_type: ResourceType,
+        resource_id: str | None,
+        role: Role | None = None,
+        *,
+        zed_token: ZedToken | None = None,
+    ) -> AsyncGenerator[Member, None]:
         """Get all users that are members of a resource, if role is None then all roles are retrieved."""
-        resource_id_str = str(resource_id)
         consistency = Consistency(at_least_as_fresh=zed_token) if zed_token else Consistency(fully_consistent=True)
         sub_filter = SubjectFilter(subject_type=ResourceType.user.value)
-        rel_filter = RelationshipFilter(
-            resource_type=resource_type,
-            optional_resource_id=resource_id_str,
-            optional_subject_filter=sub_filter,
-        )
-        if role:
-            relation = _Relation.from_role(role)
+        if resource_id is None:
+            rel_filter = RelationshipFilter(resource_type=resource_type, optional_subject_filter=sub_filter)
+        else:
             rel_filter = RelationshipFilter(
                 resource_type=resource_type,
-                optional_resource_id=resource_id_str,
-                optional_relation=relation,
+                optional_resource_id=resource_id,
                 optional_subject_filter=sub_filter,
             )
+        if role:
+            relation = _Relation.from_role(role)
+            if resource_id is None:
+                rel_filter = RelationshipFilter(
+                    resource_type=resource_type,
+                    optional_relation=relation,
+                    optional_subject_filter=sub_filter,
+                )
+            else:
+                rel_filter = RelationshipFilter(
+                    resource_type=resource_type,
+                    optional_resource_id=resource_id,
+                    optional_relation=relation,
+                    optional_subject_filter=sub_filter,
+                )
         responses: AsyncIterable[ReadRelationshipsResponse] = self.client.ReadRelationships(
             ReadRelationshipsRequest(
                 consistency=consistency,
                 relationship_filter=rel_filter,
             )
         )
-        members: list[Member] = []
+
         async for response in responses:
             # Skip "public_viewer" relationships
             if response.relationship.relation == _Relation.public_viewer.value:
                 continue
             member_role = _Relation(response.relationship.relation).to_role()
-            members.append(
-                Member(
-                    user_id=response.relationship.subject.object.object_id, role=member_role, resource_id=resource_id
-                )
+            member = Member(
+                user_id=response.relationship.subject.object.object_id,
+                role=member_role,
+                resource_id=response.relationship.resource.object_id,
             )
-        return members
+
+            yield member
 
     @staticmethod
     def authz_change(

components/renku_data_services/authz/models.py

@@ -88,7 +88,7 @@ class MembershipChange:
 
 
 class Visibility(Enum):
-    """The visisibilty of a resource."""
+    """The visibility of a resource."""
 
     PUBLIC: str = "public"
     PRIVATE: str = "private"

components/renku_data_services/message_queue/api.spec.yaml

@@ -0,0 +1,112 @@
+openapi: 3.0.2
+info:
+  title: Renku Data Services API
+  description: |
+    This service is the main backend for Renku. It provides information about users, projects,
+    cloud storage, access to compute resources and many other things.
+  version: v1
+servers:
+  - url: /api/data
+  - url: /ui-server/api/data
+paths:
+  /message_queue/reprovision:
+    post:
+      summary: Start a new reprovisioning
+      description: Only a single reprovisioning is active at any time
+      responses:
+        "201":
+          description: The reprovisioning is/will be started
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Reprovisioning"
+        "409":
+          description: A reprovisioning is already started
+        default:
+          $ref: "#/components/responses/Error"
+      tags:
+        - message_queue
+    get:
+      summary: Return status of reprovisioning
+      responses:
+        "200":
+          description: Status of reprovisioning if there's one in progress
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ReprovisioningStatus"
+        "404":
+          description: There's no active reprovisioning
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+        default:
+          $ref: "#/components/responses/Error"
+      tags:
+        - message_queue
+    delete:
+      summary: Stop an active reprovisioning
+      responses:
+        "204":
+          description: The reprovisioning was stopped or there was no one in progress
+        default:
+          $ref: "#/components/responses/Error"
+      tags:
+        - message_queue
+
+components:
+  schemas:
+    Reprovisioning:
+      description: A reprovisioning
+      type: object
+      properties:
+        id:
+          $ref: "#/components/schemas/Ulid"
+        start_date:
+          description: The date and time the reprovisioning was started (in UTC and ISO-8601 format)
+          type: string
+          format: date-time
+          example: "2023-11-01T17:32:28Z"
+      required:
+        - id
+        - start_date
+    ReprovisioningStatus:
+      description: Status of a reprovisioning
+      allOf:
+      - $ref: "#/components/schemas/Reprovisioning"
+    Ulid:
+      description: ULID identifier
+      type: string
+      minLength: 26
+      maxLength: 26
+      pattern: "^[0-7][0-9A-HJKMNP-TV-Z]{25}$" # This is case-insensitive
+    ErrorResponse:
+      type: object
+      properties:
+        error:
+          type: object
+          properties:
+            code:
+              type: integer
+              minimum: 0
+              exclusiveMinimum: true
+              example: 1404
+            detail:
+              type: string
+              example: A more detailed optional message showing what the problem was
+            message:
+              type: string
+              example: Something went wrong - please try again later
+          required:
+            - code
+            - message
+      required:
+        - error
+  responses:
+    Error:
+      description: The schema for all 4xx and 5xx responses
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/ErrorResponse"

components/renku_data_services/message_queue/apispec.py

@@ -0,0 +1,42 @@
+# generated by datamodel-codegen:
+#   filename:  api.spec.yaml
+#   timestamp: 2024-10-03T07:51:44+00:00
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional
+
+from pydantic import Field
+from renku_data_services.message_queue.apispec_base import BaseAPISpec
+
+
+class Error(BaseAPISpec):
+    code: int = Field(..., example=1404, gt=0)
+    detail: Optional[str] = Field(
+        None, example="A more detailed optional message showing what the problem was"
+    )
+    message: str = Field(..., example="Something went wrong - please try again later")
+
+
+class ErrorResponse(BaseAPISpec):
+    error: Error
+
+
+class Reprovisioning(BaseAPISpec):
+    id: str = Field(
+        ...,
+        description="ULID identifier",
+        max_length=26,
+        min_length=26,
+        pattern="^[0-7][0-9A-HJKMNP-TV-Z]{25}$",
+    )
+    start_date: datetime = Field(
+        ...,
+        description="The date and time the reprovisioning was started (in UTC and ISO-8601 format)",
+        example="2023-11-01T17:32:28Z",
+    )
+
+
+class ReprovisioningStatus(Reprovisioning):
+    pass