19181077268

Committed 07 Nov 2025 09:00PM UTC coverage: 86.352% (-0.5%) from 86.841%

Build # 19181077268

Build Type

Pull #1059

github

Committed by

web-flow

Commit Message

Merge fb47045e6 into 58a6e4765

Pull Request Pull Request #1059: fix: patching of session custom resources

Run Details

89 of 104 new or added lines in 5 files covered. (85.58%)

577 existing lines in 33 files now uncovered.

22791 of 26393 relevant lines covered (86.35%)

1.52 hits per line

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

92.7

/components/renku_data_services/notebooks/config/dynamic.py

"""Dynamic configuration."""

import json
import os
from dataclasses import dataclass, field
from enum import Enum
from io import StringIO
from typing import Any, ClassVar, Self, Union
from urllib.parse import urlunparse

import yaml

from renku_data_services.notebooks.api.schemas.config_server_options import ServerOptionsChoices, ServerOptionsDefaults
from renku_data_services.notebooks.crs import Affinity, Toleration

latest_version: str = "1.25.3"


def _parse_str_as_bool(val: Union[str, bool]) -> bool:
    if isinstance(val, str):
        return val.lower() == "true"
    elif isinstance(val, bool):
        return val
    else:
        raise ValueError(f"Unsupported data type received, expected str or bool, got {type(val)}")


def _parse_value_as_int(val: Any) -> int:
    # NOTE: That int() does not understand scientific notation
    # even stuff that is "technically" an integer like 3e10, but float does understand it
    return int(float(val))


def _parse_value_as_float(val: Any) -> float:
    return float(val)


class CPUEnforcement(str, Enum):
    """CPU enforcement policies."""

    LAX = "lax"  # CPU limit equals 3x cpu request
    STRICT = "strict"  # CPU limit equals cpu request
    OFF = "off"  # no CPU limit at all


@dataclass
class ServerOptionsConfig:
    """Config class for server options."""

    defaults: dict[str, str | bool | int | float] = field(init=False)
    ui_choices: dict[str, Any] = field(init=False)
    defaults_path: str = "/etc/renku-notebooks/server_options/server_defaults.json"
    ui_choices_path: str = "/etc/renku-notebooks/server_options/server_options.json"

    def __post_init__(self) -> None:
        with open(self.defaults_path) as f:
            self.defaults = ServerOptionsDefaults().loads(f.read())
        with open(self.ui_choices_path) as f:
            self.ui_choices = ServerOptionsChoices().loads(f.read())

    @property
    def lfs_auto_fetch_default(self) -> bool:
        """Whether lfs autofetch is enabled or not."""
        return str(self.defaults.get("lfs_auto_fetch", "false")).lower() == "true"

    @property
    def default_url_default(self) -> str:
        """Default url (path) for session."""
        return str(self.defaults.get("defaultUrl", "/lab"))

    @classmethod
    def from_env(cls) -> Self:
        """Load config from environment variables."""
        return cls(
            os.environ["NB_SERVER_OPTIONS__DEFAULTS_PATH"],
            os.environ["NB_SERVER_OPTIONS__UI_CHOICES_PATH"],
        )


@dataclass
class _SentryConfig:
    enabled: bool = False
    dsn: str | None = field(repr=False, default=None)
    env: str | None = None
    sample_rate: float = 0.2

    @classmethod
    def from_env(cls, prefix: str = "") -> Self:
        return cls(
            _parse_str_as_bool(os.environ.get(f"{prefix}SENTRY__ENABLED", False)),
            os.environ.get(f"{prefix}SENTRY__DSN", None),
            os.environ.get(f"{prefix}SENTRY__ENV", None),
            _parse_value_as_float(os.environ.get(f"{prefix}SENTRY__SAMPLE_RATE", 0.2)),
        )


@dataclass
class _GitConfig:
    url: str
    registry: str

    @classmethod
    def from_env(cls, enable_internal_gitlab: bool = True) -> Self:
        if enable_internal_gitlab:
            return cls(os.environ["NB_GIT__URL"], os.environ["NB_GIT__REGISTRY"])
        return cls("", "")


@dataclass
class _GitProxyConfig:
    renku_client_secret: str = field(repr=False)
    sentry: _SentryConfig = field(default_factory=_SentryConfig.from_env)
    port: int = 65480
    health_port: int = 65481
    image: str = f"renku/git-https-proxy:{latest_version}"
    renku_client_id: str = "renku"

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            renku_client_secret=os.environ["NB_SESSIONS__GIT_PROXY__RENKU_CLIENT_SECRET"],
            renku_client_id=os.environ.get("NB_SESSIONS__GIT_PROXY__RENKU_CLIENT_ID", "renku"),
            sentry=_SentryConfig.from_env(prefix="NB_SESSIONS__GIT_PROXY__"),
            port=_parse_value_as_int(os.environ.get("NB_SESSIONS__GIT_PROXY__PORT", 65480)),
            health_port=_parse_value_as_int(os.environ.get("NB_SESSIONS__GIT_PROXY__HEALTH_PORT", 65481)),
            image=os.environ.get("NB_SESSIONS__GIT_PROXY__IMAGE", f"renku/git-https-proxy:{latest_version}"),
        )


@dataclass
class _GitRpcServerConfig:
    sentry: _SentryConfig = field(default_factory=_SentryConfig.from_env)
    host: str = "0.0.0.0"  # nosec B104
    port: int = 4000
    image: str = f"renku/git-rpc-server:{latest_version}"

    def __post_init__(self) -> None:
        self.port = _parse_value_as_int(self.port)

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            image=os.environ.get("NB_SESSIONS__GIT_RPC_SERVER__IMAGE", f"renku/git-rpc-server:{latest_version}"),
            host=os.environ.get("NB_SESSIONS__GIT_RPC_SERVER__HOST", "0.0.0.0"),  # nosec B104
            port=_parse_value_as_int(os.environ.get("NB_SESSIONS__GIT_RPC_SERVER__PORT", 4000)),
            sentry=_SentryConfig.from_env(prefix="NB_SESSIONS__GIT_RPC_SERVER__"),
        )


