qq_lib.resubmit

 20class Resubmitter(Operator):
 21    """
 22    Handles resubmission of loop and continuous jobs.
 23    """
 24
 25    def resubmit(self) -> str:
 26        """
 27        Resubmit the job to the next cycle.
 28
 29        Returns:
 30            str: The job ID of the newly submitted job.
 31
 32        Raises:
 33            QQError: If the main node is not defined or if resubmission fails
 34                on all candidate hosts.
 35        """
 36        informer = self.get_informer()
 37        input_dir = self._info_file.parent
 38
 39        # we set the current cycle manually instead of determining it from the archive
 40        # this means that the job will always be resubmitted to the "correct" (expected) cycle,
 41        # but may fail after starting if the archive does not contain files for this cycle
 42        self._advance_loop_cycle(informer)
 43
 44        submitter = self._build_submitter(informer, input_dir)
 45        hosts = (
 46            informer.info.resubmit_from
 47            # fall back to batch system default
 48            # this is only needed to accomodate transition from loop jobs submitted using previous versions of qq
 49            or informer.batch_system.get_default_resubmit_hosts()
 50        )
 51
 52        return self._try_resubmit(submitter, informer, hosts)
 53
 54    @staticmethod
 55    def _advance_loop_cycle(informer: Informer) -> None:
 56        """
 57        Advance the loop cycle counter if the job is a loop job.
 58
 59        Args:
 60            informer: The informer instance holding job metadata.
 61        """
 62        if informer.info.loop_info:
 63            informer.info.loop_info.current += 1
 64
 65    @staticmethod
 66    def _build_submitter(informer: Informer, input_dir: Path) -> Submitter:
 67        """
 68        Construct a Submitter configured for resubmission.
 69
 70        All original job parameters are preserved, except dependencies are replaced
 71        with a single dependency on the current job. This is because the previous
 72        cycle has already run, so its original dependencies must have been satisfied.
 73
 74        Args:
 75            informer (Informer): The informer instance holding job metadata.
 76            input_dir (Path): Path to the directory containing the job's input files.
 77
 78        Returns:
 79            Submitter: A configured submitter ready to submit the job.
 80        """
 81        return Submitter(
 82            batch_system=informer.batch_system,
 83            queue=informer.info.queue,
 84            account=informer.info.account,
 85            script=input_dir / informer.info.script_name,
 86            job_type=informer.info.job_type,
 87            resources=informer.info.resources,
 88            loop_info=informer.info.loop_info,
 89            exclude=[str(x) for x in informer.info.excluded_files],
 90            include=[str(x) for x in informer.info.included_files],
 91            ignore=[str(x) for x in informer.info.ignored_files],
 92            depend=[Depend(type=DependType.AFTER_SUCCESS, jobs=[informer.info.job_id])],
 93            transfer_mode=informer.info.transfer_mode,
 94            server=informer.info.server,
 95            interpreter=informer.info.interpreter,
 96            resubmit_from=informer.info.resubmit_from
 97            # fall back to batch system default
 98            # this is only needed to accommodate transition from loop jobs submitted using previous versions of qq
 99            or informer.batch_system.get_default_resubmit_hosts(),
100        )
101
102    @staticmethod
103    def _try_resubmit(
104        submitter: Submitter,
105        informer: Informer,
106        hosts: list[ResubmitHost],
107    ) -> str:
108        """
109        Attempt resubmission on each candidate host in order.
110
111        Args:
112            submitter (Submitter): The configured submitter to use for job submission.
113            informer (Informer): The informer instance holding job metadata.
114            hosts (list[ResubmitHost]): Ordered list of candidate resubmission hosts to try.
115
116        Returns:
117            str: The job ID of the newly submitted job.
118
119        Raises:
120            QQError: If the main node is not defined in the job metadata.
121            QQError: If the list of resubmission hosts is empty.
122            QQError: If resubmission fails on all candidate hosts.
123        """
124        # get the main node for host resolution
125        # since the job should be running, the main node should be defined
126        main_node = informer.info.main_node
127        if not main_node:
128            raise QQError(
129                "Job cannot be resubmitted. The 'main_node' of the job is not defined"
130            )
131
132        if not hosts:
133            raise QQError(
134                "Job cannot be resubmitted. No resubmission hosts defined. This is a bug, please report it"
135            )
136
137        for host in hosts:
138            hostname = host.resolve(informer.info.input_machine, main_node)
139            logger.info(f"Resubmitting from host '{hostname}'.")
140            try:
141                return Retryer(
142                    submitter.submit,
143                    remote=hostname,
144                    max_tries=CFG.resubmitter.retry_tries,
145                    wait_seconds=CFG.resubmitter.retry_wait,
146                ).run()
147            except Exception as e:
148                logger.warning(f"Failed resubmission from host '{hostname}': {e}")
149
150        raise QQError("Could not resubmit the job")