Merge pull request #717 from leigaoms/lgrestapi

perf optimization for job list and detail

Author: leigaoms

src/ClusterManager/job_launcher.py

@@ -650,11 +650,6 @@ def submit_job_impl(self, job):
 
             ret["jobId"] = job_object.job_id
 
-            dataHandler.UpdateJobTextField(job_object.job_id, "jobStatus", "scheduling")
-            dataHandler.UpdateJobTextField(job_object.job_id, "jobDescriptionPath", job_description_path)
-            dataHandler.UpdateJobTextField(job_object.job_id, "jobDescription", base64.b64encode(job_description))
-            dataHandler.UpdateJobTextField(job_object.job_id, "lastUpdated", datetime.datetime.now().isoformat())
-
             jobMeta = {}
             jobMeta["jobDescriptionPath"] = job_description_path
             jobMeta["jobPath"] = job_object.job_path
@@ -663,19 +658,31 @@ def submit_job_impl(self, job):
             jobMeta["LaunchCMD"] = pods[0].spec.containers[0].command
 
             jobMetaStr = base64.b64encode(json.dumps(jobMeta))
-            dataHandler.UpdateJobTextField(job_object.job_id, "jobMeta", jobMetaStr)
+
+            dataFields = {
+                "jobStatus": "scheduling",
+                "jobDescriptionPath": job_description_path,
+                "jobDescription": base64.b64encode(job_description),
+                "lastUpdated": datetime.datetime.now().isoformat(),
+                "jobMeta": jobMetaStr
+            }
+            conditionFields = {"jobId": job_object.job_id}
+            dataHandler.UpdateJobTextFields(conditionFields, dataFields)
         except Exception as e:
             logger.error("Submit job failed: %s" % job, exc_info=True)
             ret["error"] = str(e)
             retries = dataHandler.AddandGetJobRetries(job["jobId"])
             if retries >= 5:
-                dataHandler.UpdateJobTextField(job["jobId"], "jobStatus", "error")
-                dataHandler.UpdateJobTextField(job["jobId"], "errorMsg", "Cannot submit job!" + str(e))
-
                 detail = get_job_status_detail(job)
                 detail = job_status_detail_with_finished_time(detail, "error", "Server error in job submission")
-                dataHandler.UpdateJobTextField(job["jobId"], "jobStatusDetail", base64.b64encode(json.dumps(detail)))
 
+                dataFields = {
+                    "jobStatus": "error",
+                    "errorMsg": "Cannot submit job!" + str(e),
+                    "jobStatusDetail": base64.b64encode(json.dumps(detail))
+                }
+                conditionFields = {"jobId": job["jobId"]}
+                dataHandler.UpdateJobTextFields(conditionFields, dataFields)
                 # Try to clean up the job
                 try:
                     job_deployer = JobDeployer()
@@ -705,15 +712,20 @@ def kill_job_impl(self, job_id, desired_state="killed", dataHandlerOri=None):
         job_deployer = JobDeployer()
         errors = job_deployer.delete_job(job_id, force=True)
 
+        dataFields = {
+            "jobStatusDetail": base64.b64encode(json.dumps(detail)),
+            "lastUpdated": datetime.datetime.now().isoformat()
+        }
+        conditionFields = {"jobId": job_id}
         if len(errors) == 0:
-            dataHandler.UpdateJobTextField(job_id, "jobStatus", desired_state)
-            dataHandler.UpdateJobTextField(job_id, "lastUpdated", datetime.datetime.now().isoformat())
+            dataFields["jobStatus"] = desired_state
+            dataHandler.UpdateJobTextFields(conditionFields, dataFields)
             if dataHandlerOri is None:
                 dataHandler.Close()
             return True
         else:
-            dataHandler.UpdateJobTextField(job_id, "jobStatus", "error", "{}".format(errors))
-            dataHandler.UpdateJobTextField(job_id, "lastUpdated", datetime.datetime.now().isoformat())
+            dataFields["jobStatus"] = "error"
+            dataHandler.UpdateJobTextFields(conditionFields, dataFields)
             if dataHandlerOri is None:
                 dataHandler.Close()
             logger.error("Kill job failed with errors: {}".format(errors))

src/ClusterManager/job_manager.py

@@ -209,8 +209,13 @@ def ApproveJob(redis_conn, job, dataHandlerOri=None):
         if "preemptionAllowed" in jobParams and jobParams["preemptionAllowed"] is True:
             logger.info("Job {} preemptible, approve!".format(job_id))
             detail = [{"message": "waiting for available preemptible resource."}]
