自动化任务服务执行外部命令：超时取消、输出限额与子进程清理

AI智能摘要

可靠的外部命令执行不能把父进程退出视为任务结束：未排空的输出管道会阻塞子进程，超时也未必停止进程树。文章给出 Python 实现，以参数数组且不经 shell 启动命令，分别持续读取标准输出和错误输出、仅保留设定字节限额并丢弃多余内容；超时或取消时先终止进程组，宽限期后强制清理，同时区分执行与读写错误。该方案依赖 POSIX 进程组，命令不得自行脱离进程组后台化。

— 此摘要由AI分析文章内容生成，仅供参考。

任务服务调用外部命令时，超时并不等于进程已经停止：父进程可能退出，子进程仍在运行；如果输出管道没有持续读取，标准输出或错误输出还可能让进程阻塞。可靠的执行边界需要同时处理启动、持续排空输出、超时或取消，以及整个进程组的清理。

先明确执行契约

下面的最小实现采用 Python，约定：

命令以参数数组启动，不经 shell 解析，避免字符串拼接带来的转义和注入问题。
标准输出、标准错误分别保留最多指定字节数；超过限额后丢弃多余内容，但继续读取管道，避免子进程因管道写满而卡住。
超时和取消都会先向进程组发送终止信号，等待宽限期后再强制结束。
任务拥有命令启动的整个进程组。命令不得脱离进程组自行后台化；否则服务无法通过进程组信号保证清理。
示例依赖 POSIX 进程组能力。若运行环境不支持该机制，应替换对应的进程树控制实现，不能只终止父进程后就认为清理完成。

最小实现：限额采集、超时取消与进程组清理

from __future__ import annotations

import os
import signal
import subprocess
import threading
import time
from dataclasses import dataclass
from typing import Optional, Sequence


@dataclass
class CommandResult:
    # status: success、nonzero、timeout、cancelled、launch_error 或 io_error
    status: str
    returncode: Optional[int]
    stdout: bytes
    stderr: bytes
    stdout_truncated: bool
    stderr_truncated: bool
    error: Optional[str] = None


class _Capture:
    """持续排空管道，只保留前 limit 个字节。"""

    def __init__(self, pipe, limit: int):
        self.pipe = pipe
        self.limit = limit
        self.parts: list[bytes] = []
        self.size = 0
        self.truncated = False
        self.error: Optional[str] = None

    def read_all(self) -> None:
        try:
            while True:
                chunk = self.pipe.read(8192)
                if not chunk:
                    break

                remaining = self.limit - self.size
                if remaining > 0:
                    kept = chunk[:remaining]
                    self.parts.append(kept)
                    self.size += len(kept)

                if len(chunk) > max(remaining, 0):
                    self.truncated = True
        except OSError as exc:
            self.error = str(exc)
        finally:
            self.pipe.close()

    def value(self) -> bytes:
        return b"".join(self.parts)


def _signal_group(proc: subprocess.Popen, sig: int) -> None:
    try:
        os.killpg(proc.pid, sig)
    except ProcessLookupError:
        pass


def _stop_group(proc: subprocess.Popen, grace: float) -> None:
    """先请求退出，再在宽限期结束后强制终止进程组。"""
    _signal_group(proc, signal.SIGTERM)

    deadline = time.monotonic() + grace
    while time.monotonic() < deadline:
        if proc.poll() is not None:
            # 父进程退出不代表组内子进程已经退出，因此仍保留完整宽限期。
            time.sleep(min(0.02, max(0.0, deadline - time.monotonic())))
        else:
            time.sleep(min(0.02, max(0.0, deadline - time.monotonic())))

    _signal_group(proc, signal.SIGKILL)
    proc.wait()


def _join_readers(readers: list[threading.Thread], timeout: float) -> None:
    deadline = time.monotonic() + timeout
    for reader in readers:
        reader.join(max(0.0, deadline - time.monotonic()))


def run_command(
    argv: Sequence[str],
    *,
    timeout: float,
    cancel_event: Optional[threading.Event] = None,
    output_limit: int = 1_000_000,
    terminate_grace: float = 0.5,
) -> CommandResult:
    """每个输出流最多保留 output_limit 字节；输入参数不应来自未校验的用户命令字符串。"""
    if timeout <= 0:
        raise ValueError("timeout 必须大于 0")
    if output_limit < 0:
        raise ValueError("output_limit 不能小于 0")
    if terminate_grace < 0:
        raise ValueError("terminate_grace 不能小于 0")

    try:
        proc = subprocess.Popen(
            list(argv),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=False,
            start_new_session=True,
        )
    except OSError as exc:
        return CommandResult(
            status="launch_error",
            returncode=None,
            stdout=b"",
            stderr=b"",
            stdout_truncated=False,
            stderr_truncated=False,
            error=str(exc),
        )

    assert proc.stdout is not None
    assert proc.stderr is not None

    out_capture = _Capture(proc.stdout, output_limit)
    err_capture = _Capture(proc.stderr, output_limit)
    readers = [
        threading.Thread(target=out_capture.read_all, daemon=True),
        threading.Thread(target=err_capture.read_all, daemon=True),
    ]
    for reader in readers:
        reader.start()

    deadline = time.monotonic() + timeout
    status: Optional[str] = None

    while proc.poll() is None:
        if cancel_event is not None and cancel_event.is_set():
            status = "cancelled"
            _stop_group(proc, terminate_grace)
            break

        if time.monotonic() >= deadline:
            status = "timeout"
            _stop_group(proc, terminate_grace)
            break

        time.sleep(0.02)

    # 即使命令主进程正常退出，若其子进程仍持有输出管道，读取线程也可能不结束。
    _join_readers(readers, timeout=0.2)
    if any(reader.is_alive() for reader in readers):
        _stop_group(proc, terminate_grace)
        _join_readers(readers, timeout=terminate_grace + 0.5)

    read_error = out_capture.error or err_capture.error
    if any(reader.is_alive() for reader in readers):
        read_error = read_error or "输出读取线程未能在清理期限内退出"

    returncode = proc.poll()
    if status is None:
        if read_error:
            status = "io_error"
        elif returncode == 0:
            status = "success"
        else:
            status = "nonzero"

    return CommandResult(
        status=status,
        returncode=returncode,
        stdout=out_capture.value(),
        stderr=err_capture.value(),
        stdout_truncated=out_capture.truncated,
        stderr_truncated=err_capture.truncated,
        error=read_error,
    )

