def update_node_ids(self, sub_job_num=None):
        """
        Associate Jobs with respective cluster ids.
        """
        # Build condor_q and condor_history commands
        dag_id = '%s.%s' % (self.cluster_id, sub_job_num) if sub_job_num else str(self.cluster_id)
        job_delimiter = '+++'
        attr_delimiter = ';;;'

        format = [
            '-format', '"%d' + attr_delimiter + '"', 'ClusterId',
            '-format', '"%v' + attr_delimiter + '"', 'Cmd',
            '-format', '"%v' + attr_delimiter + '"', 'Args',     # Old way
            '-format', '"%v' + job_delimiter + '"', 'Arguments'  # New way
        ]
        # Get ID, Executable, and Arguments for each job that is either started to be processed or finished in the workflow
        cmd = 'condor_q -constraint DAGManJobID=={0} {1} && condor_history -constraint DAGManJobID=={0} {1}'.format(dag_id, ' '.join(format))

        # 'condor_q -constraint DAGManJobID==1018 -format "%d\n" ClusterId -format "%s\n" CMD -format "%s\n" ARGS && condor_history -constraint DAGManJobID==1018 -format "%d\n" ClusterId -format "%s\n" CMD -format "%s\n" ARGS'
        _args = [cmd]
        out, err = self._execute(_args, shell=True, run_in_job_dir=False)

        if err:
            log.error('Error while associating ids for jobs dag %s: %s', dag_id, err)
            raise HTCondorError(err)
        if not out:
            log.warning('Error while associating ids for jobs in dag %s: No jobs found for dag.', dag_id)

        try:
            # Split into one line per job
            jobs_out = out.split(job_delimiter)

            # Match node to cluster id using combination of cmd and arguments
            for node in self._node_set:
                job = node.job

                # Skip jobs that already have cluster id defined
                if job.cluster_id != job.NULL_CLUSTER_ID:
                    continue

                for job_out in jobs_out:
                    if not job_out or attr_delimiter not in job_out:
                        continue

                    # Split line by attributes
                    cluster_id, cmd, _args, _arguments = job_out.split(attr_delimiter)

                    # If new form of arguments is used, _args will be 'undefined' and _arguments will not
                    if _args == 'undefined' and _arguments != 'undefined':
                        args = _arguments.strip()

                    # If both are undefined, then there are no arguments
                    elif _args == 'undefined' and _arguments == 'undefined':
                        args = None

                    # Otherwise, using old form and _arguments will be 'undefined' and _args will not.
                    else:
                        args = _args.strip()

                    job_cmd = job.executable
                    job_args = job.arguments.strip() if job.arguments else None

                    if job_cmd in cmd and job_args == args:
                        log.info('Linking cluster_id %s to job with command and arguments: %s %s', cluster_id,
                                  job_cmd, job_args)
                        job._cluster_id = int(cluster_id)
                        break

        except ValueError as e:
            log.warning(str(e))
def update_node_ids ( self , sub_job_num = None ) : dag_id = '%s.%s' % ( self . cluster_id , sub_job_num ) if sub_job_num else str ( self . cluster_id ) job_delimiter = '+++' attr_delimiter = ';;;' format = [ '-format' , '"%d' + attr_delimiter + '"' , 'ClusterId' , '-format' , '"%v' + attr_delimiter + '"' , 'Cmd' , '-format' , '"%v' + attr_delimiter + '"' , 'Args' , '-format' , '"%v' + job_delimiter + '"' , 'Arguments' ] cmd = 'condor_q -constraint DAGManJobID=={0} {1} && condor_history -constraint DAGManJobID=={0} {1}' . format ( dag_id , ' ' . join ( format ) ) _args = [ cmd ] out , err = self . _execute ( _args , shell = True , run_in_job_dir = False ) if err : log . error ( 'Error while associating ids for jobs dag %s: %s' , dag_id , err ) raise HTCondorError ( err ) if not out : log . warning ( 'Error while associating ids for jobs in dag %s: No jobs found for dag.' , dag_id ) try : jobs_out = out . split ( job_delimiter ) for node in self . _node_set : job = node . job if job . cluster_id != job . NULL_CLUSTER_ID : continue for job_out in jobs_out : if not job_out or attr_delimiter not in job_out : continue cluster_id , cmd , _args , _arguments = job_out . split ( attr_delimiter ) if _args == 'undefined' and _arguments != 'undefined' : args = _arguments . strip ( ) elif _args == 'undefined' and _arguments == 'undefined' : args = None else : args = _args . strip ( ) job_cmd = job . executable job_args = job . arguments . strip ( ) if job . arguments else None if job_cmd in cmd and job_args == args : log . info ( 'Linking cluster_id %s to job with command and arguments: %s %s' , cluster_id , job_cmd , job_args ) job . _cluster_id = int ( cluster_id ) break except ValueError as e : log . warning ( str ( e ) )
Associate Jobs with respective cluster ids.