kunwp1 commented on code in PR #6866: URL: https://github.com/apache/texera/pull/6866#discussion_r3810195859
########## bin/mounter/mounter.py: ########## @@ -0,0 +1,402 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +texera-mounter: a per-node privileged service that performs GeeseFS FUSE mounts on behalf +of (unprivileged) computing-unit pods. + +A CU pod POSTs /mount with {cuid, repositoryName, commitHash, jwt, fileServiceBase}. The +mounter runs GeeseFS against file-service's JWT-authenticated S3 proxy (passing the pod's +JWT as the S3 access key) and mounts read-only under MOUNT_ROOT/<cuid>/<repo>/<commit>. +That host directory is bind-mounted (mountPropagation: Bidirectional) into the mounter and +propagates back into the CU pod (mountPropagation: HostToContainer), so the CU pod sees +the mount without any privilege of its own. + +The mounter holds no LakeFS credentials; authorization stays entirely in file-service. +A background watcher unmounts a CU's directories as soon as its pod is deleted. +""" + +import json +import os +import shutil +import ssl +import subprocess +import threading +import time +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +MOUNT_ROOT = os.environ.get("MOUNT_ROOT", "/var/lib/texera-mounts") +MOUNTER_PORT = int(os.environ.get("MOUNTER_PORT", "8100")) +POOL_NAMESPACE = os.environ.get("POOL_NAMESPACE", "texera-workflow-computing-unit-pool") +MOUNT_SECRET_PLACEHOLDER = "texera-jwt-mount" +MOUNT_TIMEOUT_S = 30 +# How long the API server keeps a watch open. Each expiry re-runs the reconcile, which is +# the safety net for events missed while disconnected and for orphans still busy earlier. +WATCH_TIMEOUT_S = 300 +WATCH_RETRY_S = 10 + +SA_DIR = "/var/run/secrets/kubernetes.io/serviceaccount" +# Computing-unit pods are named "<prefix>-<cuid>"; the helm chart gives this and the +# computing-unit manager's kubernetes.compute-unit-pod-name-prefix the same value. +CU_POD_NAME_PREFIX = os.environ.get("CU_POD_NAME_PREFIX", "computing-unit") +PROC_MOUNTS = "/proc/mounts" + + +def log(msg): + print(f"[mounter] {msg}", flush=True) + + +def _unescape(field): + """Decode the four characters /proc/mounts escapes octally.""" + for code, char in (("\\040", " "), ("\\011", "\t"), ("\\012", "\n"), ("\\134", "\\")): + field = field.replace(code, char) + return field + + +def mount_targets_under(path): + """Mount targets at or under `path`, deepest first. + + Read from /proc/mounts rather than using os.path.ismount(): when a FUSE server dies + (the mounter being restarted kills every GeeseFS it started) its mount entry survives + here, but stat()ing the mount point fails with ENOTCONN, which os.path.ismount() + reports as "not a mount point". Such a dead mount would then never be unmounted, and + its directory could never be removed. + """ + path = os.path.normpath(path) + prefix = path + "/" + targets = [] + try: + with open(PROC_MOUNTS) as mounts: + for line in mounts: + fields = line.split() + if len(fields) < 2: + continue + target = _unescape(fields[1]) + if target == path or target.startswith(prefix): + targets.append(target) + except OSError as e: + log(f"reading {PROC_MOUNTS} failed: {e}") + # Deepest first, so nested mounts are detached before their parents. + return sorted(targets, key=len, reverse=True) + + +def is_mounted(path): + """True if `path` itself is a mount point, alive or dead.""" + return os.path.normpath(path) in mount_targets_under(path) + + +def _responds(path): + """True if `path` can be stat()ed — false for a mount whose FUSE server is gone.""" + try: + os.stat(path) + return True + except OSError: + return False + + +def _remove_empty_dirs(path, cuid): + """Remove `path` and any parents left empty, stopping at the CU's own directory.""" + stop_at = os.path.normpath(os.path.join(MOUNT_ROOT, cuid)) + path = os.path.normpath(path) + while path.startswith(stop_at): + try: + os.rmdir(path) + except OSError: + return # not empty (another commit is mounted here) or already gone + path = os.path.dirname(path) + + +def ensure_shared_root(): + """Make MOUNT_ROOT a shared mount so mounts created under it propagate to peers.""" + os.makedirs(MOUNT_ROOT, exist_ok=True) + if not os.path.ismount(MOUNT_ROOT): + subprocess.run(["mount", "--bind", MOUNT_ROOT, MOUNT_ROOT], check=False) + subprocess.run(["mount", "--make-rshared", MOUNT_ROOT], check=False) + + +def do_mount(cuid, repo, commit, jwt, file_service_base): + """Idempotently mount repo:commit for cuid. Returns the mount target path.""" + if not cuid or not repo or not commit or not jwt or not file_service_base: + raise ValueError("cuid, repositoryName, commitHash, jwt and fileServiceBase are required") + + target = os.path.join(MOUNT_ROOT, cuid, repo, commit) + if is_mounted(target): Review Comment: We had a discussion that `repo` and `commit` are validated by LakeFS. But I see nobody validates `cuid`. So a request can still name someone else's `cuid`. For example, a legitimate user running CU `5` can `POST /mount` with `cuid=8` and a repo their own JWT can read -> LakeFS authorizes it -> the mount lands in `MOUNT_ROOT/8/...`, and mount propagation pushes it into CU `8`'s pod -> `GET /mounts?cuid=8` similarly enumerates another CU's mounts. Also, a `cuid` like `../..` can flow straight into `os.path.join`/`os.makedirs`. I suggest deriving `cuid` from the authenticated caller rather than the request body and also reject `cuid` values that aren't a single path segment. ########## bin/mounter/mounter.py: ########## @@ -0,0 +1,402 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +texera-mounter: a per-node privileged service that performs GeeseFS FUSE mounts on behalf +of (unprivileged) computing-unit pods. + +A CU pod POSTs /mount with {cuid, repositoryName, commitHash, jwt, fileServiceBase}. The +mounter runs GeeseFS against file-service's JWT-authenticated S3 proxy (passing the pod's +JWT as the S3 access key) and mounts read-only under MOUNT_ROOT/<cuid>/<repo>/<commit>. +That host directory is bind-mounted (mountPropagation: Bidirectional) into the mounter and +propagates back into the CU pod (mountPropagation: HostToContainer), so the CU pod sees +the mount without any privilege of its own. + +The mounter holds no LakeFS credentials; authorization stays entirely in file-service. +A background watcher unmounts a CU's directories as soon as its pod is deleted. +""" + +import json +import os +import shutil +import ssl +import subprocess +import threading +import time +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +MOUNT_ROOT = os.environ.get("MOUNT_ROOT", "/var/lib/texera-mounts") +MOUNTER_PORT = int(os.environ.get("MOUNTER_PORT", "8100")) +POOL_NAMESPACE = os.environ.get("POOL_NAMESPACE", "texera-workflow-computing-unit-pool") +MOUNT_SECRET_PLACEHOLDER = "texera-jwt-mount" +MOUNT_TIMEOUT_S = 30 +# How long the API server keeps a watch open. Each expiry re-runs the reconcile, which is +# the safety net for events missed while disconnected and for orphans still busy earlier. +WATCH_TIMEOUT_S = 300 +WATCH_RETRY_S = 10 + +SA_DIR = "/var/run/secrets/kubernetes.io/serviceaccount" +# Computing-unit pods are named "<prefix>-<cuid>"; the helm chart gives this and the +# computing-unit manager's kubernetes.compute-unit-pod-name-prefix the same value. +CU_POD_NAME_PREFIX = os.environ.get("CU_POD_NAME_PREFIX", "computing-unit") +PROC_MOUNTS = "/proc/mounts" + + +def log(msg): + print(f"[mounter] {msg}", flush=True) + + +def _unescape(field): + """Decode the four characters /proc/mounts escapes octally.""" + for code, char in (("\\040", " "), ("\\011", "\t"), ("\\012", "\n"), ("\\134", "\\")): + field = field.replace(code, char) + return field + + +def mount_targets_under(path): + """Mount targets at or under `path`, deepest first. + + Read from /proc/mounts rather than using os.path.ismount(): when a FUSE server dies + (the mounter being restarted kills every GeeseFS it started) its mount entry survives + here, but stat()ing the mount point fails with ENOTCONN, which os.path.ismount() + reports as "not a mount point". Such a dead mount would then never be unmounted, and + its directory could never be removed. + """ + path = os.path.normpath(path) + prefix = path + "/" + targets = [] + try: + with open(PROC_MOUNTS) as mounts: + for line in mounts: + fields = line.split() + if len(fields) < 2: + continue + target = _unescape(fields[1]) + if target == path or target.startswith(prefix): + targets.append(target) + except OSError as e: + log(f"reading {PROC_MOUNTS} failed: {e}") + # Deepest first, so nested mounts are detached before their parents. + return sorted(targets, key=len, reverse=True) + + +def is_mounted(path): + """True if `path` itself is a mount point, alive or dead.""" + return os.path.normpath(path) in mount_targets_under(path) + + +def _responds(path): + """True if `path` can be stat()ed — false for a mount whose FUSE server is gone.""" + try: + os.stat(path) + return True + except OSError: + return False + + +def _remove_empty_dirs(path, cuid): + """Remove `path` and any parents left empty, stopping at the CU's own directory.""" + stop_at = os.path.normpath(os.path.join(MOUNT_ROOT, cuid)) + path = os.path.normpath(path) + while path.startswith(stop_at): Review Comment: I am confused with the word `stop_at`. The logic seems like it's removing `MOUNT_ROOT/<cuid>` as well. Does deleting it break mount propagation into the already-running pod? If not, the docstring and `stop_at` name are confusing since it says it stops there. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
