subprocess capture output of first 60 seconds

Question

I have a long running process (k8s job with logs) that I wish to run via subprocess.check_output. I only care about the output of the command in the first 60 seconds, so I've done this:

out = subprocess.check_output(
    shlex.split(command),
    timeout=60,
)

However, since the process runs for longer it ends with this exception. raise TimeoutExpired( subprocess.TimeoutExpired: Command ... . Is there a way to simply terminate the command at 60 seconds and capture the output. The output is important as I wish to do some post processing of the output in the first 60 seconds.

tripleee · Accepted Answer · 2022-02-15T08:27:23.627

You can make your own replacement using bare Popen but it's kind of cumbersome. Basically cribbing from Read streaming input from subprocess.communicate() and adding a timeout;

from subprocess import Popen, PIPE
import shlex
from datetime import datetime, timedelta

deadline = datetime.now() + timedelta(seconds=60)
with Popen(
    shlex.split(command), stdout=PIPE, bufsize=1,
    text=True, check=True
) as p:
    for line in p.stdout:
        print(line, end='')
        if datetime.now() >= deadline:
           break

Obviously, you can do something else with the output than print it -- probably collect it into a list so you can examine it after the loop.

The check=True is vaguely useless here, but mimics the behavior of check_output in that it will raise an exception if the subprocess fails. It would guard against typos in command etc, of course.

subprocess capture output of first 60 seconds

1 Answers1