output_limit 是每个流的上限，因此保留的输出总量最多约为其两倍。读取线程会持续排空管道，即使已经达到上限也不会停止读取。这样既限制了内存，又避免输出量大的子进程因管道反压而挂起。

调用方应根据 status 做明确处理，而不是只判断返回码：timeout 和 cancelled 表示执行被中断，nonzero 表示命令以非零状态退出，launch_error 表示命令未能启动。截断标记也应传递给日志或任务结果，避免把不完整输出误认为完整诊断信息。

异常路径与实现边界

启动失败：可执行文件不存在、权限不足等错误通过 launch_error 返回；此时没有可清理的子进程。
超时与取消同时发生：循环先检查取消事件，因此两者同时满足时结果为 cancelled。如果业务要求固定优先级，应调整检查顺序并通过测试固化。
非零退出：保留退出码和已采集的输出，不要把所有失败都压成同一种异常，否则调用方无法区分命令失败与执行器故障。
输出截断：限额只限制内存保留量，不限制命令生成的总字节数。若还要限制磁盘、CPU 或总运行时间，需要在命令执行边界增加相应策略。
进程组逃逸：进程若主动创建新的会话或把工作交给组外服务，进程组清理无法覆盖它。对这类命令，应禁止后台化，或改用平台提供的专门进程树管理能力。
取消信号来源：HTTP 请求断开、任务被撤销等事件应转化为 threading.Event，并确保执行线程能及时观察到该事件。

可复现测试：检查结果与资源边界

以下用例可以放在测试文件中，假设实现保存在 command_runner.py。它们验证成功、非零退出、输出截断、超时、取消，以及父进程结束后子进程是否继续写文件。

import os
import sys
import tempfile
import threading
import time
import unittest

from command_runner import run_command


class CommandRunnerTests(unittest.TestCase):
    def test_success_and_nonzero(self):
        ok = run_command(
            [sys.executable, "-c", "print('ok')"],
            timeout=3,
        )
        self.assertEqual(ok.status, "success")
        self.assertIn(b"ok", ok.stdout)

        failed = run_command(
            [sys.executable, "-c", "import sys; print('bad'); sys.exit(7)"],
            timeout=3,
        )
        self.assertEqual(failed.status, "nonzero")
        self.assertEqual(failed.returncode, 7)
        self.assertIn(b"bad", failed.stdout)

    def test_output_limit_keeps_draining(self):
        result = run_command(
            [
                sys.executable,
                "-c",
                "import sys; sys.stdout.write('x' * 200000); "
                "sys.stderr.write('y' * 200000)",
            ],
            timeout=3,
            output_limit=1024,
        )
        self.assertEqual(result.status, "success")
        self.assertEqual(len(result.stdout), 1024)
        self.assertEqual(len(result.stderr), 1024)
        self.assertTrue(result.stdout_truncated)
        self.assertTrue(result.stderr_truncated)

    def test_timeout_and_cancel(self):
        timed = run_command(
            [sys.executable, "-c", "import time; time.sleep(30)"],
            timeout=0.2,
            terminate_grace=0.1,
        )
        self.assertEqual(timed.status, "timeout")

        cancel = threading.Event()
        threading.Timer(0.2, cancel.set).start()
        cancelled = run_command(
            [sys.executable, "-c", "import time; time.sleep(30)"],
            timeout=5,
            cancel_event=cancel,
            terminate_grace=0.1,
        )
        self.assertEqual(cancelled.status, "cancelled")

    def test_timeout_cleans_up_child_process(self):
        with tempfile.TemporaryDirectory() as directory:
            path = os.path.join(directory, "heartbeat.txt")
            child = (
                "import time; "
                f"f = open({path!r}, 'a', buffering=1); "
                "while True: f.write('x'); time.sleep(0.03)"
            )
            parent = (
                "import subprocess, sys, time; "
                "subprocess.Popen([sys.executable, '-c', sys.argv[1]]); "
                "time.sleep(30)"
            )

            result = run_command(
                [sys.executable, "-c", parent, child],
                timeout=0.3,
                terminate_grace=0.1,
            )
            self.assertEqual(result.status, "timeout")

            size_after_return = os.path.getsize(path)
            time.sleep(0.3)
            self.assertEqual(os.path.getsize(path), size_after_return)


if __name__ == "__main__":
    unittest.main()

心跳测试不依赖进程 ID 是否暂时显示为已存在，而是验证任务返回后文件是否还在增长：若子进程仍运行，断言会失败。实际部署中还应验证日志中能区分超时、取消、非零退出和输出截断，并在压力测试里观察执行期间的内存上限与取消后的进程残留。

发布者：jacky，转转请注明出处：https://kubiyun.com/archives/4623