chore: re-enable notebooks tests and use kind (#1070)

Author: olevski

.devcontainer/devcontainer.json

@@ -29,7 +29,9 @@
 		"ghcr.io/EliiseS/devcontainer-features/bash-profile:1": {
 			"command": "alias k=kubectl"
 		},
-		"./k3d": {},
+		"ghcr.io/devcontainers-extra/features/kind:1": {
+			"version": "v0.30.0"
+		},
 		"ghcr.io/devcontainers/features/go:1": {},
 		"./rclone": {
 			"rclone_repository": "https://github.com/SwissDataScienceCenter/rclone.git",

.devcontainer/k3d/devcontainer-feature.json

@@ -84,6 +84,7 @@ result
 
 # tests
 .k3d-config.yaml
+.kind-kubeconfig.yaml
 
 # Misc
 *.pem

.devcontainer/k3d/install.sh

@@ -77,7 +77,7 @@ test_setup:  ## Prep for the tests - removes old coverage reports if one is pres
 main_tests:  ## Run the main (i.e. non-schemathesis tests)
 	DUMMY_STORES=true poetry run alembic --name common upgrade heads
 	poetry run alembic --name common check
-	poetry run pytest -m "not schemathesis" -n auto -v
+	poetry run pytest -m "not schemathesis" -n auto -v --dist loadgroup
 
 .PHONY: schemathesis_tests
 schemathesis_tests:  ## Run schemathesis checks

.gitignore

@@ -67,7 +67,7 @@ class KubeConfigEnv(KubeConfig):
     """Get a kube config from the environment."""
 
     def __init__(self) -> None:
-        super().__init__(ns=os.environ.get("K8S_NAMESPACE", "default"))
+        super().__init__(ns=os.environ.get("K8S_NAMESPACE", "default"), kubeconfig=os.environ.get("KUBECONFIG"))
 
 
 async def from_kubeconfig_file(kubeconfig_path: str) -> KubeConfig:

Makefile

@@ -189,7 +189,7 @@ async def collect_metrics(
         await metrics.session_stopped(user=user, metadata={"session_id": new_obj.meta.name})
         return
     previous_state = previous_obj.manifest.get("status", {}).get("state", None) if previous_obj else None
-    match new_obj.obj.status.state:
+    match new_obj.obj.status.get("state"):
         case State.Running.value if previous_state is None or previous_state == State.NotReady.value:
             # session starting
             resource_class_id = int(new_obj.obj.metadata.annotations.get("renku.io/resource_class_id"))

components/renku_data_services/k8s/config.py

@@ -1,6 +1,10 @@
+import asyncio
+import contextlib
 import json
 from collections.abc import AsyncGenerator, Callable
 from copy import deepcopy
+from datetime import timedelta
+from pathlib import Path
 from typing import Any, Protocol
 
 import pytest
@@ -19,9 +23,14 @@
 from renku_data_services.data_api.app import register_all_handlers
 from renku_data_services.data_api.dependencies import DependencyManager
 from renku_data_services.data_connectors.apispec import DataConnector as ApiDataConnector
+from renku_data_services.k8s.clients import K8sClusterClient
+from renku_data_services.k8s.config import from_kubeconfig_file, get_clusters
+from renku_data_services.k8s.constants import ClusterId
+from renku_data_services.k8s.watcher import K8sWatcher, k8s_object_handler
 from renku_data_services.migrations.core import run_migrations_for_app
 from renku_data_services.namespace.apispec import GroupResponse as ApiGroup
 from renku_data_services.namespace.models import UserNamespace
+from renku_data_services.notebooks.constants import JUPYTER_SESSION_GVK
 from renku_data_services.project.apispec import Project as ApiProject
 from renku_data_services.search.apispec import SearchResult
 from renku_data_services.secrets_storage_api.app import register_all_handlers as register_secrets_handlers
@@ -33,6 +42,7 @@
 from renku_data_services.users.dummy_kc_api import DummyKeycloakAPI
 from renku_data_services.users.models import UserInfo
 from renku_data_services.utils.middleware import validate_null_byte
+from test.bases.renku_data_services.data_api.utils import KindCluster, setup_amalthea
 from test.bases.renku_data_services.data_tasks.test_sync import get_kc_users
 from test.utils import SanicReusableASGITestClient, TestDependencyManager
 
@@ -727,3 +737,52 @@ def __make_headers(user: UserInfo, admin: bool = False) -> dict[str, str]:
         }
     )
     return {"Authorization": f"Bearer {access_token}"}
+
+
+@pytest.fixture(scope="session")
+def cluster_name():
+    return f"k8s-cluster-{str(ULID()).lower()}"
+
+
+@pytest.fixture(scope="session")
+def kubeconfig_path(monkeysession):
+    kconf = ".kind-kubeconfig.yaml"
+    monkeysession.setenv("KUBECONFIG", kconf)
+    return Path(kconf)
+
+
+@pytest.fixture(scope="session")
+def cluster(cluster_name, kubeconfig_path):
+    with KindCluster(cluster_name, kubeconfig=str(kubeconfig_path)) as cluster:
+        yield cluster
+
+
+@pytest.fixture(scope="session")
+def amalthea_installation(cluster):
+    setup_amalthea("amalthea", "amalthea", "0.22.0", cluster)
+
+
+@pytest_asyncio.fixture
+async def jupyter_server_k8s_watcher(cluster, amalthea_installation, app_manager_instance):
+    app_manager = app_manager_instance
+    default_kubeconfig = await from_kubeconfig_file(cluster.kubeconfig)
+    clusters: dict[ClusterId, K8sClusterClient] = {}
+    async for client in get_clusters(
+        kube_conf_root_dir=app_manager.config.k8s_config_root,
+        default_kubeconfig=default_kubeconfig,
+        cluster_repo=app_manager.cluster_repo,
+    ):
+        clusters[client.get_cluster().id] = client
+
+    # sleep to give amalthea a chance to create the CRDs, otherwise the watcher can error out
+    await asyncio.sleep(1)
+    watcher = K8sWatcher(
+        handler=k8s_object_handler(app_manager.config.nb_config.k8s_db_cache, app_manager.metrics, app_manager.rp_repo),
+        clusters=clusters,
+        kinds=[JUPYTER_SESSION_GVK],
+        db_cache=app_manager.config.nb_config.k8s_db_cache,
+    )
+    asyncio.create_task(watcher.start())
+    yield
+    with contextlib.suppress(TimeoutError):
+        await watcher.stop(timeout=timedelta(seconds=1))