16295569791

Committed 15 Jul 2025 02:07PM UTC coverage: 87.229% (+0.09%) from 87.138%

Build # 16295569791

Build Type

Pull #876

github

Committed by

web-flow

Commit Message

Merge 94cd72cbc into cd2196ae6

Pull Request Pull Request #876: Build/external resources 2

Run Details

346 of 409 new or added lines in 33 files covered. (84.6%)

14 existing lines in 7 files now uncovered.

21481 of 24626 relevant lines covered (87.23%)

1.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.88

/components/renku_data_services/notebooks/api/classes/server.py

"""Jupyter server models."""

from collections.abc import Sequence
from itertools import chain
from pathlib import PurePosixPath
from typing import Any
from urllib.parse import urlparse

from gitlab.v4.objects.projects import Project

from renku_data_services.app_config import logging
from renku_data_services.base_models import AnonymousAPIUser, AuthenticatedAPIUser
from renku_data_services.base_models.core import APIUser
from renku_data_services.notebooks.api.amalthea_patches import cloudstorage as cloudstorage_patches
from renku_data_services.notebooks.api.amalthea_patches import general as general_patches
from renku_data_services.notebooks.api.amalthea_patches import git_proxy as git_proxy_patches
from renku_data_services.notebooks.api.amalthea_patches import git_sidecar as git_sidecar_patches
from renku_data_services.notebooks.api.amalthea_patches import init_containers as init_containers_patches
from renku_data_services.notebooks.api.amalthea_patches import inject_certificates as inject_certificates_patches
from renku_data_services.notebooks.api.amalthea_patches import jupyter_server as jupyter_server_patches
from renku_data_services.notebooks.api.amalthea_patches import ssh as ssh_patches
from renku_data_services.notebooks.api.classes.cloud_storage import ICloudStorageRequest
from renku_data_services.notebooks.api.classes.k8s_client import NotebookK8sClient
from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository
from renku_data_services.notebooks.api.schemas.secrets import K8sUserSecrets
from renku_data_services.notebooks.api.schemas.server_options import ServerOptions
from renku_data_services.notebooks.config import NotebooksConfig
from renku_data_services.notebooks.constants import JUPYTER_SESSION_GVK
from renku_data_services.notebooks.crs import JupyterServerV1Alpha1
from renku_data_services.notebooks.errors.programming import DuplicateEnvironmentVariableError
from renku_data_services.notebooks.errors.user import MissingResourceError

logger = logging.getLogger(__name__)


