如何用SaltStack Python客户端API实时输出Shell脚本执行日志并传至网页

问题 #

使用SaltStack Python客户端执行Shell脚本时，无法实时输出脚本的执行内容。现有代码如下，期望循环中持续打印日志，但实际无输出：

import salt.client

selector = "test"
client = salt.client.LocalClient()


def start_job(selector):
    return client.run_job(selector, "cmd.run", arg=["""for i in {1..50}; do echo "Log line $i at $(date '+%T')"; sleep 1; done"""])


def check_job(job):
    returned = client.get_cli_returns(job["jid"], job["minions"], timeout=30, verbose=True, show_jid=True)
    print(list(returned))
    for resp in returned:
        print(f"Response: {resp}")


if __name__ == "__main__":
    print("Immediate check -- fails")
    job = start_job(selector)
    print(job)
    check_job(job)

期望的实时输出效果：

Response: Log line 1 at 11:47:08
Response: Log line 2 at 11:47:09
Response: Log line 3 at 11:47:10
...

请问如何实现上述实时输出，或如何通过Python客户端API将命令执行过程传递到网页？

解决方案 #

方法1：利用Salt事件总线实时捕获输出 #

SaltStack的事件系统可以监听任务的实时输出，修改代码订阅对应任务的事件即可：

import salt.client
import salt.utils.event

selector = "test"
client = salt.client.LocalClient()


def start_job(selector):
    # 启动任务时开启stream参数，触发实时输出事件
    return client.run_job(
        selector,
        "cmd.run",
        arg=["""for i in {1..50}; do echo "Log line $i at $(date '+%T')"; sleep 1; done"""],
        kwargs={"stream": True}
    )


def listen_for_job_output(jid, minion_id):
    # 创建事件监听器，监听master端的任务输出事件
    event = salt.utils.event.get_event(
        "master",
        sock_dir="/var/run/salt/master",  # 根据实际master配置调整路径
        transport="tcp",
        opts=client.opts
    )

    while True:
        raw_event = event.get_event(full=True)
        if not raw_event:
            continue
        
        # 筛选当前任务的输出事件
        if (raw_event["tag"] == f"salt/job/{jid}/ret/{minion_id}" and 
            "data" in raw_event and "return" in raw_event["data"]):
            output_line = raw_event["data"]["return"]
            print(f"Response: {output_line}")
            # 检测任务结束（可根据脚本逻辑调整判断条件）
            if "Log line 50" in output_line:
                break


if __name__ == "__main__":
    job = start_job(selector)
    jid = job["jid"]
    minion_id = job["minions"][0]
    print(f"Started job with JID: {jid}")
    listen_for_job_output(jid, minion_id)

方法2：轮询任务进度获取实时输出 #

如果无法使用事件总线，可通过定期轮询任务状态来获取输出：

import salt.client
import time

selector = "test"
client = salt.client.LocalClient()


def start_job(selector):
    return client.run_job(
        selector,
        "cmd.run",
        arg=["""for i in {1..50}; do echo "Log line $i at $(date '+%T')"; sleep 1; done"""],
        kwargs={"stream": True, "output_loglevel": "trace"}
    )


def poll_job_output(jid, minion_id):
    seen_lines = set()
    while True:
        # 获取任务当前的输出内容
        job_result = client.get_job(jid)
        if minion_id not in job_result["return"]:
            time.sleep(1)
            continue
        
        output = job_result["return"][minion_id]
        # 分割输出为行，过滤已打印过的内容
        lines = output.split("\n")
        for line in lines:
            if line and line not in seen_lines:
                print(f"Response: {line}")
                seen_lines.add(line)
                if "Log line 50" in line:
                    return
        time.sleep(1)


if __name__ == "__main__":
    job = start_job(selector)
    jid = job["jid"]
    minion_id = job["minions"][0]
    print(f"Started job with JID: {jid}")
    poll_job_output(jid, minion_id)

传递到网页的实现思路 #

若要将实时输出传递到网页，可结合Web框架的WebSocket功能：

• 在后端启动Salt任务并通过上述方法捕获实时输出
• 利用WebSocket将每一行输出推送给前端页面
• 前端页面接收WebSocket消息后，实时更新页面显示内容

核心逻辑是把实时捕获输出的代码集成到Web框架的异步任务或线程中，通过WebSocket实现前后端的实时通信。

内容的提问来源于stack exchange，提问作者Tony