Skip to content

Commit 093bc8d

Browse files
jiridanekjiridanek
committed
RHOAIENG-18848: chore(tests/containers): initial test that runs container on openshift/kubernetes
* install local-path provisioner on kubernetes in github actions * more careful printing of pod status in case `containerStatuses == None` * sort out how we want to work with privileged/unprivileged client * only run the new test if we have kubernetes around * add pod waiting and port forwarding utils diff --git c/.github/workflows/build-notebooks-TEMPLATE.yaml i/.github/workflows/build-notebooks-TEMPLATE.yaml index 8a98aa21..13507b78 100644 --- c/.github/workflows/build-notebooks-TEMPLATE.yaml +++ i/.github/workflows/build-notebooks-TEMPLATE.yaml @@ -290,10 +290,10 @@ jobs: - name: Install deps run: poetry install --sync - - name: Run container tests (in PyTest) + - name: Run Testcontainers container tests (in PyTest) run: | set -Eeuxo pipefail - poetry run pytest --capture=fd tests/containers --image="${{ steps.calculated_vars.outputs.OUTPUT_IMAGE }}" + poetry run pytest --capture=fd tests/containers -m 'not openshift' --image="${{ steps.calculated_vars.outputs.OUTPUT_IMAGE }}" env: DOCKER_HOST: "unix:///var/run/podman/podman.sock" TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE: "/var/run/podman/podman.sock" @@ -439,6 +439,16 @@ jobs: kubectl wait deployments --all --all-namespaces --for=condition=Available --timeout=100s kubectl wait pods --all --all-namespaces --for=condition=Ready --timeout=100s + - name: "Install local-path provisioner" + if: ${{ steps.have-tests.outputs.tests == 'true' }} + run: | + set -Eeuxo pipefail + kubectl apply -f https://raw.githubusercontent.com/rancher/local-path-provisioner/v0.0.31/deploy/local-path-storage.yaml + kubectl wait deployments --all --namespace=local-path-storage --for=condition=Available --timeout=100s + # https://kubernetes.io/docs/tasks/administer-cluster/change-default-storage-class/ + kubectl get storageclass + kubectl patch storageclass local-path -p '{"metadata": {"annotations":{"storageclass.kubernetes.io/is-default-class":"true"}}}' + - name: "Run image tests" if: ${{ steps.have-tests.outputs.tests == 'true' }} run: python3 ci/cached-builds/make_test.py --target ${{ inputs.target }} @@ -449,6 +459,18 @@ jobs: # endregion + - name: Run OpenShift container tests (in PyTest) + if: ${{ steps.have-tests.outputs.tests == 'true' }} + run: | + set -Eeuxo pipefail + poetry run pytest --capture=fd tests/containers -m 'openshift' --image="${{ steps.calculated_vars.outputs.OUTPUT_IMAGE }}" + env: + # TODO(jdanek): this Testcontainers stuff should not be necessary but currently it has to be there + DOCKER_HOST: "unix:///var/run/podman/podman.sock" + TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE: "/var/run/podman/podman.sock" + # pulling the Ryuk container from docker.io introduces CI flakiness + TESTCONTAINERS_RYUK_DISABLED: "true" + # region Trivy vulnerability scan - name: Run Trivy vulnerability scanner diff --git c/README.md i/README.md index 961e5907..22703ac0 100644 --- c/README.md +++ i/README.md @@ -105,7 +105,7 @@ sudo dnf install podman systemctl --user start podman.service systemctl --user status podman.service systemctl --user status podman.socket -DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock poetry run pytest tests/containers --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4 +DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock poetry run pytest tests/containers -m 'not openshift' --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4 # Mac OS brew install podman @@ -113,7 +113,7 @@ podman machine init podman machine set --rootful sudo podman-mac-helper install podman machine start -poetry run pytest tests/containers --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4 +poetry run pytest tests/containers -m 'not openshift' --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4 ``` When using lima on macOS, it might be useful to give yourself access to rootful podman socket diff --git c/pyproject.toml i/pyproject.toml index 9271a4c3..6440b123 100644 --- c/pyproject.toml +++ i/pyproject.toml @@ -8,6 +8,7 @@ package-mode = false [tool.poetry.dependencies] python = "~3.12" +requests = "^2.32.3" [tool.poetry.group.dev.dependencies] diff --git c/pytest.ini i/pytest.ini index 2b320d7a..aff25089 100644 --- c/pytest.ini +++ i/pytest.ini @@ -15,3 +15,5 @@ log_cli_level = INFO log_file = logs/pytest-logs.txt log_file_level = DEBUG + +markers = openshift diff --git c/tests/containers/base_image_test.py i/tests/containers/base_image_test.py index 03f3d9ae..b7e00498 100644 --- c/tests/containers/base_image_test.py +++ i/tests/containers/base_image_test.py @@ -11,12 +11,13 @@ import tempfile import textwrap from typing import TYPE_CHECKING, Any, Callable -import pytest import testcontainers.core.container import testcontainers.core.waiting_utils from tests.containers import docker_utils +import pytest + logging.basicConfig(level=logging.DEBUG) LOGGER = logging.getLogger(__name__) @@ -72,7 +73,8 @@ class TestBaseImage: if "not found" in line: unsatisfied_deps.append((dlib, line.strip())) assert output - print("OUTPUT>", json.dumps({"dir": path, "count_scanned": count_scanned, "unsatisfied": unsatisfied_deps})) + print("OUTPUT>", + json.dumps({"dir": path, "count_scanned": count_scanned, "unsatisfied": unsatisfied_deps})) try: container.start() @@ -105,18 +107,7 @@ class TestBaseImage: with subtests.test(f"{dlib=}"): pytest.fail(f"{dlib=} has unsatisfied dependencies {deps=}") - def test_oc_command_runs(self, image: str): - container = testcontainers.core.container.DockerContainer(image=image, user=23456, group_add=[0]) - container.with_command("/bin/sh -c 'sleep infinity'") - try: - container.start() - ecode, output = container.exec(["/bin/sh", "-c", "oc version"]) - finally: - docker_utils.NotebookContainer(container).stop(timeout=0) - - logging.debug(output.decode()) - assert ecode == 0 - + # @pytest.mark.environmentss("docker") def test_oc_command_runs_fake_fips(self, image: str, subtests: pytest_subtests.SubTests): """Establishes a best-effort fake FIPS environment and attempts to execute `oc` binary in it. @@ -190,7 +181,8 @@ class TestBaseImage: docker_utils.NotebookContainer(container).stop(timeout=0) -def encode_python_function_execution_command_interpreter(python: str, function: Callable[..., Any], *args: list[Any]) -> list[str]: +def encode_python_function_execution_command_interpreter(python: str, function: Callable[..., Any], *args: list[Any]) -> \ + list[str]: """Returns a cli command that will run the given Python function encoded inline. All dependencies (imports, ...) must be part of function body.""" code = textwrap.dedent(inspect.getsource(function)) diff --git c/tests/containers/cancellation_token.py i/tests/containers/cancellation_token.py new file mode 100644 index 00000000..d7d62603 --- /dev/null +++ i/tests/containers/cancellation_token.py @@ -0,0 +1,37 @@ +import os +import threading + + +class CancellationToken: + """Flag to signal a thread it should cancel itself. + This cooperative cancellation pattern is commonly used in c# and go + See https://learn.microsoft.com/en-us/dotnet/api/system.threading.cancellationtoken?view=net-9.0 + """ + + def __init__(self): + # consider using the wrapt.synchronized decorator + # https://github.com/GrahamDumpleton/wrapt/blob/develop/blog/07-the-missing-synchronized-decorator.md + self._lock = threading.Lock() + self._canceled = False + # something selectable avoids having to use short timeout in select + self._read_fd, self._write_fd = os.pipe() + + def fileno(self): + """This lets us use the token in select() calls""" + return self._read_fd + + @Property + def cancelled(self): + with self._lock: + return self._canceled + + def cancel(self): + with self._lock: + os.write(self._write_fd, b'x') + self._canceled = True + + def __del__(self): + # consider https://docs.python.org/3/library/weakref.html#weakref.finalize + with self._lock: + os.close(self._read_fd) + os.close(self._write_fd) diff --git c/tests/containers/kubernetes_utils.py i/tests/containers/kubernetes_utils.py new file mode 100644 index 00000000..66829829 --- /dev/null +++ i/tests/containers/kubernetes_utils.py @@ -0,0 +1,473 @@ +from __future__ import annotations + +import contextlib +import functools +import logging +import threading +import time +import traceback +import typing +import socket +from socket import socket +from typing import Any, Callable, Generator + +import requests + +import kubernetes +import kubernetes.dynamic.exceptions +import kubernetes.stream.ws_client +import kubernetes.dynamic.exceptions +import kubernetes.stream.ws_client +import kubernetes.client.api.core_v1_api +from kubernetes.dynamic import DynamicClient, ResourceField + +import ocp_resources.pod +import ocp_resources.deployment +import ocp_resources.service +import ocp_resources.persistent_volume_claim +import ocp_resources.project_request +import ocp_resources.namespace +import ocp_resources.project_project_openshift_io +import ocp_resources.deployment +import ocp_resources.resource +import ocp_resources.pod +import ocp_resources.namespace +import ocp_resources.project_project_openshift_io +import ocp_resources.project_request + +from tests.containers import socket_proxy + + +class TestFrameConstants: + GLOBAL_POLL_INTERVAL_MEDIUM = 10 + TIMEOUT_2MIN = 2 * 60 + + +logging.basicConfig(level=logging.DEBUG) +LOGGER = logging.getLogger(__name__) + + +# https://github.com/RedHatQE/openshift-python-wrapper/tree/main/examples + +def get_client() -> kubernetes.dynamic.DynamicClient: + try: + # client = kubernetes.dynamic.DynamicClient(client=kubernetes.config.new_client_from_config()) + # probably same as above + client = ocp_resources.resource.get_client() + return client + except kubernetes.config.ConfigException as e: + # probably bad config + logging.error(e) + except kubernetes.dynamic.exceptions.UnauthorizedError as e: + # wrong or expired credentials + logging.error(e) + except kubernetes.client.ApiException as e: + # unexpected, we catch unauthorized above + logging.error(e) + except Exception as e: + # unexpected error, assert here + logging.error(e) + + raise RuntimeError("Failed to instantiate client") + + +def get_username(client: kubernetes.dynamic.DynamicClient) -> str: + # can't just access + # > client.configuration.username + # because we normally auth using tokens, not username and password + + # this is what kubectl does (see kubectl -v8 auth whoami) + self_subject_review_resource: kubernetes.dynamic.Resource = client.resources.get( + api_version="authentication.k8s.io/v1", kind="SelfSubjectReview" + ) + self_subject_review: kubernetes.dynamic.ResourceInstance = client.create(self_subject_review_resource) + username: str = self_subject_review.status.userInfo.username + return username + + +class TestKubernetesUtils: + def test_get_username(self): + client = get_client() + username = get_username(client) + assert username is not None and len(username) > 0 + + +class TestFrame: + def __init__[T](self): + self.stack: list[tuple[T, Callable[[T], None] | None]] = [] + + def defer_resource[T: ocp_resources.resource.Resource](self, resource: T, wait=False, + destructor: Callable[[T], None] | None = None) -> T: + result = resource.deploy(wait=wait) + self.defer(resource, destructor) + return result + + def add[T](self, resource: T, destructor: Callable[[T], None] = None) -> T: + self.defer(resource, destructor) + return resource + + def defer[T](self, resource: T, destructor: Callable[[T], None] = None) -> T: + self.stack.append((resource, destructor)) + + def destroy(self, wait=False): + while self.stack: + resource, destructor = self.stack.pop() + if destructor is not None: + destructor(resource) + else: + resource.clean_up(wait=wait) + + def __enter__(self) -> TestFrame: + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.destroy(wait=True) + + +class ImageDeployment: + def __init__(self, client: kubernetes.dynamic.DynamicClient, image: str): + self.client = client + self.image = image + self.tf = TestFrame() + + def __enter__(self) -> ImageDeployment: + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.tf.destroy() + + def deploy(self, container_name: str) -> None: + LOGGER.debug(f"Deploying {self.image}") + # custom namespace is necessary, because we cannot assign a SCC to pods created in one of the default namespaces: + # default, kube-system, kube-public, openshift-node, openshift-infra, openshift. + # https://docs.openshift.com/container-platform/4.17/authentication/managing-security-context-constraints.html#role-based-access-to-ssc_configuring-internal-oauth + + # TODO(jdanek): sort out how we want to work with privileged/unprivileged client + # take inspiration from odh-tests + ns = create_namespace(privileged_client=True, name=f"test-ns-{container_name}") + self.tf.defer_resource(ns) + + pvc = ocp_resources.persistent_volume_claim.PersistentVolumeClaim( + name=container_name, + namespace=ns.name, + accessmodes=ocp_resources.persistent_volume_claim.PersistentVolumeClaim.AccessMode.RWO, + volume_mode=ocp_resources.persistent_volume_claim.PersistentVolumeClaim.VolumeMode.FILE, + size="1Gi", + ) + self.tf.defer_resource(pvc, wait=True) + deployment = ocp_resources.deployment.Deployment( + client=self.client, + name=container_name, + namespace=ns.name, + selector={"matchLabels": {"app": container_name}}, + replicas=1, + template={ + "metadata": { + "annotations": { + # This will result in the container spec having something like below, + # regardless of what kind of namespace this is being run in. + # Keep in mind that `default` is a privileged namespace and this annotation has no effect there. + # ``` + # spec: + # securityContext: + # seLinuxOptions: + # level: 's0:c34,c4' + # fsGroup: 1001130000 + # seccompProfile: + # type: RuntimeDefault + # ``` + "openshift.io/scc": "restricted-v2" + }, + "labels": { + "app": container_name, + } + }, + "spec": { + "containers": [ + { + "name": container_name, + "image": self.image, + # "command": ["/bin/sh", "-c", "while true ; do date; sleep 5; done;"], + "ports": [ + { + "containerPort": 8888, + "name": "notebook-port", + "protocol": "TCP", + } + ], + # rstudio will not start without its volume mount and it does not log the error for it + # See the testcontainers implementation of this (the tty=True part) + "volumeMounts": [ + { + "mountPath": "/opt/app-root/src", + "name": "my-workbench" + } + ], + }, + ], + "volumes": [ + { + "name": "my-workbench", + "persistentVolumeClaim": { + "claimName": container_name, + } + } + ] + } + } + ) + self.tf.defer_resource(deployment) + LOGGER.debug(f"Waiting for pods to become ready...") + PodUtils.wait_for_pods_ready(self.client, namespace_name=ns.name, label_selector=f"app={container_name}", + expect_pods_count=1) + + core_v1_api = kubernetes.client.api.core_v1_api.CoreV1Api(api_client=self.client.client) + pod_name: kubernetes.client.models.v1_pod_list.V1PodList = core_v1_api.list_namespaced_pod( + namespace=ns.name, + label_selector=f"app={container_name}" + ) + assert len(pod_name.items) == 1 + pod: kubernetes.client.models.v1_pod.V1Pod = pod_name.items[0] + + p = socket_proxy.SocketProxy(exposing_contextmanager(core_v1_api, pod), "localhost", 0) + t = threading.Thread(target=p.listen_and_serve_until_canceled) + t.start() + self.tf.defer(t, lambda thread: thread.join()) + self.tf.defer(p.cancellation_token, lambda token: token.cancel()) + + self.port = p.get_actual_port() + LOGGER.debug(f"Listening on port {self.port}") + resp = requests.get(f"http://localhost:{self.port}") + assert resp.status_code == 200 + LOGGER.debug(f"Done with portforward") + + +class PodUtils: + READINESS_TIMEOUT = TestFrameConstants.TIMEOUT_2MIN + + # consider using timeout_sampler + @staticmethod + def wait_for_pods_ready( + client: DynamicClient, namespace_name: str, label_selector: str, expect_pods_count: int + ) -> None: + """Wait for all pods in namespace to be ready + :param client: + :param namespace_name: name of the namespace + :param label_selector: + :param expect_pods_count: + """ + + # it's a dynamic client with the `resource` parameter already filled in + class ResourceType(kubernetes.dynamic.Resource, kubernetes.dynamic.DynamicClient): + pass + + resource: ResourceType = client.resources.get( + kind=ocp_resources.pod.Pod.kind, + api_version=ocp_resources.pod.Pod.api_version, + ) + + def ready() -> bool: + pods = resource.get(namespace=namespace_name, label_selector=label_selector).items + if not pods and expect_pods_count == 0: + logging.debug("All expected Pods %s in Namespace %s are ready", label_selector, namespace_name) + return True + if not pods: + logging.debug("Pods matching %s/%s are not ready", namespace_name, label_selector) + return False + if len(pods) != expect_pods_count: + logging.debug("Expected Pods %s/%s are not ready", namespace_name, label_selector) + return False + pod: ResourceField + for pod in pods: + if not Readiness.is_pod_ready(pod) and not Readiness.is_pod_succeeded(pod): + if not pod.status.containerStatuses: + pod_status = pod.status + else: + pod_status = {cs.name: cs.state for cs in pod.status.containerStatuses} + + logging.debug("Pod is not ready: %s/%s (%s)", + namespace_name, pod.metadata.name, pod_status) + return False + else: + # check all containers in pods are ready + for cs in pod.status.containerStatuses: + if not (cs.ready or cs.state.get("terminated", {}).get("reason", "") == "Completed"): + logging.debug( + f"Container {cs.getName()} of Pod {namespace_name}/{pod.metadata.name} not ready ({cs.state=})" + ) + return False + logging.info("Pods matching %s/%s are ready", namespace_name, label_selector) + return True + + Wait.until( + description=f"readiness of all Pods matching {label_selector} in Namespace {namespace_name}", + poll_interval=TestFrameConstants.GLOBAL_POLL_INTERVAL_MEDIUM, + timeout=PodUtils.READINESS_TIMEOUT, + ready=ready, + ) + + +class Wait: + @staticmethod + def until( + description: str, + poll_interval: float, + timeout: float, + ready: Callable[[], bool], + on_timeout: Callable[[], None] | None = None, + ) -> None: + """For every poll (happening once each {@code pollIntervalMs}) checks if supplier {@code ready} is true. + + If yes, the wait is closed. Otherwise, waits another {@code pollIntervalMs} and tries again. + Once the wait timeout (specified by {@code timeoutMs} is reached and supplier wasn't true until that time, + runs the {@code onTimeout} (f.e. print of logs, showing the actual value that was checked inside {@code ready}), + and finally throws {@link WaitException}. + @param description information about on what we are waiting + @param pollIntervalMs poll interval in milliseconds + @param timeoutMs timeout specified in milliseconds + @param ready {@link BooleanSupplier} containing code, which should be executed each poll, + verifying readiness of the particular thing + @param onTimeout {@link Runnable} executed once timeout is reached and + before the {@link WaitException} is thrown.""" + logging.info("Waiting for: %s", description) + deadline = time.monotonic() + timeout + + exception_message: str | None = None + previous_exception_message: str | None = None + + # in case we are polling every 1s, we want to print exception after x tries, not on the first try + # for minutes poll interval will 2 be enough + exception_appearance_count: int = 2 if (poll_interval // 60) > 0 else max(int(timeout // poll_interval // 4), 2) + exception_count: int = 0 + new_exception_appearance: int = 0 + + stack_trace_error: str | None = None + + while True: + try: + result: bool = ready() + except KeyboardInterrupt: + raise # quick exit if the user gets tired of waiting + except Exception as e: + exception_message = str(e) + + exception_count += 1 + new_exception_appearance += 1 + if ( + exception_count == exception_appearance_count + and exception_message is not None + and exception_message == previous_exception_message + ): + logging.info(f"While waiting for: {description} exception occurred: {exception_message}") + # log the stacktrace + stack_trace_error = traceback.format_exc() + elif ( + exception_message is not None + and exception_message != previous_exception_message + and new_exception_appearance == 2 + ): + previous_exception_message = exception_message + + result = False + + time_left: float = deadline - time.monotonic() + if result: + return + if time_left <= 0: + if exception_count > 1: + logging.error("Exception waiting for: %s, %s", description, exception_message) + + if stack_trace_error is not None: + # printing handled stacktrace + logging.error(stack_trace_error) + if on_timeout is not None: + on_timeout() + wait_exception: WaitException = WaitException(f"Timeout after {timeout} s waiting for {description}") + logging.error(wait_exception) + raise wait_exception + + sleep_time: float = min(poll_interval, time_left) + time.sleep(sleep_time) # noqa: FCN001 + + +class WaitException(Exception): + pass + + +class Readiness: + @staticmethod + def is_pod_ready(pod: ResourceField) -> bool: + Utils.check_not_none(value=pod, message="Pod can't be null.") + + condition = ocp_resources.pod.Pod.Condition.READY + status = ocp_resources.pod.Pod.Condition.Status.TRUE + for cond in pod.get("status", {}).get("conditions", []): + if cond["type"] == condition and cond["status"].casefold() == status.casefold(): + return True + return False + + @staticmethod + def is_pod_succeeded(pod: ResourceField) -> bool: + Utils.check_not_none(value=pod, message="Pod can't be null.") + return pod.status is not None and "Succeeded" == pod.status.phase + + +class Utils: + @staticmethod + def check_not_none(value: Any, message: str) -> None: + if value is None: + raise ValueError(message) + + [email protected] +def exposing_contextmanager( + core_v1_api: kubernetes.client.CoreV1Api, + pod: kubernetes.client.models.V1Pod +) -> Generator[socket, None, None]: + # If we e.g., specify the wrong port, the pf = portforward() call succeeds, + # but pf.connected will later flip to False + # we need to check that _everything_ works before moving on + pf = None + s = None + while not pf or not pf.connected or not s: + pf: kubernetes.stream.ws_client.PortForward = kubernetes.stream.portforward( + api_method=core_v1_api.connect_get_namespaced_pod_portforward, + name=pod.metadata.name, + namespace=pod.metadata.namespace, + ports=",".join(str(p) for p in [8888]), + ) + s: typing.Union[kubernetes.stream.ws_client.PortForward._Port._Socket, socket.socket] | None = pf.socket(8888) + assert s, "Failed to establish connection" + + try: + yield s + finally: + s.close() + pf.close() + + [email protected](ocp_resources.namespace.Namespace.__init__) +def create_namespace(privileged_client: bool = False, *args, + **kwargs) -> ocp_resources.project_project_openshift_io.Project: + if not privileged_client: + with ocp_resources.project_request.ProjectRequest(*args, **kwargs): + project = ocp_resources.project_project_openshift_io.Project(*args, **kwargs) + project.wait_for_status(status=project.Status.ACTIVE, timeout=TestFrameConstants.TIMEOUT_2MIN) + return project + else: + with ocp_resources.namespace.Namespace(*args, **kwargs) as ns: + ns.wait_for_status(status=ocp_resources.namespace.Namespace.Status.ACTIVE, + timeout=TestFrameConstants.TIMEOUT_2MIN) + return ns + + +__all__ = [ + get_client, + get_username, + exposing_contextmanager, + create_namespace, + PodUtils, + TestFrame, + TestFrameConstants, + ImageDeployment, +] diff --git c/tests/containers/socket_proxy.py i/tests/containers/socket_proxy.py new file mode 100644 index 00000000..77fc3485 --- /dev/null +++ i/tests/containers/socket_proxy.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import contextlib +import logging +import socket +import select +import threading +import subprocess +import typing + +from tests.containers.cancellation_token import CancellationToken + +"""Proxies kubernetes portforwards to a local port. + +This is implemented as a thread running select() loop and managing the sockets. + +There are alternative implementations for this. + +1) Run oc port-forward in a subprocess +* There isn't a nice way where kubectl would report in machine-readable way the + port number, kubernetes/kubectl#1190 (comment) +2) Use the socket as is, mount a custom adaptor to the requests library +* The code to do this is weird. This is what docker-py does w.r.t. the docker socket. + It defines a custom 'http+docker://' protocol, and an adaptor for it, that uses the docker socket. +3) Implement proxy using asyncio +* There are advantages to asyncio, but since we don't have Python asyncio anywhere else yet, + it is probably best to avoid using asyncio. + +Out of these, the oc port-forward subprocess is a decent alternative solution. +""" + +class SubprocessProxy: + # + def __init__(self, namespace: str, name: str, port: int): + self.namespace = namespace + self.name = name + self.port = port + + def start(self): + self.forwarder = subprocess.Popen( + ["kubectl", "port-forward", self.namespace, self.name], + text=True, + ) + self.forwarder.communicate() + + def stop(self): + self.forwarder.terminate() + + +class SocketProxy: + def __init__( + self, + remote_socket_factory: typing.ContextManager[socket.socket], + local_host: str = "localhost", + local_port: int = 0, + buffer_size: int = 4096 + ) -> None: + """ + + :param local_host: probably "localhost" would make most sense here + :param local_port: usually leave as to 0, which will make the OS choose a free port + :param remote_socket_factory: this is a context manager for kubernetes port forwarding + :param buffer_size: do not poke it, leave this at the default value + """ + self.local_host = local_host + self.local_port = local_port + self.buffer_size = buffer_size + self.remote_socket_factory = remote_socket_factory + + self.cancellation_token = CancellationToken() + + self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.server_socket.bind((self.local_host, self.local_port)) + self.server_socket.listen(1) + logging.info(f"Proxy listening on {self.local_host}:{self.local_port}") + + def listen_and_serve_until_canceled(self): + """Accepts the client, creates a new socket to the remote, and proxies the data. + + Handles at most one client at a time. """ + try: + while not self.cancellation_token.cancelled: + client_socket, addr = self.server_socket.accept() + logging.info(f"Accepted connection from {addr[0]}:{addr[1]}") + self._handle_client(client_socket) + except Exception as e: + logging.exception(f"Proxying failed to listen", exc_info=e) + raise + finally: + self.server_socket.close() + + def get_actual_port(self) -> int: + """Returns the port that the proxy is listening on. + When port number 0 was passed in, this will return the actual randomly assigned port.""" + return self.server_socket.getsockname()[1] + + def _handle_client(self, client_socket): + with client_socket as _, self.remote_socket_factory as remote_socket: + while True: + readable, _, _ = select.select([client_socket, remote_socket, self.cancellation_token], [], []) + + if self.cancellation_token.cancelled: + break + + if client_socket in readable: + data = client_socket.recv(self.buffer_size) + if not data: + break + remote_socket.send(data) + + if remote_socket in readable: + data = remote_socket.recv(self.buffer_size) + if not data: + break + client_socket.send(data) + + +if __name__ == "__main__": + """Sample application to show how this can work.""" + + + @contextlib.contextmanager + def remote_socket_factory(): + class MockServer(threading.Thread): + def __init__(self, local_host: str = "localhost", local_port: int = 0): + self.local_host = local_host + self.local_port = local_port + + self.is_socket_bound = threading.Event() + + super().__init__() + + def run(self): + self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.server_socket.bind((self.local_host, self.local_port)) + self.server_socket.listen(1) + print(f"MockServer listening on {self.local_host}:{self.local_port}") + self.is_socket_bound.set() + + client_socket, addr = self.server_socket.accept() + logging.info(f"MockServer accepted connection from {addr[0]}:{addr[1]}") + + client_socket.send(b"Hello World\n") + client_socket.close() + + def get_actual_port(self): + self.is_socket_bound.wait() + return self.server_socket.getsockname()[1] + + server = MockServer() + server.start() + + client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + client_socket.connect(("localhost", server.get_actual_port())) + + yield client_socket + + client_socket.close() + server.join() + + + proxy = SocketProxy(remote_socket_factory(), "localhost", 0) + thread = threading.Thread(target=proxy.listen_and_serve_until_canceled) + thread.start() + + client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + client_socket.connect(("localhost", proxy.get_actual_port())) + + print(client_socket.recv(1024)) # prints Hello World + + thread.join() diff --git c/tests/containers/workbenches/workbench_image_test.py i/tests/containers/workbenches/workbench_image_test.py index cbfb7dae..892f775c 100644 --- c/tests/containers/workbenches/workbench_image_test.py +++ i/tests/containers/workbenches/workbench_image_test.py @@ -21,7 +21,7 @@ import testcontainers.core.waiting_utils import pytest import pytest_subtests -from tests.containers import docker_utils, podman_machine_utils +from tests.containers import docker_utils, podman_machine_utils, kubernetes_utils class TestWorkbenchImage: @@ -108,6 +108,18 @@ class TestWorkbenchImage: finally: docker_utils.NotebookContainer(container).stop(timeout=0) + @pytest.mark.openshift + def test_image_run_on_openshift(self, image: str): + skip_if_not_workbench_image(image) + + client = kubernetes_utils.get_client() + print(client) + + username = kubernetes_utils.get_username(client) + print(username) + + with kubernetes_utils.ImageDeployment(client, image) as image: + image.deploy(container_name="notebook-tests-pod") class WorkbenchContainer(testcontainers.core.container.DockerContainer): @functools.wraps(testcontainers.core.container.DockerContainer.__init__)
1 parent 167cb7f commit 093bc8d

