Support redirect log from bohrium which will be used by dflow (#298)

dpdispatcher/dp_cloud_server.py

@@ -18,6 +18,7 @@
 
 class Bohrium(Machine):
     alias = ("Lebesgue", "DpCloudServer")
+
     def __init__(self, context):
         self.context = context
         self.input_data = context.remote_profile['input_data'].copy()
@@ -148,7 +149,13 @@ def check_status(self, job):
 
         job_state = self.map_dp_job_state(dp_job_status)
         if job_state == JobStatus.finished:
+            job_log = self.api.get_log(job_id)
+            if self.input_data.get('output_log'):
+                print(job_log, end='')
             self._download_job(job)
+        elif self.input_data.get('output_log') and job_state == JobStatus.running:
+            job_log = self.api.get_log(job_id)
+            print(job_log, end='')
         return job_state
 
     def _download_job(self, job):

dpdispatcher/dpcloudserver/client.py

@@ -34,6 +34,7 @@ def __init__(self, email=None, password=None, debug=False,
         self.config['email'] = email
         self.config['password'] = password
         self.base_url = base_url
+        self.last_log_offset = 0
 
     def post(self, url, data=None, header=None, params=None, retry=5):
         self.refresh_token()
@@ -224,6 +225,30 @@ def get_tasks(self, job_id, group_id, page=1, per_page=10):
         )
         return ret
 
+    def get_log(self, job_id):
+        url, size = self._get_job_log(job_id)
+        if not url:
+            return ''
+        if self.last_log_offset >= size:
+            return ''
+        resp = requests.get(url, headers={
+            'Range': f'bytes={self.last_log_offset}-'
+        })
+        self.last_log_offset += len(resp.content)
+        return resp.content.decode('utf-8')
+
+    def _get_job_log(self, job_id):
+        ret = self.get(
+            f'/brm/v1/job/{job_id}/log',
+            params={
+                'pageSize': 1,
+            }
+        )
+        d = ret.get('logFiles')
+        if d and len(d) != 0:
+            return d[0]['url'], d[0]['size']
+        return None, 0
+
     def get_tasks_list(self, group_id, per_page=30):
         result = []
         page = 0