17325685128

Committed 29 Aug 2025 01:52PM UTC coverage: 86.988% (-0.2%) from 87.181%

Build # 17325685128

Build Type

push

github

Committed by

web-flow

Commit Message

feat: external resources (#950)

* Add support for secrets to an external Kubernetes cluster
* Extend cluster settings

---------

Co-authored-by: Tasko Olevski <olevski90@gmail.com>

Run Details

511 of 671 new or added lines in 35 files covered. (76.15%)

24 existing lines in 9 files now uncovered.

21761 of 25016 relevant lines covered (86.99%)

1.53 hits per line

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

87.14

/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.k8s.constants import DEFAULT_K8S_CLUSTER
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)

        if cluster.id != DEFAULT_K8S_CLUSTER:
            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)
        else:
            # 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),
                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.k8s.constants import DEFAULT_K8S_CLUSTER	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
239	if cluster.id != DEFAULT_K8S_CLUSTER:	1✔
NEW 240	cluster_settings = await self.config.cluster_rp.select(cluster.id)	×
NEW 241	(	×
242	base_server_path,
243	_,
244	_,
245	host,
246	tls_secret,
247	ingress_annotations,
248	) = cluster_settings.get_ingress_parameters(self.server_name)
249	else:
250	# Fallback to global, main cluster parameters
251	host = self.config.sessions.ingress.host	1✔
252	base_server_path = self.config.sessions.ingress.base_path(self.server_name)	1✔
253	ingress_annotations = self.config.sessions.ingress.annotations	1✔
254
255	if self.config.sessions.ingress.tls_secret is not None:	1✔
NEW 256	tls_name = self.config.sessions.ingress.tls_secret	×
257	else:
258	tls_name = None	1✔
259	tls_secret = None if tls_name is None else TlsSecret(adopt=False, name=tls_name)	1✔
260
261	# Combine everything into the manifest
262	manifest = {	1✔
263	"apiVersion": JUPYTER_SESSION_GVK.group_version,
264	"kind": JUPYTER_SESSION_GVK.kind,
265	"metadata": {
266	"name": self.server_name,
267	"labels": self.get_labels(),
268	"annotations": self.get_annotations(),
269	},
270	"spec": {
271	"auth": session_auth,
272	"culling": {
273	"idleSecondsThreshold": self.idle_seconds_threshold,
274	"maxAgeSecondsThreshold": (
275	self.config.sessions.culling.registered.max_age_seconds
276	if isinstance(self._user, AuthenticatedAPIUser)
277	else self.config.sessions.culling.anonymous.max_age_seconds
278	),
279	"hibernatedSecondsThreshold": self.hibernated_seconds_threshold,
280	},
281	"jupyterServer": {
282	"defaultUrl": self.server_options.default_url,
283	"image": self.image,
284	"rootDir": self.work_dir.as_posix(),
285	"resources": self.server_options.to_k8s_resources(
286	enforce_cpu_limits=self.config.sessions.enforce_cpu_limits
287	),
288	},
289	"routing": {
290	"host": host,
291	"path": base_server_path,
292	"ingressAnnotations": ingress_annotations,
293	"tls": {
294	"enabled": tls_secret is not None,
295	"secretName": tls_secret.name if tls_secret is not None else "",
296	},
297	},
298	"storage": storage,
299	"patches": patches,
300	},
301	}
302	return manifest	1✔
303
304	def _get_renku_annotation_prefix(self) -> str:	2✔
305	return self.config.session_get_endpoint_annotations.renku_annotation_prefix	1✔
306
307	async def _get_patches(self) -> list[dict[str, Any]]:	2✔
308	return list(	1✔
309	chain(
310	general_patches.test(self),
311	general_patches.session_tolerations(self),
312	general_patches.session_affinity(self),
313	general_patches.session_node_selector(self),
314	general_patches.priority_class(self),
315	general_patches.dev_shm(self),
316	jupyter_server_patches.args(),
317	jupyter_server_patches.env(self),
318	jupyter_server_patches.image_pull_secret(self, self.internal_gitlab_user.access_token),
319	jupyter_server_patches.disable_service_links(),
320	jupyter_server_patches.rstudio_env_variables(self),
321	await git_proxy_patches.main(self),
322	await git_sidecar_patches.main(self),
323	general_patches.oidc_unverified_email(self),
324	ssh_patches.main(self.config),
325	# init container for certs must come before all other init containers
326	# so that it runs first before all other init containers
327	init_containers_patches.certificates(self.config),
328	init_containers_patches.download_image(self),
329	await init_containers_patches.git_clone(self),
330	inject_certificates_patches.proxy(self),
331	# Cloud Storage needs to patch the git clone sidecar spec and so should come after
332	# the sidecars
333	# WARN: this patch depends on the index of the sidecar and so needs to be updated
334	# if sidercars are added or removed
335	await cloudstorage_patches.main(self),
336	# NOTE: User secrets adds an init container, volume and mounts, so it may affect
337	# indices in other patches.
338	jupyter_server_patches.user_secrets(self),
339	)
340	)
341
342	def get_labels(self) -> dict[str, str \| None]:	2✔
343	"""Get the labels for the session."""
344	prefix = self._get_renku_annotation_prefix()	1✔
345	labels = {	1✔
346	"app": "jupyter",
347	"component": "singleuser-server",
348	f"{prefix}commit-sha": None,
349	f"{prefix}gitlabProjectId": None,
350	f"{prefix}safe-username": self.safe_username,
351	f"{prefix}quota": self.server_options.priority_class
352	if self.server_options.priority_class is not None
353	else "",
354	f"{prefix}userId": self._user.id,
355	}
356	return labels	1✔
357
358	def get_annotations(self) -> dict[str, str \| None]:	2✔
359	"""Get the annotations for the session."""
360	prefix = self._get_renku_annotation_prefix()	1✔
361	username = self._user.id	1✔
362	if isinstance(self.user, AuthenticatedAPIUser) and self._user.email:	1✔
363	username = self._user.email	×
364	annotations = {	1✔
365	f"{prefix}commit-sha": None,
366	f"{prefix}gitlabProjectId": None,
367	f"{prefix}safe-username": self._user.id,
368	f"{prefix}username": username,
369	f"{prefix}userId": self._user.id,
370	f"{prefix}servername": self.server_name,
371	f"{prefix}branch": None,
372	f"{prefix}git-host": None,
373	f"{prefix}namespace": None,
374	f"{prefix}projectName": None,
375	f"{prefix}requested-image": self.image,
376	f"{prefix}repository": None,
377	f"{prefix}hibernation": "",
378	f"{prefix}hibernationBranch": "",
379	f"{prefix}hibernationCommitSha": "",
380	f"{prefix}hibernationDirty": "",
381	f"{prefix}hibernationSynchronized": "",
382	f"{prefix}hibernationDate": "",
383	f"{prefix}hibernatedSecondsThreshold": str(self.hibernated_seconds_threshold),
384	f"{prefix}lastActivityDate": "",
385	f"{prefix}idleSecondsThreshold": str(self.idle_seconds_threshold),
386	}
387	if self.server_options.resource_class_id:	1✔
388	annotations[f"{prefix}resourceClassId"] = str(self.server_options.resource_class_id)	1✔
389	return annotations	1✔
390
391
392	class Renku1UserServer(UserServer):	2✔
393	"""Represents a Renku 1.0 server session."""
394
395	def __init__(	2✔
396	self,
397	user: AnonymousAPIUser \| AuthenticatedAPIUser,
398	server_name: str,
399	gl_namespace: str,
400	project: str,
401	branch: str,
402	commit_sha: str,
403	image: str \| None,
404	server_options: ServerOptions,
405	environment_variables: dict[str, str],
406	user_secrets: K8sUserSecrets \| None,
407	cloudstorage: Sequence[ICloudStorageRequest],
408	k8s_client: NotebookK8sClient,
409	workspace_mount_path: PurePosixPath,
410	work_dir: PurePosixPath,
411	config: NotebooksConfig,
412	host: str,
413	namespace: str,
414	gitlab_project: Project \| None,
415	internal_gitlab_user: APIUser,
416	using_default_image: bool = False,
417	is_image_private: bool = False,
418	**_: dict, # Required to ignore unused arguments, among which repositories
419	):
420	repositories = [	1✔
421	Repository(
422	url=p.http_url_to_repo,
423	dirname=p.path,
424	branch=branch,
425	commit_sha=commit_sha,
426	)
427	for p in [gitlab_project]
428	if p is not None
429	]
430
431	super().__init__(	1✔
432	user=user,
433	server_name=server_name,
434	image=image,
435	server_options=server_options,
436	environment_variables=environment_variables,
437	user_secrets=user_secrets,
438	cloudstorage=cloudstorage,
439	k8s_client=k8s_client,
440	workspace_mount_path=workspace_mount_path,
441	work_dir=work_dir,
442	using_default_image=using_default_image,
443	is_image_private=is_image_private,
444	repositories=repositories,
445	host=host,
446	namespace=namespace,
447	config=config,
448	internal_gitlab_user=internal_gitlab_user,
449	)
450
451	self.gl_namespace = gl_namespace	1✔
452	self.project = project	1✔
453	self.branch = branch	1✔
454	self.commit_sha = commit_sha	1✔
455	self.git_host = urlparse(config.git.url).netloc	1✔
456	self.gitlab_project = gitlab_project	1✔
457
458	def _get_start_errors(self) -> list[str]:	2✔
459	"""Check if there are any errors before starting the server."""
460	errors = super()._get_start_errors()	1✔
461	if self.gitlab_project is None:	1✔
462	errors.append(f"project {self.project} does not exist")	×
463	if not self._branch_exists():	1✔
464	errors.append(f"branch {self.branch} does not exist")	×
465	if not self._commit_sha_exists():	1✔
466	errors.append(f"commit {self.commit_sha} does not exist")	×
467	return errors	1✔
468
469	def _branch_exists(self) -> bool:	2✔
470	"""Check if a specific branch exists in the user's gitlab project.
471
472	The branch name is not required by the API and therefore
473	passing None to this function will return True.
474	"""
475	if self.branch is not None and self.gitlab_project is not None:	1✔
476	try:	1✔
477	self.gitlab_project.branches.get(self.branch)	1✔
478	except Exception as err:	×
479	logger.warning(f"Branch {self.branch} cannot be verified or does not exist. {err}")	×
480	else:
481	return True	1✔
482	return False	×
483
484	def _commit_sha_exists(self) -> bool:	2✔
485	"""Check if a specific commit sha exists in the user's gitlab project."""
486	if self.commit_sha is not None and self.gitlab_project is not None:	1✔
487	try:	1✔
488	self.gitlab_project.commits.get(self.commit_sha)	1✔
489	except Exception as err:	×
490	logger.warning(f"Commit {self.commit_sha} cannot be verified or does not exist. {err}")	×
491	else:
492	return True	1✔
493	return False	×
494
495	def get_labels(self) -> dict[str, str \| None]:	2✔
496	"""Get the labels of the jupyter server."""
497	prefix = self._get_renku_annotation_prefix()	1✔
498	labels = super().get_labels()	1✔
499	labels[f"{prefix}commit-sha"] = self.commit_sha	1✔
500	if self.gitlab_project is not None:	1✔
501	labels[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	1✔
502	return labels	1✔
503
504	def get_annotations(self) -> dict[str, str \| None]:	2✔
505	"""Get the annotations of the jupyter server."""
506	prefix = self._get_renku_annotation_prefix()	1✔
507	annotations = super().get_annotations()	1✔
508	annotations[f"{prefix}commit-sha"] = self.commit_sha	1✔
509	annotations[f"{prefix}branch"] = self.branch	1✔
510	annotations[f"{prefix}git-host"] = self.git_host	1✔
511	annotations[f"{prefix}namespace"] = self.gl_namespace	1✔
512	annotations[f"{prefix}projectName"] = self.project	1✔
513	if self.gitlab_project is not None:	1✔
514	annotations[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	1✔
515	annotations[f"{prefix}repository"] = self.gitlab_project.web_url	1✔
516	return annotations	1✔

SwissDataScienceCenter / renku-data-services / 17325685128

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