10 files changed

+738
-26
lines changed

‎.github/workflows/build-notebooks-TEMPLATE.yaml

+24-2
Original file line numberDiff line numberDiff line change
@@ -290,10 +290,10 @@ jobs:
290290
- name: Install deps
291291
run: poetry install --sync
292292

293-
- name: Run container tests (in PyTest)
293+
- name: Run Testcontainers container tests (in PyTest)
294294
run: |
295295
set -Eeuxo pipefail
296-
poetry run pytest --capture=fd tests/containers --image="${{ steps.calculated_vars.outputs.OUTPUT_IMAGE }}"
296+
poetry run pytest --capture=fd tests/containers -m 'not openshift' --image="${{ steps.calculated_vars.outputs.OUTPUT_IMAGE }}"
297297
env:
298298
DOCKER_HOST: "unix:///var/run/podman/podman.sock"
299299
TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE: "/var/run/podman/podman.sock"
@@ -439,6 +439,16 @@ jobs:
439439
kubectl wait deployments --all --all-namespaces --for=condition=Available --timeout=100s
440440
kubectl wait pods --all --all-namespaces --for=condition=Ready --timeout=100s
441441
442+
- name: "Install local-path provisioner"
443+
if: ${{ steps.have-tests.outputs.tests == 'true' }}
444+
run: |
445+
set -Eeuxo pipefail
446+
kubectl apply -f https://raw.githubusercontent.com/rancher/local-path-provisioner/v0.0.31/deploy/local-path-storage.yaml
447+
kubectl wait deployments --all --namespace=local-path-storage --for=condition=Available --timeout=100s
448+
# https://kubernetes.io/docs/tasks/administer-cluster/change-default-storage-class/
449+
kubectl get storageclass
450+
kubectl patch storageclass local-path -p '{"metadata": {"annotations":{"storageclass.kubernetes.io/is-default-class":"true"}}}'
451+
442452
- name: "Run image tests"
443453
if: ${{ steps.have-tests.outputs.tests == 'true' }}
444454
run: python3 ci/cached-builds/make_test.py --target ${{ inputs.target }}
@@ -449,6 +459,18 @@ jobs:
449459

