From d2b47e3985171726dcbcf5cbd135a2f4ac4bad59 Mon Sep 17 00:00:00 2001 From: Lei Chen Date: Sat, 9 May 2026 16:57:54 +0800 Subject: [PATCH] Init commit of jobs output/error message peek project. --- src/controller/cluster.c | 18 + src/controller/include/cluster.h | 2 + src/controller/job.c | 187 +++++++ src/docs/job_peek.md | 408 ++++++++++++++ src/frontend/apiserver/backend/clib.go | 38 ++ src/frontend/apiserver/backend/config.go | 7 + .../apiserver/backend/convert_pure.go | 28 + src/frontend/apiserver/backend/stub.go | 5 + src/frontend/apiserver/controllers/jobs.go | 2 +- src/frontend/apiserver/controllers/output.go | 245 +++++++++ .../apiserver/controllers/output_test.go | 515 ++++++++++++++++++ src/frontend/apiserver/router/routers.go | 3 + src/frontend/xcecli/cmd/jobs/jobs.go | 2 + src/frontend/xcecli/cmd/jobs/peek.go | 153 ++++++ src/frontend/xcecli/cmd/jobs/peek_test.go | 80 +++ src/frontend/xcecli/common/request/request.go | 39 ++ src/frontend/xcecli/models/jobs.go | 15 + src/include/client.h | 3 + src/include/mylock.h | 2 + src/include/request.h | 1 + src/include/typemap.h | 2 + src/include/xce/types.h | 15 + src/include/xceapi.h | 1 + src/include/xcectx.h | 9 +- src/lib/client.c | 30 + src/lib/helper.c | 2 + src/lib/meson.build | 1 + src/lib/mylock.c | 26 + src/lib/request.c | 23 + src/lib/testsuite/test.c | 2 + src/lib/testsuite/test_cbuf.c | 207 +++++++ src/lib/xceapi.c | 88 ++- src/lib/xcectx.c | 7 + src/scheduler/src/plugins/parjob_allocator.c | 4 + src/scheduler/src/schedobj.c | 8 +- src/scheduler/src/scheduler.c | 4 + src/schema/controller.capnp | 18 + src/worker/src/controller_connector.c | 88 +++ 38 files changed, 2267 insertions(+), 21 deletions(-) create mode 100644 src/docs/job_peek.md create mode 100644 src/frontend/apiserver/controllers/output.go create mode 100644 src/frontend/apiserver/controllers/output_test.go create mode 100644 src/frontend/xcecli/cmd/jobs/peek.go create mode 100644 src/frontend/xcecli/cmd/jobs/peek_test.go create mode 100644 src/lib/testsuite/test_cbuf.c diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 90d7b49e..9dc56d64 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -201,6 +201,24 @@ static int myreq_handler(vtask_t* task) case OPCode_clrpreason: rc = do_clr_preason(c, ep, &request); break; + case OPCode_fetchfile: + rc = on_fetchfile(c, ep, &request); + break; + case OPCode_reply: { + /* Reply from a worker (e.g., for fetchfile proxy) */ + xce_reply_t* workreply = request.body.replymsg; + if (workreply != NULL && workreply->opcode == OPCode_fetchfile) { + xce_fetchfile_reply_t* fr = + (workreply->data_which == RequestReply_data_fetchfilereply) + ? workreply->data.fetchfilereply + : NULL; + uint64_t jid = fr ? fr->jobid : 0; + fetchfile_signal(jid, workreply->status, fr); + /* Ownership of fr transfers to the pending waiter. */ + workreply->data.fetchfilereply = NULL; + } + break; + } default: rc = -E_XCE_BAD_OPCODE; ERROR("Unknown opcode: %d", request.opcode); diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 6bf0cedb..dcd5ac09 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -120,6 +120,8 @@ extern int release_resource_from_worker(char*, cluster_t*); extern int collect_resval_with_idx(zarray_t*, xlist_t*); extern int do_set_preason(cluster_t*, endpoint_t*, xce_request_t*); extern int do_clr_preason(cluster_t*, endpoint_t*, xce_request_t*); +extern int on_fetchfile(cluster_t*, endpoint_t*, xce_request_t*); +extern void fetchfile_signal(uint64_t jobid, int status, xce_fetchfile_reply_t*); extern int handle_apiserver_leave(cluster_t*, endpoint_t*); extern int add_apiserver(cluster_t*, char*, endpoint_t*); extern int apiserver_init(cluster_t*); diff --git a/src/controller/job.c b/src/controller/job.c index c975c1b5..f5fbdd14 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1,10 +1,12 @@ #include +#include #include #include #include "cluster.h" #include "jobj.h" #include "preason.h" +#include "request.h" #include "xce/cluster_def.h" #include "xce/errnox.h" #include "xce/types.h" @@ -1047,3 +1049,188 @@ int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request) return 0; } + +/* ---- fetchfile pending-request infrastructure ---- */ + +typedef struct fetchfile_pending { + uint64_t jobid; + pthread_mutex_t mu; + pthread_cond_t cv; + int done; + int status; + xce_fetchfile_reply_t* reply; + struct fetchfile_pending* next; +} fetchfile_pending_t; + +static pthread_mutex_t g_fetchfile_mu = PTHREAD_MUTEX_INITIALIZER; +static fetchfile_pending_t* g_fetchfile_list = NULL; + +static void fetchfile_register(fetchfile_pending_t* p) +{ + pthread_mutex_lock(&g_fetchfile_mu); + p->next = g_fetchfile_list; + g_fetchfile_list = p; + pthread_mutex_unlock(&g_fetchfile_mu); +} + +static void fetchfile_unregister(fetchfile_pending_t* p) +{ + fetchfile_pending_t** pp; + pthread_mutex_lock(&g_fetchfile_mu); + for (pp = &g_fetchfile_list; *pp != NULL; pp = &(*pp)->next) { + if (*pp == p) { + *pp = p->next; + break; + } + } + pthread_mutex_unlock(&g_fetchfile_mu); +} + +fetchfile_pending_t* fetchfile_find(uint64_t jobid) +{ + fetchfile_pending_t* p; + pthread_mutex_lock(&g_fetchfile_mu); + for (p = g_fetchfile_list; p != NULL; p = p->next) { + if (p->jobid == jobid) break; + } + pthread_mutex_unlock(&g_fetchfile_mu); + return p; +} + +/* Signal a pending fetchfile request with the worker's reply. */ +void fetchfile_signal(uint64_t jobid, int status, xce_fetchfile_reply_t* reply) +{ + fetchfile_pending_t* p = fetchfile_find(jobid); + if (p == NULL) return; + + pthread_mutex_lock(&p->mu); + p->status = status; + p->reply = reply; + p->done = 1; + pthread_cond_signal(&p->cv); + pthread_mutex_unlock(&p->mu); +} + +/* ---- on_fetchfile: proxy fetchfile from apiserver to worker ---- */ + +int on_fetchfile(cluster_t* c, endpoint_t* caller_ep, xce_request_t* request) +{ + xce_fetchfile_t* msg = request->body.fetchfilemsg; + xce_reply_t reply; + job_t* j = NULL; + worker_t* w = NULL; + int rc = 0; + char* exechost = NULL; + struct timespec deadline; + + fetchfile_pending_t pending = { + .mu = PTHREAD_MUTEX_INITIALIZER, + .cv = PTHREAD_COND_INITIALIZER, + .done = 0, .status = -ETIMEDOUT, .reply = NULL, .next = NULL + }; + + init_reply_by_req(&reply, request); + + if (msg == NULL || msg->filepath == NULL || msg->filepath[0] == '\0') { + reply.status = -EINVAL; + reply.data_which = RequestReply_data_nodata; + xce_reply(caller_ep, c->myhost, &reply); + return -EINVAL; + } + + pending.jobid = msg->jobid; + + /* Look up the job */ + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)msg->jobid, (void**)&j); + if (j != NULL) reference(j); + unlock(c->jobmgr); + + if (j == NULL) { + reply.status = -ENOENT; + reply.data_which = RequestReply_data_nodata; + xce_reply(caller_ep, c->myhost, &reply); + return -ENOENT; + } + + /* Get the first exec-host from job allocs */ + lock(j, RTTI_LOCK_READ); + if (j->job.allocs != NULL && j->job.allocs->nhosts > 0 + && j->job.allocs->allocs != NULL && j->job.allocs->allocs[0] != NULL + && j->job.allocs->allocs[0]->host != NULL) { + exechost = strdup(j->job.allocs->allocs[0]->host); + } + unlock(j); + dereference(j); + + if (exechost == NULL) { + reply.status = -ENOENT; + reply.data_which = RequestReply_data_nodata; + xce_reply(caller_ep, c->myhost, &reply); + return -ENOENT; + } + + /* Find the worker */ + lock(c->workermgr, RTTI_LOCK_READ); + rc = key_find(c->workermgr, (void*)exechost, (void**)&w); + if (w != NULL) reference(w); + unlock(c->workermgr); + free(exechost); + + if (w == NULL) { + reply.status = -ENOENT; + reply.data_which = RequestReply_data_nodata; + xce_reply(caller_ep, c->myhost, &reply); + return -ENOENT; + } + + lock(w, RTTI_LOCK_READ); + if (w->ep == NULL) { + unlock(w); + dereference(w); + reply.status = -ENOENT; + reply.data_which = RequestReply_data_nodata; + xce_reply(caller_ep, c->myhost, &reply); + return -ENOENT; + } + + /* Register pending before sending to avoid a race with the reply */ + fetchfile_register(&pending); + rc = xce_fetchfile_req(w->ep, msg); + unlock(w); + dereference(w); + + if (rc < 0) { + fetchfile_unregister(&pending); + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + xce_reply(caller_ep, c->myhost, &reply); + return rc; + } + + /* Wait up to 10 seconds for the worker to respond */ + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += 10; + + pthread_mutex_lock(&pending.mu); + while (!pending.done) { + rc = pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline); + if (rc == ETIMEDOUT) break; + } + pthread_mutex_unlock(&pending.mu); + + fetchfile_unregister(&pending); + + /* Relay the reply back to the apiserver */ + if (pending.done && pending.status >= 0 && pending.reply != NULL) { + reply.status = 0; + reply.data_which = RequestReply_data_fetchfilereply; + reply.data.fetchfilereply = pending.reply; + } else { + reply.status = pending.done ? pending.status : -ETIMEDOUT; + reply.data_which = RequestReply_data_nodata; + FREEUP(pending.reply); + } + xce_reply(caller_ep, c->myhost, &reply); + return 0; +} diff --git a/src/docs/job_peek.md b/src/docs/job_peek.md new file mode 100644 index 00000000..db628c04 --- /dev/null +++ b/src/docs/job_peek.md @@ -0,0 +1,408 @@ +# 项目实施方案：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 枚举末尾追加** +```capnp +enum OPCode { + # ... 现有枚举项 (nop @0 .. clrpreason @21) ... + fetchfile @22; # 新增 +} +``` + +**修改二：新增两个结构体** +```capnp +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 消息必须加入才能被框架正确编解码： +```capnp +struct Request $C.mapname("xce_request_t") { + # ... + body : union { + # ... 现有分支 (nomsg @3 .. hostctrlmsg @21) ... + fetchfilemsg @22: FetchFileMsg; # 新增 + } +} +``` + +`RequestReply.body` union 中同样需要追加 fetchfile 回复分支，供 worker 回复时使用： +```capnp +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/`） + +```c +// 新增函数，保留 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` + +```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 中实现代理逻辑： + +```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()` 中添加： +```c +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`（已有，追加内容） + +```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） + +```go +// 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： + ```go + 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 会挂。 diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index 4fc63897..eb66db56 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -416,3 +416,41 @@ func signalNumber(signame string, action int) int { } return 15 // SIGTERM as safe default } + +// FetchFile fetches up to maxBytes of a job output file from offset via the +// controller→worker chain. The filepath must be the fully resolved absolute +// path on the exec host. +func FetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*FetchFileResult, error) { +cFilepath := C.CString(filepath) +defer C.free(unsafe.Pointer(cFilepath)) + +msg := C.xce_fetchfile_t{ +jobid: C.uint64_t(jobid), +filepath: cFilepath, +offset: C.uint64_t(offset), +max_bytes: C.uint32_t(maxBytes), +} + +var reply *C.xce_fetchfile_reply_t +rc := C.xcecli_fetchfile(gCtx.ctx, &msg, &reply) +if rc < 0 { +return nil, fmt.Errorf("xcecli_fetchfile failed: %d", int(rc)) +} +if reply == nil { +return nil, fmt.Errorf("xcecli_fetchfile returned nil reply") +} +defer C.free(unsafe.Pointer(reply)) + +if reply.error != 0 { +return nil, fmt.Errorf("fetchfile remote error: %d", int(reply.error)) +} + +result := &FetchFileResult{ +FileSize: uint64(reply.file_size), +BytesRead: uint32(reply.bytes_read), +} +if reply.bytes_read > 0 && reply.content != nil { +result.Content = C.GoBytes(unsafe.Pointer(reply.content), C.int(reply.bytes_read)) +} +return result, nil +} diff --git a/src/frontend/apiserver/backend/config.go b/src/frontend/apiserver/backend/config.go index e1245c3a..b38f5e77 100644 --- a/src/frontend/apiserver/backend/config.go +++ b/src/frontend/apiserver/backend/config.go @@ -8,3 +8,10 @@ type BackendConfig struct { From string // identity sent to controller (default: "xce_apiserver") Key string // authentication key } + +// FetchFileResult holds the data returned by FetchFile. +type FetchFileResult struct { + Content []byte + FileSize uint64 + BytesRead uint32 +} diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index 20d4ca49..b0b6a24d 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -3,6 +3,7 @@ package backend import ( "apiserver/models" "encoding/json" + "fmt" "time" ) @@ -76,6 +77,33 @@ func IsArrayJob(id uint64) bool { return extractIDTag(id) == tagJobArray } +// FormatJobID formats a packed 64-bit job ID as a human-readable string, +// matching the C fmt_jobid() output in src/lib/utils.c. +// - Regular job: "{cluster}-{apiserver}-{timestamp}-{sequence}" +// - Array child: "{cluster}-{apiserver}-{timestamp}-{base}[{idx}]" +func FormatJobID(uniqueJobID int64) string { + id := uint64(uniqueJobID) + clusterID := extractBits(id, clusterIDShift, nClusterIDBits) + apiserverID := extractBits(id, apiserverIDShift, nApiserverIDBits) + timestamp := extractBits(id, timestampShift, nTimestampBits) + + if extractIDTag(id) == tagJobArray { + arrayBase := extractBits(id, sequenceHighShift, nSequenceBits-nArrayIdxBits) << uint(sequenceHighShift) + arrayIdx := extractBits(id, 0, nArrayIdxBits) + return fmt.Sprintf("%d-%d-%d-%d[%d]", + clusterID, apiserverID, timestamp, arrayBase, arrayIdx) + } + sequence := extractBits(id, 0, nSequenceBits) + return fmt.Sprintf("%d-%d-%d-%d", + clusterID, apiserverID, timestamp, sequence) +} + +// WorkerDefaultOutputPath returns the output file path the worker uses by default +// for the given job ID and task index, matching uv_proc.c make_job_output_path(). +func WorkerDefaultOutputPath(uniqueJobID int64, taskIdx int) string { + return fmt.Sprintf("/tmp/job_%s_task%d.out", FormatJobID(uniqueJobID), taskIdx) +} + // rawStartInfo mirrors the C job_startinfo_t fields used by the Go model. type rawStartInfo struct { cmdline string diff --git a/src/frontend/apiserver/backend/stub.go b/src/frontend/apiserver/backend/stub.go index b290eba8..d6126195 100644 --- a/src/frontend/apiserver/backend/stub.go +++ b/src/frontend/apiserver/backend/stub.go @@ -41,3 +41,8 @@ func EvalSelectForHost(_, _ string) error { return errNoCGO } // Cleanup stub — real implementation in clib.go (requires CGO). func Cleanup() {} + +// FetchFile stub — real implementation in clib.go (requires CGO). +func FetchFile(_ uint64, _ string, _ uint64, _ uint32) (*FetchFileResult, error) { + return nil, errNoCGO +} diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index af592d7b..a038b1c1 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -1434,7 +1434,7 @@ func (j JobController) CreateJob(ctx *gin.Context) { logger.Info("Generated job ID", zap.Int64("jobID", primaryJobID), zap.Int("arraySize", arraySize)) defaultValues := map[string]string{ - "job_id": fmt.Sprintf("%d", primaryJobID), + "job_id": backend.FormatJobID(primaryJobID), "user_name": submitUser, "job_group": req.JobGroup, "project_name": req.ProjectName, diff --git a/src/frontend/apiserver/controllers/output.go b/src/frontend/apiserver/controllers/output.go new file mode 100644 index 00000000..ad793bdd --- /dev/null +++ b/src/frontend/apiserver/controllers/output.go @@ -0,0 +1,245 @@ +package controllers + +import ( + "fmt" + "strconv" + "strings" + + "apiserver/backend" + "apiserver/common" + "apiserver/models" + "apiserver/store" + "apiserver/utils" + + logger "apiserver/utils" + + "github.com/gin-gonic/gin" + "go.uber.org/zap" +) + +// fetchFile is the function used to fetch remote file content. +// Overrideable in tests. +var fetchFile = backend.FetchFile + +const ( + // tailChunkSize is the maximum bytes fetched per request in tail mode. + tailChunkSize = 4 * 1024 * 1024 + // incrementalChunkSize is the maximum bytes fetched in incremental mode. + incrementalChunkSize = 4 * 1024 * 1024 +) + +// OutputData represents the response payload for job output. +type OutputData struct { + Content string `json:"content"` + Offset int64 `json:"offset"` + JobStatus string `json:"job_status"` + Done bool `json:"done"` +} + +// GetJobOutput returns the tail or incremental content of a job's output file. +// GET /xce/v1/jobs/:jobid/output?offset=0&lines=100&type=stdout +func (j JobController) GetJobOutput(ctx *gin.Context) { + logger.Debug("Enter GetJobOutput() function from ", zap.String("client_ip", ctx.ClientIP())) + + token := ctx.GetHeader("Authorization") + submitUser, err := GetUsernameFromToken(token) + if err != nil { + ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) + return + } + + if ctx.IsAborted() { + ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) + return + } + + jobidStr := ctx.Param("jobids") + offsetStr := ctx.DefaultQuery("offset", "0") + linesStr := ctx.DefaultQuery("lines", "100") + streamType := ctx.DefaultQuery("type", "stdout") + + offset, err := strconv.ParseInt(offsetStr, 10, 64) + if err != nil { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "invalid offset: "+err.Error(), nil) + return + } + + lines, err := strconv.Atoi(linesStr) + if err != nil || lines <= 0 { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "invalid lines: must be a positive integer", nil) + return + } + + if streamType != "stdout" && streamType != "stderr" { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "invalid type: must be 'stdout' or 'stderr'", nil) + return + } + + // Parse job ID + jobInfo, err := parseJobIDParam(jobidStr) + if err != nil { + ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) + return + } + + // Look up job in store + job, ok := store.Jobs.Get(uint64(jobInfo.Unique_JobID)) + if !ok { + ReturnError(ctx, common.ERROR_NOT_FOUND_CODE, common.ERROR_JOB_NOT_FOUND_MSG, nil) + return + } + + // Check ownership: users with domain_type=all can view any job + permission, perr := GetRequireResourcePermission(submitUser, "job:view") + if perr != nil || permission.DomainType != models.DomainTypeAll { + if job.SubmitUser != submitUser { + ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) + return + } + } + + // Resolve output file path + var filePath string + if streamType == "stderr" { + filePath = resolveOutputPath(job.ErrFile, job) + } else { + filePath = resolveOutputPath(job.OutFile, job) + } + + if filePath == "" { + ReturnError(ctx, common.ERROR_READ_FILE_CODE, fmt.Sprintf("job <%s>: no output file path could be determined for %s", backend.FormatJobID(job.Unique_JobID), streamType), nil) + return + } + + var content string + var newOffset int64 + + if offset == 0 { + content, newOffset, err = fetchFileTail(uint64(job.Unique_JobID), filePath, lines) + } else { + content, newOffset, err = fetchFileFromOffset(uint64(job.Unique_JobID), filePath, offset) + } + + if err != nil { + ReturnError(ctx, common.ERROR_READ_FILE_CODE, fmt.Sprintf("job <%s> (path=%s): failed to read output: %s", backend.FormatJobID(job.Unique_JobID), filePath, err.Error()), nil) + return + } + + jobStatusStr := models.JobStateStr(job.JobStatus) + done := jobStatusStr != "RUN" && jobStatusStr != "PSUSP" && jobStatusStr != "USUSP" && jobStatusStr != "SSUSP" + + ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, OutputData{ + Content: content, + Offset: newOffset, + JobStatus: jobStatusStr, + Done: done, + }) +} + +// fetchFileTail fetches the last n lines of a file via the backend. +// It uses a two-pass approach for files larger than tailChunkSize. +func fetchFileTail(jobid uint64, path string, n int) (string, int64, error) { + // First pass: fetch from offset 0 to get file size and content for small files. + result, err := fetchFile(jobid, path, 0, tailChunkSize) + if err != nil { + return "", 0, err + } + + fileSize := int64(result.FileSize) + if fileSize == 0 { + return "", 0, nil + } + + data := result.Content + + // If the file is larger than our chunk, re-fetch the tail portion. + if result.FileSize > tailChunkSize { + endOffset := result.FileSize - tailChunkSize + result, err = fetchFile(jobid, path, endOffset, tailChunkSize) + if err != nil { + return "", 0, err + } + data = result.Content + } + + return lastNLines(data, n), fileSize, nil +} + +// fetchFileFromOffset fetches content from the given byte offset via the backend. +func fetchFileFromOffset(jobid uint64, path string, offset int64) (string, int64, error) { + result, err := fetchFile(jobid, path, uint64(offset), incrementalChunkSize) + if err != nil { + return "", 0, err + } + + fileSize := int64(result.FileSize) + if int64(offset) >= fileSize { + return "", fileSize, nil + } + + return string(result.Content), fileSize, nil +} + +// lastNLines returns the last n lines from data as a string. +func lastNLines(data []byte, n int) string { + if len(data) == 0 { + return "" + } + + lineCount := 0 + start := 0 + end := len(data) - 1 + + for i := end; i >= 0; i-- { + if data[i] == '\n' { + lineCount++ + if lineCount == n+1 { + start = i + 1 + break + } + } + } + + return string(data[start:]) +} + +// resolveOutputPath replaces placeholders in the output file path. +// When path is empty, returns the worker's default output path for task 0, +// matching uv_proc.c make_job_output_path() behavior. +func resolveOutputPath(path string, job models.JobInfo) string { + if path == "" { + return backend.WorkerDefaultOutputPath(job.Unique_JobID, 0) + } + formattedID := backend.FormatJobID(job.Unique_JobID) + path = strings.ReplaceAll(path, "%J", formattedID) + path = strings.ReplaceAll(path, "%I", strconv.Itoa(job.ArrayIdx)) + path = strings.ReplaceAll(path, "%U", job.SubmitUser) + if job.CWD != "" { + path = strings.ReplaceAll(path, "%C", job.CWD) + } + return path +} + +// parseJobIDParam parses a job ID string and returns the corresponding JobInfo. +func parseJobIDParam(jobidStr string) (models.JobInfo, error) { + // Use the same parsing logic as GetJobByIDs + jobArrInfo, err := parseArrayName(jobidStr) + if err != nil { + return models.JobInfo{}, err + } + + if !utils.IsNumeric(jobArrInfo.KeyID) { + return models.JobInfo{}, fmt.Errorf("bad jobid: %s", jobArrInfo.KeyID) + } + + uniqueJobID, err := strconv.ParseInt(jobArrInfo.KeyID, 10, 64) + if err != nil { + return models.JobInfo{}, fmt.Errorf("bad jobid: %s", jobArrInfo.KeyID) + } + + return models.JobInfo{ + Unique_JobID: uniqueJobID, + ArrayIdx: 0, // peek defaults to base job; array idx can be added later + }, nil +} + diff --git a/src/frontend/apiserver/controllers/output_test.go b/src/frontend/apiserver/controllers/output_test.go new file mode 100644 index 00000000..30119edb --- /dev/null +++ b/src/frontend/apiserver/controllers/output_test.go @@ -0,0 +1,515 @@ +package controllers + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "apiserver/backend" + "apiserver/common" + "apiserver/models" + "apiserver/store" + "apiserver/utils" + + "github.com/gin-gonic/gin" + "go.uber.org/zap" +) + +// mockFetchFile simulates FetchFile by reading from the local filesystem. +// It is installed by resetJobsState() so all output endpoint tests use it. +func mockFetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*backend.FetchFileResult, error) { + data, err := os.ReadFile(filepath) + if err != nil { + return nil, err + } + fileSize := uint64(len(data)) + if offset >= fileSize { + return &backend.FetchFileResult{FileSize: fileSize, BytesRead: 0}, nil + } + chunk := data[offset:] + if maxBytes > 0 && uint32(len(chunk)) > maxBytes { + chunk = chunk[:maxBytes] + } + return &backend.FetchFileResult{ + Content: chunk, + FileSize: fileSize, + BytesRead: uint32(len(chunk)), + }, nil +} + +func resetJobsState() { + store.InitStores() + utils.Log = zap.NewNop() + initTestAuthConfig() // ensure JWT parsing works in all output tests + fetchFile = mockFetchFile +} + +func setupTestJob(t *testing.T, tmpDir string) { + t.Helper() + + // OutFile/ErrFile use %J placeholder; resolveOutputPath replaces %J with + // the formatted job ID (FormatJobID(1001) = "0-0-0-1001" for small test IDs). + outFile := filepath.Join(tmpDir, "job_%J.out") + errFile := filepath.Join(tmpDir, "job_%J.err") + + job := models.JobInfo{ + Unique_JobID: 1001, + JobID: 1001, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x02, // RUN + OutFile: outFile, + ErrFile: errFile, + CWD: tmpDir, + } + store.Jobs.Put(1001, job) + + // Write to the resolved paths (with %J replaced by formatted ID "0-0-0-1001"). + os.WriteFile(filepath.Join(tmpDir, "job_0-0-0-1001.out"), []byte("line1\nline2\nline3\nline4\nline5\n"), 0644) + os.WriteFile(filepath.Join(tmpDir, "job_0-0-0-1001.err"), []byte("err1\nerr2\n"), 0644) +} + +func setupTestRouter() *gin.Engine { + gin.SetMode(gin.TestMode) + r := gin.New() + handler := &JobController{} + r.GET("/xce/v1/jobs/:jobids/output", handler.GetJobOutput) + return r +} + +// testToken returns a signed JWT for "testuser" valid for 1 hour. +// initTestAuthConfig() must be called before using this token. +func testToken() string { + return "Bearer " + signTestJWT("testuser", "admin", time.Now().Add(time.Hour)) +} + +func decodeResponse(t *testing.T, w *httptest.ResponseRecorder, v interface{}) { + t.Helper() + t.Logf("response body: %s", w.Body.String()) + if err := json.Unmarshal(w.Body.Bytes(), v); err != nil { + t.Fatalf("decode response: %v", err) + } +} + +func TestGetJobOutput_TailLines(t *testing.T) { + resetJobsState() + + tmpDir := t.TempDir() + setupTestJob(t, tmpDir) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/1001/output?lines=2", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Code int `json:"code"` + Data OutputData `json:"data"` + } + decodeResponse(t, w, &resp) + + if resp.Data.Content != "line4\nline5\n" { + t.Errorf("content = %q, want %q", resp.Data.Content, "line4\nline5\n") + } + if resp.Data.Offset != 30 { + t.Errorf("offset = %d, want 30", resp.Data.Offset) + } + if resp.Data.Done { + t.Error("done should be false for running job") + } + if resp.Data.JobStatus != "RUN" { + t.Errorf("job_status = %q, want RUN", resp.Data.JobStatus) + } +} + +func TestGetJobOutput_FromOffset(t *testing.T) { + resetJobsState() + + tmpDir := t.TempDir() + setupTestJob(t, tmpDir) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/1001/output?offset=12", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Code int `json:"code"` + Data OutputData `json:"data"` + } + decodeResponse(t, w, &resp) + + if resp.Data.Content != "line3\nline4\nline5\n" { + t.Errorf("content = %q, want %q", resp.Data.Content, "line3\nline4\nline5\n") + } + if resp.Data.Offset != 30 { + t.Errorf("offset = %d, want 30", resp.Data.Offset) + } +} + +func TestGetJobOutput_FileNotFound(t *testing.T) { + resetJobsState() + + tmpDir := t.TempDir() + outFile := filepath.Join(tmpDir, "job_%J.out") + + job := models.JobInfo{ + Unique_JobID: 2001, + JobID: 2001, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x02, + OutFile: outFile, + } + store.Jobs.Put(2001, job) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/2001/output?lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + var resp struct { + Code int `json:"code"` + } + decodeResponse(t, w, &resp) + + if resp.Code != common.ERROR_READ_FILE_CODE { + t.Errorf("code = %d, want %d", resp.Code, common.ERROR_READ_FILE_CODE) + } +} + +func TestGetJobOutput_NoOutputFileConfigured(t *testing.T) { + resetJobsState() + + job := models.JobInfo{ + Unique_JobID: 3001, + JobID: 3001, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x02, + OutFile: "", + } + store.Jobs.Put(3001, job) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/3001/output?lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + // resolveOutputPath uses the worker default (/tmp/job_0-0-0-3001_task0.out). + // That file does not exist, so we expect a read error (501). + var resp struct { + Code int `json:"code"` + } + decodeResponse(t, w, &resp) + + if resp.Code != common.ERROR_READ_FILE_CODE { + t.Errorf("code = %d, want %d", resp.Code, common.ERROR_READ_FILE_CODE) + } +} + +// TestGetJobOutput_DefaultPathFallback verifies that when OutFile is empty, +// the worker's default output path is used and content is returned successfully. +func TestGetJobOutput_DefaultPathFallback(t *testing.T) { + resetJobsState() + + // Use a real-ish packed job ID: cluster=1, apiserver=0, ts=1778311965, seq=0 + // FormatJobID → "1-0-1778311965-0" + // WorkerDefaultOutputPath → "/tmp/job_1-0-1778311965-0_task0.out" + const uniqueID = int64(0x0100000006A04BDD<<0) // MakeJobID(1,0,1778311965,0,0) + // Use backend.MakeJobID directly for clarity; compute inline: + // clusterID=1 at bit 56, timestamp=1778311965 at bit 20 + // packed = (1 << 56) | (1778311965 << 20) + const packedID = int64((uint64(1) << 56) | (uint64(1778311965) << 20)) + + // Install a fetchFile that reads from a tmpdir-based path instead of /tmp + // to avoid touching /tmp during tests. + tmpDir := t.TempDir() + workerDefault := backend.WorkerDefaultOutputPath(packedID, 0) // /tmp/job_1-0-1778311965-0_task0.out + formattedID := backend.FormatJobID(packedID) // "1-0-1778311965-0" + localFile := filepath.Join(tmpDir, "job_"+formattedID+"_task0.out") + os.WriteFile(localFile, []byte("worker output line\n"), 0644) + + // Override fetchFile for this test to redirect /tmp paths to tmpDir. + origFetchFile := fetchFile + fetchFile = func(jobid uint64, path string, offset uint64, maxBytes uint32) (*backend.FetchFileResult, error) { + if path == workerDefault { + path = localFile + } + return mockFetchFile(jobid, path, offset, maxBytes) + } + defer func() { fetchFile = origFetchFile }() + + job := models.JobInfo{ + Unique_JobID: packedID, + JobID: packedID, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x02, // RUN + OutFile: "", // no OutFile — should fall back to worker default + } + store.Jobs.Put(uint64(packedID), job) + + router := setupTestRouter() + + w := httptest.NewRecorder() + jobIDStr := fmt.Sprintf("%d", packedID) + req, _ := http.NewRequest("GET", "/xce/v1/jobs/"+jobIDStr+"/output?lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Data OutputData `json:"data"` + } + decodeResponse(t, w, &resp) + + if resp.Data.Content != "worker output line\n" { + t.Errorf("content = %q, want %q", resp.Data.Content, "worker output line\n") + } +} + +func TestGetJobOutput_PlaceholderResolution(t *testing.T) { + resetJobsState() + + tmpDir := t.TempDir() + // FormatJobID(4001) = "0-0-0-4001" for small test IDs + outFile := filepath.Join(tmpDir, "job_0-0-0-4001.out") + + job := models.JobInfo{ + Unique_JobID: 4001, + JobID: 4001, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x02, + OutFile: filepath.Join(tmpDir, "job_%J.out"), + } + store.Jobs.Put(4001, job) + + os.WriteFile(outFile, []byte("placeholder resolved\n"), 0644) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/4001/output?lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Data OutputData `json:"data"` + } + decodeResponse(t, w, &resp) + + if resp.Data.Content != "placeholder resolved\n" { + t.Errorf("content = %q, want %q", resp.Data.Content, "placeholder resolved\n") + } +} + +func TestGetJobOutput_JobNotFound(t *testing.T) { + resetJobsState() + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/99999/output?lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusNotFound { + t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestGetJobOutput_InvalidParams(t *testing.T) { + resetJobsState() + + router := setupTestRouter() + + tests := []struct { + name string + url string + expectCode int + }{ + {"invalid offset", "/xce/v1/jobs/1001/output?offset=abc", http.StatusBadRequest}, + {"invalid lines zero", "/xce/v1/jobs/1001/output?lines=0", http.StatusBadRequest}, + {"invalid type", "/xce/v1/jobs/1001/output?type=combined", http.StatusBadRequest}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", tc.url, nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + if w.Code != tc.expectCode { + t.Errorf("expected %d, got %d: %s", tc.expectCode, w.Code, w.Body.String()) + } + }) + } +} + +func TestLastNLines(t *testing.T) { + tests := []struct { + name string + input []byte + n int + expect string + }{ + {"tail 2 of 5", []byte("a\nb\nc\nd\ne\n"), 2, "d\ne\n"}, + {"tail all", []byte("a\nb\n"), 100, "a\nb\n"}, + {"empty", []byte(""), 10, ""}, + {"no trailing newline", []byte("no newline"), 5, "no newline"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := lastNLines(tc.input, tc.n) + if got != tc.expect { + t.Errorf("lastNLines(%q, %d) = %q, want %q", tc.input, tc.n, got, tc.expect) + } + }) + } +} + +func TestResolveOutputPath(t *testing.T) { + // FormatJobID(123) = "0-0-0-123" for small packed IDs (cluster=0, apiserver=0, ts=0, seq=123) + job := models.JobInfo{ + Unique_JobID: 123, + JobID: 123, + ArrayIdx: 2, + SubmitUser: "alice", + CWD: "/home/alice/work", + } + + tests := []struct { + input string + expected string + }{ + {"/tmp/job_%J.out", "/tmp/job_0-0-0-123.out"}, + {"/tmp/job_%J_%I.out", "/tmp/job_0-0-0-123_2.out"}, + {"/tmp/job_%U.out", "/tmp/job_alice.out"}, + {"/tmp/job_%C.out", "/tmp/job_/home/alice/work.out"}, + // empty path → worker default for task 0 + {"", "/tmp/job_0-0-0-123_task0.out"}, + {"/tmp/fixed.out", "/tmp/fixed.out"}, + } + + for _, tc := range tests { + t.Run(tc.input, func(t *testing.T) { + result := resolveOutputPath(tc.input, job) + if result != tc.expected { + t.Errorf("resolveOutputPath(%q) = %q, want %q", tc.input, result, tc.expected) + } + }) + } +} + +func TestGetJobOutput_Stderr(t *testing.T) { + resetJobsState() + + tmpDir := t.TempDir() + errFile := filepath.Join(tmpDir, "job_5001.err") + + job := models.JobInfo{ + Unique_JobID: 5001, + JobID: 5001, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x02, + ErrFile: errFile, + } + store.Jobs.Put(5001, job) + + os.WriteFile(errFile, []byte("stderr output\n"), 0644) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/5001/output?type=stderr&lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Data OutputData `json:"data"` + } + decodeResponse(t, w, &resp) + + if resp.Data.Content != "stderr output\n" { + t.Errorf("content = %q, want %q", resp.Data.Content, "stderr output\n") + } +} + +func TestGetJobOutput_DoneJob(t *testing.T) { + resetJobsState() + + tmpDir := t.TempDir() + outFile := filepath.Join(tmpDir, "job_6001.out") + + job := models.JobInfo{ + Unique_JobID: 6001, + JobID: 6001, + ArrayIdx: 0, + SubmitUser: "testuser", + JobStatus: 0x08, // DONE + OutFile: outFile, + } + store.Jobs.Put(6001, job) + + os.WriteFile(outFile, []byte("finished output\n"), 0644) + + router := setupTestRouter() + + w := httptest.NewRecorder() + req, _ := http.NewRequest("GET", "/xce/v1/jobs/6001/output?lines=10", nil) + req.Header.Set("Authorization", testToken()) + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp struct { + Data OutputData `json:"data"` + } + decodeResponse(t, w, &resp) + + if !resp.Data.Done { + t.Error("done should be true for finished job") + } + if resp.Data.JobStatus != "DONE" { + t.Errorf("job_status = %q, want DONE", resp.Data.JobStatus) + } +} + diff --git a/src/frontend/apiserver/router/routers.go b/src/frontend/apiserver/router/routers.go index e999813f..b0e68ef1 100644 --- a/src/frontend/apiserver/router/routers.go +++ b/src/frontend/apiserver/router/routers.go @@ -125,6 +125,9 @@ func Router() *gin.Engine { jobsApi := r.Group("/xce/v1/jobs") { jobHandler := &controllers.JobController{} + // Note: /:jobid/output must be registered before /:jobids to avoid + // Gin matching "output" as a literal job ID. + jobsApi.GET("/:jobids/output", controllers.RequirePermission("job:view"), jobHandler.GetJobOutput) jobsApi.GET("", controllers.RequirePermission("job:view"), jobHandler.GetJobsInfo) jobsApi.GET("/:jobids", controllers.RequirePermission("job:view"), jobHandler.GetJobByIDs) jobsApi.POST("/create", controllers.RequirePermission("queue:submit"), jobHandler.CreateJob) diff --git a/src/frontend/xcecli/cmd/jobs/jobs.go b/src/frontend/xcecli/cmd/jobs/jobs.go index 95d28fe4..2ae57cc0 100755 --- a/src/frontend/xcecli/cmd/jobs/jobs.go +++ b/src/frontend/xcecli/cmd/jobs/jobs.go @@ -31,6 +31,8 @@ Use 'xce jobs ' to submit/kill/stop/resume/checkpoint/restart jobs.`, cmd.AddCommand(NewResumeJobsCmd(configManager)) cmd.AddCommand(NewCheckPointJobsCmd(configManager)) cmd.AddCommand(NewRestartJobsCmd(configManager)) + // Add peek command to view job output + cmd.AddCommand(NewPeekJobsCmd(configManager)) return cmd } diff --git a/src/frontend/xcecli/cmd/jobs/peek.go b/src/frontend/xcecli/cmd/jobs/peek.go new file mode 100644 index 00000000..cb281652 --- /dev/null +++ b/src/frontend/xcecli/cmd/jobs/peek.go @@ -0,0 +1,153 @@ +package jobs + +import ( + "fmt" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + "xcecli/common/format" + "xcecli/common/request" + "xcecli/config" + + "github.com/spf13/cobra" +) + +// PeekQuerier implements the jobs peek command. +type PeekQuerier struct { + configManager *config.ConfigManager + lines int + stderr bool + noFollow bool +} + +// NewPeekQuerier creates a new PeekQuerier. +func NewPeekQuerier(configManager *config.ConfigManager) *PeekQuerier { + return &PeekQuerier{ + configManager: configManager, + lines: 100, // default show last 100 lines + } +} + +// NewPeekJobsCmd creates the xce jobs peek command. +func NewPeekJobsCmd(configManager *config.ConfigManager) *cobra.Command { + querier := NewPeekQuerier(configManager) + + cmd := &cobra.Command{ + Use: "peek", + Short: "Peek at job output", + Long: `Peek at job output in real-time. + +Examples: + xce jobs peek jobid # stream stdout output + xce jobs peek jobid -n 50 # initial show last 50 lines + xce jobs peek jobid --err # view stderr output + xce jobs peek jobid --no-follow # one-shot snapshot, no streaming`, + RunE: func(cmd *cobra.Command, args []string) error { + if len(args) == 0 { + fmt.Println("error: job ID is required") + fmt.Println() + cmd.Help() + return nil + } + jobID := args[0] + if err := querier.Run(jobID); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + flags := cmd.Flags() + flags.IntVarP(&querier.lines, "lines", "n", 100, "number of initial lines to show") + flags.BoolVar(&querier.stderr, "err", false, "view stderr output instead of stdout") + flags.BoolVar(&querier.noFollow, "no-follow", false, "one-shot snapshot, do not follow new output") + + return cmd +} + +// Run executes the peek command. +func (q *PeekQuerier) Run(jobID string) error { + serverInfo, err := q.configManager.GetServerInfo() + if err != nil { + return err + } + + // Convert job ID to API format + apiJobID, err := format.ToAPIJobID(jobID) + if err != nil { + return fmt.Errorf("invalid job ID %q: %w", jobID, err) + } + + apiClient := request.NewAPIClient(serverInfo.URL) + apiClient.SetRootCAs(serverInfo.CACert) + + // Determine stream type + streamType := "stdout" + if q.stderr { + streamType = "stderr" + } + + // Setup signal handling for graceful exit + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + + offset := int64(0) + lines := q.lines + + for { + // Check for interrupt + select { + case <-sigChan: + fmt.Println() + return nil + default: + } + + // Build query params + params := map[string]string{ + "type": streamType, + } + if offset == 0 { + params["lines"] = strconv.Itoa(lines) + } else { + params["offset"] = strconv.FormatInt(offset, 10) + } + + // Fetch output + resp, err := apiClient.GetJobOutput(serverInfo.Token, serverInfo.CACert, apiJobID, params) + if err != nil { + return err + } + + // Check for error in response + if resp.Code != 200 { + return fmt.Errorf("job <%s>: %s", jobID, resp.Msg) + } + + // Print new content + if resp.Data.Content != "" { + fmt.Print(resp.Data.Content) + } + + // Update offset for next poll + offset = resp.Data.Offset + + // If done, exit + if resp.Data.Done { + return nil + } + + // If one-shot mode, exit after first fetch + if q.noFollow { + return nil + } + + // Poll interval + time.Sleep(500 * time.Millisecond) + } +} diff --git a/src/frontend/xcecli/cmd/jobs/peek_test.go b/src/frontend/xcecli/cmd/jobs/peek_test.go new file mode 100644 index 00000000..5db8bb47 --- /dev/null +++ b/src/frontend/xcecli/cmd/jobs/peek_test.go @@ -0,0 +1,80 @@ +package jobs + +import ( + "strings" + "testing" + + "xcecli/common/format" +) + +func TestPeekJobIDParsing(t *testing.T) { + tests := []struct { + input string + wantAPI string + }{ + {"12345", "12345"}, + {"12345[0]", "12345[0]"}, + {"12345[2]", "12345[2]"}, + {"plain_id", "plain_id"}, // ToAPIJobID passes through non-numeric strings + } + + for _, tc := range tests { + t.Run(tc.input, func(t *testing.T) { + apiID, err := format.ToAPIJobID(tc.input) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if apiID != tc.wantAPI { + t.Errorf("apiID = %q, want %q", apiID, tc.wantAPI) + } + }) + } +} + +func TestPeekDefaults(t *testing.T) { + q := NewPeekQuerier(nil) + if q.lines != 100 { + t.Errorf("default lines = %d, want 100", q.lines) + } + if q.stderr { + t.Error("default stderr should be false") + } + if q.noFollow { + t.Error("default noFollow should be false") + } +} + +func TestPeekStreamType(t *testing.T) { + q := NewPeekQuerier(nil) + + streamType := "stdout" + if q.stderr { + streamType = "stderr" + } + if streamType != "stdout" { + t.Errorf("default stream type = %q, want stdout", streamType) + } + + q2 := NewPeekQuerier(nil) + q2.stderr = true + streamType = "stdout" + if q2.stderr { + streamType = "stderr" + } + if streamType != "stderr" { + t.Errorf("stderr stream type = %q, want stderr", streamType) + } +} + +func TestPeekCommand_NoJobID(t *testing.T) { + q := NewPeekQuerier(nil) + + err := q.Run("") + if err == nil { + t.Error("expected error for empty job ID") + } + // Should fail on job ID parsing, not on config + if !strings.Contains(strings.ToLower(err.Error()), "invalid") { + t.Logf("error message (may vary): %v", err) + } +} diff --git a/src/frontend/xcecli/common/request/request.go b/src/frontend/xcecli/common/request/request.go index c3689c09..4af8f0da 100755 --- a/src/frontend/xcecli/common/request/request.go +++ b/src/frontend/xcecli/common/request/request.go @@ -170,6 +170,45 @@ func (c *APIClient) SubmitJob(token string, req *models.JobSubmitRequest) (*mode return &resp, nil } +// GetJobOutput retrieves job output from the API server. +func (c *APIClient) GetJobOutput(token, certPath, jobID string, params map[string]string) (*models.JobOutputResponse, error) { + var resp models.JobOutputResponse + baseURL := c.BaseURL + if strings.HasPrefix(baseURL, "https://") { + c.Client.SetTLSClientConfig(&tls.Config{ + RootCAs: loadRootCAs(certPath), + }) + } + if idx := strings.Index(baseURL, "/xce/v1"); idx != -1 { + baseURL = baseURL[:idx] + } + url := baseURL + "/xce/v1/jobs/" + jobID + "/output" + + req := c.Client.R(). + SetHeader("Authorization", token). + SetResult(&resp) + + for k, v := range params { + req.SetQueryParam(k, v) + } + + httpResp, err := req.Get(url) + if err != nil { + errMsg := ExtractErrorMessage(err) + return nil, fmt.Errorf("get job output request failed: [%s]", errMsg) + } + + if httpResp.StatusCode() != http.StatusOK { + logger.Info("get job output failed", zap.Any("resp", resp), zap.Int("status", httpResp.StatusCode())) + if resp.Msg != "" { + return &resp, fmt.Errorf("get job output failed: %s", resp.Msg) + } + return &resp, fmt.Errorf("get job output failed: status %d", httpResp.StatusCode()) + } + + return &resp, nil +} + // extractErrorMessage extracts a user-friendly error message from various error types func ExtractErrorMessage(err error) string { if err == nil { diff --git a/src/frontend/xcecli/models/jobs.go b/src/frontend/xcecli/models/jobs.go index 57048bf8..785573e5 100755 --- a/src/frontend/xcecli/models/jobs.go +++ b/src/frontend/xcecli/models/jobs.go @@ -515,3 +515,18 @@ func (e *ExitStatusValue) UnmarshalJSON(data []byte) error { *e = ExitStatusValue(s) return nil } + +// JobOutputResponse represents the response from the job output API. +type JobOutputResponse struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data JobOutputData `json:"data"` +} + +// JobOutputData represents the payload in a job output response. +type JobOutputData struct { + Content string `json:"content"` + Offset int64 `json:"offset"` + JobStatus string `json:"job_status"` + Done bool `json:"done"` +} diff --git a/src/include/client.h b/src/include/client.h index 7d5e21cd..0633dc7e 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -75,4 +75,7 @@ extern int xcecli_jobinfo(xcecli_ctx_t, uint64_t, xce_job_t**); #define xcecli_free_groups(ngroups, groups) xce_free_groups(ngroups, groups) #define xcecli_free_clusterconf(conf) xce_free_clusterconf(conf) +extern int xcecli_reply_fetchfile(xcecli_ctx_t, xce_request_t*, xce_fetchfile_reply_t*); +extern int xcecli_fetchfile(xcecli_ctx_t, xce_fetchfile_t*, xce_fetchfile_reply_t**); + #endif diff --git a/src/include/mylock.h b/src/include/mylock.h index 3fb81df6..dc8ac123 100644 --- a/src/include/mylock.h +++ b/src/include/mylock.h @@ -35,4 +35,6 @@ extern void cbuf_fini(cbuf_t*); extern int cbuf_put(cbuf_t*, void*); extern int cbuf_tryput(cbuf_t*, void*); extern int cbuf_get(cbuf_t*, void*); +extern int cbuf_tryget(cbuf_t*, void*); +extern int cbuf_timedget(cbuf_t*, void*, const struct timespec*); #endif diff --git a/src/include/request.h b/src/include/request.h index a48b4b1a..09e2cb7e 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -58,5 +58,6 @@ extern int xce_hostctrl_req(endpoint_t*, xce_hostctrl_t*); extern int xce_runtask_req(endpoint_t*, xce_taskrun_t*); extern int xce_set_pend_reason_req(endpoint_t*, xce_preason_set_t*); extern int xce_clear_reason_req(endpoint_t*, xce_clear_reason_t*); +extern int xce_fetchfile_req(endpoint_t*, xce_fetchfile_t*); #endif diff --git a/src/include/typemap.h b/src/include/typemap.h index e7d36eed..77090256 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -170,6 +170,7 @@ typedef struct { uint64_t jobid; xce_data_t* getdata; xce_clusterconf_t* clusterconf; + xce_fetchfile_reply_t* fetchfilereply; } data; } xce_reply_t; @@ -197,6 +198,7 @@ typedef struct { xce_preason_set_t* preasonmsg; xce_clear_reason_t* clrpreasonmsg; xce_hostctrl_t* hostctrlmsg; + xce_fetchfile_t* fetchfilemsg; } body; } xce_request_t; diff --git a/src/include/xce/types.h b/src/include/xce/types.h index d67f29b0..f2342616 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -773,4 +773,19 @@ typedef struct { xce_limit_t* limit; } xce_vres_t; +typedef struct { + uint64_t jobid; + char* filepath; + uint64_t offset; + uint32_t max_bytes; +} xce_fetchfile_t; + +typedef struct { + int32_t error; + uint8_t* content; + uint64_t file_size; + uint32_t bytes_read; + uint64_t jobid; /* echoed from request for pending-request matching */ +} xce_fetchfile_reply_t; + #endif diff --git a/src/include/xceapi.h b/src/include/xceapi.h index 0352b7cf..ad28d915 100644 --- a/src/include/xceapi.h +++ b/src/include/xceapi.h @@ -33,5 +33,6 @@ extern int xce_update_jusage(xce_ctx_t*, uint64_t, int, int*, int, int*, int, extern int xce_set_pend_reason(xce_ctx_t*, xce_preason_set_t*); extern int xce_clear_reason(xce_ctx_t*, xce_clear_reason_t*); extern int xce_jobinfo(xce_ctx_t*, uint64_t, xce_job_t**); +extern int xce_fetchfile(xce_ctx_t*, xce_fetchfile_t*, xce_fetchfile_reply_t**); #endif diff --git a/src/include/xcectx.h b/src/include/xcectx.h index a80d7f74..22769d19 100644 --- a/src/include/xcectx.h +++ b/src/include/xcectx.h @@ -30,10 +30,11 @@ typedef struct { struct { union { - char* debugmsg; - dataset_t* get; - uint64_t jobid; - xce_clusterconf_t* clusterconf; + char* debugmsg; + dataset_t* get; + uint64_t jobid; + xce_clusterconf_t* clusterconf; + xce_fetchfile_reply_t* fetchfilereply; } u; } reply; } xce_ctx_t; diff --git a/src/lib/client.c b/src/lib/client.c index fc94b86d..cf4935dc 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -673,3 +673,33 @@ int xcecli_jobinfo(xcecli_ctx_t c, uint64_t jobid, xce_job_t** job) rc = xce_jobinfo(&(ctx->ctx), jobid, job); return rc; } + +int xcecli_reply_fetchfile(xcecli_ctx_t c, xce_request_t* request, + xce_fetchfile_reply_t* freply) +{ + api_ctx_t* ctx = (api_ctx_t*)c; + xce_reply_t reply; + + if ((ctx == NULL) || (request == NULL) || (freply == NULL)) { + return -EINVAL; + } + + init_reply_by_req(&reply, request); + reply.status = freply->error; + reply.data_which = RequestReply_data_fetchfilereply; + reply.data.fetchfilereply = freply; + + return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply); +} + +int xcecli_fetchfile(xcecli_ctx_t c, xce_fetchfile_t* msg, + xce_fetchfile_reply_t** reply) +{ + api_ctx_t* ctx = (api_ctx_t*)c; + + if ((ctx == NULL) || (msg == NULL) || (reply == NULL)) { + return -EINVAL; + } + + return xce_fetchfile(&(ctx->ctx), msg, reply); +} diff --git a/src/lib/helper.c b/src/lib/helper.c index 0dc33e54..2c03497e 100644 --- a/src/lib/helper.c +++ b/src/lib/helper.c @@ -147,6 +147,8 @@ char* xce_request_op_name(xce_request_t* req) return "setpreason"; case OPCode_clrpreason: return "clrpreason"; + case OPCode_fetchfile: + return "fetchfile"; default: break; } diff --git a/src/lib/meson.build b/src/lib/meson.build index 8503f2ba..1e26a829 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -163,6 +163,7 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_wobj.c', 'testsuite/test_qobj.c', 'testsuite/test_request.c', + 'testsuite/test_cbuf.c', ] libxce_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep] diff --git a/src/lib/mylock.c b/src/lib/mylock.c index fcd6d89e..52617047 100644 --- a/src/lib/mylock.c +++ b/src/lib/mylock.c @@ -187,6 +187,32 @@ int cbuf_tryget(cbuf_t* b, void* it) return rc; } +int cbuf_timedget(cbuf_t* b, void* it, const struct timespec* deadline) +{ + if ((b == NULL) || (it == NULL) || (deadline == NULL)) { + return -EINVAL; + } + + 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; + } + } + + memcpy(it, (void*)((char*)b->data) + b->isize * b->head, b->isize); + + b->head = (b->head + 1) % b->ncap; + b->len--; + + pthread_cond_signal(&(b->condput)); + + pthread_mutex_unlock(&(b->mutex)); + return 0; +} + int my_rwlock_init(my_rwlock_t* l) { if (l == NULL) { diff --git a/src/lib/request.c b/src/lib/request.c index 0409ff54..74e578e7 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -935,3 +935,26 @@ int xce_clear_reason_req(endpoint_t* ep, xce_clear_reason_t* reason) return xce_request(ep, &request); } + +int xce_fetchfile_req(endpoint_t* ep, xce_fetchfile_t* msg) +{ + /* clang-format off */ + xce_request_t request = { + .opcode = OPCode_fetchfile, + .body_which = Request_body_fetchfilemsg, + .body = { + .fetchfilemsg = msg + } + }; + /* clang-format on */ + + if ((ep == NULL) || (msg == NULL)) { + return -EINVAL; + } + + if (msg->filepath == NULL || msg->filepath[0] == '\0') { + return -EINVAL; + } + + return xce_request(ep, &request); +} diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index 760e5b04..3383c963 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -29,6 +29,7 @@ extern void test_jobobj(void); extern void test_qobj(void); extern void test_wobj(void); extern void test_request(void); +extern void test_cbuf(void); void setUp(void) { return; } @@ -127,6 +128,7 @@ int main(int argc, char* argv[]) TEST(test_wobj); TEST(test_qobj); TEST(test_request); + TEST(test_cbuf); return UNITY_END(); } diff --git a/src/lib/testsuite/test_cbuf.c b/src/lib/testsuite/test_cbuf.c new file mode 100644 index 00000000..b9dc2e84 --- /dev/null +++ b/src/lib/testsuite/test_cbuf.c @@ -0,0 +1,207 @@ +#include +#include +#include +#include + +#include + +#include "mylock.h" + +/* ------------------------------------------------------------------ * + * helpers + * ------------------------------------------------------------------ */ + +static void make_deadline(struct timespec *ts, long ms) +{ + clock_gettime(CLOCK_REALTIME, ts); + ts->tv_sec += ms / 1000; + ts->tv_nsec += (ms % 1000) * 1000000L; + if (ts->tv_nsec >= 1000000000L) { + ts->tv_sec++; + ts->tv_nsec -= 1000000000L; + } +} + +/* ------------------------------------------------------------------ * + * cbuf_timedget: argument validation + * ------------------------------------------------------------------ */ + +static void test_cbuf_timedget_null_buf(void) +{ + int val; + struct timespec ts; + make_deadline(&ts, 100); + TEST_ASSERT_EQUAL_INT(-EINVAL, cbuf_timedget(NULL, &val, &ts)); +} + +static void test_cbuf_timedget_null_item(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + struct timespec ts; + make_deadline(&ts, 100); + TEST_ASSERT_EQUAL_INT(-EINVAL, cbuf_timedget(&b, NULL, &ts)); + cbuf_fini(&b); +} + +static void test_cbuf_timedget_null_deadline(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + int val; + TEST_ASSERT_EQUAL_INT(-EINVAL, cbuf_timedget(&b, &val, NULL)); + cbuf_fini(&b); +} + +/* ------------------------------------------------------------------ * + * cbuf_timedget: timeout on empty buffer + * ------------------------------------------------------------------ */ + +static void test_cbuf_timedget_timeout_empty(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + + int val = 0; + struct timespec ts; + make_deadline(&ts, 50); /* 50 ms */ + + int rc = cbuf_timedget(&b, &val, &ts); + TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc); + + cbuf_fini(&b); +} + +/* ------------------------------------------------------------------ * + * cbuf_timedget: normal read when item is already available + * ------------------------------------------------------------------ */ + +static void test_cbuf_timedget_success_immediate(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + + int put_val = 42; + cbuf_put(&b, &put_val); + + int got_val = 0; + struct timespec ts; + make_deadline(&ts, 1000); /* generous 1 s */ + + int rc = cbuf_timedget(&b, &got_val, &ts); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(42, got_val); + + cbuf_fini(&b); +} + +/* ------------------------------------------------------------------ * + * cbuf_timedget: producer thread scenario + * ------------------------------------------------------------------ */ + +typedef struct { + cbuf_t *buf; + int value; + long delay_ms; +} producer_arg_t; + +static void *producer_thread(void *arg) +{ + producer_arg_t *a = (producer_arg_t *)arg; + struct timespec ts = { .tv_sec = a->delay_ms / 1000, + .tv_nsec = (a->delay_ms % 1000) * 1000000L }; + nanosleep(&ts, NULL); + cbuf_put(a->buf, &a->value); + return NULL; +} + +static void test_cbuf_timedget_success_after_producer(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + + producer_arg_t arg = { .buf = &b, .value = 99, .delay_ms = 30 }; + pthread_t tid; + pthread_create(&tid, NULL, producer_thread, &arg); + + int got_val = 0; + struct timespec ts; + make_deadline(&ts, 500); /* 500 ms — producer fires at 30 ms */ + + int rc = cbuf_timedget(&b, &got_val, &ts); + pthread_join(tid, NULL); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(99, got_val); + + cbuf_fini(&b); +} + +/* ------------------------------------------------------------------ * + * cbuf_timedget: timeout beats slow producer + * ------------------------------------------------------------------ */ + +static void test_cbuf_timedget_timeout_beats_slow_producer(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + + /* Producer fires at 200 ms, but we timeout at 50 ms */ + producer_arg_t arg = { .buf = &b, .value = 7, .delay_ms = 200 }; + pthread_t tid; + pthread_create(&tid, NULL, producer_thread, &arg); + + int got_val = 0; + struct timespec ts; + make_deadline(&ts, 50); + + int rc = cbuf_timedget(&b, &got_val, &ts); + TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc); + + pthread_join(tid, NULL); + cbuf_fini(&b); +} + +/* ------------------------------------------------------------------ * + * cbuf_tryget: basic tests (also not previously declared in header) + * ------------------------------------------------------------------ */ + +static void test_cbuf_tryget_empty(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + int val; + TEST_ASSERT_EQUAL_INT(-EAGAIN, cbuf_tryget(&b, &val)); + cbuf_fini(&b); +} + +static void test_cbuf_tryget_success(void) +{ + cbuf_t b; + cbuf_init(&b, 4, sizeof(int)); + int put_val = 77; + cbuf_put(&b, &put_val); + int got_val = 0; + TEST_ASSERT_EQUAL_INT(0, cbuf_tryget(&b, &got_val)); + TEST_ASSERT_EQUAL_INT(77, got_val); + cbuf_fini(&b); +} + +/* ------------------------------------------------------------------ * + * suite entry point + * ------------------------------------------------------------------ */ + +void test_cbuf(void) +{ + printf("\n>> cbuf_timedget / cbuf_tryget tests\n\n"); + + RUN_TEST(test_cbuf_timedget_null_buf); + RUN_TEST(test_cbuf_timedget_null_item); + RUN_TEST(test_cbuf_timedget_null_deadline); + RUN_TEST(test_cbuf_timedget_timeout_empty); + RUN_TEST(test_cbuf_timedget_success_immediate); + RUN_TEST(test_cbuf_timedget_success_after_producer); + RUN_TEST(test_cbuf_timedget_timeout_beats_slow_producer); + RUN_TEST(test_cbuf_tryget_empty); + RUN_TEST(test_cbuf_tryget_success); +} diff --git a/src/lib/xceapi.c b/src/lib/xceapi.c index 96c3d384..3dc7a524 100644 --- a/src/lib/xceapi.c +++ b/src/lib/xceapi.c @@ -1,8 +1,17 @@ #include +#include +#include "mylock.h" #include "utils.h" #include "xceapi.h" +/* All synchronous API calls wait at most 10 seconds for a reply. */ +static void cbuf_deadline_10s(struct timespec* ts) +{ + clock_gettime(CLOCK_REALTIME, ts); + ts->tv_sec += 10; +} + int xce_hello(xce_ctx_t* ctx, char* msg) { int rc, status; @@ -19,7 +28,8 @@ int xce_hello(xce_ctx_t* ctx, char* msg) if (rc < 0) { return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -113,7 +123,8 @@ int xce_get_with_condition(xce_ctx_t* ctx, xce_get_t* get, void*** data) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -192,7 +203,8 @@ int xce_resval(xce_ctx_t* ctx, char* host, char* res, xce_resval_t*** vals) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -225,7 +237,8 @@ int xce_join(xce_ctx_t* ctx, int role, hostattrs_t* attrs) if (rc < 0) { return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -259,7 +272,8 @@ int xce_jobdispatch(xce_ctx_t* ctx, xce_job_dispatch_t* info) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -301,7 +315,8 @@ int xce_jobsubmit(xce_ctx_t* ctx, xce_job_submit_t* submit, uint64_t* jobid) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -414,7 +429,8 @@ int xce_clusterconf(xce_ctx_t* ctx, xce_clusterconf_t** conf) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -458,7 +474,8 @@ int xce_update_resvals(xce_ctx_t* ctx, int count, xce_resval_t** vals) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -501,7 +518,8 @@ int xce_update_jusage(xce_ctx_t* ctx, uint64_t jobid, int npids, int* pids, return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -527,7 +545,8 @@ int xce_jobctrl(xce_ctx_t* ctx, xce_jobctrl_t* ctrl) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -553,7 +572,8 @@ int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -579,7 +599,8 @@ int xce_hostctrl(xce_ctx_t* ctx, xce_hostctrl_t* ctrl) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -605,7 +626,8 @@ int xce_runtask(xce_ctx_t* ctx, xce_taskrun_t* runinfo) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -631,7 +653,8 @@ int xce_set_pend_reason(xce_ctx_t* ctx, xce_preason_set_t* reasons) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -657,7 +680,8 @@ int xce_clear_reason(xce_ctx_t* ctx, xce_clear_reason_t* reasons) return rc; } - rc = cbuf_get(&(ctx->wbuf), &status); + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); if (rc < 0) { return rc; } @@ -697,3 +721,37 @@ int xce_jobinfo(xce_ctx_t* ctx, uint64_t jobid, xce_job_t** job) return 1; } + +int xce_fetchfile(xce_ctx_t* ctx, xce_fetchfile_t* msg, xce_fetchfile_reply_t** reply) +{ + int rc; + int status; + + if ((ctx == NULL) || (msg == NULL) || (reply == NULL)) { + return -EINVAL; + } + + if (ctx->state != XCE_CTX_RUN) { + return -EINVAL; + } + + ctx->reply.u.fetchfilereply = NULL; + + rc = xce_fetchfile_req(ctx->ep, msg); + if (rc < 0) { + return rc; + } + + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); + if (rc < 0) { + return rc; + } + + if (status < 0) { + return status; + } + + MOVE_PTR(*reply, ctx->reply.u.fetchfilereply); + return 0; +} diff --git a/src/lib/xcectx.c b/src/lib/xcectx.c index 7aa3f267..257d6b83 100644 --- a/src/lib/xcectx.c +++ b/src/lib/xcectx.c @@ -49,6 +49,13 @@ static int handle_reply(xce_ctx_t* ctx, xce_reply_t* reply) rc = 1; } break; + case OPCode_fetchfile: + rc = -1; + if (reply->data_which == RequestReply_data_fetchfilereply) { + MOVE_PTR(ctx->reply.u.fetchfilereply, reply->data.fetchfilereply); + rc = 1; + } + break; default: rc = -ENOENT; break; diff --git a/src/scheduler/src/plugins/parjob_allocator.c b/src/scheduler/src/plugins/parjob_allocator.c index faa80431..9e731668 100644 --- a/src/scheduler/src/plugins/parjob_allocator.c +++ b/src/scheduler/src/plugins/parjob_allocator.c @@ -59,6 +59,10 @@ static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) if (obj->tag == XCE_RESREQ_PRIMARY) { xce_primary_resreq_t* primary = obj->u.primary->resreq; + if (primary == NULL) { + return 0; + } + if (primary->maxTasks == 1) { /* do not care sequence job */ diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 253ba463..ba02aa81 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -168,7 +168,13 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) return NULL; } - rc = parse_resreq(job->resreq, j->submit->resreq, sched->resmgr, &errmsg); + { + char* resreq_str = j->submit->resreq; + if (NULLSTR(resreq_str)) { + resreq_str = "type == any"; + } + rc = parse_resreq(job->resreq, resreq_str, sched->resmgr, &errmsg); + } if (rc < 0) { ERROR("%s parse_resreq failed, %s", __func__, errmsg); FREEUP(errmsg); diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 4749860f..077c7b93 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -111,6 +111,10 @@ static int sched_make_primary_allocation(xce_scheduler_t* sched, xce_sched_ctx_t ctx; int rc = 0; + if (primary == NULL) { + return -EINVAL; + } + sched_ctx_init(&ctx, sched, obj); sched_ctx_set_primary(&ctx, primary); rc = sched_make_primary_alloc_(sched, &ctx); diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 0309fd06..e8645aa6 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -30,6 +30,7 @@ enum OPCode { runtasks @19; setpreason @20; clrpreason @21; + fetchfile @22; } struct PeerInfo $C.mapname("xce_peerinfo_t") { @@ -536,6 +537,7 @@ struct RequestReply $C.mapname("xce_reply_t"){ getdata @5: Data; jobid @6: UInt64; clusterconf @7: ClusterConf; + fetchfilereply @8: FetchFileReply; } } @@ -569,6 +571,21 @@ struct ClearPReason $C.mapname("xce_clear_reason_t") { jobs @0: List(UInt64) $C.mapname("jobs") $C.maplistcount("njobs"); } +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"); + content @1: List(UInt8) $C.mapname("content") $C.maplistcount("bytes_read"); + fileSize @2: UInt64 $C.mapname("file_size"); + bytesRead @3: UInt32 $C.mapname("bytes_read"); + jobid @4: UInt64 $C.mapname("jobid"); +} + struct Request $C.mapname("xce_request_t"){ opcode @0: OPCode; seq @1: UInt64; @@ -593,6 +610,7 @@ struct Request $C.mapname("xce_request_t"){ preasonmsg @19: PendReasonSet; clrpreasonmsg @20: ClearPReason; hostctrlmsg @21: HostControl; + fetchfilemsg @22: FetchFileMsg; } } diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 6ccc3f16..5624c13f 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -1,3 +1,4 @@ +#include "client.h" #include "controller_connector.h" #include "netio.h" #include "uv_ipc.h" @@ -11,6 +12,7 @@ #include #include #include +#include #ifdef _WIN32 #include @@ -1097,6 +1099,89 @@ static int handle_runtasks(xce_request_t* request) return 0; } +/** + * @brief Handles the fetchfile request from the controller. + * + * Reads up to msg->max_bytes bytes from msg->filepath starting at msg->offset + * and sends the data back to the controller as an OPCode_reply. + * + * @param request Pointer to the xce_request_t containing the fetchfile msg. + * @return int 0 on success, -1 on failure. + */ +static int handle_fetchfile(xce_request_t* request) +{ + xce_fetchfile_t* msg = request->body.fetchfilemsg; + xce_fetchfile_reply_t freply; + struct stat st; + uint8_t* buf = NULL; + int fd; + ssize_t nread; + uint32_t maxbytes; + + XCECD_LOG(XCE_LOG_DEBUG, + "Fetchfile request: jobid=%lu path=%s offset=%lu max=%u", + msg->jobid, msg->filepath, msg->offset, msg->max_bytes); + + memset(&freply, 0, sizeof(freply)); + freply.jobid = msg->jobid; + + if (msg->filepath == NULL || msg->filepath[0] == '\0') { + XCECD_LOG(XCE_LOG_ERROR, "Fetchfile: empty filepath"); + freply.error = -EINVAL; + goto send; + } + + /* Bound max_bytes to avoid huge allocations */ + maxbytes = (msg->max_bytes == 0 || msg->max_bytes > 4 * 1024 * 1024) + ? 4 * 1024 * 1024 + : msg->max_bytes; + + fd = open(msg->filepath, O_RDONLY); + if (fd < 0) { + XCECD_LOG(XCE_LOG_ERROR, "Fetchfile: open(%s) failed: %s", + msg->filepath, strerror(errno)); + freply.error = -errno; + goto send; + } + + if (fstat(fd, &st) < 0) { + freply.error = -errno; + close(fd); + goto send; + } + freply.file_size = (uint64_t)st.st_size; + + buf = malloc(maxbytes); + if (buf == NULL) { + freply.error = -ENOMEM; + close(fd); + goto send; + } + + nread = pread(fd, buf, maxbytes, (off_t)msg->offset); + close(fd); + + if (nread < 0) { + XCECD_LOG(XCE_LOG_ERROR, "Fetchfile: pread failed: %s", + strerror(errno)); + freply.error = -errno; + freply.bytes_read = 0; + free(buf); + buf = NULL; + } else { + freply.error = 0; + freply.bytes_read = (uint32_t)nread; + freply.content = buf; + buf = NULL; /* ownership passed to freply */ + } + +send: + xcecli_reply_fetchfile(xcecli_ctx, request, &freply); + free(freply.content); + free(buf); + return 0; +} + /** * @brief Handles the runtasks request from the controller. * @@ -1234,6 +1319,9 @@ static int myreq_handler(xce_request_t* request, void* args) case OPCode_jobcontrol: ret = handle_jobcontrol(request); break; + case OPCode_fetchfile: + ret = handle_fetchfile(request); + break; default: break; }