Mlo 133 create admission controller

README.md

@@ -18,6 +18,14 @@ pip install -r requirements-test.txt
 pytest -vv tests
 ```
 
+NOTE:
+if you are running tests on a macOS,
+you should install the coreutils, so tests which are using the `du` cmd will work properly:
+
+```shell
+brew install coreutils
+```
+
 ### Running integration tests locally
 Make sure you have docker installed.
 To build docker image where test run, you should set the following ENV variables:

charts/platform-storage/templates/_helpers.tpl

@@ -126,3 +126,7 @@ app: {{ include "platformStorage.name" . }}
 release: {{ .Release.Name }}
 service: platform-storage-metrics
 {{- end -}}
+
+{{- define "platformStorage.kubeAuthMountRoot" -}}
+{{- printf "/var/run/secrets/kubernetes.io/serviceaccount" -}}
+{{- end -}}

charts/platform-storage/templates/deployment.yaml

@@ -55,6 +55,16 @@ spec:
         {{- end }}
         env:
           {{- include "platformStorage.env" . | nindent 10 }}
+          - name: NP_STORAGE_API_K8S_API_URL
+            value: https://kubernetes.default:443
+          - name: NP_STORAGE_API_K8S_AUTH_TYPE
+            value: token
+          - name: NP_STORAGE_API_K8S_CA_PATH
+            value: {{ include "platformStorage.kubeAuthMountRoot" . }}/ca.crt
+          - name: NP_STORAGE_API_K8S_TOKEN_PATH
+            value: {{ include "platformStorage.kubeAuthMountRoot" . }}/token
+          - name: NP_STORAGE_API_K8S_NS
+            value: {{ .Values.kube.namespace | default "default" | quote }}
         {{- if .Values.storages }}
         volumeMounts:
           {{- include "platformStorage.volumeMounts" . | nindent 10 }}

charts/platform-storage/values.yaml

@@ -38,6 +38,9 @@ service:
 
 storages: []
 
+kube:
+  namespace: "default"
+
 sentry:
   appName: platform-storage
   sampleRate: 0.002

pyproject.toml

@@ -19,7 +19,6 @@ select = [
 ignore = [
   "A003",  # Class attribute "..." is shadowing a Python builtin
   "N818",
-  "PT005"
 ]
 
 [tool.ruff.lint.isort]

setup.cfg

@@ -18,6 +18,7 @@ install_requires =
     aiodns==3.2.0
     aiofiles==24.1.0
     aiohttp==3.10.10
+    apolo-kube-client==25.2.1
     cbor==1.0.0
     cchardet==2.1.7
     fastapi==0.115.8
@@ -43,6 +44,7 @@ console_scripts =
     platform-storage-api = platform_storage_api.api:main
     platform-storage-metrics = platform_storage_api.metrics:main
     platform-storage-worker = platform_storage_api.worker:main
+    platform-storage-admission-controller = platform_storage_api.admission_controller
 
 [options.extras_require]
 dev =
@@ -122,3 +124,6 @@ ignore_missing_imports = true
 
 [mypy-botocore.*]
 ignore_missing_imports = true
+
+[mypy-apolo_kube_client.*]
+ignore_missing_imports = true

src/platform_storage_api/admission_controller/__main__.py

@@ -0,0 +1,30 @@
+import asyncio
+import logging
+
+from aiohttp import web
+from neuro_logging import init_logging, setup_sentry
+
+from platform_storage_api.admission_controller.app import create_app
+from platform_storage_api.config import Config
+
+
+logger = logging.getLogger(__name__)
+
+
+def main() -> None:
+    init_logging()
+    config = Config.from_environ()
+    logging.info("Loaded config: %r", config)
+
+    setup_sentry(
+        health_check_url_path="/api/v1/ping",
+        ignore_errors=[web.HTTPNotFound],
+    )
+
+    loop = asyncio.get_event_loop()
+    app = loop.run_until_complete(create_app(config))
+    web.run_app(app, host=config.server.host, port=config.server.port)
+
+
+if __name__ == "__main__":
+    main()

src/platform_storage_api/admission_controller/api.py

@@ -0,0 +1,155 @@
+import base64
+import dataclasses
+import json
+import logging
+from enum import Enum
+from typing import Any, Optional
+
+from aiohttp import web
+
+from platform_storage_api.admission_controller.app_keys import VOLUME_RESOLVER_KEY
+from platform_storage_api.admission_controller.volume_resolver import (
+    KubeVolumeResolver,
+    VolumeResolverError,
+)
+from platform_storage_api.config import Config
+
+
+logger = logging.getLogger(__name__)
+
+
+LABEL_APOLO_ORG_NAME = "platform.apolo.us/org"
+LABEL_APOLO_PROJECT_NAME = "platform.apolo.us/project"
+LABEL_APOLO_STORAGE_MOUNT_PATH = "platform.apolo.us/storage/mountPath"
+LABEL_APOLO_STORAGE_HOST_PATH = "platform.apolo.us/storage/hostPath"
+
+POD_INJECTED_VOLUME_NAME = "storage-auto-injected-volume"
+
+
+class AdmissionReviewPatchType(str, Enum):
+    JSON = "JSONPatch"
+
+
+class AdmissionControllerApi:
+    def __init__(self, app: web.Application, config: Config) -> None:
+        self._app = app
+        self._config = config
+
+    @property
+    def _volume_resolver(self) -> KubeVolumeResolver:
+        return self._app[VOLUME_RESOLVER_KEY]
+
+    def register(self, app: web.Application) -> None:
+        app.add_routes([web.post("/mutate", self.handle_post_mutate)])
+
+    async def handle_post_mutate(self, request: web.Request) -> Any:
+        payload: dict[str, Any] = await request.json()
+
+        uid = payload["request"]["uid"]
+        response = AdmissionReviewResponse(uid=uid)
+
+        obj = payload["request"]["object"]
+        kind = obj.get("kind")
+
+        if kind != "Pod":
+            # not a pod creation request. early-exit
+            return web.json_response(response.to_dict())
+
+        metadata = obj.get("metadata", {})
+        annotations = metadata.get("annotations", {})
+
+        mount_path_value = annotations.get(LABEL_APOLO_STORAGE_MOUNT_PATH)
+        host_path_value = annotations.get(LABEL_APOLO_STORAGE_HOST_PATH)
+
+        if not (mount_path_value and host_path_value):
+            # a pod does not request storage. we can do early-exit here
+            return web.json_response(response.to_dict())
+
+        pod_spec = obj["spec"]
+        containers = pod_spec.get("containers") or []
+
+        if not containers:
+            # pod does not define any containers. we can exit
+            return web.json_response(response.to_dict())
+
+        # now let's try to resolve a path which POD wants to mount
+        try:
+            volume_spec = await self._volume_resolver.resolve_to_mount_volume(
+                path=host_path_value
+            )
+        except VolumeResolverError:
+            # report an error and disallow spawning a POD
+            logger.exception("unable to resolve a volume for a provided path")
+            response.allowed = False
+            return web.json_response(response.to_dict())
+
+        # ensure volumes
+        if "volumes" not in pod_spec:
+            response.add_patch(
+                path="/spec/volumes",
+                value=[]
+            )
+
+        # add a volume host path
+        response.add_patch(
+            path="/spec/volumes/-",
+            value={
+                "name": POD_INJECTED_VOLUME_NAME,
+                **volume_spec.to_kube(),
+            }
+        )
+
+        # add a volumeMount with mount path for all the POD containers
+        for idx, container in enumerate(containers):
+            if "volumeMounts" not in container:
+                response.add_patch(
+                    path=f"/spec/containers/{idx}/volumeMounts",
+                    value=[]
+                )
+
+            response.add_patch(
+                path=f"/spec/containers/{idx}/volumeMounts/-",
+                value={
+                    "name": POD_INJECTED_VOLUME_NAME,
+                    "mountPath": mount_path_value,
+                }
+            )
+
+        return web.json_response(response.to_dict())
+
+
+@dataclasses.dataclass
+class AdmissionReviewResponse:
+    uid: str
+    allowed: bool = True
+    patch: Optional[list[dict[str, Any]]] = None
+    patch_type: AdmissionReviewPatchType = AdmissionReviewPatchType.JSON
+
+    def add_patch(self, path: str, value: Any) -> None:
+        if self.patch is None:
+            self.patch = []
+
+        self.patch.append({
+            "op": "add",
+            "path": path,
+            "value": value,
+        })
+
+    def to_dict(self) -> dict[str, Any]:
+        patch: Optional[str] = None
+
+        if self.patch is not None:
+            # convert patch changes to a b64
+            dumped = json.dumps(self.patch).encode()
+            patch = base64.b64encode(dumped).decode()
+
+        return {
+            "apiVersion": "admission.k8s.io/v1",
+            "kind": "AdmissionReview",
+            "response": {
+                "uid": self.uid,
+                "allowed": self.allowed,
+                "patch": patch,
+                "patchType": AdmissionReviewPatchType.JSON.value
+            }
+        }

src/platform_storage_api/admission_controller/app.py

@@ -0,0 +1,69 @@
+import logging
+from collections.abc import AsyncIterator
+from contextlib import AsyncExitStack
+from typing import cast
+
+from aiohttp import web
+
+from platform_storage_api.admission_controller.api import AdmissionControllerApi
+from platform_storage_api.admission_controller.app_keys import (
+    API_V1_KEY,
+    VOLUME_RESOLVER_KEY,
+)
+from platform_storage_api.admission_controller.volume_resolver import (
+    KubeVolumeResolver,
+    create_kube_client,
+)
+from platform_storage_api.api import ApiHandler, handle_exceptions
+from platform_storage_api.config import Config, KubeConfig
+from platform_storage_api.fs.local import LocalFileSystem
+from platform_storage_api.storage import create_path_resolver
+
+
+logger = logging.getLogger(__name__)
+
+
+async def create_app(config: Config) -> web.Application:
+    app = web.Application(
+        middlewares=[handle_exceptions],
+        handler_args={"keepalive_timeout": config.server.keep_alive_timeout_s},
+    )
+
+    async def _init_app(app: web.Application) -> AsyncIterator[None]:
+        async with AsyncExitStack() as exit_stack:
+            fs = await exit_stack.enter_async_context(
+                LocalFileSystem(
+                    executor_max_workers=config.storage.fs_local_thread_pool_size
+                )
+            )
+            path_resolver = create_path_resolver(config, fs)
+
+            kube_config = cast(KubeConfig, config.kube)
+            kube_client = await exit_stack.enter_async_context(
+                create_kube_client(kube_config)
+            )
+            volume_resolver = await exit_stack.enter_async_context(
+                KubeVolumeResolver(
+                    kube_client=kube_client,
+                    path_resolver=path_resolver,
+                )
+            )
+            app[API_V1_KEY][VOLUME_RESOLVER_KEY] = volume_resolver
+
+            yield
+
+    app.cleanup_ctx.append(_init_app)
+
+    api_v1_app = web.Application()
+    api_v1_handler = ApiHandler()
+    api_v1_handler.register(api_v1_app)
+    app[API_V1_KEY] = api_v1_app
+
+    admission_controller_app = web.Application()
+    admission_controller_api = AdmissionControllerApi(api_v1_app, config)
+    admission_controller_api.register(admission_controller_app)
+
+    api_v1_app.add_subapp("/admission-controller", admission_controller_app)
+    app.add_subapp("/api/v1", api_v1_app)
+
+    return app

src/platform_storage_api/admission_controller/app_keys.py

@@ -0,0 +1,7 @@
+from aiohttp import web
+
+from platform_storage_api.admission_controller.volume_resolver import KubeVolumeResolver
+
+
+API_V1_KEY = web.AppKey("api_v1", web.Application)
+VOLUME_RESOLVER_KEY = web.AppKey("volume_resolver", KubeVolumeResolver)