Added support for customizing path for job mount file system (#368)

Author: lu-ohai

ads/common/dsc_file_system.py

@@ -5,6 +5,7 @@
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 import ads
 import oci
+import os
 import ipaddress
 
 from dataclasses import dataclass
@@ -20,6 +21,7 @@ class DSCFileSystem:
     dest: str = None
     storage_type: str = None
     destination_directory_name: str = None
+    destination_path: str = None
 
     def update_to_dsc_model(self) -> dict:
         """Updates arguments to dsc model.
@@ -47,6 +49,29 @@ def update_from_dsc_model(cls, dsc_model) -> dict:
         """
         pass
 
+    @staticmethod
+    def get_destination_path_and_name(dest: str) -> (str, str):
+        """Gets the destination path and destination directory name from dest.
+        Example:
+            dir - "fss" & path - "/opc" : mount happens under "/opc/fss"
+            dir - "fss" & path - "/" : mount happens under "/fss"
+            dir - "fss" & path - omitted : mount happens under "/mnt/fss" (for backward compatibility)
+
+        Parameters
+        ----------
+        dest: str
+            The dest path to which to mount the file system.
+
+        Returns
+        -------
+        tuple
+            A tuple of destination path and destination directory name.
+        """
+        return (
+            os.path.dirname(dest.rstrip("/")) or None, # when destination path is omitted, oci api requires it to be None
+            os.path.basename(dest.rstrip("/"))
+        )
+
 
 @dataclass
 class OCIFileStorage(DSCFileSystem):
@@ -65,8 +90,10 @@ def update_to_dsc_model(self) -> dict:
         dict:
             A dictionary of arguments.
         """
+        path, directory_name = self.get_destination_path_and_name(self.dest)
         arguments = {
-            "destinationDirectoryName" : self.dest,
+            "destinationDirectoryName" : directory_name,
+            "destinationPath" : path,
             "storageType" : self.storage_type
         }
 
@@ -177,10 +204,14 @@ def update_from_dsc_model(cls, dsc_model) -> dict:
             raise ValueError(
                 "Missing parameter `destination_directory_name` from service. Check service log to see the error."
             )
+        
+        dest = dsc_model.destination_directory_name
+        if dsc_model.destination_path:
+            dest = f"{dsc_model.destination_path.rstrip('/')}/{dsc_model.destination_directory_name}"
 
         return {
             "src" : f"{dsc_model.mount_target_id}:{dsc_model.export_id}",
-            "dest" : dsc_model.destination_directory_name
+            "dest" : dest
         }
 
 @dataclass
@@ -189,8 +220,10 @@ class OCIObjectStorage(DSCFileSystem):
     storage_type: str = OBJECT_STORAGE_TYPE
 
     def update_to_dsc_model(self) -> dict:
+        path, directory_name = self.get_destination_path_and_name(self.dest)
         arguments = {
-            "destinationDirectoryName" : self.dest,
+            "destinationDirectoryName" : directory_name,
+            "destinationPath" : path,
             "storageType" : self.storage_type
         }
         src_list = self.src.split("@")
@@ -220,9 +253,13 @@ def update_from_dsc_model(cls, dsc_model) -> dict:
                 "Missing parameter `destination_directory_name` from service. Check service log to see the error."
             )
 
+        dest = dsc_model.destination_directory_name
+        if dsc_model.destination_path:
+            dest = f"{dsc_model.destination_path.rstrip('/')}/{dsc_model.destination_directory_name}"
+
         return {
             "src" : f"oci://{dsc_model.bucket}@{dsc_model.namespace}/{dsc_model.prefix or ''}",
-            "dest" : dsc_model.destination_directory_name
+            "dest" : dest
         }
 
 

ads/jobs/builders/infrastructure/dsc_job.py

@@ -1482,7 +1482,7 @@ def _update_job_infra(self, dsc_job: DSCJob) -> DataScienceJob:
 
         if self.storage_mount:
             if not hasattr(
-                oci.data_science.models, "JobStorageMountConfigurationDetails"
+                oci.data_science.models, "StorageMountConfigurationDetails"
             ):
                 raise EnvironmentError(
                     "Storage mount hasn't been supported in the current OCI SDK installed."

tests/unitary/default_setup/jobs/test_jobs_mount_file_system.py

@@ -15,7 +15,6 @@
 from ads.jobs.builders.runtimes.python_runtime import PythonRuntime
 
 try:
-    from oci.data_science.models import JobStorageMountConfigurationDetails
     from oci.data_science.models import FileStorageMountConfigurationDetails
     from oci.data_science.models import ObjectStorageMountConfigurationDetails
 except (ImportError, AttributeError) as e:
@@ -50,6 +49,7 @@
         FileStorageMountConfigurationDetails(
             **{
                 "destination_directory_name": "test_destination_directory_name_from_dsc",
+                "destination_path": "/test_destination_path",
                 "export_id": "export_id_from_dsc",
                 "mount_target_id": "mount_target_id_from_dsc",
                 "storage_type": "FILE_STORAGE",
@@ -58,6 +58,7 @@
         FileStorageMountConfigurationDetails(
             **{
                 "destination_directory_name": "test_destination_directory_name_from_dsc",
+                "destination_path": "/test_destination_path",
                 "export_id": "export_id_from_dsc",
                 "mount_target_id": "mount_target_id_from_dsc",
                 "storage_type": "FILE_STORAGE",
@@ -80,15 +81,15 @@
         .with_storage_mount(
             {
                 "src" : "1.1.1.1:test_export_path_one",
-                "dest" : "test_mount_one",
+                "dest" : "/test_path_one/test_mount_one",
             },
             {
                 "src" : "2.2.2.2:test_export_path_two",
-                "dest" : "test_mount_two",
+                "dest" : "/test_path_two/test_mount_two",
             }, 
             {
                 "src" : "oci://bucket_name@namespace/synthetic/",
-                "dest" : "test_mount_three",
+                "dest" : "/test_path_three/test_mount_three",
             } 
         )
     )
@@ -114,11 +115,11 @@
       shapeName: VM.Standard.E3.Flex
       storageMount:
       - src: 1.1.1.1:test_export_path_one
-        dest: test_mount_one
+        dest: /test_path_one/test_mount_one
       - src: 2.2.2.2:test_export_path_two
-        dest: test_mount_two
+        dest: /test_path_two/test_mount_two
       - src: oci://bucket_name@namespace/synthetic/
-        dest: test_mount_three
+        dest: /test_path_three/test_mount_three
       subnetId: ocid1.subnet.oc1.iad.xxxx
     type: dataScienceJob
   name: My Job
@@ -142,17 +143,17 @@ def test_data_science_job_initialize(self):
         dsc_file_storage_one = job.infrastructure.storage_mount[0]
         assert isinstance(dsc_file_storage_one, dict)
         assert dsc_file_storage_one["src"] == "1.1.1.1:test_export_path_one"
-        assert dsc_file_storage_one["dest"] == "test_mount_one"
+        assert dsc_file_storage_one["dest"] == "/test_path_one/test_mount_one"
 
         dsc_file_storage_two = job.infrastructure.storage_mount[1]
         assert isinstance(dsc_file_storage_two, dict)
         assert dsc_file_storage_two["src"] == "2.2.2.2:test_export_path_two"
-        assert dsc_file_storage_two["dest"] == "test_mount_two"
+        assert dsc_file_storage_two["dest"] == "/test_path_two/test_mount_two"
 
         dsc_object_storage = job.infrastructure.storage_mount[2]
         assert isinstance(dsc_object_storage, dict)
         assert dsc_object_storage["src"] == "oci://bucket_name@namespace/synthetic/"
-        assert dsc_object_storage["dest"] == "test_mount_three"
+        assert dsc_object_storage["dest"] == "/test_path_three/test_mount_three"
 
     def test_data_science_job_from_yaml(self):
         job_from_yaml = Job.from_yaml(job_yaml_string)
@@ -161,17 +162,17 @@ def test_data_science_job_from_yaml(self):
         dsc_file_storage_one = job_from_yaml.infrastructure.storage_mount[0]
         assert isinstance(dsc_file_storage_one, dict)
         assert dsc_file_storage_one["src"] == "1.1.1.1:test_export_path_one"
-        assert dsc_file_storage_one["dest"] == "test_mount_one"
+        assert dsc_file_storage_one["dest"] == "/test_path_one/test_mount_one"
 
         dsc_file_storage_two = job.infrastructure.storage_mount[1]
         assert isinstance(dsc_file_storage_two, dict)
         assert dsc_file_storage_two["src"] == "2.2.2.2:test_export_path_two"
-        assert dsc_file_storage_two["dest"] == "test_mount_two"
+        assert dsc_file_storage_two["dest"] == "/test_path_two/test_mount_two"
 
         dsc_object_storage = job.infrastructure.storage_mount[2]
         assert isinstance(dsc_object_storage, dict)
         assert dsc_object_storage["src"] == "oci://bucket_name@namespace/synthetic/"
-        assert dsc_object_storage["dest"] == "test_mount_three"
+        assert dsc_object_storage["dest"] == "/test_path_three/test_mount_three"
 
     def test_data_science_job_to_dict(self):
         assert job.to_dict() == {
@@ -201,15 +202,15 @@ def test_data_science_job_to_dict(self):
                         "storageMount": [
                             {
                                 "src" : "1.1.1.1:test_export_path_one",
-                                "dest" : "test_mount_one",
+                                "dest" : "/test_path_one/test_mount_one",
                             },
                             {
                                 "src" : "2.2.2.2:test_export_path_two",
-                                "dest" : "test_mount_two",
+                                "dest" : "/test_path_two/test_mount_two",
                             },
                             {
                                 "src" : "oci://bucket_name@namespace/synthetic/",
-                                "dest" : "test_mount_three",
+                                "dest" : "/test_path_three/test_mount_three",
                             } 
                         ],
                     },
@@ -260,11 +261,11 @@ def test_update_storage_mount_from_dsc_model(
         assert isinstance(infrastructure.storage_mount[1], dict)
         assert infrastructure.storage_mount[0] == {
             "src" : "mount_target_id_from_dsc:export_id_from_dsc",
-            "dest" : "test_destination_directory_name_from_dsc"
+            "dest" : "/test_destination_path/test_destination_directory_name_from_dsc"
         }
         assert infrastructure.storage_mount[1] == {
             "src" : "mount_target_id_from_dsc:export_id_from_dsc",
-            "dest" : "test_destination_directory_name_from_dsc"
+            "dest" : "/test_destination_path/test_destination_directory_name_from_dsc"
         }
 
     @patch.object(OCIFileStorage, "update_to_dsc_model")
@@ -276,6 +277,7 @@ def test_update_job_infra(
 
         mock_update_to_dsc_model.return_value = {
             "destinationDirectoryName": "test_destination_directory_name_from_dsc",
+            "destination_path": "/test_destination_path",
             "exportId": "test_export_id_one",
             "mountTargetId": "test_mount_target_id_one",
             "storageType": "FILE_STORAGE",
@@ -293,6 +295,7 @@ def test_update_job_infra(
             0
         ] == {
             "destinationDirectoryName": "test_destination_directory_name_from_dsc",
+            "destination_path": "/test_destination_path",
             "exportId": "test_export_id_one",
             "mountTargetId": "test_mount_target_id_one",
             "storageType": "FILE_STORAGE",
@@ -358,7 +361,7 @@ def test_file_manager_process_data_error(self):
     def test_dsc_object_storage(self):
         object_storage = OCIObjectStorage(
             src="oci://bucket@namespace/prefix",
-            dest="test_dest",
+            dest="/test_path/test_dest",
         )
 
         result = object_storage.update_to_dsc_model()
@@ -367,10 +370,12 @@ def test_dsc_object_storage(self):
         assert result["prefix"] == "prefix"
         assert result["storageType"] == "OBJECT_STORAGE"
         assert result["destinationDirectoryName"] == "test_dest"
+        assert result["destinationPath"] == "/test_path"
 
         dsc_model = ObjectStorageMountConfigurationDetails(
             **{
                 "destination_directory_name": "test_destination_directory_name_from_dsc",
+                "destination_path": "/test_destination_path",
                 "storage_type": "OBJECT_STORAGE",
                 "bucket": "bucket",
                 "namespace": "namespace",
@@ -380,17 +385,18 @@ def test_dsc_object_storage(self):
 
         result = OCIObjectStorage.update_from_dsc_model(dsc_model)
         assert result["src"] == "oci://bucket@namespace/prefix"
-        assert result["dest"] == "test_destination_directory_name_from_dsc"
+        assert result["dest"] == "/test_destination_path/test_destination_directory_name_from_dsc"
 
     def test_dsc_object_storage_error(self):
         error_messages = {
             "namespace" : "Missing parameter `namespace` from service. Check service log to see the error.",
             "bucket" : "Missing parameter `bucket` from service. Check service log to see the error.",
-            "destination_directory_name" : "Missing parameter `destination_directory_name` from service. Check service log to see the error."
+            "destination_directory_name" : "Missing parameter `destination_directory_name` from service. Check service log to see the error.",
         }
 
         dsc_model_dict = {
             "destination_directory_name": "test_destination_directory_name_from_dsc",
+            "destination_path": "/test_path",
             "storage_type": "OBJECT_STORAGE",
             "bucket": "bucket",
             "namespace": "namespace",
@@ -412,19 +418,20 @@ def test_dsc_object_storage_error(self):
     def test_dsc_file_storage(self, mock_search_resources):
         file_storage = OCIFileStorage(
             src="ocid1.mounttarget.oc1.iad.xxxx:ocid1.export.oc1.iad.xxxx",
-            dest="test_dest",
+            dest="/test_path/test_dest",
         )
         file_storage = file_storage.update_to_dsc_model()
         assert file_storage == {
             "destinationDirectoryName" : "test_dest",
+            "destinationPath" : "/test_path",
             "exportId" : "ocid1.export.oc1.iad.xxxx",
             "mountTargetId" : "ocid1.mounttarget.oc1.iad.xxxx",
             "storageType" : "FILE_STORAGE"
         }
 
         file_storage = OCIFileStorage(
             src="1.1.1.1:/test_export",
-            dest="test_dest",
+            dest="/test_path/test_dest",
         )
 
         items = [
@@ -477,6 +484,7 @@ def test_dsc_file_storage(self, mock_search_resources):
         file_storage = file_storage.update_to_dsc_model()
         assert file_storage == {
             "destinationDirectoryName" : "test_dest",
+            "destinationPath" : "/test_path",
             "exportId" : "ocid1.export.oc1.iad.xxxx",
             "mountTargetId" : "ocid1.mounttarget.oc1.iad.xxxx",
             "storageType" : "FILE_STORAGE"
@@ -485,24 +493,26 @@ def test_dsc_file_storage(self, mock_search_resources):
         dsc_model = FileStorageMountConfigurationDetails(
             **{
                 "destination_directory_name": "test_dest",
+                "destination_path" : "/test_path",
                 "storage_type": "FILE_STORAGE",
                 "export_id": "ocid1.export.oc1.iad.xxxx",
                 "mount_target_id": "ocid1.mounttarget.oc1.iad.xxxx"
             }
         )
         result = OCIFileStorage.update_from_dsc_model(dsc_model)
         assert result["src"] == "ocid1.mounttarget.oc1.iad.xxxx:ocid1.export.oc1.iad.xxxx"
-        assert result["dest"] == "test_dest"
+        assert result["dest"] == "/test_path/test_dest"
 
     def test_dsc_file_storage_error(self):
         error_messages = {
             "mount_target_id" : "Missing parameter `mount_target_id` from service. Check service log to see the error.",
             "export_id" : "Missing parameter `export_id` from service. Check service log to see the error.",
-            "destination_directory_name" : "Missing parameter `destination_directory_name` from service. Check service log to see the error."
+            "destination_directory_name" : "Missing parameter `destination_directory_name` from service. Check service log to see the error.",
         }
 
         dsc_model_dict = {
             "destination_directory_name": "test_destination_directory_name_from_dsc",
+            "destination_path": "/test_path",
             "storage_type": "FILE_STORAGE",
             "mount_target_id": "ocid1.mounttarget.oc1.iad.xxxx",
             "export_id": "ocid1.export.oc1.iad.xxxx",
@@ -517,4 +527,25 @@ def test_dsc_file_storage_error(self):
                 dsc_model_copy.pop(error)
                 OCIFileStorage.update_from_dsc_model(
                     FileStorageMountConfigurationDetails(**dsc_model_copy)
-                )
+                )
+
+    def test_get_destination_path_and_name(self):
+        path, directory = OCIFileStorage.get_destination_path_and_name("abc")
+
+        assert path == None
+        assert directory == "abc"
+
+        path, directory = OCIFileStorage.get_destination_path_and_name("/abc")
+
+        assert path == "/"
+        assert directory == "abc"
+
+        path, directory = OCIFileStorage.get_destination_path_and_name("/abc/def")
+
+        assert path == "/abc"
+        assert directory == "def"
+
+        path, directory = OCIFileStorage.get_destination_path_and_name("/abc/def/ghi")
+
+        assert path == "/abc/def"
+        assert directory == "ghi"