17241336915

Committed 26 Aug 2025 02:30PM UTC coverage: 87.027% (-0.2%) from 87.181%

Build # 17241336915

Build Type

Pull #950

github

Committed by

web-flow

Commit Message

Merge 5c1c82e69 into 86bb5f056

Pull Request Pull Request #950: Build/external resources demo

Run Details

513 of 670 new or added lines in 36 files covered. (76.57%)

23 existing lines in 8 files now uncovered.

21769 of 25014 relevant lines covered (87.03%)

1.53 hits per line

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

87.68

/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.errors import errors
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.cr_amalthea_session import TlsSecret
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,
        namespace: 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.__namespace = namespace
        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.__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)
        try:
            cluster_settings = await self.config.cluster_rp.select(cluster.id)
            (
                base_server_path,
                _,
                _,
                host,
                tls_secret,
                ingress_annotations,
            ) = cluster_settings.get_ingress_parameters(self.server_name)
        except errors.MissingResourceError:
            # Fallback to global, main cluster parameters
            host = self.config.sessions.ingress.host
            base_server_path = self.config.sessions.ingress.base_path(self.server_name)
            ingress_annotations = self.config.sessions.ingress.annotations

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

SwissDataScienceCenter / renku-data-services / 17241336915

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