myxce/src/docs/job_peek.md

项目实施方案：xcecli jobs peek

背景

用户需要在 XCE 集群管理系统中添加 xcecli jobs peek 子命令，用于实时查看运行中作业的标准输出和标准错误输出。

架构问题

输出文件由 worker 进程（xcecd）通过 dup2() 重定向写入，文件位于 计算节点的本地文件系统 上（默认 /tmp/ 或配置路径）。API server 运行在独立节点上，无法直接访问 远程 worker 节点的文件。

解决方案

采用 方案 B：通过 controller → worker 协议链路远程拉取文件内容。需要新增 OPCode 和 cbuf 超时机制。

当前实施状态

Phase 2（Go 层）已完成，Phase 1（C 层）尚未开始。

现有 controllers/output.go 采用方案 A（本地读文件），直接 os.Open() 读取本地文件系统。这在 apiserver 与 worker 同节点时可用，但多节点部署下不可靠。Phase 1 完成后，Phase 2 需要重写 output.go，将底层数据源切换为 CGO → controller → worker 链路。

分阶段目标

1. 第一阶段：cbuf 超时机制 + 新 OPCode（libxce/controller/worker）— 待完成
2. 第二阶段：将 output.go 切换为 CGO/FetchFile 实现（Plan A → Plan B）— 待完成
3. 第三阶段：测试验证 — Go 层单元测试已完成，C 层及集成测试待完成

完整架构

┌──────────────────────────────────────────────────────────────────┐
│                        用户终端 (xcecli)                          │
│                        HTTP/REST                                 │
└──────────────────────────────┬───────────────────────────────────┘
                               │
                               ▼
┌───────────────────────────────────────────────────────────────────┐
│                    API Server (Go, Gin)                            │
│                                                                   │
│  ┌─────────────────┐    ┌─────────────────┐                      │
│  │ controllers/    │    │ backend/        │                      │
│  │ output.go       │    │ clib.go         │                      │
│  │                 │    │                 │  ┌─────────────────┐  │
│  │ 1. 解析 jobid   │───>│ 2. CGO 调用     │  │ 独立 ctx (peek) │  │
│  │ 2. 查 exec_hosts│    │ xce_fetchfile() │  │ 不阻塞主连接     │  │
│  │ 3. 找目标 worker│    │ (新函数)        │  │ timeout=10s     │  │
│  └─────────────────┘    └────────┬────────┘  └─────────────────┘  │
│                                 │                                  │
│  store.Jobs                     │ 主 ctx（submit/kill/heartbeat）  │
│  JobInfo{                       │                                  │
│    OutFile, ErrFile,            │                                  │
│    ExecHosts, JobStatus         │                                  │
│  }                              │                                  │
└─────────────────────────────────┼──────────────────────────────────┘
                                  │ TCP/TLS (libxce)
                                  ▼
┌───────────────────────────────────────────────────────────────────┐
│                    Controller (C, xcecd)                           │
│                                                                   │
│  ┌─────────────┐    ┌─────────────┐    ┌──────────────────────┐  │
│  │ cluster.c   │    │ job.c       │    │ vtaskmgr (8 vCPU)    │  │
│  │             │    │             │    │                      │  │
│  │ 查找 worker │───>│ 转发到      │    │ 并发处理不同 client   │  │
│  │ by hostname │    │ worker ep   │    │ 的请求               │  │
│  │             │    │             │    │                      │  │
│  │ OPCode      │    │ OPCode      │    │                      │  │
│  │ fetchfile   │    │ fetchfile   │    │                      │  │
│  │ (新增)      │    │ (新增)      │    │                      │  │
│  └─────────────┘    └──────┬──────┘    └──────────────────────┘  │
└────────────────────────────┼──────────────────────────────────────┘
                             │ TCP/TLS (libxce)
                             ▼
