uniworx/k8s-gitlab-borg
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

nonblocking operation

Browse Source
This commit is contained in:
Gregor Kleen 2023-05-29 17:06:18 +02:00
parent 9027c83f94
commit 6be4f57a49
1 changed files with 39 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
k8s_gitlab_borg/__main__.py
View File

@ -459,6 +459,10 @@ def main():
stdout_logger = proc_logger.getChild("stdout") stdout_logger = proc_logger.getChild("stdout")
stderr_logger = proc_logger.getChild("stderr") stderr_logger = proc_logger.getChild("stderr")
os.set_blocking(proc.stdin.fileno(), False)
os.set_blocking(proc.stdout.fileno(), False)
os.set_blocking(proc.stderr.fileno(), False)
poll = select.poll() poll = select.poll()
poll.register( poll.register(
proc.stdin, select.POLLOUT | select.POLLHUP | select.POLLERR proc.stdin, select.POLLOUT | select.POLLHUP | select.POLLERR
@ -482,6 +486,8 @@ def main():
events = poll.poll() events = poll.poll()
logger.debug("Done, %d event(s)", len(events)) logger.debug("Done, %d event(s)", len(events))
pipe_buffer = b"" pipe_buffer = b""
stdout_line_buffer = bytearray(b"")
stderr_line_buffer = bytearray(b"")
while pollc > 0 and len(events) > 0: while pollc > 0 and len(events) > 0:
for rfd, event in events: for rfd, event in events:
logger.debug("rfd=%d, event=%x", rfd, event) logger.debug("rfd=%d, event=%x", rfd, event)
@ -503,15 +509,33 @@ def main():
proc.stdin.close() proc.stdin.close()
if event & select.POLLIN or event & select.POLLPRI: if event & select.POLLIN or event & select.POLLPRI:
if rfd == proc.stdout.fileno(): if rfd == proc.stdout.fileno():
logger.debug("Reading line from stdout...") logger.debug("Reading from stdout...")
if line := proc.stdout.readline(): if chunk := proc.stdout.read(PIPE_BUF):
logger.debug("Done") logger.debug("Done, length %d", len(chunk))
stdout_logger.info(line[:-1]) stdout_line_buffer.append(chunk)
while True:
line, sep, rest = stdout_line_buffer.partition(
b"\n"
)
if sep:
stdout_logger.info(line)
stdout_line_buffer = rest
else:
break
if rfd == proc.stderr.fileno(): if rfd == proc.stderr.fileno():
logger.debug("Reading line from stderr...") logger.debug("Reading from stderr...")
if line := proc.stderr.readline(): if chunk := proc.stderr.read(PIPE_BUF):
logger.debug("Done") logger.debug("Done, length %d", len(chunk))
stderr_logger.info(line[:-1]) stderr_line_buffer.append(chunk)
while True:
line, sep, rest = stderr_line_buffer.partition(
b"\n"
)
if sep:
stderr_logger.info(line)
stderr_line_buffer = rest
else:
break
if event & select.POLLERR: if event & select.POLLERR:
if rfd == proc.stdin.fileno(): if rfd == proc.stdin.fileno():
logger.error("STDIN error") logger.error("STDIN error")
@ -536,6 +560,13 @@ def main():
else: else:
logger.debug("Nothing left to poll") logger.debug("Nothing left to poll")
if stdout_line_buffer:
for line in stdout_line_buffer.split(b"\n"):
stdout_logger.info(line)
if stderr_line_buffer:
for line in stderr_line_buffer.split(b"\n"):
stderr_logger.info(line)
for handler in proc_logger.handlers: for handler in proc_logger.handlers:
handler.flush() handler.flush()