@dataclass
class _GitCloneConfig:
    image: str = f"renku/git-clone:{latest_version}"
    sentry: _SentryConfig = field(default_factory=lambda: _SentryConfig(enabled=False))

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            image=os.environ.get("NB_SESSIONS__GIT_CLONE__IMAGE", f"renku/git-clone:{latest_version}"),
            sentry=_SentryConfig.from_env(prefix="NB_SESSIONS__GIT_CLONE__"),
        )


@dataclass
class _SessionStorageConfig:
    pvs_enabled: bool = True
    pvs_storage_class: str | None = None
    use_empty_dir_size_limit: bool = False

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            pvs_enabled=_parse_str_as_bool(os.environ.get("NB_SESSIONS__STORAGE__PVS_ENABLED", True)),
            pvs_storage_class=os.environ.get("NB_SESSIONS__STORAGE__PVS_STORAGE_CLASS"),
            use_empty_dir_size_limit=_parse_str_as_bool(
                os.environ.get("NB_SESSIONS__STORAGE__USE_EMPTY_DIR_SIZE_LIMIT", False)
            ),
        )


@dataclass
class _SessionOidcConfig:
    client_secret: str = field(repr=False)
    token_url: str
    auth_url: str
    issuer_url: str
    client_id: str = "renku-jupyterserver"
    allow_unverified_email: Union[str, bool] = False

    def __post_init__(self) -> None:
        self.allow_unverified_email = _parse_str_as_bool(self.allow_unverified_email)

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            token_url=os.environ["NB_SESSIONS__OIDC__TOKEN_URL"],
            auth_url=os.environ["NB_SESSIONS__OIDC__AUTH_URL"],
            client_secret=os.environ["NB_SESSIONS__OIDC__CLIENT_SECRET"],
            allow_unverified_email=_parse_str_as_bool(
                os.environ.get("NB_SESSIONS__OIDC__ALLOW_UNVERIFIED_EMAIL", False)
            ),
            client_id=os.environ.get("NB_SESSIONS__OIDC__CLIENT_ID", "renku-jupyterserver"),
            issuer_url=os.environ["NB_SESSIONS__OIDC__ISSUER_URL"],
        )


@dataclass
class _CustomCaCertsConfig:
    image: str = "renku/certificates:0.0.2"
    path: str = "/usr/local/share/ca-certificates"
    secrets: list[dict[str, str]] = field(default_factory=list)

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            image=os.environ.get("NB_SESSIONS__CA_CERTS__IMAGE", "renku/certificates:0.0.2"),
            path=os.environ.get("NB_SESSIONS__CA_CERTS__PATH", "/usr/local/share/ca-certificates"),
            secrets=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__CA_CERTS__SECRETS", "[]"))),
        )


@dataclass
class _AmaltheaConfig:
    cache_url: str
    group: str = "amalthea.dev"
    version: str = "v1alpha1"
    plural: str = "jupyterservers"

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            cache_url=os.environ["NB_AMALTHEA__CACHE_URL"],
            group=os.environ.get("NB_AMALTHEA__GROUP", "amalthea.dev"),
            version=os.environ.get("NB_AMALTHEA__VERSION", "v1alpha1"),
            plural=os.environ.get("NB_AMALTHEA__PLURAL", "jupyterservers"),
        )


@dataclass
class _AmaltheaV2Config:
    cache_url: str
    group: str = "amalthea.dev"
    version: str = "v1alpha1"
    plural: str = "amaltheasessions"

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            cache_url=os.environ["NB_AMALTHEA_V2__CACHE_URL"],
            group=os.environ.get("NB_AMALTHEA_V2__GROUP", "amalthea.dev"),
            version=os.environ.get("NB_AMALTHEA_V2__VERSION", "v1alpha1"),
            plural=os.environ.get("NB_AMALTHEA_V2__PLURAL", "amaltheasessions"),
        )


@dataclass
class _SessionIngress:
    host: str
    tls_secret: str | None = None
    class_name: str | None = None
    annotations: dict[str, str] = field(default_factory=dict)

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            host=os.environ["NB_SESSIONS__INGRESS__HOST"],
            tls_secret=os.environ["NB_SESSIONS__INGRESS__TLS_SECRET"],
            class_name=os.environ.get("NB_SESSIONS__INGRESS__CLASS_NAME", None),
            annotations=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__INGRESS__ANNOTATIONS", "{}"))),
        )

    @staticmethod
    def base_path(server_name: str) -> str:
        return f"/sessions/{server_name}"

    def base_url(self, server_name: str, force_https: bool = False) -> str:
        scheme = "https" if self.tls_secret else "http"
        if force_https:
            scheme = "https"
        return str(urlunparse((scheme, self.host, self.base_path(server_name), None, None, None)))


@dataclass
class _GenericCullingConfig:
    idle_seconds: int = 86400
    max_age_seconds: int = 0
    pending_seconds: int = 0
    failed_seconds: int = 0
    hibernated_seconds: int = 86400

    def __post_init__(self) -> None:
        self.idle_seconds = _parse_value_as_int(self.idle_seconds)
        self.max_age_seconds = _parse_value_as_int(self.max_age_seconds)
        self.pending_seconds = _parse_value_as_int(self.pending_seconds)
        self.failed_seconds = _parse_value_as_int(self.failed_seconds)
        self.hibernated_seconds = _parse_value_as_int(self.hibernated_seconds)

    @classmethod
    def from_env(cls, prefix: str = "") -> Self:
        return cls(
            idle_seconds=_parse_value_as_int(os.environ.get(f"{prefix}IDLE_SECONDS", 86400)),
            max_age_seconds=_parse_value_as_int(os.environ.get(f"{prefix}MAX_AGE_SECONDS", 0)),
            pending_seconds=_parse_value_as_int(os.environ.get(f"{prefix}PENDING_SECONDS", 0)),
            failed_seconds=_parse_value_as_int(os.environ.get(f"{prefix}FAILED_SECONDS", 0)),
            hibernated_seconds=_parse_value_as_int(os.environ.get(f"{prefix}HIBERNATED_SECONDS", 86400)),
        )


