Merge pull request #136 from KZHIWEI/master

Machine support Lebesgue V2 API

Author: felix5572

dpdispatcher/dp_cloud_server.py

@@ -13,6 +13,9 @@ 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)
+        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)
@@ -23,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
@@ -42,30 +45,58 @@ 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),
+                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 = 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
-        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)
+        job_id = 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_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
+        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,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")
             time.sleep(60)
-            retry_return = api.get_tasks(job.job_id)
+            retry_return = None
+            if self.api_version == 2:
+                retry_return = api.get_tasks_v2(job_id,group_id)
+            else:
+                retry_return = api.get_tasks(job_id)
             try:
                 dp_job_status = retry_return[0]["status"]
             except IndexError as e:
@@ -91,8 +122,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/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,6 +14,7 @@
 from .retcode import RETCODE
 from .config import HTTP_TIME_OUT, API_HOST
 token = ''
+
 def get(url, params):
     global token
     headers = {'Authorization': "jwt " + token}
@@ -118,6 +119,28 @@ 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, group_id=None):
+    post_data = {
+                'job_type': job_type,
+                'oss_path': oss_path,
+            }
+    if program_id is not None:
+        post_data["program_id"] = program_id
+    if group_id is not None:
+        post_data["job_group_id"] = group_id
+    if input_data.get('command') is not None:
+        post_data["cmd"] = input_data.get('command')
+    if input_data.get('backward_files') is not None:
+        post_data["out_files"] = input_data.get('backward_files')
+    input_keys = ['job_group_id', 'job_name', 'rerun', 'image_name', 'disk_size', 'scass_type',
+                  'instance_group_id', 'log_file', 'platform', 'region', 'zone', 'on_demand']
+    for key in input_keys:
+        if key in input_data:
+            post_data[key] = input_data[key]
+    ret = post('/data/v2/insert_job', post_data)
+    group_id = ret.get('job_group_id')
+    return ret['job_id'], group_id
+
 def get_jobs(page=1, per_page=10):
     ret = get(
         '/data/jobs',
@@ -139,5 +162,21 @@ 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]
+    if len(ret['items']) != 0:
+        return get_tasks_v2(job_id, group_id, page=page+1)
+    return []
+
 #%%