class UserServer:
    """Represents a Renku server session."""

    def __init__(
        self,
        user: AnonymousAPIUser | AuthenticatedAPIUser,
        server_name: str,
        image: str | None,
        server_options: ServerOptions,
        environment_variables: dict[str, str],
        user_secrets: K8sUserSecrets | None,
        cloudstorage: Sequence[ICloudStorageRequest],
        k8s_client: NotebookK8sClient[JupyterServerV1Alpha1],
        workspace_mount_path: PurePosixPath,
        work_dir: PurePosixPath,
        config: NotebooksConfig,
        internal_gitlab_user: APIUser,
        host: str,
        using_default_image: bool = False,
        is_image_private: bool = False,
        repositories: list[Repository] | None = None,
    ):
        self._user = user
        self.server_name = server_name
        self._k8s_client = k8s_client
        self.safe_username = self._user.id
        self.image = image
        self.server_options = server_options
        self.environment_variables = environment_variables
        self.user_secrets = user_secrets
        self.using_default_image = using_default_image
        self.workspace_mount_path = workspace_mount_path
        self.work_dir = work_dir
        self.cloudstorage = cloudstorage
        self.is_image_private = is_image_private
        self.host = host
        self.config = config
        self.internal_gitlab_user = internal_gitlab_user

        if self.server_options.idle_threshold_seconds is not None:
            self.idle_seconds_threshold = self.server_options.idle_threshold_seconds
        else:
            self.idle_seconds_threshold = (
                config.sessions.culling.registered.idle_seconds
                if isinstance(self._user, AuthenticatedAPIUser)
                else config.sessions.culling.anonymous.idle_seconds
            )

        if self.server_options.hibernation_threshold_seconds is not None:
            self.hibernated_seconds_threshold: int = self.server_options.hibernation_threshold_seconds
        else:
            self.hibernated_seconds_threshold = (
                config.sessions.culling.registered.hibernated_seconds
                if isinstance(user, AuthenticatedAPIUser)
                else config.sessions.culling.anonymous.hibernated_seconds
            )
        self._repositories: list[Repository] = repositories or []
        self._git_providers: list[GitProvider] | None = None
        self._has_configured_git_providers = False

        self.server_url = f"https://{self.host}/sessions/{self.server_name}"
        if not self._user.is_authenticated:
            self.server_url = f"{self.server_url}?token={self._user.id}"

    def k8s_namespace(self) -> str:
        """Get the preferred namespace for a server."""
        return self._k8s_client.namespace()

    @property
    def user(self) -> AnonymousAPIUser | AuthenticatedAPIUser:
        """Getter for server's user."""
        return self._user

    async def repositories(self) -> list[Repository]:
        """Get the list of repositories in the project."""
        # Configure git repository providers based on matching URLs.
        if not self._has_configured_git_providers:
            git_providers = await self.git_providers()
            for repo in self._repositories:
                found_provider = None
                for provider in git_providers:
                    if urlparse(provider.url).netloc == urlparse(repo.url).netloc:
                        found_provider = provider
                        break
                if found_provider is not None:
                    repo.provider = found_provider.id
            self._has_configured_git_providers = True

        return self._repositories

    async def git_providers(self) -> list[GitProvider]:
        """The list of git providers."""
        if self._git_providers is None:
            self._git_providers = await self.config.git_provider_helper.get_providers(user=self.user)
        return self._git_providers

    async def required_git_providers(self) -> list[GitProvider]:
        """The list of required git providers."""
        repositories = await self.repositories()
        required_provider_ids: set[str] = set(r.provider for r in repositories if r.provider)
        providers = await self.git_providers()
        return [p for p in providers if p.id in required_provider_ids]

    def __str__(self) -> str:
        return f"<UserServer user: {self._user.id} server_name: {self.server_name}>"

    async def start(self) -> JupyterServerV1Alpha1 | None:
        """Create the jupyterserver resource in k8s."""
        errors = self._get_start_errors()
        if errors:
            raise MissingResourceError(
                message=(
                    "Cannot start the session because the following Git "
                    f"or Docker resources are missing: {', '.join(errors)}"
                )
            )
        session_manifest = await self._get_session_manifest()
        manifest = JupyterServerV1Alpha1.model_validate(session_manifest)
        return await self._k8s_client.create_session(manifest, self.user)

    @staticmethod
    def _check_environment_variables_overrides(patches_list: list[dict[str, Any]]) -> None:
        """Check if any patch overrides server's environment variables.

        Checks if it overrides with a different value or if two patches create environment variables with different
        values.
        """
        env_vars: dict[tuple[str, str], str] = {}

        for patch_list in patches_list:
            patches = patch_list["patch"]

            for patch in patches:
                path = str(patch["path"]).lower()
                if path.endswith("/env/-"):
                    name = str(patch["value"]["name"])
                    value = str(patch["value"]["value"])
                    key = (path, name)

                    if key in env_vars and env_vars[key] != value:
                        raise DuplicateEnvironmentVariableError(
                            message=f"Environment variable {path}::{name} is being overridden by multiple patches"
                        )
                    else:
                        env_vars[key] = value

    def _get_start_errors(self) -> list[str]:
        """Check if there are any errors before starting the server."""
        errors: list[str] = []
        if self.image is None:
            errors.append(f"image {self.image} does not exist or cannot be accessed")
        return errors

    async def _get_session_manifest(self) -> dict[str, Any]:
        """Compose the body of the user session for the k8s operator."""
        patches = await self._get_patches()
        self._check_environment_variables_overrides(patches)

        # Storage
        if self.config.sessions.storage.pvs_enabled:
            storage: dict[str, Any] = {
                "size": self.server_options.storage,
                "pvc": {
                    "enabled": True,
                    # We should check against the cluster, but as this is only used by V1 sessions, we ignore this
                    # use-case.
                    "storageClassName": self.config.sessions.storage.pvs_storage_class,
                    "mountPath": self.workspace_mount_path.as_posix(),
                },
            }
        else:
            storage_size = self.server_options.storage if self.config.sessions.storage.use_empty_dir_size_limit else ""
            storage = {
                "size": storage_size,
                "pvc": {
                    "enabled": False,
                    "mountPath": self.workspace_mount_path.as_posix(),
                },
            }
        # Authentication
        if isinstance(self._user, AuthenticatedAPIUser):
            session_auth = {
                "token": "",
                "oidc": {
                    "enabled": True,
                    "clientId": self.config.sessions.oidc.client_id,
                    "clientSecret": {"value": self.config.sessions.oidc.client_secret},
                    "issuerUrl": self.config.sessions.oidc.issuer_url,
                    "authorizedEmails": [self._user.email],
                },
            }
        else:
            session_auth = {
                "token": self._user.id,
                "oidc": {"enabled": False},
            }

        cluster = await self.config.k8s_client.cluster_by_class_id(self.server_options.resource_class_id, self._user)
        (
            base_server_path,
            base_server_url,
            base_server_https_url,
            host,
            tls_secret,
            ingress_annotations,
        ) = await cluster.get_ingress_parameters(
            self._user, self.config.cluster_rp, self.config.sessions.ingress, self.server_name
        )

        # Combine everything into the manifest
        manifest = {
            "apiVersion": JUPYTER_SESSION_GVK.group_version,
            "kind": JUPYTER_SESSION_GVK.kind,
            "metadata": {
                "name": self.server_name,
                "labels": self.get_labels(),
                "annotations": self.get_annotations(),
            },
            "spec": {
                "auth": session_auth,
                "culling": {
                    "idleSecondsThreshold": self.idle_seconds_threshold,
                    "maxAgeSecondsThreshold": (
                        self.config.sessions.culling.registered.max_age_seconds
                        if isinstance(self._user, AuthenticatedAPIUser)
                        else self.config.sessions.culling.anonymous.max_age_seconds
                    ),
                    "hibernatedSecondsThreshold": self.hibernated_seconds_threshold,
                },
                "jupyterServer": {
                    "defaultUrl": self.server_options.default_url,
                    "image": self.image,
                    "rootDir": self.work_dir.as_posix(),
                    "resources": self.server_options.to_k8s_resources(
                        enforce_cpu_limits=self.config.sessions.enforce_cpu_limits
                    ),
                },
                "routing": {
                    "host": host,
                    "path": base_server_path,
                    "ingressAnnotations": ingress_annotations,
                    "tls": {
                        "enabled": tls_secret is not None,
                        "secretName": tls_secret.name if tls_secret is not None else "",
                    },
                },
                "storage": storage,
                "patches": patches,
            },
        }
        return manifest

    def _get_renku_annotation_prefix(self) -> str:
        return self.config.session_get_endpoint_annotations.renku_annotation_prefix

    async def _get_patches(self) -> list[dict[str, Any]]:
        return list(
            chain(
                general_patches.test(self),
                general_patches.session_tolerations(self),
                general_patches.session_affinity(self),
                general_patches.session_node_selector(self),
                general_patches.priority_class(self),
                general_patches.dev_shm(self),
                jupyter_server_patches.args(),
                jupyter_server_patches.env(self),
                jupyter_server_patches.image_pull_secret(self, self.internal_gitlab_user.access_token),
                jupyter_server_patches.disable_service_links(),
                jupyter_server_patches.rstudio_env_variables(self),
                await git_proxy_patches.main(self),
                await git_sidecar_patches.main(self),
                general_patches.oidc_unverified_email(self),
                ssh_patches.main(self.config),
                # init container for certs must come before all other init containers
                # so that it runs first before all other init containers
                init_containers_patches.certificates(self.config),
                init_containers_patches.download_image(self),
                await init_containers_patches.git_clone(self),
                inject_certificates_patches.proxy(self),
                # Cloud Storage needs to patch the git clone sidecar spec and so should come after
                # the sidecars
                # WARN: this patch depends on the index of the sidecar and so needs to be updated
                # if sidercars are added or removed
                await cloudstorage_patches.main(self),
                # NOTE: User secrets adds an init container, volume and mounts, so it may affect
                # indices in other patches.
                jupyter_server_patches.user_secrets(self),
            )
        )

    def get_labels(self) -> dict[str, str | None]:
        """Get the labels for the session."""
        prefix = self._get_renku_annotation_prefix()
        labels = {
            "app": "jupyter",
            "component": "singleuser-server",
            f"{prefix}commit-sha": None,
            f"{prefix}gitlabProjectId": None,
            f"{prefix}safe-username": self.safe_username,
            f"{prefix}quota": self.server_options.priority_class
            if self.server_options.priority_class is not None
            else "",
            f"{prefix}userId": self._user.id,
        }
        return labels

    def get_annotations(self) -> dict[str, str | None]:
        """Get the annotations for the session."""
        prefix = self._get_renku_annotation_prefix()
        username = self._user.id
        if isinstance(self.user, AuthenticatedAPIUser) and self._user.email:
            username = self._user.email
        annotations = {
            f"{prefix}commit-sha": None,
            f"{prefix}gitlabProjectId": None,
            f"{prefix}safe-username": self._user.id,
            f"{prefix}username": username,
            f"{prefix}userId": self._user.id,
            f"{prefix}servername": self.server_name,
            f"{prefix}branch": None,
            f"{prefix}git-host": None,
            f"{prefix}namespace": None,
            f"{prefix}projectName": None,
            f"{prefix}requested-image": self.image,
            f"{prefix}repository": None,
            f"{prefix}hibernation": "",
            f"{prefix}hibernationBranch": "",
            f"{prefix}hibernationCommitSha": "",
            f"{prefix}hibernationDirty": "",
            f"{prefix}hibernationSynchronized": "",
            f"{prefix}hibernationDate": "",
            f"{prefix}hibernatedSecondsThreshold": str(self.hibernated_seconds_threshold),
            f"{prefix}lastActivityDate": "",
            f"{prefix}idleSecondsThreshold": str(self.idle_seconds_threshold),
        }
        if self.server_options.resource_class_id:
            annotations[f"{prefix}resourceClassId"] = str(self.server_options.resource_class_id)
        return annotations