@dataclass
class _SessionCullingConfig:
    anonymous: _GenericCullingConfig
    registered: _GenericCullingConfig

    def __post_init__(self) -> None:
        # NOTE: We don't allow hibernating anonymous users' sessions. However, when these sessions are
        # culled, they are hibernated automatically by Amalthea. To delete them as quickly as possible
        # after hibernation, we set the threshold to the minimum possible value. Since zero means don't
        # delete, 1 is the minimum threshold value.
        self.anonymous.hibernated_seconds = 1

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            anonymous=_GenericCullingConfig.from_env("NB_SESSIONS__CULLING__ANONYMOUS__"),
            registered=_GenericCullingConfig.from_env("NB_SESSIONS__CULLING__REGISTERED__"),
        )


@dataclass
class _SessionContainers:
    anonymous: list[str] = field(default_factory=list)
    registered: list[str] = field(default_factory=list)
    anonymous_default: ClassVar[list[str]] = [
        "jupyter-server",
        "passthrough-proxy",
        "git-proxy",
    ]
    registered_default: ClassVar[list[str]] = [
        "jupyter-server",
        "oauth2-proxy",
        "git-proxy",
        "git-sidecar",
    ]

    def __post_init__(self) -> None:
        if not self.anonymous:
            self.anonymous = self.anonymous_default
        if not self.registered:
            self.registered = self.registered_default

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            anonymous=json.loads(
                os.environ.get("NB_SESSIONS__ANONYMOUS_CONTAINERS", json.dumps(cls.anonymous_default))
            ),
            registered=json.loads(
                os.environ.get("NB_SESSIONS__REGISTERED_CONTAINERS", json.dumps(cls.registered_default))
            ),
        )


@dataclass
class _SessionSshConfig:
    enabled: bool = False
    service_port: int = 22
    container_port: int = 2022
    host_key_secret: str | None = None
    host_key_location: str = "/opt/ssh/ssh_host_keys"

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            enabled=_parse_str_as_bool(os.environ.get("NB_SESSIONS__SSH__ENABLED", False)),
            service_port=_parse_value_as_int(os.environ.get("NB_SESSIONS__SSH__SERVICE_PORT", 22)),
            container_port=_parse_value_as_int(os.environ.get("NB_SESSIONS__SSH__CONTAINER_PORT", 2022)),
            host_key_secret=os.environ.get("NB_SESSIONS__SSH__HOST_KEY_SECRET"),
            host_key_location=os.environ.get("NB_SESSIONS__SSH__HOST_KEY_LOCATION", "/opt/ssh/ssh_host_keys"),
        )


@dataclass
class _SessionConfig:
    culling: _SessionCullingConfig
    git_proxy: _GitProxyConfig
    git_rpc_server: _GitRpcServerConfig
    git_clone: _GitCloneConfig
    ingress: _SessionIngress
    ca_certs: _CustomCaCertsConfig
    oidc: _SessionOidcConfig
    storage: _SessionStorageConfig
    containers: _SessionContainers
    ssh: _SessionSshConfig
    default_image: str = "renku/singleuser:latest"
    enforce_cpu_limits: CPUEnforcement = CPUEnforcement.OFF
    termination_warning_duration_seconds: int = 12 * 60 * 60
    image_default_workdir: str = "/home/jovyan"
    node_selector: dict[str, str] = field(default_factory=dict)
    affinity: dict[str, Any] = field(default_factory=dict)
    tolerations: list[dict[str, str]] = field(default_factory=list)
    init_containers: list[str] = field(
        default_factory=lambda: [
            "init-certificates",
            "download-image",
            "git-clone",
        ]
    )

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            culling=_SessionCullingConfig.from_env(),
            git_proxy=_GitProxyConfig.from_env(),
            git_rpc_server=_GitRpcServerConfig.from_env(),
            git_clone=_GitCloneConfig.from_env(),
            ingress=_SessionIngress.from_env(),
            ca_certs=_CustomCaCertsConfig.from_env(),
            oidc=_SessionOidcConfig.from_env(),
            storage=_SessionStorageConfig.from_env(),
            containers=_SessionContainers.from_env(),
            ssh=_SessionSshConfig.from_env(),
            default_image=os.environ.get("NB_SESSIONS__DEFAULT_IMAGE", "renku/singleuser:latest"),
            enforce_cpu_limits=CPUEnforcement(os.environ.get("NB_SESSIONS__ENFORCE_CPU_LIMITS", "off")),
            termination_warning_duration_seconds=_parse_value_as_int(os.environ.get("", 12 * 60 * 60)),
            image_default_workdir="/home/jovyan",
            node_selector=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__NODE_SELECTOR", "{}"))),
            affinity=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__AFFINITY", "{}"))),
            tolerations=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__TOLERATIONS", "[]"))),
        )

    @classmethod
    def _for_testing(cls) -> Self:
        return cls(
            culling=_SessionCullingConfig.from_env(),
            git_proxy=_GitProxyConfig(renku_client_secret="not-defined"),  # nosec B106
            git_rpc_server=_GitRpcServerConfig.from_env(),
            git_clone=_GitCloneConfig.from_env(),
            ingress=_SessionIngress(host="localhost", tls_secret="some-secret"),  # nosec: B106
            ca_certs=_CustomCaCertsConfig.from_env(),
            oidc=_SessionOidcConfig(
                client_id="not-defined",
                client_secret="not-defined",  # nosec B106
                token_url="http://not.defined",
                auth_url="http://not.defined",
                issuer_url="http://not.defined",
            ),
            storage=_SessionStorageConfig.from_env(),
            containers=_SessionContainers.from_env(),
            ssh=_SessionSshConfig.from_env(),
            default_image=os.environ.get("", "renku/singleuser:latest"),
            enforce_cpu_limits=CPUEnforcement(os.environ.get("", "off")),
            termination_warning_duration_seconds=_parse_value_as_int(os.environ.get("", 12 * 60 * 60)),
            image_default_workdir="/home/jovyan",
            node_selector=yaml.safe_load(StringIO(os.environ.get("", "{}"))),
            affinity=yaml.safe_load(StringIO(os.environ.get("", "{}"))),
            tolerations=yaml.safe_load(StringIO(os.environ.get("", "[]"))),
        )

    @property
    def affinity_model(self) -> Affinity:
        return Affinity.model_validate(self.affinity)

    @property
    def tolerations_model(self) -> list[Toleration]:
        return [Toleration.model_validate(tol) for tol in self.tolerations]