┌───────────────────────────────────────────────────────────────────┐
│                    Worker Node (C, xcecd worker)                   │
│                                                                   │
│  ┌─────────────────────────────────────────────────────────────┐  │
│  │ controller_connector.c                                      │  │
│  │ myreq_handler()                                             │  │
│  │                                                             │  │
│  │ case OPCode_fetchfile:  (新增)                              │  │
│  │   1. 解析请求: jobid, filepath, offset, maxBytes            │  │
│  │   2. os.Open(filepath)                                      │  │
│  │   3. Seek(offset)                                           │  │
│  │   4. Read(maxBytes)                                         │  │
│  │   5. 回复: content, fileSize, error_code                    │  │
│  └─────────────────────────────────────────────────────────────┘  │
│                                                                   │
│  输出文件: /tmp/job_XXX.out  (本地文件系统)                        │
│  由 process_manager.c / uv_proc.c 通过 dup2() 创建                │
└───────────────────────────────────────────────────────────────────┘

实施详情

Phase 1: C 层 — libxce + Controller + Worker

1.1 Cap'n Proto 协议定义

文件: src/schema/controller.capnp（注意：不是 src/include/xce/，没有独立的 opcodes.h，OPCode 枚举内联在 capnp 文件中）

修改一：OPCode 枚举末尾追加

enum OPCode {
  # ... 现有枚举项 (nop @0 .. clrpreason @21) ...
  fetchfile @22;  # 新增
}

修改二：新增两个结构体

struct FetchFileMsg $C.mapname("xce_fetchfile_t") {
  jobid     @0: UInt64  $C.mapname("jobid");
  filepath  @1: Text    $C.mapname("filepath");
  offset    @2: UInt64  $C.mapname("offset");
  maxBytes  @3: UInt32  $C.mapname("max_bytes");
}

struct FetchFileReply $C.mapname("xce_fetchfile_reply_t") {
  error     @0: Int32   $C.mapname("error");   # 0=成功, -1=不存在, -2=读取错误, -3=权限拒绝
  content   @1: Data    $C.mapname("content");
  fileSize  @2: UInt64  $C.mapname("file_size");
  bytesRead @3: UInt32  $C.mapname("bytes_read");
}

修改三：在 Request.body union 中追加新分支（⚠️ 关键，计划原版遗漏）

Request 结构体的 body 字段是一个 union（@3 到 @21），fetchfile 消息必须加入才能被框架正确编解码：

struct Request $C.mapname("xce_request_t") {
  # ...
  body : union {
    # ... 现有分支 (nomsg @3 .. hostctrlmsg @21) ...
    fetchfilemsg  @22: FetchFileMsg;   # 新增
  }
}

RequestReply.body union 中同样需要追加 fetchfile 回复分支，供 worker 回复时使用：

struct RequestReply $C.mapname("xce_reply_t") {
  body : union {
    # ... 现有分支 ...
    fetchfilereply @N: FetchFileReply;  # 新增，N 为现有最大值+1
  }
}

1.2 cbuf 超时机制 — cbuf_timedget()

文件: src/lib/mylock.c、src/include/mylock.h（注意：头文件路径是 src/include/，不是 src/lib/include/）

// 新增函数，保留 cbuf_get() 不变
int cbuf_timedget(cbuf_t *b, void *item, struct timespec *deadline) {
    pthread_mutex_lock(&b->mutex);
    while (b->len == 0) {
        int rc = pthread_cond_timedwait(&b->condget, &b->mutex, deadline);
        if (rc == ETIMEDOUT) {
            pthread_mutex_unlock(&b->mutex);
            return -ETIMEDOUT;
        }
    }
    // ... 正常读取逻辑（同 cbuf_get）
    pthread_mutex_unlock(&b->mutex);
    return 0;
}

受影响场景分析:

