modify submission.json format; will contain machine information in submission.json (#152)

* modify submission.json format; will contain machine information in submission.json

* fix bug in machine deserialize; add unittest

* only try one line

Author: felix5572

dpdispatcher/dp_cloud_server_context.py

@@ -26,6 +26,8 @@ def __init__ (self,
         *args,
         **kwargs,
     ):
+        self.init_local_root = local_root
+        self.init_remote_root = remote_root
         self.temp_local_root = os.path.abspath(local_root)
         self.remote_profile = remote_profile
         email = remote_profile.get("email", None)

dpdispatcher/hdfs_context.py

@@ -14,6 +14,8 @@ def __init__(self,
                 ):
 
         assert(type(local_root) == str)
+        self.init_local_root = local_root
+        self.init_remote_root = remote_root
         self.temp_local_root = os.path.abspath(local_root)
         self.temp_remote_root = remote_root
         self.remote_profile = remote_profile

dpdispatcher/lazy_local_context.py

@@ -31,6 +31,8 @@ def __init__ (self,
         remote_profile:
         """
         assert(type(local_root) == str)
+        self.init_local_root = local_root
+        self.init_remote_root = remote_root
         self.temp_local_root = os.path.abspath(local_root)
         self.temp_remote_root = os.path.abspath(local_root)
         self.remote_profile = remote_profile

dpdispatcher/local_context.py

@@ -55,6 +55,8 @@ def __init__(self,
         remote_profile:
         """
         assert(type(local_root) == str)
+        self.init_local_root = local_root
+        self.init_remote_root = remote_root
         self.temp_local_root = os.path.abspath(local_root)
         self.temp_remote_root = os.path.abspath(remote_root)
         self.remote_profile = remote_profile

dpdispatcher/machine.py

@@ -124,6 +124,26 @@ def load_from_dict(cls, machine_dict):
         machine = machine_class(context=context)
         return machine
 
+    def serialize(self, if_empty_remote_profile=False):
+        machine_dict = {}
+        machine_dict['batch_type'] = self.__class__.__name__
+        machine_dict['context_type'] = self.context.__class__.__name__
+        machine_dict['local_root'] = self.context.init_local_root
+        machine_dict['remote_root'] = self.context.init_remote_root
+        if not if_empty_remote_profile:
+            machine_dict['remote_profile'] = self.context.remote_profile
+        else:
+            machine_dict['remote_profile'] = {}
+        return machine_dict
+
+    def __eq__(self, other):
+        return self.serialize() == other.serialize()
+
+    @classmethod
+    def deserialize(cls, machine_dict):
+        machine = cls.load_from_dict(machine_dict=machine_dict)
+        return machine
+
     def check_status(self, job) :
         raise NotImplementedError('abstract method check_status should be implemented by derived class')        
 

dpdispatcher/ssh_context.py

@@ -186,9 +186,12 @@ def __init__ (self,
                 **kwargs,
                 ):
         assert(type(local_root) == str)
+        self.init_local_root = local_root
+        self.init_remote_root = remote_root
         self.temp_local_root = os.path.abspath(local_root)
         assert os.path.isabs(remote_root), f"remote_root must be a abspath"
         self.temp_remote_root = remote_root
+        self.remote_profile = remote_profile
 
         # self.job_uuid = None
         self.clean_asynchronously = clean_asynchronously
@@ -258,6 +261,12 @@ def bind_submission(self, submission):
         # self.remote_root = os.path.join(self.temp_remote_root, self.submission.submission_hash, self.submission.work_base )
         self.remote_root = pathlib.PurePath(os.path.join(self.temp_remote_root, self.submission.submission_hash)).as_posix()
 
+        sftp = self.ssh_session.ssh.open_sftp()
+        try:
+            sftp.mkdir(self.remote_root)
+        except OSError:
+            pass
+
         # self.job_uuid = submission.submission_hash
         # dlog.debug("debug:SSHContext.bind_submission"
         #     "{submission.submission_hash}; {self.local_root}; {self.remote_root")

dpdispatcher/submission.py

@@ -88,10 +88,14 @@ def deserialize(cls, submission_dict, machine=None):
             backward_common_files=submission_dict['backward_common_files'])
         submission.belonging_jobs = [Job.deserialize(job_dict=job_dict) for job_dict in submission_dict['belonging_jobs']]
         submission.submission_hash = submission.get_hash()
-        submission.bind_machine(machine=machine)
+        if machine is not None:
+            submission.bind_machine(machine=machine)
+        else:
+            machine = Machine.deserialize(machine_dict=submission_dict['machine'])
+            submission.bind_machine(machine)
         return submission
 
-    def serialize(self, if_static=False, if_none_local_root=False):
+    def serialize(self, if_static=False):
         """convert the Submission class instance to a dictionary.
 
         Parameters
@@ -105,11 +109,17 @@ def serialize(self, if_static=False, if_none_local_root=False):
             the dictionary converted from the Submission class instance
         """
         submission_dict = {}
-        if if_none_local_root:
-            submission_dict['local_root'] = None
-        else:
-            submission_dict['local_root'] = self.local_root
+        # if if_none_local_root:
+        #     submission_dict['local_root'] = None
+        # else:
+        #     submission_dict['local_root'] = self.local_root
+
         submission_dict['work_base'] = self.work_base
+        machine = getattr(self, 'machine', None)
+        if machine is None:
+            submission_dict['machine'] = {}
+        else:
+            submission_dict['machine'] = machine.serialize()
         submission_dict['resources'] = self.resources.serialize()
         submission_dict['forward_common_files'] = self.forward_common_files
         submission_dict['backward_common_files'] = self.backward_common_files
@@ -333,7 +343,7 @@ def try_recover_from_json(self):
             if self == submission:
                 self.belonging_jobs = submission.belonging_jobs
                 self.bind_machine(machine=self.machine)
-                dlog.info(f"Find old submission; recover from json; "
+                dlog.info(f"Find old submission; recover submission from json file;"
                     f"submission.submission_hash:{submission.submission_hash}; "
                     f"machine.context.remote_root:{self.machine.context.remote_root}; "
                     f"submission.work_base:{submission.work_base};")

tests/jsons/submission.json

@@ -1,6 +1,12 @@
 {
-    "local_root": null,
     "work_base": "0_md/",
+    "machine": {
+        "batch_type": "PBS",
+        "context_type": "LocalContext",
+        "local_root": "test_pbs_dir/",
+        "remote_root": "tmp_pbs_dir/",
+        "remote_profile": {}
+    },
     "resources": {
         "number_node": 1,
         "cpu_per_node": 4,
@@ -128,4 +134,4 @@
             }
         }
     ]
-}
+}