@dataclass
class _K8sConfig:
    """Defines the k8s client and namespace."""

    renku_namespace: str = "default"

    @classmethod
    def from_env(cls) -> Self:
        return cls(renku_namespace=os.environ.get("KUBERNETES_NAMESPACE", "default"))


@dataclass
class _DynamicConfig:
    server_options: ServerOptionsConfig
    sessions: _SessionConfig
    amalthea: _AmaltheaConfig
    sentry: _SentryConfig
    git: _GitConfig
    anonymous_sessions_enabled: bool = False
    ssh_enabled: bool = False
    service_prefix: str = "/notebooks"
    version: str = "0.0.0"

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            server_options=ServerOptionsConfig.from_env(),
            sessions=_SessionConfig.from_env(),
            amalthea=_AmaltheaConfig.from_env(),
            sentry=_SentryConfig.from_env("NB_SENTRY_"),
            git=_GitConfig.from_env(),
            anonymous_sessions_enabled=_parse_str_as_bool(os.environ.get("NB_ANONYMOUS_SESSIONS_ENABLED", False)),
            ssh_enabled=_parse_str_as_bool(os.environ.get("NB_SESSIONS__SSH__ENABLED", False)),
            service_prefix=os.environ.get("NB_SERVICE_PREFIX", "/notebooks"),
            version=os.environ.get("NB_VERSION", "0.0.0"),
        )


@dataclass
class _CloudStorage:
    enabled: Union[str, bool] = False
    storage_class: str = "csi-rclone"
    mount_folder: str = "/cloudstorage"

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            enabled=_parse_str_as_bool(os.environ.get("NB_CLOUD_STORAGE__ENABLED", False)),
            storage_class=os.environ.get("NB_CLOUD_STORAGE__STORAGE_CLASS", "csi-rclone"),
            mount_folder=os.environ.get("NB_CLOUD_STORAGE__MOUNT_FOLDER", "/cloudstorage"),
        )


@dataclass
class _UserSecrets:
    image: str = f"renku/secrets_mount:{latest_version}"
    secrets_storage_service_url: str = "http://renku-secrets-storage"

    def __post_init__(self) -> None:
        self.secrets_storage_service_url = self.secrets_storage_service_url.rstrip("/")

    @classmethod
    def from_env(cls) -> Self:
        return cls(
            image=os.environ.get("NB_USER_SECRETS__IMAGE", f"renku/secrets_mount:{latest_version}"),
            secrets_storage_service_url=os.environ.get(
                "NB_USER_SECRETS__SECRETS_STORAGE_SERVICE_URL", "http://renku-secrets-storage"
            ),
        )