调用方	文件	是否改用 timed	原因
xceapi.c（15处，非13处）	src/lib/xceapi.c	是	客户端 RPC，超时防止 goroutine 泄漏
scheduler 发布决策	src/scheduler/src/scheduler.c	是	通过 xceapi.c 路径，自动受益
scheduler 内部协调	src/scheduler/src/scheduler.c:133,233	否	内部线程，无限等待合理
fcfs plugin	src/scheduler/src/plugins/fcfs.c:213	否	内部触发，无限等待合理
controller 主循环	src/controller/cluster.c:768	否	守护进程主循环，设计意图
vtaskmgr (4处)	src/lib/vtaskmgr.c	否	内部线程池，由 pthread_cancel 保护

超时值设置: XCE_RPC_TIMEOUT = 10s（定义在 src/include/xce/config.h）

• submit/kill 等操作：通常 <100ms，10s 足够
• fetchfile 操作：需要 worker 读文件，可能稍慢，但不超过 10s
• scheduler 发布决策：通常 <1s，10s 足够
• 如果超时，返回 -ETIMEDOUT，调用方可以区分超时和其他错误

1.3 libxce 客户端函数

文件: src/lib/request.c, src/lib/xceapi.c

// src/lib/xceapi.c — 新增
int xce_fetchfile(xcecli_ctx_t *ctx, fetchfile_req_t *req, fetchfile_rsp_t *rsp) {
    // 1. 编码请求
    // 2. endpoint_enq() 发送
    // 3. cbuf_timedget() 等待回复（超时 10s）
    // 4. 解码回复到 rsp
}

// 分块读取辅助函数
int xce_fetchfile_chunked(xcecli_ctx_t *ctx, const char *hostname,
                          const char *filepath, uint64_t offset,
                          uint32_t maxBytes, char **content, uint64_t *totalSize);

1.4 Controller 代理

文件: src/controller/cluster.c（主路由）+ src/controller/job.c（具体实现）

所有 OPCode 的 case 路由都在 cluster.c 的消息分发 switch 中。在 cluster.c 中添加分发，在 job.c 中实现代理逻辑：

// cluster.c — 分发
case OPCode_fetchfile:
    on_fetchfile(c, ep, request);
    break;

// job.c — 代理实现
void on_fetchfile(cluster_t *c, endpoint_t *ep, xce_request_t *req) {
    xce_fetchfile_t *fetchmsg = &req->body.fetchfilemsg;

    // 1. 根据 jobid 或 filepath 内嵌的 hostname 查找 worker endpoint
    endpoint_t *worker_ep = cluster_find_worker_by_jobid(c, fetchmsg->jobid);
    if (!worker_ep) {
        send_error_reply(ep, OPCode_fetchfile, -ENOENT);
        return;
    }

    // 2. 转发请求到 worker（保留原始 seq，便于追踪）
    // 3. 等待 worker 回复（通过独立 cbuf 或直接透传）
    // 4. 将 worker 的 FetchFileReply 回复给 apiserver
}

1.5 Worker 处理

文件: src/worker/src/controller_connector.c

在 myreq_handler() 中添加：

case OPCode_fetchfile: {
    fetchfile_req_t req;
    decode_fetchfile_req(msg, &req);

    // 安全校验：只允许读取作业相关的输出文件
    // 1. 通过 job_information.h 中的 job_id 字段验证 jobid 属于此 worker
    //    （worker 用 job_info_t 结构体维护本地运行作业列表，字段 job_info_t.job_id）
    // 2. 验证 filepath 是作业的 out_file 或 err_file 路径
    // 3. 防止路径穿越（检查路径中是否包含 ".."）

    fetchfile_rsp_t rsp = {0};
    rsp.fileSize = get_file_size(req.filepath);

    if (req.offset >= rsp.fileSize) {
        rsp.error = 0;  // 没有新内容
        rsp.bytesRead = 0;
    } else {
        FILE *f = fopen(req.filepath, "r");
        if (!f) {
            rsp.error = -1;  // 文件不存在
        } else {
            fseek(f, req.offset, SEEK_SET);
            rsp.bytesRead = fread(rsp.content, 1, req.maxBytes, f);
            fclose(f);
            rsp.error = 0;
        }
    }

    send_reply(ep, OPCode_reply, &rsp);
}

