support v2

Author: KZHIWEI

dpdispatcher/dp_cloud_server.py

@@ -14,6 +14,8 @@ def __init__(self, context):
         self.context = context
         self.input_data = context.remote_profile['input_data'].copy()
         self.api_version = self.input_data.get('api_version', 1)
+        self.grouped = self.input_data.get('grouped', False)
+        self.group_id = None
 
     def gen_script(self, job):
         shell_script = super(DpCloudServer, self).gen_script(job)
@@ -24,10 +26,10 @@ def gen_script_header(self, job):
         return shell_script_header
 
     def gen_local_script(self, job):
-        script_str = self.gen_script(job) 
+        script_str = self.gen_script(job)
         script_file_name = job.script_file_name
         self.context.write_local_file(
-            fname=script_file_name, 
+            fname=script_file_name,
             write_str=script_str
         )
         return script_file_name
@@ -49,9 +51,11 @@ def do_submit(self, job):
                 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)
+                program_id=self.context.remote_profile.get('program_id', None),
+                group_id=self.group_id
             )
-            self.input_data['job_group_id'] = 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:
@@ -68,26 +72,29 @@ def check_status(self, job):
         if job.job_id == '':
             return JobStatus.unsubmitted
         job_id = job.job_id
-        if type(job.job_id) is str and ':job_group_id' in job.job_id:
+        group_id = None
+        if isinstance(job.job_id,str) and ':job_group_id' in job.job_id:
+            group_id = None
             ids = job.job_id.split(":job_group_id")
-            job.job_id ,self.input_data["job_group_id"] = int(ids[0]), int(ids[1])
+            job_id, group_id = int(ids[0]), int(ids[1])
+            if self.input_data.get('grouped') and 'job_group_id' not in self.input_data:
+                self.group_id = group_id
             self.api_version = 2
-            job_id = job.job_id
-        dlog.debug(f"debug: check_status; job.job_id:{job.job_id}; job.job_hash:{job.job_hash}")
+        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 = api.get_tasks_v2(job_id,self.input_data.get('job_group_id'))
+            check_return = api.get_tasks_v2(job_id,group_id)
         else:
             check_return = api.get_tasks(job_id)
         try:
             dp_job_status = check_return[0]["status"]
         except IndexError as e:
-            dlog.error(f"cannot find job information in check_return. job {job_id}. check_return:{check_return}; retry one more time after 60 seconds")
+            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
-            retry_return = api.get_tasks(job_id)
             if self.api_version == 2:
-                retry_return = api.get_tasks_v2(job_id,self.input_data.get('job_group_id'))
+                retry_return = api.get_tasks_v2(job_id,group_id)
             else:
                 retry_return = api.get_tasks(job_id)
             try:

dpdispatcher/dp_cloud_server_context.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # coding: utf-8
-#%%
+# %%
 from dpdispatcher.base_context import BaseContext
 import os
 # from dpdispatcher import dlog
@@ -55,7 +55,6 @@ def bind_submission(self, submission):
         self.submission_hash = submission.submission_hash
 
         self.machine = submission.machine
-    
 
         # def zip_files(self, submission):
         #     file_uuid = uuid.uuid1().hex
@@ -91,7 +90,6 @@ def upload(self, submission):
                 zip_task_file,
                 file_list=upload_file_list
             )
-
             result = api.upload(oss_task_zip, upload_zip, ENDPOINT, BUCKET_NAME)
         return result
         # return oss_task_zip

dpdispatcher/dpcloudserver/api.py

@@ -14,7 +14,6 @@
 from .retcode import RETCODE
 from .config import HTTP_TIME_OUT, API_HOST
 token = ''
-group_id = None
 
 def get(url, params):
     global token
@@ -120,8 +119,7 @@ def job_create(job_type, oss_path, input_data, program_id=None):
     return ret['job_id']
 
 
-def job_create_v2(job_type, oss_path, input_data, program_id=None):
-    global group_id
+def job_create_v2(job_type, oss_path, input_data, program_id=None, group_id=None):
     post_data = {
                 'job_type': job_type,
                 'oss_path': oss_path,
@@ -130,8 +128,7 @@ def job_create_v2(job_type, oss_path, input_data, program_id=None):
         post_data["program_id"] = program_id
     if input_data.get('job_group_id') is not None:
         post_data["job_group_id"] = input_data.get('job_group_id')
-    elif input_data.get('grouped') is not None and input_data.get('grouped'):
-        if group_id is not None:
+    if group_id is not None:
             post_data["job_group_id"] = group_id
     if input_data.get('job_name') is not None:
         post_data["job_name"] = input_data.get('job_name')
@@ -196,6 +193,8 @@ def get_tasks_v2(job_id, group_id, page=1, per_page=10):
     for each in ret['items']:
         if job_id == each["task_id"]:
             return [each]
+    if len(ret['items']) != 0:
+        return get_tasks_v2(job_id, group_id, page=2)
     return []
 
 #%%