class Renku1UserServer(UserServer):
    """Represents a Renku 1.0 server session."""

    def __init__(
        self,
        user: AnonymousAPIUser | AuthenticatedAPIUser,
        server_name: str,
        gl_namespace: str,
        project: str,
        branch: str,
        commit_sha: str,
        image: str | None,
        server_options: ServerOptions,
        environment_variables: dict[str, str],
        user_secrets: K8sUserSecrets | None,
        cloudstorage: Sequence[ICloudStorageRequest],
        k8s_client: NotebookK8sClient,
        workspace_mount_path: PurePosixPath,
        work_dir: PurePosixPath,
        config: NotebooksConfig,
        host: str,
        gitlab_project: Project | None,
        internal_gitlab_user: APIUser,
        using_default_image: bool = False,
        is_image_private: bool = False,
        **_: dict,  # Required to ignore unused arguments, among which repositories
    ):
        repositories = [
            Repository(
                url=p.http_url_to_repo,
                dirname=p.path,
                branch=branch,
                commit_sha=commit_sha,
            )
            for p in [gitlab_project]
            if p is not None
        ]

        super().__init__(
            user=user,
            server_name=server_name,
            image=image,
            server_options=server_options,
            environment_variables=environment_variables,
            user_secrets=user_secrets,
            cloudstorage=cloudstorage,
            k8s_client=k8s_client,
            workspace_mount_path=workspace_mount_path,
            work_dir=work_dir,
            using_default_image=using_default_image,
            is_image_private=is_image_private,
            repositories=repositories,
            host=host,
            config=config,
            internal_gitlab_user=internal_gitlab_user,
        )

        self.gl_namespace = gl_namespace
        self.project = project
        self.branch = branch
        self.commit_sha = commit_sha
        self.git_host = urlparse(config.git.url).netloc
        self.gitlab_project = gitlab_project

    def _get_start_errors(self) -> list[str]:
        """Check if there are any errors before starting the server."""
        errors = super()._get_start_errors()
        if self.gitlab_project is None:
            errors.append(f"project {self.project} does not exist")
        if not self._branch_exists():
            errors.append(f"branch {self.branch} does not exist")
        if not self._commit_sha_exists():
            errors.append(f"commit {self.commit_sha} does not exist")
        return errors

    def _branch_exists(self) -> bool:
        """Check if a specific branch exists in the user's gitlab project.

        The branch name is not required by the API and therefore
        passing None to this function will return True.
        """
        if self.branch is not None and self.gitlab_project is not None:
            try:
                self.gitlab_project.branches.get(self.branch)
            except Exception as err:
                logger.warning(f"Branch {self.branch} cannot be verified or does not exist. {err}")
            else:
                return True
        return False

    def _commit_sha_exists(self) -> bool:
        """Check if a specific commit sha exists in the user's gitlab project."""
        if self.commit_sha is not None and self.gitlab_project is not None:
            try:
                self.gitlab_project.commits.get(self.commit_sha)
            except Exception as err:
                logger.warning(f"Commit {self.commit_sha} cannot be verified or does not exist. {err}")
            else:
                return True
        return False

    def get_labels(self) -> dict[str, str | None]:
        """Get the labels of the jupyter server."""
        prefix = self._get_renku_annotation_prefix()
        labels = super().get_labels()
        labels[f"{prefix}commit-sha"] = self.commit_sha
        if self.gitlab_project is not None:
            labels[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)
        return labels

    def get_annotations(self) -> dict[str, str | None]:
        """Get the annotations of the jupyter server."""
        prefix = self._get_renku_annotation_prefix()
        annotations = super().get_annotations()
        annotations[f"{prefix}commit-sha"] = self.commit_sha
        annotations[f"{prefix}branch"] = self.branch
        annotations[f"{prefix}git-host"] = self.git_host
        annotations[f"{prefix}namespace"] = self.gl_namespace
        annotations[f"{prefix}projectName"] = self.project
        if self.gitlab_project is not None:
            annotations[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)
            annotations[f"{prefix}repository"] = self.gitlab_project.web_url
        return annotations