Phase 2: Go 层 — 将 output.go 切换为 Plan B 实现

注意：以下文件均已存在并有完整的 Plan A 实现（本地读文件）。 Phase 2 的任务是在 Phase 1 C 层完成后，将底层数据源替换为 CGO → controller → worker 链路。 CLI 端（peek.go）、路由注册、models、request.go 无需改动。

2.1 API Server CGO 绑定

文件: src/frontend/apiserver/backend/clib.go（已有，追加内容）

// peekCtx 是专用于 peek/fetchfile 的独立 controller 连接。
// 独立连接避免 fetchfile 的阻塞等待占用主连接的 vtaskmgr 槽位。
//
// HostRole 决策：复用 HostRole_apiserver（controller 的 join 处理已支持
// 同角色多连接）。若未来需区分，可在 capnp 中新增 apiserverPeek @5，
// 并同步修改 controller 的 join/leave 处理。
var peekCtx C.backend_ctx_t

func initPeekContext(config *BackendConfig) error {
    // 创建第 2 个连接，专用于 peek；join 以 HostRole_apiserver 身份
}

// FetchFile 通过 controller→worker 链路读取作业输出文件的一段内容。
func FetchFile(jobID uint64, filePath string, offset uint64, maxBytes uint32) (content []byte, fileSize uint64, err error) {
    // CGO 调用 xce_fetchfile()
}

文件: src/frontend/apiserver/backend/stub.go（已有，必须同步追加 stub）

// FetchFile stub — 供 go:build !cgo 的单元测试编译。
func FetchFile(jobID uint64, filePath string, offset uint64, maxBytes uint32) ([]byte, uint64, error) {
    return nil, 0, fmt.Errorf("stub: FetchFile not implemented")
}

2.2 output.go 重写要点

文件: src/frontend/apiserver/controllers/output.go（已存在，需重写底层读取逻辑）

当前 Plan A 实现直接读本地文件（os.Open → readFileTail / readFromOffset）。

Plan B 重写后：

1. 解析 job.ExecHosts（类型 Jsonb = []byte）获取 worker hostname：

   var hosts []models.JobExecHosts
   json.Unmarshal([]byte(job.ExecHosts), &hosts)
   hostname := hosts[0].HostName  // JobExecHosts.HostName 字段
   

2. 调用 backend.FetchFile(job.JobID, filePath, uint64(offset), 65536)
3. 返回 content、新 offset（= offset + bytesRead）、fileSize

当前 Bug：Plan A 的 readFileTail 调用 os.ReadFile(path) 将整个文件读入内存，大文件会 OOM。Plan B 通过 maxBytes 参数天然规避，重写时不必保留 readFileTail。

Phase 3: 测试验证

3.1 C 层测试

测试文件位于 src/lib/testsuite/（已有 test_request.c 等）。

• 新建 src/lib/testsuite/test_cbuf.c：
  • cbuf_timedget 正常读取测试
  • cbuf_timedget 超时返回 -ETIMEDOUT 测试
  • 与 cbuf_put 的并发测试
• OPCode_fetchfile 端到端测试（含 controller testsuite）
• Worker 安全测试：路径穿越防护（.. 构造的恶意路径）

3.2 Go 层测试

• output_test.go（已有 15 个用例，覆盖 Plan A）：Plan B 重写后需更新为 mock backend.FetchFile
• peek_test.go（已有）：无需大改，轮询逻辑不变
• stub.go 中的 FetchFile stub 必须能让现有单元测试通过编译

文件清单

需新建的文件

文件	说明
src/lib/testsuite/test_cbuf.c	cbuf_timedget 单元测试

需修改的文件（C 层 — Phase 1）

