Merge pull request #8729 from OpenMined/yash/swfs-svc-acc

Add service account to seaweedfs + fixes

packages/grid/helm/syft/templates/seaweedfs/seaweedfs-service-account.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: seaweedfs-service-account
+  labels:
+    {{- include "common.labels" . | nindent 4 }}
+    app.kubernetes.io/component: seaweedfs

packages/grid/helm/syft/templates/seaweedfs/seaweedfs-statefulset.yaml

@@ -31,6 +31,7 @@ spec:
       {{- if .Values.seaweedfs.nodeSelector }}
       nodeSelector: {{- .Values.seaweedfs.nodeSelector | toYaml | nindent 8 }}
       {{- end }}
+      serviceAccountName: seaweedfs-service-account
       containers:
         - name: seaweedfs-container
           image: {{ .Values.global.registry }}/openmined/grid-seaweedfs:{{ .Values.global.version }}

packages/grid/seaweedfs/config/app/supervisord.conf

@@ -34,16 +34,5 @@ stdout_logfile=/dev/stdout
 stdout_logfile_maxbytes=0
 redirect_stderr=true
 
-[program:automount]
-priority=3
-command=sh -c ". scripts/wait_for_swfs.sh && exec python -m src.cli.automount"
-autostart=true
-autorestart=false
-startretries=0
-startsecs=0
-stdout_logfile=/dev/stdout
-stdout_logfile_maxbytes=0
-redirect_stderr=true
-
 [include]
 files = /data/mounts/**/*.conf /etc/mounts/**/*.conf

packages/grid/seaweedfs/src/api.py

@@ -1,19 +1,33 @@
 # stdlib
 import logging
+from pathlib import Path
 import subprocess
 
 # third party
 from fastapi import FastAPI
 from fastapi import HTTPException
 
 # first party
+from src.automount import automount
 from src.mount import mount_bucket
 from src.mount_options import MountOptions
 
-app = FastAPI(title="SeaweedFS Remote Mount API")
+# Automount all configured buckets
+AUTOMOUNT_CONFIG = Path("./automount.yaml")
+
+# keeping it away from /data/ to avoid these being persisted in a volume
+PRIVATE_CONF_DIR = Path("/etc/mounts/")
+PRIVATE_CONF_DIR.mkdir(mode=0o600, parents=True, exist_ok=True)
 
 logger = logging.getLogger("uvicorn.error")
 
+app = FastAPI(title="SeaweedFS Remote Mount API")
+
+
+@app.on_event("startup")
+async def startup_event() -> None:
+    automount(automount_conf=AUTOMOUNT_CONFIG, mount_conf_dir=PRIVATE_CONF_DIR)
+
 
 @app.get("/")
 def ping() -> dict:
@@ -32,15 +46,15 @@ def mount(opts: MountOptions, overwrite: bool = False) -> dict:
             "name": result["name"],
         }
     except FileExistsError as e:
-        raise HTTPException(status_code=409, detail=str(e))
+        raise HTTPException(status_code=400, detail=str(e))
     except subprocess.CalledProcessError as e:
         logger.error(
-            f"Subprocess error: code={e.returncode} stdout={e.stdout} stderr={e.stderr}"
+            f"Mount error: code={e.returncode} stdout={e.stdout} stderr={e.stderr}"
         )
-        raise HTTPException(status_code=503, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))
     except Exception as e:
         logger.error(f"Unhandled exception: {e}")
-        raise HTTPException(status_code=503, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))
 
 
 @app.post("/configure_azure", deprecated=True)

packages/grid/seaweedfs/src/automount.py

@@ -0,0 +1,42 @@
+# stdlib
+import logging
+import logging.config
+from pathlib import Path
+import subprocess
+
+# third party
+import yaml
+
+# relative
+from .mount import mount_bucket
+from .mount_options import MountOptions
+
+logger = logging.getLogger(__name__)
+
+
+def automount(automount_conf: Path, mount_conf_dir: Path) -> None:
+    with automount_conf.open() as f:
+        config = yaml.safe_load(f)
+
+    mounts = config.get("mounts", [])
+    for mount_opts in mounts:
+        mount_opts = MountOptions(**mount_opts)
+        try:
+            logger.info(
+                f"Auto mount type={mount_opts.remote_bucket.type} "
+                f"bucket={mount_opts.remote_bucket.bucket_name}"
+            )
+            result = mount_bucket(
+                mount_opts,
+                conf_dir=mount_conf_dir,
+                overwrite=True,
+            )
+            logger.info(f"Auto mount success. {result}")
+        except FileExistsError as e:
+            logger.info(e)
+        except subprocess.CalledProcessError as e:
+            logger.error(
+                f"Mount error: code={e.returncode} stdout={e.stdout} stderr={e.stderr}"
+            )
+        except Exception as e:
+            logger.error(f"Unhandled exception: {e}")

packages/grid/seaweedfs/src/mount.py

@@ -81,11 +81,16 @@ def mount_bucket(
         check=True,
         capture_output=True,
     )
-    logger.info(
-        f"Mount {opts.remote_bucket.bucket_name} configured.",
-        f'stdout="{proc.stdout!r}"',
-        f'stderr="{proc.stderr!r}"',
-    )
+    # weed shell will be successful, but the internal command might throw error
+    if b"error" in proc.stdout or b"error" in proc.stderr:
+        raise subprocess.CalledProcessError(
+            proc.returncode,
+            proc.args,
+            output=proc.stdout,
+            stderr=proc.stderr,
+        )
+
+    logger.info(f"Mount {opts.remote_bucket.bucket_name} configured.")
 
     # create sync command
     sync_cmd = create_sync_cmd(opts.local_bucket)
@@ -110,9 +115,9 @@ def mount_bucket(
         capture_output=True,
     )
     logger.info(
-        "Supervisor updated.",
-        f'stdout="{proc.stdout!r}"',
-        f'stderr="{proc.stderr!r}"',
+        "Supervisor updated. ",
+        f"stdout={proc.stdout.decode()} ",
+        f"stderr={proc.stderr.decode()}",
     )
 
     return {"name": swfs_config_name, "path": mount_conf_dir}