沙箱指定路径日志采集最佳实践（Python）

方案原理

本方案在用户自定义Sandbox镜像基础上增加“日志采集”能力，通过旁路进程监听指定目录的文件变化，增量读取新增日志行，并通过LTS Producer SDK异步上报到华为云LTS。

整体工作流程如下：

1. 业务写入日志：用户应用按原有方式将日志写入指定目录（如/var/log/app/）
2. watchdog监听变化：监听进程通过watchdog库检测到目录内文件的创建和修改事件
3. 增量读取：从状态文件记录的上次偏移量位置读取新增的完整日志行
4. 异步上报：通过Producer SDK将日志行批量发送到LTS API，由LTS完成存储和索引

图1 工作流程

前提条件

场景化示例

假设有一个AI推理沙箱，业务应用将推理日志写入/var/log/inference/目录，包含以下文件：

• app.log：主应用运行日志，持续追加
• tasks/：子目录，存放每个推理任务的独立日志
• audit.log：审计日志，记录每次API调用

配置LOG_WATCH_DIR=/var/log/inference后，watcher会递归监听该目录下所有文件，将新增日志行上报到LTS。在LTS控制台可以通过path字段区分日志来源（如/var/log/inference/app.log或/var/log/inference/tasks/task-001.log）。

步骤一：准备SDK源码和依赖包

获取LTS Python SDK源码和运行所需的Python依赖包，为补丁镜像提供日志上报和文件监听能力。

1. 获取LTS Python SDK源码

   git clone https://gitee.com/lordstar-habile/huaweicloud-lts-python-sdk.git 

   Producer组件位于源码的producer/目录下。将SDK源码目录（包含producer/、utils/、auth/等）放入构建上下文中。