-            dataHandler.UpdateJobTextField(job["jobId"], "jobStatusDetail", base64.b64encode(json.dumps(detail)))
-            dataHandler.UpdateJobTextField(job_id, "jobStatus", "queued")
+
+            dataFields = {
+                "jobStatusDetail": base64.b64encode(json.dumps(detail)),
+                "jobStatus": "queued"
+            }
+            conditionFields = {"jobId": job_id}
+            dataHandler.UpdateJobTextFields(conditionFields, dataFields)
             update_job_state_latency(redis_conn, job_id, "approved")
             if dataHandlerOri is None:
                 dataHandler.Close()
@@ -250,8 +255,13 @@ def ApproveJob(redis_conn, job, dataHandlerOri=None):
                 return False
 
         detail = [{"message": "waiting for available resource."}]
-        dataHandler.UpdateJobTextField(job["jobId"], "jobStatusDetail", base64.b64encode(json.dumps(detail)))
-        dataHandler.UpdateJobTextField(job_id, "jobStatus", "queued")
+
+        dataFields = {
+            "jobStatusDetail": base64.b64encode(json.dumps(detail)),
+            "jobStatus": "queued"
+        }
+        conditionFields = {"jobId": job_id}
+        dataHandler.UpdateJobTextFields(conditionFields, dataFields)
         update_job_state_latency(redis_conn, job_id, "approved")
         if dataHandlerOri is None:
             dataHandler.Close()
@@ -293,8 +303,13 @@ def UpdateJobStatus(redis_conn, launcher, job, notifier=None, dataHandlerOri=Non
         # TODO: Refactor
         detail = get_job_status_detail(job)
         detail = job_status_detail_with_finished_time(detail, "finished")
-        dataHandler.UpdateJobTextField(job["jobId"], "jobStatusDetail", base64.b64encode(json.dumps(detail)))
-        dataHandler.UpdateJobTextField(job["jobId"], "jobStatus", "finished")
+
+        dataFields = {
+            "jobStatusDetail": base64.b64encode(json.dumps(detail)),
+            "jobStatus": "finished"
+        }
+        conditionFields = {"jobId": job["jobId"]}
+        dataHandler.UpdateJobTextFields(conditionFields, dataFields)
 
         # Retain the old code for reference
         # if jobDescriptionPath is not None and os.path.isfile(jobDescriptionPath):
@@ -311,8 +326,13 @@ def UpdateJobStatus(redis_conn, launcher, job, notifier=None, dataHandlerOri=Non
         if job["jobStatus"] != "running":
             started_at = k8sUtils.localize_time(datetime.datetime.now())
             detail = [{"startedAt": started_at, "message": "started at: {}".format(started_at)}]
-            dataHandler.UpdateJobTextField(job["jobId"], "jobStatusDetail", base64.b64encode(json.dumps(detail)))
-            dataHandler.UpdateJobTextField(job["jobId"], "jobStatus", "running")
+
+            dataFields = {
+                "jobStatusDetail": base64.b64encode(json.dumps(detail)),
+                "jobStatus":"running"
+            }
+            conditionFields = {"jobId": job["jobId"]}
+            dataHandler.UpdateJobTextFields(conditionFields, dataFields)
 
     elif result == "Failed":
         logger.warning("Job %s fails, cleaning...", job["jobId"])
@@ -326,9 +346,14 @@ def UpdateJobStatus(redis_conn, launcher, job, notifier=None, dataHandlerOri=Non
         # TODO: Refactor
         detail = get_job_status_detail(job)
         detail = job_status_detail_with_finished_time(detail, "failed")
-        dataHandler.UpdateJobTextField(job["jobId"], "jobStatusDetail", base64.b64encode(json.dumps(detail)))
-        dataHandler.UpdateJobTextField(job["jobId"], "jobStatus", "failed")
-        dataHandler.UpdateJobTextField(job["jobId"], "errorMsg", "pod failed")
+
+        dataFields = {
+            "jobStatusDetail": base64.b64encode(json.dumps(detail)),
+            "jobStatus": "failed",
+            "errorMsg": "pod failed"
+        }
+        conditionFields = {"jobId": job["jobId"]}
+        dataHandler.UpdateJobTextFields(conditionFields, dataFields)
 
         # Retain the old code for reference
         # if jobDescriptionPath is not None and os.path.isfile(jobDescriptionPath):