Hostpath volume

paddlecloud/paddlecloud/settings.py

@@ -261,18 +261,28 @@
 #if Paddle cloud use CephFS as backend storage, configure CEPHFS_CONFIGURATION
 #the following is an example:
 #DATACENTERS = {
-#    "datacenter1":{
-#        "monitors_addr": "172.19.32.166:6789",
-#        "secret": "ceph-secret",
-#        "user": "admin",
-#        "mount_path": "/pfs/datacenter1/home/%s/", # mount_path % username
-#        "cephfs_path": "/%s" # cephfs_path % username
-#        "admin_key": "/certs/admin.secret"
+#   "datacenter1":{
+#       "fstype": "cephfs",
+#       "monitors_addr": "172.19.32.166:6789",
+#       "secret": "ceph-secret",
+#       "user": "admin",
+#       "mount_path": "/pfs/datacenter1/home/%s/", # mount_path % username
+#       "cephfs_path": "/%s" # cephfs_path % username
+#       "admin_key": "/certs/admin.secret"
+#   }
+#}
+#for HostPath example:
+#DATACENTERS = {
+#   ...
+#   "dc1":{
+#       "fstype": "hostpath",
+#       "host_path": "/mnt/hdfs/",
+#       "mount_path" "/pfs/dc1/home/%s/" # mount_path % username
 #    }
 #}
 DATACENTERS = {
     "datacenter1":{
-        "type": "cephfs",
+        "fstype": "cephfs",
         "monitors_addr": "172.19.32.166:6789",
         "secret": "ceph-secret",
         "user": "admin",

paddlecloud/paddlejob/__init__.py

@@ -1,4 +1,3 @@
 from paddle_job import PaddleJob
-from cephfs_volume import CephFSVolume
-
-__all__ = ["CephFSVolume", "PaddleJob"]
+import volumes
+__all__ = ["volumes", "PaddleJob"]

paddlecloud/paddlejob/paddle_job.py

@@ -1,7 +1,7 @@
 import kubernetes
 from kubernetes import client, config
 import os
-from cephfs_volume import CephFSVolume
+from volumes import *
 
 __all__ = ["PaddleJob"]
 DEFAULT_PADDLE_PORT=7164
@@ -99,13 +99,13 @@ def _get_trainer_labels(self):
     def _get_trainer_volumes(self):
         volumes = []
         for item in self._volumes:
-            volumes.append(item.volume)
+            volumes.append(item["volume"])
         return volumes
 
     def _get_trainer_volume_mounts(self):
         volume_mounts = []
         for item in self._volumes:
-            volume_mounts.append(item.volume_mount)
+            volume_mounts.append(item["volume_mount"])
         return volume_mounts
 
     def new_trainer_job(self):

paddlecloud/paddlejob/tests/test_paddle_job.py

@@ -1,6 +1,5 @@
 import unittest
 from paddle_job import PaddleJob
-from paddlejob import CephFSVolume
 class PaddleJobTest(unittest.TestCase):
     def __new_paddle_job(self):
         return PaddleJob(
@@ -14,14 +13,7 @@ def __new_paddle_job(self):
             pscpu=1,
             psmemory="1Gi",
             topology="train.py",
-            volumes=[
-                CephFSVolume(
-                    monitors_addr="192.168.1.123:6789",
-                    user="admin",
-                    secret_name="cephfs-secret",
-                    mount_path="/mnt/cephfs",
-                    cephfs_path="/")
-            ])
+            volumes=[])
     def test_runtime_image(self):
         paddle_job=self.__new_paddle_job()
         self.assertEqual(paddle_job.pservers, 3)

paddlecloud/paddlejob/tests/test_volume.py

@@ -0,0 +1,15 @@
+import unittest
+from volume import get_volume_config
+class GetVolumeConfigTest(unittest.TestCase):
+    def test_hostpath(self):
+        volume = get_volume_config(
+            fstype="hostpath", name="abc",mount_path="/pfs/dc1/home/yanxu05",
+            host_path= "/mnt/hdfs_mulan")
+        self.assertEqual(volume["volume"]["name"], "abc")
+    def test_cephfs(self):
+        volume = get_volume_config(
+            fstype="cephfs", name="cephfs",
+            monitors_addr="192.168.2.1:6789,182.68.2.2:6789".split(","),
+            cephfs_path="/a/d", user="admin", secret="ceph-secret",
+            mount_path="/pfs/dc1/home/yanxu05")
+        self.assertEqual(volume["volume"]["name"], "cephfs")

paddlecloud/paddlejob/views.py

@@ -3,7 +3,7 @@
 from django.conf import settings
 from kubernetes import client, config
 from kubernetes.client.rest import ApiException
-from . import PaddleJob, CephFSVolume
+from . import PaddleJob
 from rest_framework.authtoken.models import Token
 from rest_framework import viewsets, generics, permissions
 from rest_framework.response import Response
@@ -39,16 +39,8 @@ def post(self, request, format=None):
             return utils.simple_response(500, "no datacenter specified")
         dc = obj.get("datacenter")
         volumes = []
-        for name, cfg in settings.DATACENTERS.items():
-            if cfg["type"] == "cephfs":
-                volumes.append(CephFSVolume(
-                    monitors_addr = cfg["monitors_addr"],
-                    user = cfg["user"],
-                    secret_name = cfg["secret"],
-                    mount_path = cfg["mount_path"] % username,
-                    cephfs_path = cfg["cephfs_path"] % username
-                ))
-
+        if dc in settings.DATACENTERS:
+            volumes.append(volume.get_volume_config(settings.DATACENTERS[dc]))
 
         paddle_job = PaddleJob(
             name = obj.get("name", "paddle-cluster-job"),

paddlecloud/paddlejob/volume.py

@@ -0,0 +1,67 @@
+import json
+
+__all__=["get_volume_config"]
+
+tmpl_volume = {
+    "hostpath": "{\"name\": $NAME, \"hostPath\":{\"path\": $HOST_PATH}}",
+    "cephfs":"{\"name\": $NAME,\"cephfs\":{\"name\": \"cephfs\", \
+               \"monitors\": $MONITORS_ADDR,\"path\": $CEPHFS_PATH, \
+               \"user\": $USER, \"secretRef\": {\"name\": $SECRET}}}"
+}
+tmpl_volume_mount = {
+    "hostpath": "{\"name\": $NAME, \"mountPath\":$MOUNT_PATH}",
+    "cephfs":"{\"mount_path\": $MOUNT_PATH, \"name\": $NAME}"
+}
+
+def __render(tmpl, **kwargs):
+    tmpl.replace("$NAME", "sfsdf")
+    for k, v in kwargs.items():
+        tmpl_k = "$%s" % k.upper()
+        if tmpl_k in tmpl:
+            if type(v) is str:
+                tmpl = tmpl.replace(tmpl_k, "\"%s\"" % v)
+            elif type(v) is list:
+                tmpl = tmpl.replace(tmpl_k, json.dumps(v))
+            else:
+                pass
+    return tmpl
+
+def __get_template(tmpls, fstype):
+    if fstype in tmpls.keys():
+        return tmpls[fstype]
+    else:
+        return ""
+
+def get_volume_config(fstype, **kwargs):
+    """
+    :param fstype: which filesystem type
+    :type fstype: str
+
+    if fstype is host_path:
+
+    :param name: a unique name for a Kubernetes job configuration
+    :type name: str
+    :param mount_path: path in pod
+    :type mount_path: str
+    :param host_path: path no the host
+    :type host_path: str
+
+    if fstype is cephfs:
+
+    :param name: unique name for a Kubernetes Job configuration
+    :type name: str
+    :param monitors_addr: the CephFS monitors address
+    :type monitors_addr: list
+    :param cephfs_path: CephFS Path
+    :type cephfs_path: str
+    :param user: ceph cluster user
+    :type user: str
+    :param secret: Kubernetes Secret for Ceph secret
+    :type secret: str
+    :param mount_path: mount path in Pod
+    :type mount_path: str
+    """
+    tmpl_v = __get_template(tmpl_volume, fstype)
+    tmpl_vm = __get_template(tmpl_volume_mount, fstype)
+    return {"volume":json.loads(__render(tmpl=tmpl_v, **kwargs)),
+            "volume_mount": json.loads(__render(tmpl=tmpl_vm, **kwargs))}