add job group id

Author: KZHIWEI

dpdispatcher/dp_cloud_server.py

@@ -13,6 +13,7 @@ class DpCloudServer(Machine):
     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)
 
     def gen_script(self, job):
         shell_script = super(DpCloudServer, self).gen_script(job)
@@ -42,30 +43,53 @@ def do_submit(self, job):
         input_data['job_resources'] = job_resources
         input_data['command'] = f"bash {job.script_file_name}"
 
-
-        job_id = 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.job_id = job_id
+        job_id = None
+        if self.api_version == 2:
+            job_id, group_id = 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)
+            )
+            self.input_data['job_group_id'] = group_id
+            job.job_id = str(job_id) + ':job_group_id' + str(group_id)
+            job_id = job.job_id
+        else:
+            job_id = 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.job_state = JobStatus.waiting
         return job_id
 
     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:
+            ids = job.job_id.split(":job_group_id")
+            job.job_id ,self.input_data["job_group_id"] = int(ids[0]), int(ids[1])
+            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}")
-
-        check_return = api.get_tasks(job.job_id)
+        check_return = None
+        if self.api_version == 2:
+            check_return = api.get_tasks_v2(job_id,self.input_data.get('job_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.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_id}. check_return:{check_return}; retry one more time after 60 seconds")
             time.sleep(60)
-            retry_return = api.get_tasks(job.job_id)
+            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'))
+            else:
+                retry_return = api.get_tasks(job_id)
             try:
                 dp_job_status = retry_return[0]["status"]
             except IndexError as e:
@@ -91,8 +115,13 @@ def map_dp_job_state(status):
             -1:JobStatus.terminated,
             0:JobStatus.waiting,
             1:JobStatus.running,
-            2:JobStatus.finished
+            2:JobStatus.finished,
+            3:JobStatus.waiting,
+            4:JobStatus.running,
+            5:JobStatus.terminated
         }
+        if status not in map_dict:
+            return JobStatus.unknown
         return map_dict[status]
 
 

dpdispatcher/dpcloudserver/api.py

@@ -15,6 +15,7 @@
 from .config import HTTP_TIME_OUT, API_HOST
 token = ''
 group_id = None
+
 def get(url, params):
     global token
     headers = {'Authorization': "jwt " + token}
@@ -108,9 +109,6 @@ def upload(oss_task_zip, zip_task_file, endpoint, bucket_name):
 
 
 def job_create(job_type, oss_path, input_data, program_id=None):
-    api_version = input_data.get('api_version',1)
-    if api_version == 2:
-        return job_create_v2(job_type,oss_path,input_data, program_id=program_id)
     post_data = {
                 'job_type': job_type,
                 'oss_path': oss_path,
@@ -137,14 +135,14 @@ def job_create_v2(job_type, oss_path, input_data, program_id=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')
+    if input_data.get('rerun') is not None:
+        post_data["rerun"] = input_data.get('rerun')
     if input_data.get('image_name') is not None:
         post_data["image_name"] = input_data.get('image_name')
     if input_data.get('disk_size') is not None:
         post_data["disk_size"] = input_data.get('disk_size')
     if input_data.get('scass_type') is not None:
         post_data["scass_type"] = input_data.get('scass_type')
-    if input_data.get('scass_type') is not None:
-        post_data["scass_type"] = input_data.get('scass_type')
     if input_data.get('instance_group_id') is not None:
         post_data["instance_group_id"] = input_data.get('instance_group_id')
     if input_data.get('command') is not None:
@@ -163,7 +161,7 @@ def job_create_v2(job_type, oss_path, input_data, program_id=None):
         post_data["on_demand"] = input_data.get('on_demand')
     ret = post('/data/v2/insert_job', post_data)
     group_id = ret.get('job_group_id')
-    return ret['job_id']
+    return ret['job_id'], group_id
 
 def get_jobs(page=1, per_page=10):
     ret = get(
@@ -186,5 +184,19 @@ def get_tasks(job_id, page=1, per_page=10):
     )
     return ret['items']
 
+
+def get_tasks_v2(job_id, group_id, page=1, per_page=10):
+    ret = get(
+        f'data/job/{group_id}/tasks',
+        {
+            'page': page,
+            'per_page': per_page,
+        }
+    )
+    for each in ret['items']:
+        if job_id == each["task_id"]:
+            return [each]
+    return []
+
 #%%