def _handle_failure()

in src/buildstream/_frontend/app.py [0:0]

• 72 lines of code
• 19 McCabe index (conditional complexity)

    def _handle_failure(self, element, task, failure):
        full_name = task.full_name

        # Handle non interactive mode setting of what to do when a job fails.
        if not self._interactive_failures:

            if self.context.sched_error_action == _SchedulerErrorAction.TERMINATE:
                self.stream.terminate()
            elif self.context.sched_error_action == _SchedulerErrorAction.QUIT:
                self.stream.quit()
            elif self.context.sched_error_action == _SchedulerErrorAction.CONTINUE:
                pass
            return

        # Interactive mode for element failures
        with self._interrupted():

            summary = (
                "\n{} failure on element: {}\n".format(failure.action_name, full_name)
                + "\n"
                + "Choose one of the following options:\n"
                + "  (c)ontinue  - Continue queueing jobs as much as possible\n"
                + "  (q)uit      - Exit after all ongoing jobs complete\n"
                + "  (t)erminate - Terminate any ongoing jobs and exit\n"
                + "  (r)etry     - Retry this job\n"
            )
            if failure.logfile:
                summary += "  (l)og       - View the full log file\n"
            if failure.sandbox:
                summary += "  (s)hell     - Drop into a shell in the failed build sandbox\n"
            summary += "\nPressing ^C will terminate jobs and exit\n"

            choices = ["continue", "quit", "terminate", "retry"]
            if failure.logfile:
                choices += ["log"]
            if failure.sandbox:
                choices += ["shell"]

            choice = ""
            while choice not in ["continue", "quit", "terminate", "retry"]:
                click.echo(summary, err=True)

                self._notify("BuildStream failure", "{} on element {}".format(failure.action_name, full_name))

                try:
                    choice = click.prompt(
                        "Choice:", default="continue", err=True, value_proc=_prefix_choice_value_proc(choices)
                    )
                except (click.Abort, SystemError):
                    # In some cases, the readline buffer underlying the prompt gets corrupted on the second CTRL+C
                    # This throws a SystemError, which doesn't seem to be problematic for the rest of the program

                    # Ensure a newline after automatically printed '^C'
                    click.echo("", err=True)
                    choice = "terminate"

                # Handle choices which you can come back from
                #
                if choice == "shell":
                    click.echo("\nDropping into an interactive shell in the failed build sandbox\n", err=True)
                    try:
                        unique_id, _ = element
                        self.stream.shell(
                            None,
                            _Scope.BUILD,
                            self.shell_prompt,
                            isolate=True,
                            usebuildtree=True,
                            unique_id=unique_id,
                        )
                    except BstError as e:
                        click.echo("Error while attempting to create interactive shell: {}".format(e), err=True)
                elif choice == "log":
                    with open(failure.logfile, "r", encoding="utf-8") as logfile:
                        content = logfile.read()
                        click.echo_via_pager(content)

            if choice == "terminate":
                click.echo("\nTerminating all jobs\n", err=True)
                self.stream.terminate()
            else:
                if choice == "quit":
                    click.echo("\nCompleting ongoing tasks before quitting\n", err=True)
                    self.stream.quit()
                elif choice == "continue":
                    click.echo("\nContinuing with other non failing elements\n", err=True)
                elif choice == "retry":
                    click.echo("\nRetrying failed job\n", err=True)
                    unique_id = element[0]
                    self.stream.retry_job(task.action_name, unique_id)