18305748766

Committed 07 Oct 2025 07:46AM UTC coverage: 83.391% (-3.4%) from 86.821%

Build # 18305748766

Build Type

Pull #1015

github

Committed by

web-flow

Commit Message

Merge 8299447b8 into c4e28e8fb

Pull Request Pull Request #1015: fix: loading kube configs

Run Details

24 of 71 new or added lines in 8 files covered. (33.8%)

893 existing lines in 41 files now uncovered.

21494 of 25775 relevant lines covered (83.39%)

1.49 hits per line

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

26.32

/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 GitProviderHelperProto, 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,
        git_provider_helper: GitProviderHelperProto,
        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.git_provider_helper = git_provider_helper
        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.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

            tls_name = self.config.sessions.ingress.tls_secret
            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 sidecars 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,
        git_provider_helper: GitProviderHelperProto,
        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,
            git_provider_helper=git_provider_helper,
            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 GitProviderHelperProto, 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	git_provider_helper: GitProviderHelperProto,
58	using_default_image: bool = False,
59	is_image_private: bool = False,
60	repositories: list[Repository] \| None = None,
61	):
UNCOV 62	self._user = user	×
UNCOV 63	self.server_name = server_name	×
UNCOV 64	self._k8s_client = k8s_client	×
UNCOV 65	self.safe_username = self._user.id	×
UNCOV 66	self.image = image	×
UNCOV 67	self.server_options = server_options	×
UNCOV 68	self.environment_variables = environment_variables	×
UNCOV 69	self.user_secrets = user_secrets	×
UNCOV 70	self.using_default_image = using_default_image	×
UNCOV 71	self.workspace_mount_path = workspace_mount_path	×
UNCOV 72	self.work_dir = work_dir	×
UNCOV 73	self.cloudstorage = cloudstorage	×
UNCOV 74	self.is_image_private = is_image_private	×
UNCOV 75	self.host = host	×
UNCOV 76	self.__namespace = namespace	×
UNCOV 77	self.config = config	×
UNCOV 78	self.git_provider_helper = git_provider_helper	×
UNCOV 79	self.internal_gitlab_user = internal_gitlab_user	×
80
UNCOV 81	if self.server_options.idle_threshold_seconds is not None:	×
82	self.idle_seconds_threshold = self.server_options.idle_threshold_seconds	×
83	else:
UNCOV 84	self.idle_seconds_threshold = (	×
85	config.sessions.culling.registered.idle_seconds
86	if isinstance(self._user, AuthenticatedAPIUser)
87	else config.sessions.culling.anonymous.idle_seconds
88	)
89
UNCOV 90	if self.server_options.hibernation_threshold_seconds is not None:	×
91	self.hibernated_seconds_threshold: int = self.server_options.hibernation_threshold_seconds	×
92	else:
UNCOV 93	self.hibernated_seconds_threshold = (	×
94	config.sessions.culling.registered.hibernated_seconds
95	if isinstance(user, AuthenticatedAPIUser)
96	else config.sessions.culling.anonymous.hibernated_seconds
97	)
UNCOV 98	self._repositories: list[Repository] = repositories or []	×
UNCOV 99	self._git_providers: list[GitProvider] \| None = None	×
UNCOV 100	self._has_configured_git_providers = False	×
101
UNCOV 102	self.server_url = f"https://{self.host}/sessions/{self.server_name}"	×
UNCOV 103	if not self._user.is_authenticated:	×
104	self.server_url = f"{self.server_url}?token={self._user.id}"	×
105
106	def k8s_namespace(self) -> str:	2✔
107	"""Get the preferred namespace for a server."""
UNCOV 108	return self.__namespace	×
109
110	@property	2✔
111	def user(self) -> AnonymousAPIUser \| AuthenticatedAPIUser:	2✔
112	"""Getter for server's user."""
UNCOV 113	return self._user	×
114
115	async def repositories(self) -> list[Repository]:	2✔
116	"""Get the list of repositories in the project."""
117	# Configure git repository providers based on matching URLs.
UNCOV 118	if not self._has_configured_git_providers:	×
UNCOV 119	git_providers = await self.git_providers()	×
UNCOV 120	for repo in self._repositories:	×
UNCOV 121	found_provider = None	×
UNCOV 122	for provider in git_providers:	×
123	if urlparse(provider.url).netloc == urlparse(repo.url).netloc:	×
124	found_provider = provider	×
125	break	×
UNCOV 126	if found_provider is not None:	×
127	repo.provider = found_provider.id	×
UNCOV 128	self._has_configured_git_providers = True	×
129
UNCOV 130	return self._repositories	×
131
132	async def git_providers(self) -> list[GitProvider]:	2✔
133	"""The list of git providers."""
UNCOV 134	if self._git_providers is None:	×
UNCOV 135	self._git_providers = await self.git_provider_helper.get_providers(user=self.user)	×
UNCOV 136	return self._git_providers	×
137
138	async def required_git_providers(self) -> list[GitProvider]:	2✔
139	"""The list of required git providers."""
UNCOV 140	repositories = await self.repositories()	×
UNCOV 141	required_provider_ids: set[str] = set(r.provider for r in repositories if r.provider)	×
UNCOV 142	providers = await self.git_providers()	×
UNCOV 143	return [p for p in providers if p.id in required_provider_ids]	×
144
145	def __str__(self) -> str:	2✔
146	return f"<UserServer user: {self._user.id} server_name: {self.server_name}>"	×
147
148	async def start(self) -> JupyterServerV1Alpha1 \| None:	2✔
149	"""Create the jupyterserver resource in k8s."""
UNCOV 150	errors = self._get_start_errors()	×
UNCOV 151	if errors:	×
152	raise MissingResourceError(	×
153	message=(
154	"Cannot start the session because the following Git "
155	f"or Docker resources are missing: {', '.join(errors)}"
156	)
157	)
UNCOV 158	session_manifest = await self._get_session_manifest()	×
UNCOV 159	manifest = JupyterServerV1Alpha1.model_validate(session_manifest)	×
UNCOV 160	return await self._k8s_client.create_session(manifest, self.user)	×
161
162	@staticmethod	2✔
163	def _check_environment_variables_overrides(patches_list: list[dict[str, Any]]) -> None:	2✔
164	"""Check if any patch overrides server's environment variables.
165
166	Checks if it overrides with a different value or if two patches create environment variables with different
167	values.
168	"""
UNCOV 169	env_vars: dict[tuple[str, str], str] = {}	×
170
UNCOV 171	for patch_list in patches_list:	×
UNCOV 172	patches = patch_list["patch"]	×
173
UNCOV 174	for patch in patches:	×
UNCOV 175	path = str(patch["path"]).lower()	×
UNCOV 176	if path.endswith("/env/-"):	×
UNCOV 177	name = str(patch["value"]["name"])	×
UNCOV 178	value = str(patch["value"]["value"])	×
UNCOV 179	key = (path, name)	×
180
UNCOV 181	if key in env_vars and env_vars[key] != value:	×
182	raise DuplicateEnvironmentVariableError(	×
183	message=f"Environment variable {path}::{name} is being overridden by multiple patches"
184	)
185	else:
UNCOV 186	env_vars[key] = value	×
187
188	def _get_start_errors(self) -> list[str]:	2✔
189	"""Check if there are any errors before starting the server."""
UNCOV 190	errors: list[str] = []	×
UNCOV 191	if self.image is None:	×
192	errors.append(f"image {self.image} does not exist or cannot be accessed")	×
UNCOV 193	return errors	×
194
195	async def _get_session_manifest(self) -> dict[str, Any]:	2✔
196	"""Compose the body of the user session for the k8s operator."""
UNCOV 197	patches = await self._get_patches()	×
UNCOV 198	self._check_environment_variables_overrides(patches)	×
199
200	# Storage
UNCOV 201	if self.config.sessions.storage.pvs_enabled:	×
UNCOV 202	storage: dict[str, Any] = {	×
203	"size": self.server_options.storage,
204	"pvc": {
205	"enabled": True,
206	# We should check against the cluster, but as this is only used by V1 sessions, we ignore this
207	# use-case.
208	"storageClassName": self.config.sessions.storage.pvs_storage_class,
209	"mountPath": self.workspace_mount_path.as_posix(),
210	},
211	}
212	else:
213	storage_size = self.server_options.storage if self.config.sessions.storage.use_empty_dir_size_limit else ""	×
214	storage = {	×
215	"size": storage_size,
216	"pvc": {
217	"enabled": False,
218	"mountPath": self.workspace_mount_path.as_posix(),
219	},
220	}
221	# Authentication
UNCOV 222	if isinstance(self._user, AuthenticatedAPIUser):	×
223	session_auth = {	×
224	"token": "",
225	"oidc": {
226	"enabled": True,
227	"clientId": self.config.sessions.oidc.client_id,
228	"clientSecret": {"value": self.config.sessions.oidc.client_secret},
229	"issuerUrl": self.config.sessions.oidc.issuer_url,
230	"authorizedEmails": [self._user.email],
231	},
232	}
233	else:
UNCOV 234	session_auth = {	×
235	"token": self._user.id,
236	"oidc": {"enabled": False},
237	}
238
UNCOV 239	cluster = await self.config.k8s_client.cluster_by_class_id(self.server_options.resource_class_id, self._user)	×
240
UNCOV 241	if cluster.id != DEFAULT_K8S_CLUSTER:	×
242	cluster_settings = await self.config.cluster_rp.select(cluster.id)	×
243	(	×
244	base_server_path,
245	_,
246	_,
247	host,
248	tls_secret,
249	ingress_annotations,
250	) = cluster_settings.get_ingress_parameters(self.server_name)
251	else:
252	# Fallback to global, main cluster parameters
UNCOV 253	host = self.config.sessions.ingress.host	×
UNCOV 254	base_server_path = self.config.sessions.ingress.base_path(self.server_name)	×
UNCOV 255	ingress_annotations = self.config.sessions.ingress.annotations	×
256
NEW 257	tls_name = self.config.sessions.ingress.tls_secret	×
258	tls_secret = None if tls_name is None else TlsSecret(adopt=False, name=tls_name)	×
259
260	# Combine everything into the manifest
UNCOV 261	manifest = {	×
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	}
UNCOV 301	return manifest	×
302
303	def _get_renku_annotation_prefix(self) -> str:	2✔
UNCOV 304	return self.config.session_get_endpoint_annotations.renku_annotation_prefix	×
305
306	async def _get_patches(self) -> list[dict[str, Any]]:	2✔
UNCOV 307	return list(	×
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	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 sidecars 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."""
UNCOV 343	prefix = self._get_renku_annotation_prefix()	×
UNCOV 344	labels = {	×
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	}
UNCOV 355	return labels	×
356
357	def get_annotations(self) -> dict[str, str \| None]:	2✔
358	"""Get the annotations for the session."""
UNCOV 359	prefix = self._get_renku_annotation_prefix()	×
UNCOV 360	username = self._user.id	×
UNCOV 361	if isinstance(self.user, AuthenticatedAPIUser) and self._user.email:	×
362	username = self._user.email	×
UNCOV 363	annotations = {	×
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	}
UNCOV 386	if self.server_options.resource_class_id:	×
UNCOV 387	annotations[f"{prefix}resourceClassId"] = str(self.server_options.resource_class_id)	×
UNCOV 388	return annotations	×
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	git_provider_helper: GitProviderHelperProto,
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	):
UNCOV 420	repositories = [	×
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
UNCOV 431	super().__init__(	×
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	git_provider_helper=git_provider_helper,
443	using_default_image=using_default_image,
444	is_image_private=is_image_private,
445	repositories=repositories,
446	host=host,
447	namespace=namespace,
448	config=config,
449	internal_gitlab_user=internal_gitlab_user,
450	)
451
UNCOV 452	self.gl_namespace = gl_namespace	×
UNCOV 453	self.project = project	×
UNCOV 454	self.branch = branch	×
UNCOV 455	self.commit_sha = commit_sha	×
UNCOV 456	self.git_host = urlparse(config.git.url).netloc	×
UNCOV 457	self.gitlab_project = gitlab_project	×
458
459	def _get_start_errors(self) -> list[str]:	2✔
460	"""Check if there are any errors before starting the server."""
UNCOV 461	errors = super()._get_start_errors()	×
UNCOV 462	if self.gitlab_project is None:	×
463	errors.append(f"project {self.project} does not exist")	×
UNCOV 464	if not self._branch_exists():	×
465	errors.append(f"branch {self.branch} does not exist")	×
UNCOV 466	if not self._commit_sha_exists():	×
467	errors.append(f"commit {self.commit_sha} does not exist")	×
UNCOV 468	return errors	×
469
470	def _branch_exists(self) -> bool:	2✔
471	"""Check if a specific branch exists in the user's gitlab project.
472
473	The branch name is not required by the API and therefore
474	passing None to this function will return True.
475	"""
UNCOV 476	if self.branch is not None and self.gitlab_project is not None:	×
UNCOV 477	try:	×
UNCOV 478	self.gitlab_project.branches.get(self.branch)	×
479	except Exception as err:	×
480	logger.warning(f"Branch {self.branch} cannot be verified or does not exist. {err}")	×
481	else:
UNCOV 482	return True	×
483	return False	×
484
485	def _commit_sha_exists(self) -> bool:	2✔
486	"""Check if a specific commit sha exists in the user's gitlab project."""
UNCOV 487	if self.commit_sha is not None and self.gitlab_project is not None:	×
UNCOV 488	try:	×
UNCOV 489	self.gitlab_project.commits.get(self.commit_sha)	×
490	except Exception as err:	×
491	logger.warning(f"Commit {self.commit_sha} cannot be verified or does not exist. {err}")	×
492	else:
UNCOV 493	return True	×
494	return False	×
495
496	def get_labels(self) -> dict[str, str \| None]:	2✔
497	"""Get the labels of the jupyter server."""
UNCOV 498	prefix = self._get_renku_annotation_prefix()	×
UNCOV 499	labels = super().get_labels()	×
UNCOV 500	labels[f"{prefix}commit-sha"] = self.commit_sha	×
UNCOV 501	if self.gitlab_project is not None:	×
UNCOV 502	labels[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	×
UNCOV 503	return labels	×
504
505	def get_annotations(self) -> dict[str, str \| None]:	2✔
506	"""Get the annotations of the jupyter server."""
UNCOV 507	prefix = self._get_renku_annotation_prefix()	×
UNCOV 508	annotations = super().get_annotations()	×
UNCOV 509	annotations[f"{prefix}commit-sha"] = self.commit_sha	×
UNCOV 510	annotations[f"{prefix}branch"] = self.branch	×
UNCOV 511	annotations[f"{prefix}git-host"] = self.git_host	×
UNCOV 512	annotations[f"{prefix}namespace"] = self.gl_namespace	×
UNCOV 513	annotations[f"{prefix}projectName"] = self.project	×
UNCOV 514	if self.gitlab_project is not None:	×
UNCOV 515	annotations[f"{prefix}gitlabProjectId"] = str(self.gitlab_project.id)	×
UNCOV 516	annotations[f"{prefix}repository"] = self.gitlab_project.web_url	×
UNCOV 517	return annotations	×

SwissDataScienceCenter / renku-data-services / 18305748766

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