1	"""Dynamic configuration."""
2
3	import json	2✔
4	import os	2✔
5	from dataclasses import dataclass, field	2✔
6	from enum import Enum	2✔
7	from io import StringIO	2✔
8	from typing import Any, ClassVar, Self, Union	2✔
9	from urllib.parse import urlunparse	2✔
10
11	import yaml	2✔
12
13	from renku_data_services.notebooks.api.schemas.config_server_options import ServerOptionsChoices, ServerOptionsDefaults	2✔
14	from renku_data_services.notebooks.crs import Affinity, Toleration	2✔
15
16	latest_version: str = "1.25.3"	2✔
17
18
19	def _parse_str_as_bool(val: Union[str, bool]) -> bool:	2✔
20	if isinstance(val, str):	2✔
21	return val.lower() == "true"	2✔
22	elif isinstance(val, bool):	2✔
23	return val	2✔
24	else:
25	raise ValueError(f"Unsupported data type received, expected str or bool, got {type(val)}")	×
26
27
28	def _parse_value_as_int(val: Any) -> int:	2✔
29	# NOTE: That int() does not understand scientific notation
30	# even stuff that is "technically" an integer like 3e10, but float does understand it
31	return int(float(val))	2✔
32
33
34	def _parse_value_as_float(val: Any) -> float:	2✔
35	return float(val)	2✔
36
37
38	class CPUEnforcement(str, Enum):	2✔
39	"""CPU enforcement policies."""
40
41	LAX = "lax" # CPU limit equals 3x cpu request	2✔
42	STRICT = "strict" # CPU limit equals cpu request	2✔
43	OFF = "off" # no CPU limit at all	2✔
44
45
46	@dataclass	2✔
47	class ServerOptionsConfig:	2✔
48	"""Config class for server options."""
49
50	defaults: dict[str, str \| bool \| int \| float] = field(init=False)	2✔
51	ui_choices: dict[str, Any] = field(init=False)	2✔
52	defaults_path: str = "/etc/renku-notebooks/server_options/server_defaults.json"	2✔
53	ui_choices_path: str = "/etc/renku-notebooks/server_options/server_options.json"	2✔
54
55	def __post_init__(self) -> None:	2✔
56	with open(self.defaults_path) as f:	2✔
57	self.defaults = ServerOptionsDefaults().loads(f.read())	2✔
58	with open(self.ui_choices_path) as f:	2✔
59	self.ui_choices = ServerOptionsChoices().loads(f.read())	2✔
60
61	@property	2✔
62	def lfs_auto_fetch_default(self) -> bool:	2✔
63	"""Whether lfs autofetch is enabled or not."""
64	return str(self.defaults.get("lfs_auto_fetch", "false")).lower() == "true"	×
65
66	@property	2✔
67	def default_url_default(self) -> str:	2✔
68	"""Default url (path) for session."""
69	return str(self.defaults.get("defaultUrl", "/lab"))	×
70
71	@classmethod	2✔
72	def from_env(cls) -> Self:	2✔
73	"""Load config from environment variables."""
74	return cls(	2✔
75	os.environ["NB_SERVER_OPTIONS__DEFAULTS_PATH"],
76	os.environ["NB_SERVER_OPTIONS__UI_CHOICES_PATH"],
77	)
78
79
80	@dataclass	2✔
81	class _SentryConfig:	2✔
82	enabled: bool = False	2✔
83	dsn: str \| None = field(repr=False, default=None)	2✔
84	env: str \| None = None	2✔
85	sample_rate: float = 0.2	2✔
86
87	@classmethod	2✔
88	def from_env(cls, prefix: str = "") -> Self:	2✔
89	return cls(	2✔
90	_parse_str_as_bool(os.environ.get(f"{prefix}SENTRY__ENABLED", False)),
91	os.environ.get(f"{prefix}SENTRY__DSN", None),
92	os.environ.get(f"{prefix}SENTRY__ENV", None),
93	_parse_value_as_float(os.environ.get(f"{prefix}SENTRY__SAMPLE_RATE", 0.2)),
94	)
95
96
97	@dataclass	2✔
98	class _GitConfig:	2✔
99	url: str	2✔
100	registry: str	2✔
101
102	@classmethod	2✔
103	def from_env(cls, enable_internal_gitlab: bool = True) -> Self:	2✔
104	if enable_internal_gitlab:	×
105	return cls(os.environ["NB_GIT__URL"], os.environ["NB_GIT__REGISTRY"])	×
106	return cls("", "")	×
107
108
109	@dataclass	2✔
110	class _GitProxyConfig:	2✔
111	renku_client_secret: str = field(repr=False)	2✔
112	sentry: _SentryConfig = field(default_factory=_SentryConfig.from_env)	2✔
113	port: int = 65480	2✔
114	health_port: int = 65481	2✔
115	image: str = f"renku/git-https-proxy:{latest_version}"	2✔
116	renku_client_id: str = "renku"	2✔
117
118	@classmethod	2✔
119	def from_env(cls) -> Self:	2✔
120	return cls(	×
121	renku_client_secret=os.environ["NB_SESSIONS__GIT_PROXY__RENKU_CLIENT_SECRET"],
122	renku_client_id=os.environ.get("NB_SESSIONS__GIT_PROXY__RENKU_CLIENT_ID", "renku"),
123	sentry=_SentryConfig.from_env(prefix="NB_SESSIONS__GIT_PROXY__"),
124	port=_parse_value_as_int(os.environ.get("NB_SESSIONS__GIT_PROXY__PORT", 65480)),
125	health_port=_parse_value_as_int(os.environ.get("NB_SESSIONS__GIT_PROXY__HEALTH_PORT", 65481)),
126	image=os.environ.get("NB_SESSIONS__GIT_PROXY__IMAGE", f"renku/git-https-proxy:{latest_version}"),
127	)
128
129
130	@dataclass	2✔
131	class _GitRpcServerConfig:	2✔
132	sentry: _SentryConfig = field(default_factory=_SentryConfig.from_env)	2✔
133	host: str = "0.0.0.0" # nosec B104	2✔
134	port: int = 4000	2✔
135	image: str = f"renku/git-rpc-server:{latest_version}"	2✔
136
137	def __post_init__(self) -> None:	2✔
138	self.port = _parse_value_as_int(self.port)	2✔
139
140	@classmethod	2✔
141	def from_env(cls) -> Self:	2✔
142	return cls(	2✔
143	image=os.environ.get("NB_SESSIONS__GIT_RPC_SERVER__IMAGE", f"renku/git-rpc-server:{latest_version}"),
144	host=os.environ.get("NB_SESSIONS__GIT_RPC_SERVER__HOST", "0.0.0.0"), # nosec B104
145	port=_parse_value_as_int(os.environ.get("NB_SESSIONS__GIT_RPC_SERVER__PORT", 4000)),
146	sentry=_SentryConfig.from_env(prefix="NB_SESSIONS__GIT_RPC_SERVER__"),
147	)
148
149
150	@dataclass	2✔
151	class _GitCloneConfig:	2✔
152	image: str = f"renku/git-clone:{latest_version}"	2✔
153	sentry: _SentryConfig = field(default_factory=lambda: _SentryConfig(enabled=False))	2✔
154
155	@classmethod	2✔
156	def from_env(cls) -> Self:	2✔
157	return cls(	2✔
158	image=os.environ.get("NB_SESSIONS__GIT_CLONE__IMAGE", f"renku/git-clone:{latest_version}"),
159	sentry=_SentryConfig.from_env(prefix="NB_SESSIONS__GIT_CLONE__"),
160	)
161
162
163	@dataclass	2✔
164	class _SessionStorageConfig:	2✔
165	pvs_enabled: bool = True	2✔
166	pvs_storage_class: str \| None = None	2✔
167	use_empty_dir_size_limit: bool = False	2✔
168
169	@classmethod	2✔
170	def from_env(cls) -> Self:	2✔
171	return cls(	2✔
172	pvs_enabled=_parse_str_as_bool(os.environ.get("NB_SESSIONS__STORAGE__PVS_ENABLED", True)),
173	pvs_storage_class=os.environ.get("NB_SESSIONS__STORAGE__PVS_STORAGE_CLASS"),
174	use_empty_dir_size_limit=_parse_str_as_bool(
175	os.environ.get("NB_SESSIONS__STORAGE__USE_EMPTY_DIR_SIZE_LIMIT", False)
176	),
177	)
178
179
180	@dataclass	2✔
181	class _SessionOidcConfig:	2✔
182	client_secret: str = field(repr=False)	2✔
183	token_url: str	2✔
184	auth_url: str	2✔
185	issuer_url: str	2✔
186	client_id: str = "renku-jupyterserver"	2✔
187	allow_unverified_email: Union[str, bool] = False	2✔
188
189	def __post_init__(self) -> None:	2✔
190	self.allow_unverified_email = _parse_str_as_bool(self.allow_unverified_email)	2✔
191
192	@classmethod	2✔
193	def from_env(cls) -> Self:	2✔
194	return cls(	×
195	token_url=os.environ["NB_SESSIONS__OIDC__TOKEN_URL"],
196	auth_url=os.environ["NB_SESSIONS__OIDC__AUTH_URL"],
197	client_secret=os.environ["NB_SESSIONS__OIDC__CLIENT_SECRET"],
198	allow_unverified_email=_parse_str_as_bool(
199	os.environ.get("NB_SESSIONS__OIDC__ALLOW_UNVERIFIED_EMAIL", False)
200	),
201	client_id=os.environ.get("NB_SESSIONS__OIDC__CLIENT_ID", "renku-jupyterserver"),
202	issuer_url=os.environ["NB_SESSIONS__OIDC__ISSUER_URL"],
203	)
204
205
206	@dataclass	2✔
207	class _CustomCaCertsConfig:	2✔
208	image: str = "renku/certificates:0.0.2"	2✔
209	path: str = "/usr/local/share/ca-certificates"	2✔
210	secrets: list[dict[str, str]] = field(default_factory=list)	2✔
211
212	@classmethod	2✔
213	def from_env(cls) -> Self:	2✔
214	return cls(	2✔
215	image=os.environ.get("NB_SESSIONS__CA_CERTS__IMAGE", "renku/certificates:0.0.2"),
216	path=os.environ.get("NB_SESSIONS__CA_CERTS__PATH", "/usr/local/share/ca-certificates"),
217	secrets=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__CA_CERTS__SECRETS", "[]"))),
218	)
219
220
221	@dataclass	2✔
222	class _AmaltheaConfig:	2✔
223	cache_url: str	2✔
224	group: str = "amalthea.dev"	2✔
225	version: str = "v1alpha1"	2✔
226	plural: str = "jupyterservers"	2✔
227
228	@classmethod	2✔
229	def from_env(cls) -> Self:	2✔
230	return cls(	×
231	cache_url=os.environ["NB_AMALTHEA__CACHE_URL"],
232	group=os.environ.get("NB_AMALTHEA__GROUP", "amalthea.dev"),
233	version=os.environ.get("NB_AMALTHEA__VERSION", "v1alpha1"),
234	plural=os.environ.get("NB_AMALTHEA__PLURAL", "jupyterservers"),
235	)
236
237
238	@dataclass	2✔
239	class _AmaltheaV2Config:	2✔
240	cache_url: str	2✔
241	group: str = "amalthea.dev"	2✔
242	version: str = "v1alpha1"	2✔
243	plural: str = "amaltheasessions"	2✔
244
245	@classmethod	2✔
246	def from_env(cls) -> Self:	2✔
247	return cls(	×
248	cache_url=os.environ["NB_AMALTHEA_V2__CACHE_URL"],
249	group=os.environ.get("NB_AMALTHEA_V2__GROUP", "amalthea.dev"),
250	version=os.environ.get("NB_AMALTHEA_V2__VERSION", "v1alpha1"),
251	plural=os.environ.get("NB_AMALTHEA_V2__PLURAL", "amaltheasessions"),
252	)
253
254
255	@dataclass	2✔
256	class _SessionIngress:	2✔
257	host: str	2✔
258	tls_secret: str \| None = None	2✔
259	class_name: str \| None = None	2✔
260	annotations: dict[str, str] = field(default_factory=dict)	2✔
261
262	@classmethod	2✔
263	def from_env(cls) -> Self:	2✔
264	return cls(	×
265	host=os.environ["NB_SESSIONS__INGRESS__HOST"],
266	tls_secret=os.environ["NB_SESSIONS__INGRESS__TLS_SECRET"],
267	class_name=os.environ.get("NB_SESSIONS__INGRESS__CLASS_NAME", None),
268	annotations=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__INGRESS__ANNOTATIONS", "{}"))),
269	)
270
271	@staticmethod	2✔
272	def base_path(server_name: str) -> str:	2✔
UNCOV 273	return f"/sessions/{server_name}"	×
274
275	def base_url(self, server_name: str, force_https: bool = False) -> str:	2✔
276	scheme = "https" if self.tls_secret else "http"	×
277	if force_https:	×
278	scheme = "https"	×
279	return str(urlunparse((scheme, self.host, self.base_path(server_name), None, None, None)))	×
280
281
282	@dataclass	2✔
283	class _GenericCullingConfig:	2✔
284	idle_seconds: int = 86400	2✔
285	max_age_seconds: int = 0	2✔
286	pending_seconds: int = 0	2✔
287	failed_seconds: int = 0	2✔
288	hibernated_seconds: int = 86400	2✔
289
290	def __post_init__(self) -> None:	2✔
291	self.idle_seconds = _parse_value_as_int(self.idle_seconds)	2✔
292	self.max_age_seconds = _parse_value_as_int(self.max_age_seconds)	2✔
293	self.pending_seconds = _parse_value_as_int(self.pending_seconds)	2✔
294	self.failed_seconds = _parse_value_as_int(self.failed_seconds)	2✔
295	self.hibernated_seconds = _parse_value_as_int(self.hibernated_seconds)	2✔
296
297	@classmethod	2✔
298	def from_env(cls, prefix: str = "") -> Self:	2✔
299	return cls(	2✔
300	idle_seconds=_parse_value_as_int(os.environ.get(f"{prefix}IDLE_SECONDS", 86400)),
301	max_age_seconds=_parse_value_as_int(os.environ.get(f"{prefix}MAX_AGE_SECONDS", 0)),
302	pending_seconds=_parse_value_as_int(os.environ.get(f"{prefix}PENDING_SECONDS", 0)),
303	failed_seconds=_parse_value_as_int(os.environ.get(f"{prefix}FAILED_SECONDS", 0)),
304	hibernated_seconds=_parse_value_as_int(os.environ.get(f"{prefix}HIBERNATED_SECONDS", 86400)),
305	)
306
307
308	@dataclass	2✔
309	class _SessionCullingConfig:	2✔
310	anonymous: _GenericCullingConfig	2✔
311	registered: _GenericCullingConfig	2✔
312
313	def __post_init__(self) -> None:	2✔
314	# NOTE: We don't allow hibernating anonymous users' sessions. However, when these sessions are
315	# culled, they are hibernated automatically by Amalthea. To delete them as quickly as possible
316	# after hibernation, we set the threshold to the minimum possible value. Since zero means don't
317	# delete, 1 is the minimum threshold value.
318	self.anonymous.hibernated_seconds = 1	2✔
319
320	@classmethod	2✔
321	def from_env(cls) -> Self:	2✔
322	return cls(	2✔
323	anonymous=_GenericCullingConfig.from_env("NB_SESSIONS__CULLING__ANONYMOUS__"),
324	registered=_GenericCullingConfig.from_env("NB_SESSIONS__CULLING__REGISTERED__"),
325	)
326
327
328	@dataclass	2✔
329	class _SessionContainers:	2✔
330	anonymous: list[str] = field(default_factory=list)	2✔
331	registered: list[str] = field(default_factory=list)	2✔
332	anonymous_default: ClassVar[list[str]] = [	2✔
333	"jupyter-server",
334	"passthrough-proxy",
335	"git-proxy",
336	]
337	registered_default: ClassVar[list[str]] = [	2✔
338	"jupyter-server",
339	"oauth2-proxy",
340	"git-proxy",
341	"git-sidecar",
342	]
343
344	def __post_init__(self) -> None:	2✔
345	if not self.anonymous:	2✔
346	self.anonymous = self.anonymous_default	×
347	if not self.registered:	2✔
348	self.registered = self.registered_default	×
349
350	@classmethod	2✔
351	def from_env(cls) -> Self:	2✔
352	return cls(	2✔
353	anonymous=json.loads(
354	os.environ.get("NB_SESSIONS__ANONYMOUS_CONTAINERS", json.dumps(cls.anonymous_default))
355	),
356	registered=json.loads(
357	os.environ.get("NB_SESSIONS__REGISTERED_CONTAINERS", json.dumps(cls.registered_default))
358	),
359	)
360
361
362	@dataclass	2✔
363	class _SessionSshConfig:	2✔
364	enabled: bool = False	2✔
365	service_port: int = 22	2✔
366	container_port: int = 2022	2✔
367	host_key_secret: str \| None = None	2✔
368	host_key_location: str = "/opt/ssh/ssh_host_keys"	2✔
369
370	@classmethod	2✔
371	def from_env(cls) -> Self:	2✔
372	return cls(	2✔
373	enabled=_parse_str_as_bool(os.environ.get("NB_SESSIONS__SSH__ENABLED", False)),
374	service_port=_parse_value_as_int(os.environ.get("NB_SESSIONS__SSH__SERVICE_PORT", 22)),
375	container_port=_parse_value_as_int(os.environ.get("NB_SESSIONS__SSH__CONTAINER_PORT", 2022)),
376	host_key_secret=os.environ.get("NB_SESSIONS__SSH__HOST_KEY_SECRET"),
377	host_key_location=os.environ.get("NB_SESSIONS__SSH__HOST_KEY_LOCATION", "/opt/ssh/ssh_host_keys"),
378	)
379
380
381	@dataclass	2✔
382	class _SessionConfig:	2✔
383	culling: _SessionCullingConfig	2✔
384	git_proxy: _GitProxyConfig	2✔
385	git_rpc_server: _GitRpcServerConfig	2✔
386	git_clone: _GitCloneConfig	2✔
387	ingress: _SessionIngress	2✔
388	ca_certs: _CustomCaCertsConfig	2✔
389	oidc: _SessionOidcConfig	2✔
390	storage: _SessionStorageConfig	2✔
391	containers: _SessionContainers	2✔
392	ssh: _SessionSshConfig	2✔
393	default_image: str = "renku/singleuser:latest"	2✔
394	enforce_cpu_limits: CPUEnforcement = CPUEnforcement.OFF	2✔
395	termination_warning_duration_seconds: int = 12 * 60 * 60	2✔
396	image_default_workdir: str = "/home/jovyan"	2✔
397	node_selector: dict[str, str] = field(default_factory=dict)	2✔
398	affinity: dict[str, Any] = field(default_factory=dict)	2✔
399	tolerations: list[dict[str, str]] = field(default_factory=list)	2✔
400	init_containers: list[str] = field(	2✔
401	default_factory=lambda: [
402	"init-certificates",
403	"download-image",
404	"git-clone",
405	]
406	)
407
408	@classmethod	2✔
409	def from_env(cls) -> Self:	2✔
410	return cls(	×
411	culling=_SessionCullingConfig.from_env(),
412	git_proxy=_GitProxyConfig.from_env(),
413	git_rpc_server=_GitRpcServerConfig.from_env(),
414	git_clone=_GitCloneConfig.from_env(),
415	ingress=_SessionIngress.from_env(),
416	ca_certs=_CustomCaCertsConfig.from_env(),
417	oidc=_SessionOidcConfig.from_env(),
418	storage=_SessionStorageConfig.from_env(),
419	containers=_SessionContainers.from_env(),
420	ssh=_SessionSshConfig.from_env(),
421	default_image=os.environ.get("NB_SESSIONS__DEFAULT_IMAGE", "renku/singleuser:latest"),
422	enforce_cpu_limits=CPUEnforcement(os.environ.get("NB_SESSIONS__ENFORCE_CPU_LIMITS", "off")),
423	termination_warning_duration_seconds=_parse_value_as_int(os.environ.get("", 12 * 60 * 60)),
424	image_default_workdir="/home/jovyan",
425	node_selector=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__NODE_SELECTOR", "{}"))),
426	affinity=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__AFFINITY", "{}"))),
427	tolerations=yaml.safe_load(StringIO(os.environ.get("NB_SESSIONS__TOLERATIONS", "[]"))),
428	)
429
430	@classmethod	2✔
431	def _for_testing(cls) -> Self:	2✔
432	return cls(	2✔
433	culling=_SessionCullingConfig.from_env(),
434	git_proxy=_GitProxyConfig(renku_client_secret="not-defined"), # nosec B106
435	git_rpc_server=_GitRpcServerConfig.from_env(),
436	git_clone=_GitCloneConfig.from_env(),
437	ingress=_SessionIngress(host="localhost", tls_secret="some-secret"), # nosec: B106
438	ca_certs=_CustomCaCertsConfig.from_env(),
439	oidc=_SessionOidcConfig(
440	client_id="not-defined",
441	client_secret="not-defined", # nosec B106
442	token_url="http://not.defined",
443	auth_url="http://not.defined",
444	issuer_url="http://not.defined",
445	),
446	storage=_SessionStorageConfig.from_env(),
447	containers=_SessionContainers.from_env(),
448	ssh=_SessionSshConfig.from_env(),
449	default_image=os.environ.get("", "renku/singleuser:latest"),
450	enforce_cpu_limits=CPUEnforcement(os.environ.get("", "off")),
451	termination_warning_duration_seconds=_parse_value_as_int(os.environ.get("", 12 * 60 * 60)),
452	image_default_workdir="/home/jovyan",
453	node_selector=yaml.safe_load(StringIO(os.environ.get("", "{}"))),
454	affinity=yaml.safe_load(StringIO(os.environ.get("", "{}"))),
455	tolerations=yaml.safe_load(StringIO(os.environ.get("", "[]"))),
456	)
457
458	@property	2✔
459	def affinity_model(self) -> Affinity:	2✔
460	return Affinity.model_validate(self.affinity)	1✔
461
462	@property	2✔
463	def tolerations_model(self) -> list[Toleration]:	2✔
464	return [Toleration.model_validate(tol) for tol in self.tolerations]	1✔
465
466
467	@dataclass	2✔
468	class _K8sConfig:	2✔
469	"""Defines the k8s client and namespace."""
470
471	renku_namespace: str = "default"	2✔
472
473	@classmethod	2✔
474	def from_env(cls) -> Self:	2✔
475	return cls(renku_namespace=os.environ.get("KUBERNETES_NAMESPACE", "default"))	2✔
476
477
478	@dataclass	2✔
479	class _DynamicConfig:	2✔
480	server_options: ServerOptionsConfig	2✔
481	sessions: _SessionConfig	2✔
482	amalthea: _AmaltheaConfig	2✔
483	sentry: _SentryConfig	2✔
484	git: _GitConfig	2✔
485	anonymous_sessions_enabled: bool = False	2✔
486	ssh_enabled: bool = False	2✔
487	service_prefix: str = "/notebooks"	2✔
488	version: str = "0.0.0"	2✔
489
490	@classmethod	2✔
491	def from_env(cls) -> Self:	2✔
492	return cls(	×
493	server_options=ServerOptionsConfig.from_env(),
494	sessions=_SessionConfig.from_env(),
495	amalthea=_AmaltheaConfig.from_env(),
496	sentry=_SentryConfig.from_env("NB_SENTRY_"),
497	git=_GitConfig.from_env(),
498	anonymous_sessions_enabled=_parse_str_as_bool(os.environ.get("NB_ANONYMOUS_SESSIONS_ENABLED", False)),
499	ssh_enabled=_parse_str_as_bool(os.environ.get("NB_SESSIONS__SSH__ENABLED", False)),
500	service_prefix=os.environ.get("NB_SERVICE_PREFIX", "/notebooks"),
501	version=os.environ.get("NB_VERSION", "0.0.0"),
502	)
503
504
505	@dataclass	2✔
506	class _CloudStorage:	2✔
507	enabled: Union[str, bool] = False	2✔
508	storage_class: str = "csi-rclone"	2✔
509	mount_folder: str = "/cloudstorage"	2✔
510
511	@classmethod	2✔
512	def from_env(cls) -> Self:	2✔
513	return cls(	2✔
514	enabled=_parse_str_as_bool(os.environ.get("NB_CLOUD_STORAGE__ENABLED", False)),
515	storage_class=os.environ.get("NB_CLOUD_STORAGE__STORAGE_CLASS", "csi-rclone"),
516	mount_folder=os.environ.get("NB_CLOUD_STORAGE__MOUNT_FOLDER", "/cloudstorage"),
517	)
518
519
520	@dataclass	2✔
521	class _UserSecrets:	2✔
522	image: str = f"renku/secrets_mount:{latest_version}"	2✔
523	secrets_storage_service_url: str = "http://renku-secrets-storage"	2✔
524
525	def __post_init__(self) -> None:	2✔
526	self.secrets_storage_service_url = self.secrets_storage_service_url.rstrip("/")	2✔
527
528	@classmethod	2✔
529	def from_env(cls) -> Self:	2✔
530	return cls(	2✔
531	image=os.environ.get("NB_USER_SECRETS__IMAGE", f"renku/secrets_mount:{latest_version}"),
532	secrets_storage_service_url=os.environ.get(
533	"NB_USER_SECRETS__SECRETS_STORAGE_SERVICE_URL", "http://renku-secrets-storage"
534	),
535	)

SwissDataScienceCenter / renku-data-services / 19181077268

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