tests/script_gen_json.py

@@ -21,10 +21,12 @@
 with open('jsons/resources.json', 'w') as f:
     json.dump(resources_dict, f, indent=4)
 
-submission_dict = SampleClass.get_sample_submission_dict()
+pbs = SampleClass.get_sample_pbs_local_context()
+submission = SampleClass.get_sample_submission()
+submission.bind_machine(machine=pbs)
 assert os.path.isfile('jsons/submission.json') is False
 with open('jsons/submission.json', 'w') as f:
-    json.dump(submission_dict, f, indent=4)
+    json.dump(submission.serialize(), f, indent=4)
 
 job_dict = SampleClass.get_sample_job_dict()
 assert os.path.isfile('jsons/job.json') is False

tests/test_class_machine.py

@@ -0,0 +1,31 @@
+import os,sys,json,glob,shutil,uuid,time
+from socket import gaierror
+import unittest
+from unittest.mock import MagicMock, patch, PropertyMock
+
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+__package__ = 'tests'
+from .context import LocalContext
+from .context import BaseContext
+from .context import PBS
+from .context import JobStatus
+from .context import LazyLocalContext, LocalContext, SSHContext
+from .context import LSF, Slurm, PBS, Shell
+from .context import Machine
+from .context import dargs
+from .context import DistributedShell, HDFSContext
+from .sample_class import SampleClass
+from dargs.dargs import ArgumentKeyError
+
+class TestMachineInit(unittest.TestCase):
+    def setUp(self):
+        self.maxDiff = None
+    
+    def test_machine_serialize_deserialize(self):
+        pbs = SampleClass.get_sample_pbs_local_context()
+        self.assertEqual(pbs, Machine.deserialize(pbs.serialize()))
+
+    def test_machine_load_from_dict(self):
+        pbs = SampleClass.get_sample_pbs_local_context()
+        self.assertEqual(pbs, PBS.load_from_dict(pbs.serialize()))