文件	修改内容
src/schema/controller.capnp	① OPCode 枚举加 fetchfile @22；② 新增 FetchFileMsg、FetchFileReply 结构体；③ Request.body union 加 fetchfilemsg @22；④ RequestReply.body union 加 fetchfile 回复分支
src/lib/mylock.c	新增 cbuf_timedget() 实现
src/include/mylock.h	声明 cbuf_timedget()
src/lib/xceapi.c	15处 cbuf_get 改用 cbuf_timedget()；新增 xce_fetchfile()
src/lib/request.c	新增 fetchfile 请求编码/解码函数
src/controller/cluster.c	在 OPCode switch 中添加 case OPCode_fetchfile 分发
src/controller/job.c	实现 on_fetchfile() controller→worker 代理逻辑
src/worker/src/controller_connector.c	在 myreq_handler() 中添加 case OPCode_fetchfile 处理，含安全校验

需修改的文件（Go 层 — Phase 2，已有文件重写/追加）

文件	当前状态	修改内容
src/frontend/apiserver/backend/clib.go	已有	新增 peekCtx（独立连接）、FetchFile() CGO 绑定、initPeekContext()
src/frontend/apiserver/backend/stub.go	已有	新增 FetchFile() stub（⚠️ 否则单元测试编译失败）
src/frontend/apiserver/controllers/output.go	已有（Plan A）	重写底层读取：os.Open → backend.FetchFile + ExecHosts 解析
src/frontend/apiserver/controllers/output_test.go	已有（Plan A）	更新 mock：本地文件 mock → FetchFile stub

已完成、无需改动的文件

文件	说明
src/frontend/apiserver/router/routers.go	✅ 路由 GET /:jobid/output 已注册
src/frontend/xcecli/cmd/jobs/peek.go	✅ peek 命令完整实现
src/frontend/xcecli/cmd/jobs/peek_test.go	✅ 单元测试已完成
src/frontend/xcecli/cmd/jobs/jobs.go	✅ peek 子命令已注册
src/frontend/xcecli/models/jobs.go	✅ JobOutputResponse 已定义
src/frontend/xcecli/common/request/request.go	✅ GetJobOutput 已实现

性能影响总结

指标	现有系统	添加后
单次 peek 延迟	N/A	~10-50ms
并发 peek (10 用户)	N/A	~300ms 排队
并发 peek (100 用户)	N/A	~3s 排队（建议加连接池）
controller vCPU 负载	不变	微小增加（转发请求）
worker 线程池	不变	微小增加（读文件操作）
主连接影响	不变	独立 ctx，零影响
现有 xceapi.c 调用	无超时	10s 超时，防止永久阻塞

风险点

1. cbuf_timedget 超时误判：如果 controller 处理慢（如大量作业同时调度），可能误超时。10s 是保守值，可通过配置调整。
2. controller 代理性能：controller 需要额外转发请求到 worker，增加 controller 负载。但在 8 vCPU 线程池下，影响可控。
3. 大文件传输：通过 maxBytes 限制单次传输大小，CLI 通过 offset 分块获取，不会 OOM。（当前 Plan A 实现的 readFileTail 有整文件读入内存的 Bug，Plan B 天然修复）
4. Worker 安全性：必须校验 filepath 是作业的输出文件，防止任意文件读取攻击。通过 job_info_t.job_id 和 out/err 文件路径双重验证。
5. 独立 peekCtx 并发：单个 peekCtx 下，并发 peek 请求通过 cbuf 串行化（100 用户 ≈ 3s 排队）。若需更高并发，可维护一个小型 ctx 连接池（2～4个），但目前阶段不必实现。
6. Plan A → Plan B 切换：重写 output.go 后，已有的 output_test.go（15 个用例）需同步更新为 mock backend.FetchFile，否则测试会因找不到本地文件而失败。
7. stub.go 漏改：若只改 clib.go 而忘记同步 stub.go，make test（无 CGO）编译失败，CI 会挂。