如何实现HashiCorp Nomad raw_exec任务日志的增量拉取

核心问题在于你之前的请求没有指定具体日志文件，且全局复用一个偏移量——Nomad的日志轮转会生成新的归档文件（如stdout.0、stdout.1），每个文件的偏移量是独立的，必须跟踪每个文件的读取位置才能实现可靠的增量拉取。

关键解决方案步骤 #

1. 使用file参数指定具体日志文件：Nomad的/client/fs/logs API支持file参数，可指定拉取stdout、stderr或归档的stdout.N/stderr.N文件。
2. 跟踪每个文件的偏移量：用字典维护每个日志文件的上次读取偏移量，归档文件读取完毕（偏移量等于文件大小）后不再重复处理。
3. 定期扫描日志文件列表：每次轮询时先获取最新的日志文件列表，识别新增的归档文件并从0开始读取。

代码实现示例 #

import requests
import time

# 配置参数
NOMAD_API_URL = "http://your-nomad-client:4646/v1/"
ALLOC_ID = "your-alloc-id"
TASK_NAME = "your-task-name"
POLL_INTERVAL = 15  # 15秒轮询一次
LOG_TYPES = ["stdout", "stderr"]

# 状态维护：键为日志文件名（如"stdout"、"stdout.0"），值为上次读取的偏移量
log_file_offsets = {}
# 记录已读取完毕的归档文件（无需再轮询）
completed_files = set()

def get_log_files(log_type):
    """获取指定日志类型的所有文件列表"""
    resp = requests.get(
        f"{NOMAD_API_URL}client/fs/ls/{ALLOC_ID}",
        params={"task": TASK_NAME, "path": "logs"}
    )
    resp.raise_for_status()
    files = []
    for entry in resp.json():
        if entry["Name"].startswith(log_type):
            files.append(entry["Name"])
    # 按文件编号排序：stdout.0 < stdout.1 < stdout（最新）
    files.sort(key=lambda x: int(x.split(".")[-1]) if "." in x else -1)
    return files

def get_file_size(log_file):
    """获取日志文件的大小"""
    resp = requests.get(
        f"{NOMAD_API_URL}client/fs/stat/{ALLOC_ID}",
        params={"task": TASK_NAME, "path": f"logs/{log_file}"}
    )
    resp.raise_for_status()
    return resp.json()["Size"]

def pull_file_logs(log_type, log_file, offset):
    """拉取指定文件的增量日志"""
    resp = requests.get(
        f"{NOMAD_API_URL}client/fs/logs/{ALLOC_ID}",
        params={
            "task": TASK_NAME,
            "type": log_type,
            "origin": "start",
            "offset": offset,
            "file": log_file
        }
    )
    resp.raise_for_status()
    data = resp.json()
    return data["Data"], data["LastOffset"]

def main():
    while True:
        for log_type in LOG_TYPES:
            current_files = get_log_files(log_type)
            for log_file in current_files:
                # 跳过已读取完毕的归档文件
                if log_file in completed_files:
                    continue

                # 初始化新文件的偏移量
                if log_file not in log_file_offsets:
                    log_file_offsets[log_file] = 0

                # 拉取增量日志
                log_content, last_offset = pull_file_logs(log_type, log_file, log_file_offsets[log_file])
                if log_content:
                    print(f"[{log_type}] {log_file}:\n{log_content}")

                # 更新偏移量
                log_file_offsets[log_file] = last_offset

                # 检查归档文件是否已读取完毕
                if "." in log_file:
                    file_size = get_file_size(log_file)
                    if last_offset >= file_size:
                        completed_files.add(log_file)

        time.sleep(POLL_INTERVAL)

if __name__ == "__main__":
    main()

关键细节说明 #

• 文件排序：归档文件编号越大越旧，排序时优先处理旧的归档文件，再处理当前的stdout/stderr，保证日志顺序正确。
• 归档文件标记：带编号的归档文件不会再新增内容，当读取偏移量等于文件大小时，标记为已完成，避免重复请求。
• 错误处理：示例中使用raise_for_status()处理API请求错误，实际生产环境可根据需求添加重试或异常捕获逻辑。

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