remove lebesgue version 1 api support. add progress bar, and fix resume cause upload. (#162)

* add envs , module purge and lebesgue context

* fix format

* fix backup

* add module_purge to test

* fix test case

* refract backup

* fix unexpected lebesgue job status

* change logger method to exception

* fix dlog

* fix type

* remove version 1 and add tqdm

* newline

* rm newline

* change execption to dlog

* add tqdm

* fix uploading to lebesgue

* add more log

* improve upload

* fix raise

* more info

* fix check exist

* remove some log

* fix rmdir

* fix keep backup

* remove

Author: KZHIWEI

dpdispatcher/dp_cloud_server.py

@@ -1,5 +1,8 @@
+import shutil
+
 from dpdispatcher.JobStatus import JobStatus
 from dpdispatcher import dlog
+from dpdispatcher.dpcloudserver import zip_file
 from dpdispatcher.machine import Machine
 from dpdispatcher.dpcloudserver.api import API
 from dpdispatcher.dpcloudserver.config import ALI_OSS_BUCKET_URL
@@ -18,9 +21,9 @@ def __init__(self, context):
         self.input_data = context.remote_profile['input_data'].copy()
         self.api_version = 2
         if 'api_version' in self.input_data:
-            self.api_version = self.input_data.get('api_version')
+            self.api_version = self.input_data.get('api_version', 2)
         if 'lebesgue_version' in self.input_data:
-            self.api_version = self.input_data.get('lebesgue_version')
+            self.api_version = self.input_data.get('lebesgue_version', 2)
         self.grouped = self.input_data.get('grouped', False)
         email = context.remote_profile.get("email", None)
         username = context.remote_profile.get('username', None)
@@ -33,7 +36,7 @@ def __init__(self, context):
         if password is None:
             raise ValueError("can not find password in remote_profile, please check your machine file.")
         if self.api_version == 1:
-            warnings.warn('api version 1 is deprecated and will be removed in a future version. Use version 2 instead.', DeprecationWarning)
+            raise DeprecationWarning('api version 1 is deprecated. Use version 2 instead.')
         self.api = API(email, password)
         self.group_id = None
 
@@ -88,26 +91,17 @@ def do_submit(self, job):
         # input_data['backward_files'] = self._gen_backward_files_list(job)
         if self.context.remote_profile.get('program_id') is None:
             warnings.warn('program_id will be compulsory in the future.')
-        job_id = None
-        if self.api_version == 2:
-            job_id, group_id = self.api.job_create_v2(
-                job_type=input_data['job_type'],
-                oss_path=input_data['job_resources'],
-                input_data=input_data,
-                program_id=self.context.remote_profile.get('program_id', None),
-                group_id=self.group_id
-            )
-            if self.grouped:
-                self.group_id = group_id
-            job.job_id = str(job_id) + ':job_group_id:' + str(group_id)
-            job_id = job.job_id
-        else:
-            job_id = self.api.job_create(
-                job_type=input_data['job_type'],
-                oss_path=input_data['job_resources'],
-                input_data=input_data,
-                program_id=self.context.remote_profile.get('program_id', None)
-            )
+        job_id, group_id = self.api.job_create(
+            job_type=input_data['job_type'],
+            oss_path=input_data['job_resources'],
+            input_data=input_data,
+            program_id=self.context.remote_profile.get('program_id', None),
+            group_id=self.group_id
+        )
+        if self.grouped:
+            self.group_id = group_id
+        job.job_id = str(job_id) + ':job_group_id:' + str(group_id)
+        job_id = job.job_id
         job.job_state = JobStatus.waiting
         return job_id
 
@@ -126,28 +120,40 @@ def check_status(self, job):
         dlog.debug(f"debug: check_status; job.job_id:{job_id}; job.job_hash:{job.job_hash}")
         check_return = None
         # print("api",self.api_version,self.input_data.get('job_group_id'),job.job_id)
-        if self.api_version == 2:
-            check_return = self.api.get_tasks_v2(job_id,group_id)
-        else:
-            check_return = self.api.get_tasks(job_id)
+        check_return = self.api.get_tasks(job_id,group_id)
         try:
-            dp_job_status = check_return[0]["status"]
+            dp_job_status = check_return["status"]
         except IndexError as e:
             dlog.error(f"cannot find job information in check_return. job {job.job_id}. check_return:{check_return}; retry one more time after 60 seconds")
             time.sleep(60)
-            retry_return = None
-            if self.api_version == 2:
-                retry_return = self.api.get_tasks_v2(job_id, group_id)
-            else:
-                retry_return = self.api.get_tasks(job_id)
+            retry_return = self.api.get_tasks(job_id, group_id)
             try:
-                dp_job_status = retry_return[0]["status"]
+                dp_job_status = retry_return["status"]
             except IndexError as e:
                 raise RuntimeError(f"cannot find job information in dpcloudserver's database for job {job.job_id} {check_return} {retry_return}")
 
         job_state = self.map_dp_job_state(dp_job_status)
+        if job_state == JobStatus.finished:
+            self._download_job(job)
         return job_state
 
+
+    def _download_job(self, job):
+        job_url = self.api.get_job_result_url(job.job_id)
+        if not job_url:
+            return
+        job_hash = job.job_hash
+        result_filename = job_hash + '_back.zip'
+        target_result_zip = os.path.join(self.context.local_root, result_filename)
+        self.api.download_from_url(job_url, target_result_zip)
+        zip_file.unzip_file(target_result_zip, out_dir=self.context.local_root)
+        try:
+            os.makedirs(os.path.join(self.context.local_root, 'backup'), exist_ok=True)
+            shutil.move(target_result_zip,
+                        os.path.join(self.context.local_root, 'backup', os.path.split(target_result_zip)[1]))
+        except (OSError, shutil.Error) as e:
+            dlog.exception("unable to backup file, " + str(e))
+
     def check_finish_tag(self, job):
         job_tag_finished = job.job_hash + '_job_tag_finished'
         dlog.info('check if job finished: ',job.job_id, job_tag_finished)

dpdispatcher/dp_cloud_server_context.py

@@ -13,6 +13,7 @@
 from .dpcloudserver.api import API
 from .dpcloudserver import zip_file
 import shutil
+import tqdm
 # from zip_file import zip_files
 DP_CLOUD_SERVER_HOME_DIR = os.path.join(
     os.path.expanduser('~'),
@@ -89,8 +90,17 @@ def upload(self, submission):
 
         # zip_path = "/home/felix/workplace/22_dpdispatcher/dpdispatcher-yfb/dpdispatcher/dpcloudserver/t.txt"
         # zip_path = self.local_root
-
+        bar_format = "{l_bar}{bar}| {n:.02f}/{total:.02f} %  [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
+        job_to_be_uploaded = []
+        result = None
+        dlog.info("checking all job has been uploaded")
         for job in submission.belonging_jobs:
+            if not self.api.check_job_has_uploaded(job.job_id):
+                job_to_be_uploaded.append(job)
+        if len(job_to_be_uploaded) == 0:
+            dlog.info("all job has been uploaded, continue")
+            return result
+        for job in tqdm.tqdm(job_to_be_uploaded, desc="Uploading to Lebesgue", bar_format=bar_format):
             self.machine.gen_local_script(job)
             zip_filename = job.job_hash + '.zip'
             oss_task_zip = self._gen_oss_path(job, zip_filename)
@@ -113,7 +123,7 @@ def upload(self, submission):
                 file_list=upload_file_list
             )
             result = self.api.upload(oss_task_zip, upload_zip, ENDPOINT, BUCKET_NAME)
-            self._backup(self.local_root, upload_zip, keep_backup=self.remote_profile.get('keep_backup', True))
+            self._backup(self.local_root, upload_zip)
         return result
         # return oss_task_zip
         # api.upload(self.oss_task_dir, zip_task_file)
@@ -124,15 +134,12 @@ def download(self, submission):
         group_id = None
         job_infos = {}
         for job in jobs:
-            if isinstance(job.job_id, str) and ':job_group_id:' in job.job_id:
-                ids = job.job_id.split(":job_group_id:")
-                jid, gid = int(ids[0]), int(ids[1])
-                job_hashs[jid] = job.job_hash
-                group_id = gid
-            else:
-                job_infos[job.job_hash] = self.api.get_tasks(job.job_id)[0]
+            ids = job.job_id.split(":job_group_id:")
+            jid, gid = int(ids[0]), int(ids[1])
+            job_hashs[jid] = job.job_hash
+            group_id = gid
         if group_id is not None:
-            job_result = self.api.get_tasks_v2_list(group_id)
+            job_result = self.api.get_tasks_list(group_id)
             for each in job_result:
                 if 'result_url' in each and each['result_url'] != '' and each['status'] == 2:
                     job_hash = ''
@@ -143,26 +150,40 @@ def download(self, submission):
                     else:
                         job_hash = job_hashs[each['task_id']]
                     job_infos[job_hash] = each
-        for job_hash, info in job_infos.items():
+        bar_format = "{l_bar}{bar}| {n:.02f}/{total:.02f} %  [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
+        for job_hash, info in tqdm.tqdm(job_infos.items(), desc="Validating download file from Lebesgue", bar_format=bar_format):
             result_filename = job_hash + '_back.zip'
             target_result_zip = os.path.join(self.local_root, result_filename)
+            if self._check_if_job_has_already_downloaded(target_result_zip, self.local_root):
+                continue
             self.api.download_from_url(info['result_url'], target_result_zip)
             zip_file.unzip_file(target_result_zip, out_dir=self.local_root)
-            self._backup(self.local_root, target_result_zip, keep_backup=self.remote_profile.get('keep_backup', True))
+            self._backup(self.local_root, target_result_zip)
+        self._clean_backup(self.local_root, keep_backup=self.remote_profile.get('keep_backup', True))
         return True
 
-    def _backup(self, local_root, target, keep_backup=True):
+    def _check_if_job_has_already_downloaded(self, target, local_root):
+        backup_file_location = os.path.join(local_root, 'backup', os.path.split(target)[1])
+        if os.path.exists(backup_file_location):
+            return True
+        else:
+            return False
+
+    def _backup(self, local_root, target):
         try:
-            if keep_backup:
-                # move to backup directory
-                os.makedirs(os.path.join(local_root, 'backup'), exist_ok=True)
-                shutil.move(target,
-                            os.path.join(local_root, 'backup', os.path.split(target)[1]))
-            else:
-                os.remove(target)
+            # move to backup directory
+            os.makedirs(os.path.join(local_root, 'backup'), exist_ok=True)
+            shutil.move(target,
+                        os.path.join(local_root, 'backup', os.path.split(target)[1]))
         except (OSError, shutil.Error) as e:
             dlog.exception("unable to backup file, " + str(e))
 
+    def _clean_backup(self, local_root, keep_backup=True):
+        if not keep_backup:
+            dir_to_be_removed = os.path.join(local_root, 'backup')
+            if os.path.exists(dir_to_be_removed):
+                shutil.rmtree(dir_to_be_removed)
+
     def write_file(self, fname, write_str):
         result = self.write_home_file(fname, write_str)
         return result

dpdispatcher/dpcloudserver/api.py

@@ -170,18 +170,8 @@ def upload(self, oss_task_zip, zip_task_file, endpoint, bucket_name):
         # print('debug:upload_result:', result, dir())
         return result
 
-    def job_create(self, job_type, oss_path, input_data, program_id=None):
-        post_data = {
-            'job_type': job_type,
-            'oss_path': oss_path,
-            'input_data': input_data,
-        }
-        if program_id is not None:
-            post_data["program_id"] = program_id
-        ret = self.post('/data/insert_job', post_data)
-        return ret['job_id']
 
-    def job_create_v2(self, job_type, oss_path, input_data, program_id=None, group_id=None):
+    def job_create(self, job_type, oss_path, input_data, program_id=None, group_id=None):
         post_data = {
             'job_type': job_type,
             'oss_path': oss_path,
@@ -210,17 +200,7 @@ def get_jobs(self, page=1, per_page=10):
         )
         return ret['items']
 
-    def get_tasks(self, job_id, page=1, per_page=10):
-        ret = self.get(
-            f'data/job/{job_id}/tasks',
-            {
-                'page': page,
-                'per_page': per_page,
-            }
-        )
-        return ret['items']
-
-    def get_tasks_v2(self, job_id, group_id, page=1, per_page=10):
+    def get_tasks(self, job_id, group_id, page=1, per_page=10):
         ret = self.get(
             f'data/job/{group_id}/tasks',
             {
@@ -230,12 +210,12 @@ def get_tasks_v2(self, job_id, group_id, page=1, per_page=10):
         )
         for each in ret['items']:
             if job_id == each["task_id"]:
-                return [each]
+                return each
         if len(ret['items']) != 0:
-            return self.get_tasks_v2(job_id, group_id, page=page + 1)
-        return []
+            return self.get_tasks(job_id, group_id, page=page + 1)
+        return None
 
-    def get_tasks_v2_list(self, group_id, per_page=30):
+    def get_tasks_list(self, group_id, per_page=30):
         result = []
         page = 0
         while True:
@@ -253,4 +233,38 @@ def get_tasks_v2_list(self, group_id, per_page=30):
             page += 1
         return result
 
+    def check_job_has_uploaded(self, job_id):
+        try:
+            if not job_id:
+                return False
+            if 'job_group_id' in job_id:
+                ids = job_id.split(":job_group_id:")
+                job_id, _ = int(ids[0]), int(ids[1])
+            ret = self.get(f'data/job/{job_id}', {})
+            if len(ret) == 0:
+                return False
+            if ret.get('input_data'):
+                return True
+            else:
+                return False
+        except ValueError as e:
+            dlog.error(e)
+            return False
+
+    def get_job_result_url(self, job_id):
+        try:
+            if not job_id:
+                return None
+            if 'job_group_id' in job_id:
+                ids = job_id.split(":job_group_id:")
+                job_id, _ = int(ids[0]), int(ids[1])
+            ret = self.get(f'data/job/{job_id}', {})
+            if 'result_url' in ret and len(ret['result_url']) != 0:
+                return ret.get('result_url')
+            else:
+                return None
+        except ValueError as e:
+            dlog.error(e)
+            return None
+
 # %%

requirements.txt

@@ -1,3 +1,4 @@
 paramiko
 dargs>=0.2.6
 oss2
+tqdm

setup.py

@@ -19,7 +19,7 @@
 with open(path.join(NAME, '_date.py'), 'w') as fp :
     fp.write('date = \'%s\'' % today)
 
-install_requires=['paramiko', 'dargs>=0.2.6', 'requests']
+install_requires=['paramiko', 'dargs>=0.2.6', 'requests', 'tqdm']
 
 setuptools.setup(
     name=NAME,
@@ -41,7 +41,7 @@
     install_requires=install_requires,    
     extras_require={
         'docs': ['sphinx', 'recommonmark', 'sphinx_rtd_theme>=1.0.0rc1', 'numpydoc'],
-        "cloudserver": ["oss2"],
+        "cloudserver": ["oss2", "tqdm"],
         ":python_version<'3.7'": ["typing_extensions"],
     },
         entry_points={