1	"""Jupyter server models."""
2
3	from collections.abc import Sequence	2✔
4	from itertools import chain	2✔
5	from pathlib import PurePosixPath	2✔
6	from typing import Any	2✔
7	from urllib.parse import urlparse	2✔
8
9	from gitlab.v4.objects.projects import Project	2✔
10
11	from renku_data_services.app_config import logging	2✔
12	from renku_data_services.base_models import AnonymousAPIUser, AuthenticatedAPIUser	2✔
13	from renku_data_services.base_models.core import APIUser	2✔
14	from renku_data_services.notebooks.api.amalthea_patches import cloudstorage as cloudstorage_patches	2✔
15	from renku_data_services.notebooks.api.amalthea_patches import general as general_patches	2✔
16	from renku_data_services.notebooks.api.amalthea_patches import git_proxy as git_proxy_patches	2✔
17	from renku_data_services.notebooks.api.amalthea_patches import git_sidecar as git_sidecar_patches	2✔
18	from renku_data_services.notebooks.api.amalthea_patches import init_containers as init_containers_patches	2✔
19	from renku_data_services.notebooks.api.amalthea_patches import inject_certificates as inject_certificates_patches	2✔
20	from renku_data_services.notebooks.api.amalthea_patches import jupyter_server as jupyter_server_patches	2✔
21	from renku_data_services.notebooks.api.amalthea_patches import ssh as ssh_patches	2✔
22	from renku_data_services.notebooks.api.classes.cloud_storage import ICloudStorageRequest	2✔
23	from renku_data_services.notebooks.api.classes.k8s_client import NotebookK8sClient	2✔
24	from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository	2✔
25	from renku_data_services.notebooks.api.schemas.secrets import K8sUserSecrets	2✔
26	from renku_data_services.notebooks.api.schemas.server_options import ServerOptions	2✔
27	from renku_data_services.notebooks.config import NotebooksConfig	2✔
28	from renku_data_services.notebooks.constants import JUPYTER_SESSION_GVK	2✔
29	from renku_data_services.notebooks.crs import JupyterServerV1Alpha1	2✔
30	from renku_data_services.notebooks.errors.programming import DuplicateEnvironmentVariableError	2✔
31	from renku_data_services.notebooks.errors.user import MissingResourceError	2✔
32
33	logger = logging.getLogger(__name__)	2✔
34
35
36	class UserServer:	2✔
37	"""Represents a Renku server session."""
38
39	def __init__(	2✔
40	self,
41	user: AnonymousAPIUser \| AuthenticatedAPIUser,
42	server_name: str,
43	image: str \| None,
44	server_options: ServerOptions,
45	environment_variables: dict[str, str],
46	user_secrets: K8sUserSecrets \| None,
47	cloudstorage: Sequence[ICloudStorageRequest],
48	k8s_client: NotebookK8sClient[JupyterServerV1Alpha1],
49	workspace_mount_path: PurePosixPath,
50	work_dir: PurePosixPath,
51	config: NotebooksConfig,
52	internal_gitlab_user: APIUser,
53	host: str,
54	using_default_image: bool = False,
55	is_image_private: bool = False,
56	repositories: list[Repository] \| None = None,
57	):
58	self._user = user	1✔
59	self.server_name = server_name	1✔
60	self._k8s_client = k8s_client	1✔
61	self.safe_username = self._user.id	1✔
62	self.image = image	1✔
63	self.server_options = server_options	1✔
64	self.environment_variables = environment_variables	1✔
65	self.user_secrets = user_secrets	1✔
66	self.using_default_image = using_default_image	1✔
67	self.workspace_mount_path = workspace_mount_path	1✔
68	self.work_dir = work_dir	1✔
69	self.cloudstorage = cloudstorage	1✔
70	self.is_image_private = is_image_private	1✔
71	self.host = host	1✔
72	self.config = config	1✔
73	self.internal_gitlab_user = internal_gitlab_user	1✔
74
75	if self.server_options.idle_threshold_seconds is not None:	1✔
76	self.idle_seconds_threshold = self.server_options.idle_threshold_seconds	×
77	else:
78	self.idle_seconds_threshold = (	1✔
79	config.sessions.culling.registered.idle_seconds
80	if isinstance(self._user, AuthenticatedAPIUser)
81	else config.sessions.culling.anonymous.idle_seconds
82	)
83
84	if self.server_options.hibernation_threshold_seconds is not None:	1✔
85	self.hibernated_seconds_threshold: int = self.server_options.hibernation_threshold_seconds	×
86	else:
87	self.hibernated_seconds_threshold = (	1✔
88	config.sessions.culling.registered.hibernated_seconds
89	if isinstance(user, AuthenticatedAPIUser)
90	else config.sessions.culling.anonymous.hibernated_seconds
91	)
92	self._repositories: list[Repository] = repositories or []	1✔
93	self._git_providers: list[GitProvider] \| None = None	1✔
94	self._has_configured_git_providers = False	1✔
95
96	self.server_url = f"https://{self.host}/sessions/{self.server_name}"	1✔
97	if not self._user.is_authenticated:	1✔
NEW 98	self.server_url = f"{self.server_url}?token={self._user.id}"	×
99
100	def k8s_namespace(self) -> str:	2✔
101	"""Get the preferred namespace for a server."""
102	return self._k8s_client.namespace()	1✔
103
104	@property	2✔
105	def user(self) -> AnonymousAPIUser \| AuthenticatedAPIUser:	2✔
106	"""Getter for server's user."""
107	return self._user	1✔
108
109	async def repositories(self) -> list[Repository]:	2✔
110	"""Get the list of repositories in the project."""
111	# Configure git repository providers based on matching URLs.
112	if not self._has_configured_git_providers:	1✔
113	git_providers = await self.git_providers()	1✔
114	for repo in self._repositories:	1✔
115	found_provider = None	1✔
116	for provider in git_providers:	1✔
117	if urlparse(provider.url).netloc == urlparse(repo.url).netloc:	×
118	found_provider = provider	×
119	break	×
120	if found_provider is not None:	1✔
121	repo.provider = found_provider.id	×
122	self._has_configured_git_providers = True	1✔
123
124	return self._repositories	1✔
125
126	async def git_providers(self) -> list[GitProvider]:	2✔
127	"""The list of git providers."""
128	if self._git_providers is None:	1✔
129	self._git_providers = await self.config.git_provider_helper.get_providers(user=self.user)	1✔
130	return self._git_providers	1✔
131
132	async def required_git_providers(self) -> list[GitProvider]:	2✔
133	"""The list of required git providers."""
134	repositories = await self.repositories()	1✔
135	required_provider_ids: set[str] = set(r.provider for r in repositories if r.provider)	1✔
136	providers = await self.git_providers()	1✔
137	return [p for p in providers if p.id in required_provider_ids]	1✔
138
139	def __str__(self) -> str:	2✔
140	return f"<UserServer user: {self._user.id} server_name: {self.server_name}>"	×
141
142	async def start(self) -> JupyterServerV1Alpha1 \| None:	2✔
143	"""Create the jupyterserver resource in k8s."""
144	errors = self._get_start_errors()	1✔
145	if errors:	1✔
146	raise MissingResourceError(	×
147	message=(
148	"Cannot start the session because the following Git "
149	f"or Docker resources are missing: {', '.join(errors)}"
150	)
151	)
152	session_manifest = await self._get_session_manifest()	1✔
153	manifest = JupyterServerV1Alpha1.model_validate(session_manifest)	1✔
154	return await self._k8s_client.create_session(manifest, self.user)	1✔
155
156	@staticmethod	2✔
157	def _check_environment_variables_overrides(patches_list: list[dict[str, Any]]) -> None:	2✔
158	"""Check if any patch overrides server's environment variables.
159
160	Checks if it overrides with a different value or if two patches create environment variables with different
161	values.
162	"""
163	env_vars: dict[tuple[str, str], str] = {}	1✔
164
165	for patch_list in patches_list:	1✔
166	patches = patch_list["patch"]	1✔
167
168	for patch in patches:	1✔
169	path = str(patch["path"]).lower()	1✔
170	if path.endswith("/env/-"):	1✔
171	name = str(patch["value"]["name"])	1✔
172	value = str(patch["value"]["value"])	1✔
173	key = (path, name)	1✔
174
175	if key in env_vars and env_vars[key] != value:	1✔
176	raise DuplicateEnvironmentVariableError(	×
177	message=f"Environment variable {path}::{name} is being overridden by multiple patches"
178	)
179	else:
180	env_vars[key] = value	1✔
181
182	def _get_start_errors(self) -> list[str]:	2✔
183	"""Check if there are any errors before starting the server."""
184	errors: list[str] = []	1✔
185	if self.image is None:	1✔
186	errors.append(f"image {self.image} does not exist or cannot be accessed")	×
187	return errors	1✔
188
189	async def _get_session_manifest(self) -> dict[str, Any]:	2✔
190	"""Compose the body of the user session for the k8s operator."""
191	patches = await self._get_patches()	1✔
192	self._check_environment_variables_overrides(patches)	1✔
193
194	# Storage
195	if self.config.sessions.storage.pvs_enabled:	1✔
196	storage: dict[str, Any] = {	1✔
197	"size": self.server_options.storage,
198	"pvc": {
199	"enabled": True,
200	# We should check against the cluster, but as this is only used by V1 sessions, we ignore this
201	# use-case.
202	"storageClassName": self.config.sessions.storage.pvs_storage_class,
203	"mountPath": self.workspace_mount_path.as_posix(),
204	},
205	}
206	else:
207	storage_size = self.server_options.storage if self.config.sessions.storage.use_empty_dir_size_limit else ""	×
208	storage = {	×
209	"size": storage_size,
210	"pvc": {
211	"enabled": False,
212	"mountPath": self.workspace_mount_path.as_posix(),
213	},
214	}
215	# Authentication
216	if isinstance(self._user, AuthenticatedAPIUser):	1✔
217	session_auth = {	×
218	"token": "",
219	"oidc": {
220	"enabled": True,
221	"clientId": self.config.sessions.oidc.client_id,
222	"clientSecret": {"value": self.config.sessions.oidc.client_secret},
223	"issuerUrl": self.config.sessions.oidc.issuer_url,
224	"authorizedEmails": [self._user.email],
225	},
226	}
227	else:
228	session_auth = {	1✔
229	"token": self._user.id,
230	"oidc": {"enabled": False},
231	}
232
233	cluster = await self.config.k8s_client.cluster_by_class_id(self.server_options.resource_class_id, self._user)	1✔
234	(	1✔
235	base_server_path,
236	base_server_url,
237	base_server_https_url,
238	host,
239	tls_secret,
240	ingress_annotations,
241	) = await cluster.get_ingress_parameters(
242	self._user, self.config.cluster_rp, self.config.sessions.ingress, self.server_name
243	)
244
245	# Combine everything into the manifest
246	manifest = {	1✔
247	"apiVersion": JUPYTER_SESSION_GVK.group_version,
248	"kind": JUPYTER_SESSION_GVK.kind,
249	"metadata": {
250	"name": self.server_name,
251	"labels": self.get_labels(),
252	"annotations": self.get_annotations(),
253	},
254	"spec": {
255	"auth": session_auth,
256	"culling": {
257	"idleSecondsThreshold": self.idle_seconds_threshold,
258	"maxAgeSecondsThreshold": (
259	self.config.sessions.culling.registered.max_age_seconds
260	if isinstance(self._user, AuthenticatedAPIUser)
261	else self.config.sessions.culling.anonymous.max_age_seconds
262	),
263	"hibernatedSecondsThreshold": self.hibernated_seconds_threshold,
264	},
265	"jupyterServer": {
266	"defaultUrl": self.server_options.default_url,
267	"image": self.image,
268	"rootDir": self.work_dir.as_posix(),
269	"resources": self.server_options.to_k8s_resources(
270	enforce_cpu_limits=self.config.sessions.enforce_cpu_limits
271	),
272	},
273	"routing": {
274	"host": host,
275	"path": base_server_path,
276	"ingressAnnotations": ingress_annotations,
277	"tls": {
278	"enabled": tls_secret is not None,
279	"secretName": tls_secret.name if tls_secret is not None else "",
280	},
281	},
282	"storage": storage,
283	"patches": patches,
284	},
285	}
286	return manifest	1✔
287
288	def _get_renku_annotation_prefix(self) -> str:	2✔
289	return self.config.session_get_endpoint_annotations.renku_annotation_prefix	1✔
290
291	async def _get_patches(self) -> list[dict[str, Any]]:	2✔
292	return list(	1✔
293	chain(
294	general_patches.test(self),
295	general_patches.session_tolerations(self),
296	general_patches.session_affinity(self),
297	general_patches.session_node_selector(self),
298	general_patches.priority_class(self),
299	general_patches.dev_shm(self),
300	jupyter_server_patches.args(),
301	jupyter_server_patches.env(self),
302	jupyter_server_patches.image_pull_secret(self, self.internal_gitlab_user.access_token),
303	jupyter_server_patches.disable_service_links(),
304	jupyter_server_patches.rstudio_env_variables(self),
305	await git_proxy_patches.main(self),
306	await git_sidecar_patches.main(self),
307	general_patches.oidc_unverified_email(self),
308	ssh_patches.main(self.config),
309	# init container for certs must come before all other init containers
310	# so that it runs first before all other init containers
311	init_containers_patches.certificates(self.config),
312	init_containers_patches.download_image(self),
313	await init_containers_patches.git_clone(self),
314	inject_certificates_patches.proxy(self),
315	# Cloud Storage needs to patch the git clone sidecar spec and so should come after
316	# the sidecars
317	# WARN: this patch depends on the index of the sidecar and so needs to be updated
318	# if sidercars are added or removed
319	await cloudstorage_patches.main(self),
320	# NOTE: User secrets adds an init container, volume and mounts, so it may affect
321	# indices in other patches.
322	jupyter_server_patches.user_secrets(self),
323	)
324	)
325
326	def get_labels(self) -> dict[str, str \| None]:	2✔
327	"""Get the labels for the session."""
328	prefix = self._get_renku_annotation_prefix()	1✔
329	labels = {	1✔
330	"app": "jupyter",
331	"component": "singleuser-server",
332	f"{prefix}commit-sha": None,
333	f"{prefix}gitlabProjectId": None,
334	f"{prefix}safe-username": self.safe_username,
335	f"{prefix}quota": self.server_options.priority_class
336	if self.server_options.priority_class is not None
337	else "",
338	f"{prefix}userId": self._user.id,
339	}
340	return labels	1✔
341
342	def get_annotations(self) -> dict[str, str \| None]:	2✔
343	"""Get the annotations for the session."""
344	prefix = self._get_renku_annotation_prefix()	1✔
345	username = self._user.id	1✔
346	if isinstance(self.user, AuthenticatedAPIUser) and self._user.email:	1✔
347	username = self._user.email	×
348	annotations = {	1✔
349	f"{prefix}commit-sha": None,
350	f"{prefix}gitlabProjectId": None,
351	f"{prefix}safe-username": self._user.id,
352	f"{prefix}username": username,
353	f"{prefix}userId": self._user.id,
354	f"{prefix}servername": self.server_name,
355	f"{prefix}branch": None,
356	f"{prefix}git-host": None,
357	f"{prefix}namespace": None,
358	f"{prefix}projectName": None,
359	f"{prefix}requested-image": self.image,
360	f"{prefix}repository": None,
361	f"{prefix}hibernation": "",
362	f"{prefix}hibernationBranch": "",
363	f"{prefix}hibernationCommitSha": "",
364	f"{prefix}hibernationDirty": "",
365	f"{prefix}hibernationSynchronized": "",
366	f"{prefix}hibernationDate": "",
367	f"{prefix}hibernatedSecondsThreshold": str(self.hibernated_seconds_threshold),
368	f"{prefix}lastActivityDate": "",
369	f"{prefix}idleSecondsThreshold": str(self.idle_seconds_threshold),
370	}
371	if self.server_options.resource_class_id:	1✔
372	annotations[f"{prefix}resourceClassId"] = str(self.server_options.resource_class_id)	1✔
373	return annotations	1✔
374
375
376	class Renku1UserServer(UserServer):	2✔
377	"""Represents a Renku 1.0 server session."""
378
379	def __init__(	2✔
380	self,
381	user: AnonymousAPIUser \| AuthenticatedAPIUser,
382	server_name: str,
383	gl_namespace: str,
384	project: str,
385	branch: str,
386	commit_sha: str,
387	image: str \| None,
388	server_options: ServerOptions,
389	environment_variables: dict[str, str],
390	user_secrets: K8sUserSecrets \| None,
391	cloudstorage: Sequence[ICloudStorageRequest],
392	k8s_client: NotebookK8sClient,
393	workspace_mount_path: PurePosixPath,
394	work_dir: PurePosixPath,
395	config: NotebooksConfig,
396	host: str,
397	gitlab_project: Project \| None,
398	internal_gitlab_user: APIUser,
399	using_default_image: bool = False,
400	is_image_private: bool = False,
401	**_: dict, # Required to ignore unused arguments, among which repositories
402	):
403	repositories = [	1✔
404	Repository(
405	url=p.http_url_to_repo,
406	dirname=p.path,
407	branch=branch,
408	commit_sha=commit_sha,
409	)
410	for p in [gitlab_project]
411	if p is not None
412	]
413
414	super().__init__(	1✔
415	user=user,
416	server_name=server_name,
417	image=image,
418	server_options=server_options,
419	environment_variables=environment_variables,
420	user_secrets=user_secrets,
421	cloudstorage=cloudstorage,
422	k8s_client=k8s_client,
423	workspace_mount_path=workspace_mount_path,
424	work_dir=work_dir,
425	using_default_image=using_default_image,
426	is_image_private=is_image_private,
427	repositories=repositories,
428	host=host,
429	config=config,
430	internal_gitlab_user=internal_gitlab_user,
431	)
432
433	self.gl_namespace = gl_namespace	1✔
434	self.project = project	1✔
435	self.branch = branch	1✔
436	self.commit_sha = commit_sha	1✔
437	self.git_host = urlparse(config.git.url).netloc	1✔
438	self.gitlab_project = gitlab_project	1✔
439
440	def _get_start_errors(self) -> list[str]:	2✔
441	"""Check if there are any errors before starting the server."""
442	errors = super()._get_start_errors()	1✔
443	if self.gitlab_project is None:	1✔
444	errors.append(f"project {self.project} does not exist")	×
445	if not self._branch_exists():	1✔
446	errors.append(f"branch {self.branch} does not exist")	×
447	if not self._commit_sha_exists():	1✔
448	errors.append(f"commit {self.commit_sha} does not exist")	×
449	return errors	1✔
450
451	def _branch_exists(self) -> bool:	2✔
452	"""Check if a specific branch exists in the user's gitlab project.
453
454	The branch name is not required by the API and therefore
455	passing None to this function will return True.
456	"""
457	if self.branch is not None and self.gitlab_project is not None:	1✔
458	try:	1✔
459	self.gitlab_project.branches.get(self.branch)	1✔
460	except Exception as err:	×
461	logger.warning(f"Branch {self.branch} cannot be verified or does not exist. {err}")	×
462	else:
463	return True	1✔
464	return False	×
465
466	def _commit_sha_exists(self) -> bool:	2✔
467	"""Check if a specific commit sha exists in the user's gitlab project."""
468	if self.commit_sha is not None and self.gitlab_project is not None:	1✔
469	try:	1✔
470	self.gitlab_project.commits.get(self.commit_sha)	1✔
471	except Exception as err:	×
472	logger.warning(f"Commit {self.commit_sha} cannot be verified or does not exist. {err}")	×
473	else:
474	return True	1✔
475	return False	×
476
477	def get_labels(self) -> dict[str, str \| None]:	2✔
478	"""Get the labels of the jupyter server."""
479	prefix = self._get_renku_annotation_prefix()	1✔
480	labels = super().get_labels()	1✔
481	labels[f"{prefix}commit-sha"] = self.commit_sha	1✔
482	if self.gitlab_project is not None:	1✔
483	labels[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	1✔
484	return labels	1✔
485
486	def get_annotations(self) -> dict[str, str \| None]:	2✔
487	"""Get the annotations of the jupyter server."""
488	prefix = self._get_renku_annotation_prefix()	1✔
489	annotations = super().get_annotations()	1✔
490	annotations[f"{prefix}commit-sha"] = self.commit_sha	1✔
491	annotations[f"{prefix}branch"] = self.branch	1✔
492	annotations[f"{prefix}git-host"] = self.git_host	1✔
493	annotations[f"{prefix}namespace"] = self.gl_namespace	1✔
494	annotations[f"{prefix}projectName"] = self.project	1✔
495	if self.gitlab_project is not None:	1✔
496	annotations[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	1✔
497	annotations[f"{prefix}repository"] = self.gitlab_project.web_url	1✔
498	return annotations	1✔

SwissDataScienceCenter / renku-data-services / 16295569791

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous