DATAUP-499 Switch parent_job_id for batch_id for 'true' batch jobs (#404)

* Switch parent_job_id for batch_id for 'true' batch jobs

Per the request of the front end devs, we want a way to distinguish 'true'
batch jobs created by run_job_batch from 'manual' batch jobs created by
including a parent_job_id in the job input params.

As such, true batch jobs now have a batch_id field in the job status dict,
replacing the parent_job_id field. That field remains in the job_input.

A job must only ever have zero or one of the fields, never both.

* run black

* Match spec, change some spec keys

Changed the abandon_children return dict to match the spec

Updated the check_jobs_batch spec to match nomenclature elsewhere

* Update names and docs to batch from parent

Author: MrCreosote

execution_engine2.html

@@ -181,12 +181,12 @@
         } BatchParams;
 
         typedef structure {
-            job_id parent_job_id;
+            job_id batch_id;
             list<job_id> child_job_ids;
         } BatchSubmission;
 
         typedef structure {
-            job_id parent_job_id;
+            job_id batch_id;
             list<job_id> child_job_ids;
             boolean as_admin;
         } AbandonChildren;
@@ -460,7 +460,7 @@
         retry_ids - list - list of jobs that are retried based off of this job
         retry_parent - str - job_id of the parent this retry is based off of. Not available on a retry_parent itself
 
-        parent_job_id - str - job_id taken from job_input.parent_job_id
+        batch_id - str - the coordinating job, if the job is a child job created via run_job_batch
         batch_job - bool - whether or not this is a batch parent container
         child_jobs - array - Only parent container should have child job ids
 
@@ -501,6 +501,7 @@
             int error_code;
             string errormsg;
             int terminated_code;
+            string batch_id;
 
         } JobState;
 
@@ -510,12 +511,12 @@
         funcdef check_job(CheckJobParams params) returns (JobState job_state) authentication required;
 
         /*
-            parent_job - state of parent job
-            job_states - states of child jobs
+            batch_jobstate - state of the coordinating job for the batch
+            child_jobstates - states of child jobs
             IDEA: ADD aggregate_states - count of all available child job states, even if they are zero
         */
         typedef structure {
-            JobState parent_jobstate;
+            JobState batch_jobstate;
             list<JobState> child_jobstates;
         } CheckJobBatchResults;
 

execution_engine2.spec

@@ -165,6 +165,8 @@ class JobInput(EmbeddedDocument):
     service_ver = StringField(required=True)
     app_id = StringField()
     source_ws_objects = ListField()
+    # this ID is for jobs submitted via run_job with a parent_job_id field included by the
+    # client. For this case, the parent job is not updated at all.
     parent_job_id = StringField()
     requirements = EmbeddedDocumentField(JobRequirements)
     narrative_cell_info = EmbeddedDocumentField(Meta, required=True)
@@ -320,7 +322,12 @@ class Job(Document):
     job_input = EmbeddedDocumentField(JobInput, required=True)
     job_output = DynamicField()
     condor_job_ads = DynamicField()
-    child_jobs = ListField()  # Only parent container should have child jobs
+    # this is the ID of the coordinating job created as part of run_job_batch. Only child jobs
+    # in a "true" batch job maintained by EE2 should have this field. Coordinating jobs will
+    # be updated with the child ID in child_jobs, unlike "fake" batch jobs that are created
+    # outside of the EE2 codebase using the 'parent_job_id' field.
+    batch_id = StringField()
+    child_jobs = ListField()  # Only coordinating jobs should have child jobs
     # batch_parent_container = BooleanField(default=False) # Only parent container should have this
     retry_ids = ListField()  # The retry_parent has been used to launch these jobs
     # Only present on a retried job, not it's parent. If attempting to retry this job, use its parent instead

lib/execution_engine2/db/models/models.py

@@ -53,6 +53,7 @@
 _REQUIREMENTS_LIST = "requirements_list"
 _METHOD = "method"
 _APP_ID = "app_id"
+_BATCH_ID = "batch_id"
 _PARENT_JOB_ID = "parent_job_id"
 _PARENT_RETRY_JOB_ID = "retry_parent"
 _RETRY_IDS = "retry_ids"
@@ -98,6 +99,7 @@ def _init_job_rec(
         {_SERVICE_VER} (app version)
         {_APP_ID} (app UI)
         {_SOURCE_WS_OBJECTS} (collected workspace objects for this app)
+        {_BATCH_ID} (parent of the job for EE2 batch jobs, the parent should be updated)
         {_PARENT_JOB_ID} (parent of this job, doesn't update/notify the parent)
         {_META} (narrative cell information)
 
@@ -158,6 +160,7 @@ def _init_job_rec(
         parent_retry_job_id = params.get(_PARENT_RETRY_JOB_ID)
         if parent_retry_job_id:
             job.retry_parent = str(parent_retry_job_id)
+        job.batch_id = str(params.get(_BATCH_ID)) if params.get(_BATCH_ID) else None
 
         job_id = self.sdkmr.save_job(job)
         self.sdkmr.get_kafka_client().send_kafka_message(
@@ -241,7 +244,10 @@ def _prepare_to_run(self, params, concierge_params=None) -> JobSubmissionParamet
             AppInfo(params[_METHOD], params.get(_APP_ID)),
             params[_JOB_REQUIREMENTS],
             UserCreds(self.sdkmr.get_user_id(), self.sdkmr.get_token()),
-            parent_job_id=params.get(_PARENT_JOB_ID),
+            # a job should have a parent ID or a batch ID or nothing, but never both
+            # Do we want to distinguish between the two cases in the sub params?
+            # It's informational only for Condor
+            parent_job_id=params.get(_BATCH_ID) or params.get(_PARENT_JOB_ID),
             wsid=params.get(_WORKSPACE_ID),
             source_ws_objects=params.get(_SOURCE_WS_OBJECTS),
         )
@@ -287,7 +293,7 @@ def _abort_child_jobs(self, child_job_ids):
                 # TODO Maybe add a retry here?
                 self.logger.error(f"Couldn't cancel child job {e}")
 
-    def _create_parent_job(self, wsid, meta):
+    def _create_batch_job(self, wsid, meta):
         """
         This creates the parent job for all children to mark as their ancestor
         :param params:
@@ -320,11 +326,11 @@ def _create_parent_job(self, wsid, meta):
         )
         return j
 
-    def _run_batch(self, parent_job: Job, params):
+    def _run_batch(self, batch_job: Job, params):
         child_jobs = []
 
         for job_param in params:
-            job_param[_PARENT_JOB_ID] = str(parent_job.id)
+            job_param[_BATCH_ID] = str(batch_job.id)
             try:
                 child_jobs.append(str(self._run(params=job_param)))
             except Exception as e:
@@ -334,8 +340,8 @@ def _run_batch(self, parent_job: Job, params):
                 self._abort_child_jobs(child_jobs)
                 raise e
 
-        parent_job.child_jobs = child_jobs
-        self.sdkmr.save_job(parent_job)
+        batch_job.child_jobs = child_jobs
+        self.sdkmr.save_job(batch_job)
 
         return child_jobs
 
@@ -365,12 +371,12 @@ def run_batch(
             self._check_workspace_permissions_list(wsids)
 
         self._add_job_requirements(params, bool(as_admin))  # as_admin checked above
-        self._check_job_arguments(params, has_parent_job=True)
+        self._check_job_arguments(params, batch_job=True)
 
-        parent_job = self._create_parent_job(wsid=wsid, meta=meta)
-        children_jobs = self._run_batch(parent_job=parent_job, params=params)
+        batch_job = self._create_batch_job(wsid=wsid, meta=meta)
+        children_jobs = self._run_batch(batch_job=batch_job, params=params)
 
-        return {_PARENT_JOB_ID: str(parent_job.id), "child_job_ids": children_jobs}
+        return {_BATCH_ID: str(batch_job.id), "child_job_ids": children_jobs}
 
     # modifies the jobs in place
     def _add_job_requirements(self, jobs: List[Dict[str, Any]], is_write_admin: bool):
@@ -464,7 +470,7 @@ def _rethrow_incorrect_params_with_error_prefix(
             raise error
         raise IncorrectParamsException(f"{error_prefix}{error.args[0]}") from error
 
-    def _check_job_arguments(self, jobs, has_parent_job=False):
+    def _check_job_arguments(self, jobs, batch_job=False):
         # perform sanity checks before creating any jobs, including the parent job for batch jobs
         for i, job in enumerate(jobs):
             # Could make an argument checker method, or a class that doesn't require a job id.
@@ -482,7 +488,7 @@ def _check_job_arguments(self, jobs, has_parent_job=False):
                 )
             except IncorrectParamsException as e:
                 self._rethrow_incorrect_params_with_error_prefix(e, pre)
-            if has_parent_job and job.get(_PARENT_JOB_ID):
+            if batch_job and job.get(_PARENT_JOB_ID):
                 raise IncorrectParamsException(
                     f"{pre}batch jobs may not specify a parent job ID"
                 )
@@ -535,12 +541,11 @@ def _validate_retry_presubmit(self, job_id: str, as_admin: bool = False):
         job = self.sdkmr.get_job_with_permission(
             job_id, JobPermissions.WRITE, as_admin=as_admin
         )  # type: Job
-        job_input = job.job_input  # type: JobInput
 
-        parent_job = None
-        if job_input.parent_job_id:
-            parent_job = self.sdkmr.get_job_with_permission(
-                job_input.parent_job_id, JobPermissions.WRITE, as_admin=as_admin
+        batch_job = None
+        if job.batch_id:
+            batch_job = self.sdkmr.get_job_with_permission(
+                job.batch_id, JobPermissions.WRITE, as_admin=as_admin
             )
 
         if job.batch_job:
@@ -553,9 +558,9 @@ def _validate_retry_presubmit(self, job_id: str, as_admin: bool = False):
                 f"Error retrying job {job_id} with status {job.status}: can only retry jobs with status 'error' or 'terminated'"
             )
 
-        return job, parent_job
+        return job, batch_job
 
-    def _retry(self, job_id: str, job: Job, parent_job: Job, as_admin: bool = False):
+    def _retry(self, job_id: str, job: Job, batch_job: Job, as_admin: bool = False):
         # Cannot retry a retried job, you must retry the retry_parent
         if job.retry_parent:
             return self.retry(str(job.retry_parent), as_admin=as_admin)
@@ -569,13 +574,15 @@ def _retry(self, job_id: str, job: Job, parent_job: Job, as_admin: bool = False)
         retry_job_id = self.run(params=run_job_params, as_admin=as_admin)
 
         # Save that the job has been retried, and increment the count. Notify the parent(s)
-        # 1) Notify the parent container that it has a new child..
-        if parent_job:
+        # 1) Notify the batch container that it has a new child. Note that the parent jobs of
+        # 'manual' batch jobs using the job_input.parent_job_id field *are not* modified to
+        # include their children, so we don't do that here either.
+        if batch_job:
             try:
-                parent_job.modify(add_to_set__child_jobs=retry_job_id)
+                batch_job.modify(add_to_set__child_jobs=retry_job_id)
             except Exception as e:
                 self._db_update_failure(
-                    job_that_failed_operation=str(parent_job.id),
+                    job_that_failed_operation=str(batch_job.id),
                     job_to_abort=retry_job_id,
                     exception=e,
                 )
@@ -611,11 +618,11 @@ def retry(self, job_id: str, as_admin=False) -> Dict[str, Optional[str]]:
         :param as_admin: Run with admin permission
         :return: The child job id that has been retried
         """
-        job, parent_job = self._validate_retry_presubmit(
+        job, batch_job = self._validate_retry_presubmit(
             job_id=job_id, as_admin=as_admin
         )
         return self._retry(
-            job_id=job_id, job=job, parent_job=parent_job, as_admin=as_admin
+            job_id=job_id, job=job, batch_job=batch_job, as_admin=as_admin
         )
 
     def retry_multiple(
@@ -646,14 +653,14 @@ def retry_multiple(
         # Check all inputs before attempting to start submitting jobs
         retried_jobs = []
         jobs = []
-        parent_jobs = []
+        batch_jobs = []
         for job_id in job_ids:
             try:
-                job, parent_job = self._validate_retry_presubmit(
+                job, batch_job = self._validate_retry_presubmit(
                     job_id=job_id, as_admin=as_admin
                 )
                 jobs.append(job)
-                parent_jobs.append(parent_job)
+                batch_jobs.append(batch_job)
             except Exception as e:
                 raise RetryFailureException(e)
 
@@ -664,7 +671,7 @@ def retry_multiple(
                     self._retry(
                         job_id=job_id,
                         job=jobs[i],
-                        parent_job=parent_jobs[i],
+                        batch_job=batch_jobs[i],
                         as_admin=as_admin,
                     )
                 )
@@ -810,6 +817,8 @@ def get_job_params(self, job_id, as_admin=False):
         job_params["service_ver"] = job_input.service_ver
         job_params[_APP_ID] = job_input.app_id
         job_params[_WORKSPACE_ID] = job_input.wsid
+        # This is specfically the data in the job params, which includes any manually submitted
+        # parent job information but does not include batch job information
         job_params[_PARENT_JOB_ID] = job_input.parent_job_id
         job_params[_SOURCE_WS_OBJECTS] = job_input.source_ws_objects
 

lib/execution_engine2/execution_engine2Impl.py

@@ -474,9 +474,7 @@ def check_jobs(
                 del mongo_rec["_id"]
                 mongo_rec["retry_count"] = len(job["retry_ids"])
                 mongo_rec["job_id"] = str(job.id)
-                mongo_rec["parent_job_id"] = (
-                    job.job_input.parent_job_id if job.job_input else None
-                )
+                mongo_rec["batch_id"] = job.batch_id
                 mongo_rec["created"] = int(job.id.generation_time.timestamp() * 1000)
                 mongo_rec["updated"] = int(job.updated * 1000)
                 if job.estimating:
@@ -560,26 +558,28 @@ def _send_exec_stats_to_catalog(self, job_id):
 
         self.sdkmr.get_catalog().log_exec_stats(log_exec_stats_params)
 
-    def abandon_children(self, parent_job_id, child_job_ids, as_admin=False) -> Dict:
-        if not parent_job_id:
-            raise ValueError("Please provide valid parent_job id")
+    def abandon_children(self, batch_id, child_job_ids, as_admin=False) -> Dict:
+        # Note this does not work for 'manual' batch jobs as the parent job is
+        # never updated with the child jobs. It will only work with batch jobs specifically
+        # created by the run_job_batch endpoint.
+        if not batch_id:
+            raise ValueError("Please provide valid batch_id")
         if not child_job_ids:
             raise ValueError("Please provide job_ids of children to abandon")
 
         job = self.sdkmr.get_job_with_permission(
-            parent_job_id, JobPermissions.WRITE, as_admin=as_admin
+            batch_id, JobPermissions.WRITE, as_admin=as_admin
         )  # type: Job
         for child_job_id in child_job_ids:
             if child_job_id not in job.child_jobs:
                 raise ChildrenNotFoundError(
                     f"Couldn't find {child_job_id} in {child_job_ids}"
                 )
 
-        with self.sdkmr.get_mongo_util().mongo_engine_connection():
-            job.update(pull_all__child_jobs=child_job_ids)
-            job.reload()
+        job.update(pull_all__child_jobs=child_job_ids)
+        job.reload()
 
-        return {"parent_job_id": parent_job_id, "child_jobs": job.child_jobs}
+        return {"batch_id": batch_id, "child_job_ids": job.child_jobs}
 
     def start_job(self, job_id, skip_estimation=True, as_admin=False):
         """

lib/execution_engine2/sdk/EE2Runjob.py

@@ -358,10 +358,10 @@ def start_job(self, job_id, skip_estimation=True, as_admin=False):
         )
 
     # Endpoints: Changing a job's status
-    def abandon_children(self, parent_job_id, child_job_ids, as_admin=False):
+    def abandon_children(self, batch_id, child_job_ids, as_admin=False):
         """Authorization Required Read/Write"""
         return self.get_jobs_status().abandon_children(
-            parent_job_id=parent_job_id, child_job_ids=child_job_ids, as_admin=as_admin
+            batch_id=batch_id, child_job_ids=child_job_ids, as_admin=as_admin
         )
 
     def update_job_status(self, job_id, status, as_admin=False):
@@ -433,7 +433,7 @@ def get_job_status_field(self, job_id, as_admin=False):
 
     def check_job_batch(
         self,
-        parent_job_id,
+        batch_id,
         check_permission=True,
         exclude_fields=None,
         as_admin=False,
@@ -448,7 +448,7 @@ def check_job_batch(
             raise ValueError("You can't exclude child jobs from this endpoint")
 
         parent_job_status = self.get_jobs_status().check_job(
-            job_id=parent_job_id,
+            job_id=batch_id,
             check_permission=check_permission,
             exclude_fields=exclude_fields,
         )
@@ -462,7 +462,7 @@ def check_job_batch(
                 return_list=1,
             )["job_states"]
         return {
-            "parent_jobstate": parent_job_status,
+            "batch_jobstate": parent_job_status,
             "child_jobstates": child_job_states,
         }