2. 准备Python依赖

   Producer SDK依赖loguru库，文件监听依赖watchdog库。根据沙箱的网络情况选择对应方式安装。

   • 方式A：沙箱可联网——直接在Dockerfile中安装：

     RUN pip install watchdog loguru requests

   • 方式B：沙箱无法联网——需要提前在能联网的机器上下载依赖。

     推荐使用--target目录方式：

     pip install watchdog loguru requests \
         --target ./python-packages \
         --no-cache-dir 

     将python-packages/目录放入构建上下文，运行时通过PYTHONPATH指定即可。

     备选方式：下载whl包后离线安装

     pip download loguru watchdog requests -d ./wheels 

     将wheels/目录上传后，在Dockerfile中离线安装：

     COPY wheels/ /opt/sandbox-log/wheels/
     RUN pip install --no-index /opt/sandbox-log/wheels/*.whl

   

   如果目标镜像的Python大版本或CPU架构与下载机器不同，请在与目标运行环境一致的机器上重新下载依赖，否则可能出现兼容性问题。

3. 验证方法
   在构建上下文目录中确认以下文件存在：

   文件	说明
   huaweicloud-lts-python-sdk/producer/	LTS SDK的Producer组件
   python-packages/watchdog/	文件监听库
   python-packages/loguru/	日志库

步骤二：编写日志服务配置文件

将LTS认证信息和采集参数集中在一个配置文件中，便于管理和权限控制。配置文件包含敏感凭据，权限设为0600，仅业务用户可读。

1. 创建logging.env文件，填入华为云配置信息：

   # LTS认证配置（必填）
   LTS_ACCESS_KEY='你的AK'
   LTS_ACCESS_SECRET='你的SK'
   LTS_PROJECT_ID='你的ProjectID'
   LTS_LOG_GROUP_ID='你的日志组ID'
   LTS_LOG_STREAM_ID='你的日志流ID'
   
   # LTS接入点配置
   LTS_ENDPOINT='https://lts-access.cn-south-1.myhuaweicloud.com:8102'
   LTS_REGION_ID='cn-south-1'
   
   # 采集配置
   LOG_WATCH_DIR='/var/log/app'                     # 要监听的目录（递归监听）
   LOG_STATE_FILE='/var/lib/log-watcher/state.json' # 状态文件路径
   LOG_SOURCE='sandbox-file-watcher'                # 日志来源标识
   LOG_TAIL_FROM_END='false'                        # true: 只采集启动后的新增内容
   LOG_WATCH_USER='node'                            # 运行业务的用户（可选）

   

   高危操作提示：logging.env包含AK/SK等敏感凭据，泄露后可能导致云资源被未授权访问。请务必：

   • 不要提交到源码仓库，建议在.dockerignore中排除
   • 构建日志、watcher日志和诊断命令只输出配置是否存在，绝不输出配置值
   • 配置发生变更时，重新构建并发布补丁镜像
2. 验证方法
   执行以下命令确认配置文件内容正确（不输出具体值）：

   # 确认必填项已设置（只检查变量是否存在，不输出值）
   grep -c 'LTS_ACCESS_KEY\|LTS_ACCESS_SECRET\|LTS_PROJECT_ID\|LTS_LOG_GROUP_ID\|LTS_LOG_STREAM_ID' logging.env 
   # 预期输出：5

步骤三：编写文件监听程序

监听程序是日志采集的核心组件，负责监听目录变化、增量读取日志行、通过Producer SDK异步上报到LTS。它使用watchdog库监听文件系统事件，通过状态文件记录每个文件的读取偏移量，确保重启后不重复发送。

创建log-watcher.py文件：

#!/usr/bin/env python3
"""Send appended lines from a Sandbox directory to Huawei Cloud LTS."""

import json
import logging
import os
import queue
import socket
import tempfile
import time
from pathlib import Path

from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer

from producer.core.producer import Producer
from producer.model import lts_store
from producer.model.config import Config


REQUIRED = (
    "LTS_ACCESS_KEY",
    "LTS_ACCESS_SECRET",
    "LTS_PROJECT_ID",
    "LTS_LOG_GROUP_ID",
    "LTS_LOG_STREAM_ID",
)

class Events(FileSystemEventHandler):
    def __init__(self, changed: queue.Queue[str]):
        self.changed = changed

    def _add(self, path: str, is_directory: bool) -> None:
        if not is_directory:
            self.changed.put(path)

    def on_created(self, event: FileSystemEvent) -> None:
        self._add(event.src_path, event.is_directory)

    def on_modified(self, event: FileSystemEvent) -> None:
        self._add(event.src_path, event.is_directory)

    def on_moved(self, event: FileSystemEvent) -> None:
        self._add(event.dest_path, event.is_directory)


def configure_logging() -> None:
    logging.basicConfig(
        level=os.getenv("LOG_WATCH_LOG_LEVEL", "INFO").upper(),
        format="%(asctime)s %(levelname)s %(message)s",
    )


def required(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"missing required configuration: {name}")
    return value


def make_producer() -> tuple[Producer, str, str]:
    missing = [name for name in REQUIRED if not os.getenv(name)]
    if missing:
        raise RuntimeError("missing required configuration: " + ", ".join(missing))

    config = Config()
    config.endpoint = os.getenv("LTS_ENDPOINT", "https://lts-access.cn-south-1.myhuaweicloud.com:8102")
    config.region_id = os.getenv("LTS_REGION_ID", "cn-south-1")
    config.project_id = required("LTS_PROJECT_ID")
    config.access_key = required("LTS_ACCESS_KEY")
    config.access_secret = required("LTS_ACCESS_SECRET")
    producer = Producer.init_producer(config)
    producer.start_producer()
    return producer, required("LTS_LOG_GROUP_ID"), required("LTS_LOG_STREAM_ID")


def load_state(path: Path) -> dict[str, dict]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data.get("files", {}) if isinstance(data, dict) else {}
    except FileNotFoundError:
        return {}
    except (OSError, ValueError) as error:
        logging.warning("cannot read state file %s: %s", path, error)
        return {}


def save_state(path: Path, state: dict[str, dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump({"version": 1, "files": state}, handle, ensure_ascii=False)
            handle.write("\n")
        os.replace(temporary, path)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def file_key(stat: os.stat_result) -> str:
    return f"{stat.st_dev}:{stat.st_ino}"


def read_complete_lines(path: Path, offset: int) -> tuple[bytes, int]:
    with path.open("rb") as handle:
        handle.seek(offset)
        data = handle.read()
    newline = data.rfind(b"\n")
    if newline < 0:
        return b"", offset
    complete = data[: newline + 1]
    return complete, offset + len(complete)


def forward(
    producer: Producer,
    group_id: str,
    stream_id: str,
    path: Path,
    state: dict[str, dict],
    tail_from_end: bool,
) -> bool:
    try:
        stat = path.stat()
        if not path.is_file():
            return False
    except OSError:
        return False
    key = file_key(stat)
    entry = state.setdefault(
        key,
        {"offset": stat.st_size if tail_from_end else 0, "line_no": 0, "path": str(path)},
    )
    offset = int(entry["offset"])
    if stat.st_size < offset:
        offset = 0
        entry["line_no"] = 0
    try:
        data, next_offset = read_complete_lines(path, offset)
    except OSError as error:
        logging.warning("cannot read %s: %s", path, error)
        return False
    if not data:
        return False

    labels = {
        "source": os.getenv("LOG_SOURCE", "sandbox-file-watcher"),
        "path": str(path),
        "host_name": socket.gethostname(),
        "ip": os.getenv("LOG_SOURCE_IP", "127.0.0.1"),
    }
    line_no = int(entry["line_no"])
    for raw_line in data.splitlines():
        if not raw_line:
            continue
        line_no += 1
        message = raw_line.decode("utf-8", errors="replace")
        result = producer.send_log(
            group_id,
            stream_id,
            lts_store.generate_log([message], {**labels, "line_no": str(line_no)}),
        )
        if getattr(result, "error_code", 0):
            logging.warning("LTS enqueue failed for %s line %d", path, line_no)
            return False

    entry.update({"offset": next_offset, "line_no": line_no, "path": str(path)})
    logging.info("forwarded %s through line %d", path, line_no)
    return True


def drain(changed: queue.Queue[str]) -> set[Path]:
    paths: set[Path] = set()
    while True:
        try:
            paths.add(Path(changed.get_nowait()))
        except queue.Empty:
            return paths


def ensure_watch_dir(watch_dir: Path) -> bool:
    if watch_dir.is_dir():
        return True
    try:
        watch_dir.mkdir(parents=True, exist_ok=True)
        logging.info("created watch directory: %s", watch_dir)
        return True
    except OSError as error:
        logging.error("cannot create watch directory %s: %s", watch_dir, error)
        logging.error("please check permissions or create the directory manually")
        return False


def run() -> None:
    configure_logging()
    watch_dir = Path(os.getenv("LOG_WATCH_DIR", "/var/log/app"))
    state_path = Path(os.getenv("LOG_STATE_FILE", "/var/lib/log-watcher/state.json"))
    tail_from_end = os.getenv("LOG_TAIL_FROM_END", "false").lower() in {"1", "true", "yes", "on"}
    producer, group_id, stream_id = make_producer()
    state = load_state(state_path)

    if not ensure_watch_dir(watch_dir):
        return

    while True:
        if not watch_dir.is_dir():
            logging.warning("watch directory removed: %s, attempting to recreate", watch_dir)
            if not ensure_watch_dir(watch_dir):
                return

        changed: queue.Queue[str] = queue.Queue()
        observer = Observer()
        observer.schedule(Events(changed), str(watch_dir), recursive=True)
        observer.start()
        logging.info("watching log directory: %s", watch_dir)
        try:
            pending = {path for path in watch_dir.rglob("*") if path.is_file()}
            while watch_dir.is_dir():
                try:
                    pending.add(Path(changed.get(timeout=1)))
                except queue.Empty:
                    pass
                pending.update(drain(changed))
                saved = False
                for path in list(pending):
                    try:
                        saved = forward(producer, group_id, stream_id, path, state, tail_from_end) or saved
                    except OSError as error:
                        logging.warning("error processing %s: %s", path, error)
                        pending.discard(path)
                pending.clear()
                if saved:
                    save_state(state_path, state)
        finally:
            observer.stop()
            observer.join(timeout=5)
if __name__ == "__main__":
    run()

核心机制说明：

• 文件追踪：使用dev:ino（设备号:inode号）作为文件唯一标识，正确处理文件轮转和重命名
• 增量读取：通过状态文件保存每个文件的读取偏移量，重启后从上次位置继续
• 完整行处理：只发送以换行符结束的完整行，避免截断日志
• 目录自动创建：如果监听目录不存在，程序会尝试自动创建
• 状态持久化：状态文件使用原子写入（先写临时文件再重命名），避免数据损坏

步骤四：编写包装入口脚本

包装入口脚本是补丁镜像与业务镜像的桥梁。它在后台启动监听进程，然后启动原始业务入口，确保业务程序按原有方式运行，同时采集进程并行工作。当业务进程退出时，自动清理采集进程。

创建log-entrypoint.sh文件：

您需要将ORIGINAL_ENTRYPOINT的默认值改为基础镜像的原始入口脚本路径。也可通过环境变量覆盖，无需改脚本。

#!/usr/bin/env bash
set -euo pipefail
readonly LOGGING_ENV_FILE="/opt/sandbox-log/logging.env"
readonly ORIGINAL_ENTRYPOINT="${ORIGINAL_ENTRYPOINT:-/opt/onyxclaw/bin/entrypoint.sh}"
readonly RUN_AS_USER="${LOG_WATCH_USER:-node}"

if [[ ! -r "${LOGGING_ENV_FILE}" ]]; then
  echo "log collector configuration is missing: ${LOGGING_ENV_FILE}" >&2
  exit 1
fi

set -a
# shellcheck disable=SC1090
source "${LOGGING_ENV_FILE}"
set +a

watcher_pid=""
business_pid=""

stop_children() {
  for pid in "${watcher_pid}" "${business_pid}"; do
    if [[ -n "${pid}" ]] && kill -0 "${pid}" 2>/dev/null; then
      kill "${pid}" 2>/dev/null || true
    fi
  done
}

trap stop_children INT TERM

start_watcher() {
  echo "starting sandbox log watcher for ${LOG_WATCH_DIR}"
  setpriv --reuid="${RUN_AS_USER}" --regid="${RUN_AS_USER}" --init-groups \
    env PYTHONPATH="/opt/sandbox-log/python-packages:/opt/sandbox-log/lts-sdk${PYTHONPATH:+:${PYTHONPATH}}" \
    python3 -u /opt/sandbox-log/log-watcher.py &
  watcher_pid=$!
}

start_watcher
"${ORIGINAL_ENTRYPOINT}" "$@" &
business_pid=$!

set +e
wait "${business_pid}"
status=$?
set -e
stop_children
wait "${watcher_pid}" 2>/dev/null || true
exit "${status}"

步骤五：编写Dockerfile

继承用户原始镜像，只叠加采集所需的文件和依赖，不删除、覆盖或替换原始业务程序。构建时自动验证依赖可导入，失败会中断构建，避免问题留到运行时。

创建Dockerfile文件：

ARG BASE_IMAGE
FROM ${BASE_IMAGE}

ARG APP_USER=node
USER root

# 如果基础镜像没有tini和setpriv，取消下面的注释安装
# RUN apt-get update && apt-get install -y tini util-linux && rm -rf /var/lib/apt/lists/*

# 复制离线依赖和SDK
COPY python-packages/ /opt/sandbox-log/python-packages/
COPY huaweicloud-lts-python-sdk/ /opt/sandbox-log/lts-sdk/

# 复制采集脚本和入口脚本
COPY log-watcher.py /opt/sandbox-log/log-watcher.py
COPY log-entrypoint.sh /opt/sandbox-log/log-entrypoint.sh
COPY logging.env /opt/sandbox-log/logging.env

# 设置权限并验证依赖可导入
RUN chmod 0755 /opt/sandbox-log/log-watcher.py /opt/sandbox-log/log-entrypoint.sh \
 && chown -R ${APP_USER}:${APP_USER} /opt/sandbox-log \
 && chmod 0600 /opt/sandbox-log/logging.env \
 && PYTHONPATH=/opt/sandbox-log/python-packages:/opt/sandbox-log/lts-sdk \
    python3 -c "import watchdog; from producer.core.producer import Producer; print('OK')"

ENTRYPOINT ["tini", "-s", "--", "/bin/bash", "/opt/sandbox-log/log-entrypoint.sh"]

关键要点：

• FROM必须是用户需要保留的业务镜像
• 不要删除、覆盖或替换原始业务程序
• 构建时验证watchdog和LTS SDK可被导入，失败会中断构建
• logging.env权限设为0600，仅业务用户可读
• 使用tini作为init进程，正确处理信号转发

构建上下文的完整文件结构如下：

构建上下文/
├── Dockerfile                        # 补丁镜像构建文件
├── log-entrypoint.sh                 # 包装入口脚本
├── log-watcher.py                    # 文件监听程序
├── logging.env                       # 日志服务配置（不要提交到源码仓库）
├── python-packages/                  # 离线Python依赖
│   ├── watchdog/
│   ├── loguru/
│   ├── requests/
│   ├── certifi/
│   ├── charset_normalizer/
│   ├── idna/
│   ├── urllib3/
│   └── six.py
└── huaweicloud-lts-python-sdk/       # LTS Python SDK源码
    ├── producer/
    ├── utils/
    └── auth/

步骤六：构建镜像

基于用户基础镜像构建包含日志采集能力的补丁镜像。构建过程中会自动验证依赖可用性，确保镜像可用后再发布。

执行构建命令：

docker build \
  --build-arg BASE_IMAGE='<用户基础镜像>:<版本>' \
  --build-arg APP_USER='node' \
  -t '<镜像仓库>/<名称>:<日志补丁版本>' \
  .

验证方法：构建成功后，执行以下命令验证镜像可用：

# 验证镜像构建成功
docker images | grep '<镜像仓库>/<名称>'

# 验证依赖可导入
docker run --rm '<镜像仓库>/<名称>:<日志补丁版本>' \
  python3 -c "import watchdog; from producer.core.producer import Producer; print('OK')"
# 预期输出：OK

步骤七：发布并验证日志采集

将补丁镜像推送到镜像仓库，创建新的Sandbox模板，通过实际业务日志写入验证端到端的采集链路是否通畅。

1. 推送补丁镜像到镜像仓库
2. 基于补丁镜像创建新的Sandbox模板
3. 使用新模板创建全新的Sandbox
4. 触发真实业务日志写入到LOG_WATCH_DIR指定目录
5. 在LTS控制台查询日志，验证采集结果

镜像变更后必须创建新的Sandbox。已有Sandbox不会自动获得新的入口脚本或新的依赖。

配置参数说明

LTS认证配置

LTS接入点配置

采集行为配置

日志记录格式

每条上报到LTS的记录包含以下字段：

path必须是实际路径，不能是通配符，这样才能在LTS中按业务文件追踪来源。

状态文件与持久化策略

状态文件决定重启后是否重复发送或遗漏日志。

状态文件使用dev:ino（设备号:inode号）作为文件标识，正确处理文件轮转和重命名。状态文件不包含日志服务访问凭据。

常见问题

构建阶段问题

运行阶段问题

日志质量问题

DNS配置问题

如果沙箱内DNS无法解析lts-access域名，需要手动配置hosts：

echo "100.125.164.33 lts-access.<region>.myhuaweicloud.com" >> /etc/hosts