nagios.py 文件源码

def run_plugin(executable: str, args: List[str], timeout: int) -> Tuple[str, List[str]]:
    run_args = [executable] + args
    try:
        proc = await asyncio.create_subprocess_exec(
            *run_args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    except FileNotFoundError:
        raise NagiosError('executable not found')
    stdin_data, stderr_data = await proc.communicate()
    std_data = stdin_data + stderr_data
    await proc.wait()
    if proc.returncode not in [STATUS_OK, STATUS_WARNING, STATUS_CRITICAL]:
        raise MonitorFailedError(std_data)
    text, perf = parse_plugin_output(std_data)
    if proc.returncode not in [STATUS_OK, STATUS_WARNING]:
        raise MonitorFailedError(text)
    return text, perf