@@ -0,0 +1,2424 @@
+from twisted.internet import defer, error, protocol%0A%0Aimport json%0Aimport logging%0Aimport sys%0A%0A%0ALOG = logging.getLogger(__name__)%0A%0A%0Aclass ExecutorMonitor(protocol.ProcessProtocol):%0A    def __init__(self, data_to_send, log_file=None):%0A        self.data_to_send = data_to_send%0A        self.log_file = log_file%0A%0A        self.job_id_deferred = defer.Deferred()%0A        self.job_ended_deferred = defer.Deferred()%0A%0A        self._job_id_data = %5B%5D%0A        self._log_file_handle = None%0A%0A    def connectionMade(self):%0A        self.transport.write(json.dumps(self.data_to_send))%0A        self.transport.closeStdin()%0A%0A    @property%0A    def log_file_handle(self):%0A        if not self._log_file_handle:%0A            try:%0A                LOG.debug('Trying to open log file in %25s: %25s',%0A                        self.__class__.__name__, self.log_file)%0A                self._log_file_handle = open(self.log_file, 'a')%0A            except:%0A                LOG.info('Could not open log file %25r', self.log_file)%0A                self._log_file_handle = sys.stderr%0A%0A        return self._log_file_handle%0A%0A    def close_log_file(self):%0A        if self._log_file_handle:%0A            self._log_file_handle.close()%0A            self._log_file_handle = None%0A%0A%0A    def childDataReceived(self, childFD, data):%0A        if childFD == 3:%0A            self._job_id_data.append(data)%0A        else:%0A            self.log_file_handle.write(data)%0A            self.log_file_handle.flush()%0A%0A    def childConnectionLost(self, childFD):%0A        if childFD == 3:%0A            self._job_id_connection_lost()%0A        else:%0A            LOG.warning('Lost connection with child FD %25d', childFD)%0A%0A    def _job_id_connection_lost(self):%0A        job_id = ''.join(self._job_id_data)%0A        if job_id:%0A            self.job_id_deferred.callback(job_id)%0A        else:%0A            self.job_id_deferred.errback(RuntimeError('No job id received'))%0A%0A    def processEnded(self, exit_status):%0A        try:%0A            if isinstance(exit_status.value, error.ProcessDone):%0A                self.job_ended_deferred.callback(None)%0A            else:%0A                # XXX Is this a very useful log message?  Add a timestamp?%0A                self.log_file_handle.write('Executor failed with exit_status %25r'%0A                        %25 exit_status)%0A                self.log_file_handle.flush()%0A                self.job_ended_deferred.errback(exit_status)%0A        finally:%0A            self.close_log_file()%0A

add ProcessProtocol for monitoring executors