450460
# endregion
451461

462+
- name: Run OpenShift container tests (in PyTest)
463+
if: ${{ steps.have-tests.outputs.tests == 'true' }}
464+
run: |
465+
set -Eeuxo pipefail
466+
poetry run pytest --capture=fd tests/containers -m 'openshift' --image="${{ steps.calculated_vars.outputs.OUTPUT_IMAGE }}"
467+
env:
468+
# TODO(jdanek): this Testcontainers stuff should not be necessary but currently it has to be there
469+
DOCKER_HOST: "unix:///var/run/podman/podman.sock"
470+
TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE: "/var/run/podman/podman.sock"
471+
# pulling the Ryuk container from docker.io introduces CI flakiness
472+
TESTCONTAINERS_RYUK_DISABLED: "true"
473+
452474
# region Trivy vulnerability scan
453475

454476
- name: Run Trivy vulnerability scanner

‎README.md

+2-2
Original file line numberDiff line numberDiff line change
@@ -105,15 +105,15 @@ sudo dnf install podman
105105
systemctl --user start podman.service
106106
systemctl --user status podman.service
107107
systemctl --user status podman.socket
108-
DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock poetry run pytest tests/containers --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4
108+
DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock poetry run pytest tests/containers -m 'not openshift' --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4
109109
110110
# Mac OS
111111
brew install podman
112112
podman machine init
113113
podman machine set --rootful
114114
sudo podman-mac-helper install
115115
podman machine start
116-
poetry run pytest tests/containers --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4
116+
poetry run pytest tests/containers -m 'not openshift' --image quay.io/opendatahub/workbench-images@sha256:e98d19df346e7abb1fa3053f6d41f0d1fa9bab39e49b4cb90b510ca33452c2e4
117117
```
118118

119119
When using lima on macOS, it might be useful to give yourself access to rootful podman socket

‎poetry.lock

+6-6
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎pyproject.toml

+1
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ package-mode = false
88

99
[tool.poetry.dependencies]
1010
python = "~3.12"
11+
requests = "^2.32.3"
1112

1213

1314
[tool.poetry.group.dev.dependencies]

‎pytest.ini

+2
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,5 @@ log_cli_level = INFO
1515

1616
log_file = logs/pytest-logs.txt
1717
log_file_level = DEBUG
18+
19+
markers = openshift

‎tests/containers/base_image_test.py

+7-15
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,13 @@
1111
import textwrap
1212
from typing import TYPE_CHECKING, Any, Callable
1313

14-
import pytest
1514
import testcontainers.core.container
1615
import testcontainers.core.waiting_utils
1716

1817
from tests.containers import docker_utils
1918

19+
import pytest
20+
2021
logging.basicConfig(level=logging.DEBUG)
2122
LOGGER = logging.getLogger(__name__)
2223

@@ -72,7 +73,8 @@ def check_elf_file():
7273
if "not found" in line:
7374
unsatisfied_deps.append((dlib, line.strip()))
7475
assert output
75-
print("OUTPUT>", json.dumps({"dir": path, "count_scanned": count_scanned, "unsatisfied": unsatisfied_deps}))
76+
print("OUTPUT>",
77+
json.dumps({"dir": path, "count_scanned": count_scanned, "unsatisfied": unsatisfied_deps}))
7678

7779
try:
7880
container.start()
@@ -105,18 +107,7 @@ def check_elf_file():
105107
with subtests.test(f"{dlib=}"):
106108
pytest.fail(f"{dlib=} has unsatisfied dependencies {deps=}")
107109

108-
def test_oc_command_runs(self, image: str):
109-
container = testcontainers.core.container.DockerContainer(image=image, user=23456, group_add=[0])
110-
container.with_command("/bin/sh -c 'sleep infinity'")
111-
try:
112-
container.start()
113-
ecode, output = container.exec(["/bin/sh", "-c", "oc version"])
114-
finally:
115-
docker_utils.NotebookContainer(container).stop(timeout=0)
116-
117-
logging.debug(output.decode())
118-
assert ecode == 0
119-
110+
# @pytest.mark.environmentss("docker")
120111
def test_oc_command_runs_fake_fips(self, image: str, subtests: pytest_subtests.SubTests):
121112
"""Establishes a best-effort fake FIPS environment and attempts to execute `oc` binary in it.
122113
@@ -190,7 +181,8 @@ def test_pip_install_cowsay_runs(self, image: str):
190181
docker_utils.NotebookContainer(container).stop(timeout=0)
191182

192183

193-
def encode_python_function_execution_command_interpreter(python: str, function: Callable[..., Any], *args: list[Any]) -> list[str]:
184+
def encode_python_function_execution_command_interpreter(python: str, function: Callable[..., Any], *args: list[Any]) -> \
185+
list[str]:
194186
"""Returns a cli command that will run the given Python function encoded inline.
195187
All dependencies (imports, ...) must be part of function body."""
196188
code = textwrap.dedent(inspect.getsource(function))

‎tests/containers/cancellation_token.py

+37
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
import os
2+
import threading
3+
4+
5+
class CancellationToken:
6+
"""Flag to signal a thread it should cancel itself.
7+
This cooperative cancellation pattern is commonly used in c# and go
8+
See https://learn.microsoft.com/en-us/dotnet/api/system.threading.cancellationtoken?view=net-9.0
9+
"""
10+
11+
def __init__(self):
12+
# consider using the wrapt.synchronized decorator
13+
# https://github.com/GrahamDumpleton/wrapt/blob/develop/blog/07-the-missing-synchronized-decorator.md
14+
self._lock = threading.Lock()
15+
self._canceled = False
16+
# something selectable avoids having to use short timeout in select
17+
self._read_fd, self._write_fd = os.pipe()
18+
19+
def fileno(self):
20+
"""This lets us use the token in select() calls"""
21+
return self._read_fd
22+
23+
@property
24+
def cancelled(self):
25+
with self._lock:
26+
return self._canceled
27+
28+
def cancel(self):
29+
with self._lock:
30+
os.write(self._write_fd, b'x')
31+
self._canceled = True
32+
33+
def __del__(self):
34+
# consider https://docs.python.org/3/library/weakref.html#weakref.finalize
35+
with self._lock:
36+
os.close(self._read_fd)
37+
os.close(self._write_fd)

0 commit comments

Comments
 (0)