Repository: ansible/awx
File: awx/main/tasks/receptor.py (Python)

Code before:
                    detail = unit_status.get('Detail', None)
                    state_name = unit_status.get('StateName', None)
                    stdout_size = unit_status.get('StdoutSize', 0)
                except Exception:
                    detail = ''
                    state_name = ''
                    stdout_size = 0
                    logger.exception(f'An error was encountered while getting status for work unit {self.unit_id}')

                if 'exceeded quota' in detail:
                    logger.warning(detail)
                    log_name = self.task.instance.log_format
                    logger.warning(f"Could not launch pod for {log_name}. Exceeded quota.")
                    self.task.update_model(self.task.instance.pk, status='pending')
                    return

                try:
                    if state_name == 'Failed' and self.task.runner_callback.event_ct == 0:
                        # if receptor work unit failed and no events were emitted, work results may
                        # contain useful information about why the job failed. In case stdout is
                        # massive, only ask for last 1000 bytes
                        startpos = max(stdout_size - 1000, 0)
                        resultsock = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_sockfile=True)
                        lines = resultsock.readlines()
                        receptor_output = b"".join(lines).decode()
                        if receptor_output:
                            self.task.runner_callback.delay_update(result_traceback=receptor_output)
                    elif detail:
                        self.task.runner_callback.delay_update(result_traceback=detail)
                    else:
                        logger.warning(f'No result details or output from {self.task.instance.log_format}, status:\n{state_name}')
                except Exception:
                    raise RuntimeError(detail)

        return res

    # Spawned in a thread so Receptor can start reading before we finish writing, we
    # write our payload to the left side of our socketpair.
    @cleanup_new_process
    def transmit(self, _socket):
        try:
            ansible_runner.interface.run(streamer='transmit', _output=_socket.makefile('wb'), **self.runner_params)
        finally:
            # Socket must be shutdown here, or the reader will hang forever.
            _socket.shutdown(socket.SHUT_WR)

    @cleanup_new_process
    def processor(self, resultfile):
        return ansible_runner.interface.run(
            streamer='process',
            quiet=True,

Review comment (suggestion):
I'm worried about this hanging if the stream doesn't end with a `\n`. We may need to put the socket in non-blocking mode before doing this and just read the raw bytes.

Code after:
                result = namedtuple('result', ['status', 'rc'])
                res = result('canceled', 1)
            finally:
                signal_state.raise_exception = False

            if res.status == 'error':
                # If ansible-runner ran, but an error occured at runtime, the traceback information
                # is saved via the status_handler passed in to the processor.
                if 'result_traceback' in self.task.runner_callback.extra_update_fields:
                    return res

                try:
                    unit_status = receptor_ctl.simple_command(f'work status {self.unit_id}')
                    detail = unit_status.get('Detail', None)
                    state_name = unit_status.get('StateName', None)
                    stdout_size = unit_status.get('StdoutSize', 0)
                except Exception:
                    detail = ''
                    state_name = ''
                    stdout_size = 0
                    logger.exception(f'An error was encountered while getting status for work unit {self.unit_id}')

                if 'exceeded quota' in detail:
                    logger.warning(detail)
                    log_name = self.task.instance.log_format
                    logger.warning(f"Could not launch pod for {log_name}. Exceeded quota.")
                    self.task.update_model(self.task.instance.pk, status='pending')
                    return

                try:
                    receptor_output = ''
                    if state_name == 'Failed' and self.task.runner_callback.event_ct == 0:
                        # if receptor work unit failed and no events were emitted, work results may
                        # contain useful information about why the job failed. In case stdout is
                        # massive, only ask for last 1000 bytes
                        startpos = max(stdout_size - 1000, 0)
                        resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_socket=True, return_sockfile=True)
                        resultsock.setblocking(False)  # this makes resultfile reads non blocking
                        lines = resultfile.readlines()
                        receptor_output = b"".join(lines).decode()
                    if receptor_output:
                        self.task.runner_callback.delay_update(result_traceback=receptor_output)
                    elif detail:
                        self.task.runner_callback.delay_update(result_traceback=detail)
                    else:
                        logger.warning(f'No result details or output from {self.task.instance.log_format}, status:\n{state_name}')
                except Exception:
                    raise RuntimeError(detail)

        return res