feat(scheduler): working default record-dir, improved REPL output, job ID formatting
- Change default record_dir from /var/log/xce/ to /tmp (always exists) - Add mkdir(record_dir, 0755) at startup for custom --record-dir paths - Split validation_error_text into 6 distinct per-errno messages - Accept human-readable job IDs in 'info job' via parse_jobid() - Add 'info jobs' command to list all jobs in sim REPL - Format job IDs with fmt_jobid() for xce_test-consistent output - Replace printf with fprintf(stdout, ...) in all REPL files to bypass -Wl,--wrap=printf linker interception
This commit is contained in:
parent
d9d0613598
commit
c705cadadc
SSH key fingerprint:
SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
8 changed files with 164 additions and 94 deletions
271
src/docs/archive/cluster-config-split-analysis.md
Normal file
271
src/docs/archive/cluster-config-split-analysis.md
Normal file
|
|
@ -0,0 +1,271 @@
|
|||
# ClusterConfig Schema Split Analysis
|
||||
|
||||
> Analysis date: 2026-06-08. Branch: persistent_ha.
|
||||
> Source schema: `src/schema/controller.capnp:680`, struct `ClusterConfig`.
|
||||
|
||||
## Summary
|
||||
|
||||
The `ClusterConfig` struct (20 fields) mixes three distinct configuration concerns:
|
||||
cluster-wide business config, HA peer membership, and per-controller performance
|
||||
tunables. These should be split into separate objects so each can evolve
|
||||
independently without forcing a re-push of the full config blob to all standbys.
|
||||
|
||||
## Current State Discovery
|
||||
|
||||
### Peer list is already structurally separate
|
||||
|
||||
The `--peers` CLI flag (`main.c:178-210`) parses into `myconf.peers_cfg`
|
||||
(`xce_cluster_config_t`), which is a **separate struct** from the config blob's
|
||||
`cluster_config_t`. At startup, `cluster_init_phase1` uses `peers_cfg.candidates`
|
||||
to determine `my_name`, `my_port`, `my_priority`, and the peer list — all before
|
||||
the config blob is loaded (`cluster.c:865-880`). The config blob's `candidates`
|
||||
field is **not consulted** for HA peer discovery; it is carried for backward
|
||||
compatibility in the blob but superseded by `--peers`.
|
||||
|
||||
### HA tunables are hardcoded at phase1
|
||||
|
||||
`cluster_init_phase1` uses compile-time defaults for all HA timing parameters
|
||||
(`cluster.c:854-863`):
|
||||
```c
|
||||
ha_config_t ha_cfg = {
|
||||
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
|
||||
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
|
||||
.election_timeout_ms = HA_DEFAULT_ELECTION_TIMEOUT_MS,
|
||||
...
|
||||
};
|
||||
```
|
||||
|
||||
The `heartbeatInterval` and `inactiveHeartbeatCount` fields in
|
||||
`ClusterConfig` are consumed later (e.g., `cluster.c:581-585` writes them into
|
||||
a config key-value store for runtime access), but the HA init path does not
|
||||
read them. The heartbeat timeout validation in `startup_validation.c:144` works
|
||||
against the hardcoded defaults, not the config blob values.
|
||||
|
||||
### vcpus is already deprecated
|
||||
|
||||
Field `@7 vcpus` is marked deprecated in the schema comment: "use --vcpus CLI
|
||||
flag; kept for config blob backward compatibility." `cluster_init_phase2`
|
||||
reads `c->myconf->vcpus` (set from CLI), not the blob's `vcpus` field
|
||||
(`cluster.c:1045`).
|
||||
|
||||
## Proposed Split: Three Configuration Objects
|
||||
|
||||
```
|
||||
┌──────────────────────────────┐
|
||||
│ Current ClusterConfig │
|
||||
│ (20 fields) │
|
||||
└──────────────┬───────────────┘
|
||||
│
|
||||
┌───────────────────┼───────────────────┐
|
||||
▼ ▼ ▼
|
||||
┌───────────────────┐ ┌─────────────────┐ ┌──────────────────────┐
|
||||
│ ClusterConfig │ │ PeersConfig │ │ ControllerConfig │
|
||||
│ (cluster-wide) │ │ (HA membership) │ │ (per-controller) │
|
||||
└───────────────────┘ └─────────────────┘ └──────────────────────┘
|
||||
```
|
||||
|
||||
### Group 1: Cluster-Wide Business Config (keep in `ClusterConfig`)
|
||||
|
||||
These fields define the cluster's behavior, resources, and policy. They **must
|
||||
be identical** across all controllers. When modified via `OP_upgradeConfig`,
|
||||
the new blob is pushed to all standbys and committed on majority.
|
||||
|
||||
| Field | Ord | Type | Rationale |
|
||||
|-------|-----|------|-----------|
|
||||
| `clusterName` | @0 | Text | Cluster identity — must be identical cluster-wide |
|
||||
| `resources` | @8 | List(Resource) | Resource catalog — shared state across all nodes |
|
||||
| `queues` | @9 | List(QueueConf) | Queue definitions — cluster scheduling policy |
|
||||
| `ugroups` | @10 | List(GroupConfig) | User group membership — access control |
|
||||
| `hgroups` | @11 | List(GroupConfig) | Host group membership — scheduling constraints |
|
||||
| `hostConfigs` | @12 | List(HostConfig) | Per-host overrides — cluster-wide host knowledge |
|
||||
| `hproviders` | @18 | List(ScalerConfig) | Autoscaler provider config — cluster infrastructure |
|
||||
| `limits` | @19 | List(LimitConfig) | Resource limits — cluster policy enforcement |
|
||||
| `configGeneration` | @20 | UInt64 | Monotonic version counter — replication consistency |
|
||||
|
||||
**How consumed**: `cluster_init_phase2` (after election and config load)
|
||||
iterates these to populate `c->resmgr`, `c->jobmgr`, queue tables, host
|
||||
groups, etc.
|
||||
|
||||
### Group 2: HA Peer Membership (extract to `PeersConfig`)
|
||||
|
||||
These fields define **who** is in the cluster. They change independently when
|
||||
controllers are added, removed, or re-IP'd — without any change to the
|
||||
business config.
|
||||
|
||||
| Field | Ord | Type | Rationale |
|
||||
|-------|-----|------|-----------|
|
||||
| `candidates` | @2 | List(MasterCandidate) | HA peer list — changes independently of business config |
|
||||
| `port` | @1 | UInt32 | Fallback listener port — derived from candidates; per-controller in practice |
|
||||
|
||||
**Already structurally separate**: The `--peers` CLI flag populates
|
||||
`myconf.peers_cfg` (`xce_cluster_config_t` in `src/include/clusterconfig.h`),
|
||||
a struct distinct from the config blob's `cluster_config_t`. The candidates
|
||||
list in the blob is carried for backward compatibility but is not read for HA
|
||||
peer discovery at startup. Making the separation formal means:
|
||||
- `PeersConfig` travels via its own capnp message type and replication path
|
||||
- Cluster config upgrades via `OP_upgradeConfig` no longer need to carry the candidate list
|
||||
- Peer list changes (add/remove controllers) are a dedicated operation with its own quorum semantics
|
||||
|
||||
**Candidate list has special quorum requirements**: Changing the peer set is a
|
||||
Raft cluster membership change. It must not use the same
|
||||
`repl_wait_majority_ack` path as business config because the majority
|
||||
denominator itself is changing. A separate message type enables a two-phase
|
||||
membership change protocol (joint consensus) in the future.
|
||||
|
||||
**Impact of removing `candidates` from `ClusterConfig` (verified against codebase):**
|
||||
|
||||
`cluster_build_config()` (`cluster.c:575–609`) is the **only** place that reads
|
||||
`cluster_config->candidates` and `cluster_config->ncandidates` — it copies them
|
||||
into the runtime KV store as `ncandidates` and `candidate_N_{name,host,port}`.
|
||||
No other component (scheduler, apiserver, opget, sched) reads those KV keys.
|
||||
The live peer list used by `ha.c:3266–3272` and `cluster_open_listener` comes
|
||||
exclusively from `peers_cfg.candidates[]` (parsed by `--peers`). Removing
|
||||
candidates from the blob is therefore a **zero-blast-radius** change.
|
||||
|
||||
| File | Location | Change |
|
||||
|------|----------|--------|
|
||||
| `src/controller/cluster.c` | 587–606 | Remove `ncandidates` + `candidate_N_*` KV loop from `cluster_build_config()` |
|
||||
| `src/include/clusterconfig.h` | 34–35 | Remove `int ncandidates` and `master_candidate_t** candidates` from `cluster_config_t` |
|
||||
| `src/schema/controller.capnp` | 683 | Remove `candidates @2: List(MasterCandidate)` from `ClusterConfig` |
|
||||
|
||||
**`MasterCandidate` fields**:
|
||||
```
|
||||
struct MasterCandidate {
|
||||
name @0: Text; // controller identity (e.g., "ctrl-1")
|
||||
host @1: Text; // hostname or IP
|
||||
port @2: UInt32; // peer listener port
|
||||
}
|
||||
```
|
||||
|
||||
### Group 3: Per-Controller Performance Tunables (extract to `ControllerConfig`)
|
||||
|
||||
These fields configure **how** an individual controller process behaves. They
|
||||
can legitimately differ across controllers (e.g., a lower-memory node with
|
||||
fewer vcpus, or a WAN-connected standby with a longer heartbeat timeout).
|
||||
|
||||
| Field | Ord | Type | Rationale |
|
||||
|-------|-----|------|-----------|
|
||||
| `heartbeatInterval` | @3 | UInt32 | Primary heartbeat send interval — can be tuned per primary |
|
||||
| `inactiveHeartbeatCount` | @4 | UInt32 | Missed-heartbeat tolerance multiplier |
|
||||
| `fetchJobInterval` | @5 | UInt32 | Job fetch poll interval — local scheduler interaction |
|
||||
| `netioTimeout` | @6 | UInt32 | Network I/O timeout — environment-specific |
|
||||
| `vcpus` | @7 | UInt32 | **Already deprecated** — use `--vcpus` CLI flag; remove from schema |
|
||||
| `jobmgr` | @13 | HashZArrayConfig | Job manager hash table sizing (nHashSlots, delta) |
|
||||
| `workermgr` | @14 | HashZArrayConfig | Worker manager hash table sizing |
|
||||
| `qmgr` | @15 | HashZArrayConfig | Queue manager hash table sizing |
|
||||
| `resmgr` | @16 | ResmgrConfig | Resource manager hash + instance table sizing |
|
||||
| `xstrings` | @17 | HashZArrayConfig | String interning table sizing |
|
||||
|
||||
**How consumed**:
|
||||
- `jobmgr`/`workermgr`/`qmgr`/`resmgr`/`xstrings` are read once at
|
||||
`cluster_init_phase2` to construct the internal key-value stores
|
||||
(`cluster.c:803-808`). After construction, the config values are not referenced again.
|
||||
- `heartbeatInterval`/`inactiveHeartbeatCount` are read at runtime by the
|
||||
heartbeat and liveness tasks (`cluster.c:1180, 1195`).
|
||||
- `fetchJobInterval` drives the scheduler polling loop.
|
||||
- `netioTimeout` is used by the iomgr layer for TCP timeouts.
|
||||
|
||||
**Why these can differ per controller**: A controller on a high-memory machine
|
||||
can be configured with larger hash tables. A controller on a high-latency WAN
|
||||
link can have a longer netioTimeout. These differences do not affect cluster
|
||||
correctness — they only affect local process behavior.
|
||||
|
||||
**`vcpus` removal**: Already deprecated. The `--vcpus` CLI flag
|
||||
(`main.c:1045`) is the canonical source. Remove field @7 in the next
|
||||
schema revision.
|
||||
|
||||
## Fields With Dual Usage (require care)
|
||||
|
||||
### `port` (@1)
|
||||
|
||||
**Decision: Remove. No fallback bind port.**
|
||||
|
||||
`cluster_config->port` currently serves as a fallback in `cluster_open_listener`
|
||||
(`cluster.c:953–975`) when `--peers` does not specify a port for the matching
|
||||
candidate. This fallback is removed: the `--peers` port is mandatory.
|
||||
|
||||
**Affected code (all changes mechanical):**
|
||||
|
||||
| File | Location | Change |
|
||||
|------|----------|--------|
|
||||
| `src/controller/cluster.c` | 953–975 | Remove the `if (my_port == 0) { cluster_config->port }` branch; return `EINVAL` immediately if no port found in `peers_cfg` |
|
||||
| `src/controller/myconf.c` | 74 | Remove `SET_DEFAULT_IF_UNSET(c->port, 0, 9990)` |
|
||||
| `src/controller/config_repl.c` | 55 | Remove `SET_DEFAULT_IF_UNSET(c->port, 0, 9990)` |
|
||||
| `src/include/clusterconfig.h` | 33 | Remove `int port` from `cluster_config_t` |
|
||||
| `src/schema/controller.capnp` | 682 | Remove `port @1: UInt32` from `ClusterConfig` |
|
||||
|
||||
Note: `"port"` in `testsuite/test.c:124,266` is the scheduler/worker
|
||||
connect-to-primary port (via `get_setting("port")`), not the listener bind
|
||||
port. It is unaffected.
|
||||
|
||||
## Implementation Approach
|
||||
|
||||
### Phase 1: Remove `port` and `candidates` from `ClusterConfig` (immediate)
|
||||
|
||||
No backward-compat transition needed for these two fields — the live code
|
||||
already ignores both (peer discovery uses `peers_cfg`; port fallback is being
|
||||
deleted). All changes are mechanical.
|
||||
|
||||
1. `src/schema/controller.capnp` — Remove `port @1` and `candidates @2` from
|
||||
`ClusterConfig`; mark or remove `MasterCandidate` struct if no longer
|
||||
referenced elsewhere
|
||||
2. `src/include/clusterconfig.h` — Remove `int port`, `int ncandidates`,
|
||||
`master_candidate_t** candidates` from `cluster_config_t`
|
||||
3. `src/controller/cluster.c` — Remove `if (my_port == 0)` fallback block in
|
||||
`cluster_open_listener` (lines 967–975); remove `ncandidates` + `candidate_N_*`
|
||||
KV loop in `cluster_build_config` (lines 587–606)
|
||||
4. `src/controller/myconf.c` and `src/controller/config_repl.c` — Remove
|
||||
`SET_DEFAULT_IF_UNSET(c->port, 0, 9990)` from `fill_default_cluster_config`
|
||||
5. Regenerate capnp bindings
|
||||
|
||||
### Phase 2: Schema split (wire-compatible)
|
||||
|
||||
1. Define `PeersConfig` and `ControllerConfig` structs in `controller.capnp`
|
||||
2. Add them as new fields in the top-level `ConfigData` union alongside
|
||||
the existing `clusterconfig @2`
|
||||
3. Regenerate capnp bindings
|
||||
4. **Keep the remaining `ClusterConfig` fields intact** during transition —
|
||||
new fields are additive
|
||||
|
||||
### Phase 3: Controller reads from new structs
|
||||
|
||||
1. `cluster_init_phase2`: read `ControllerConfig` for tunables
|
||||
2. `cluster_config_t` retains business config fields only
|
||||
3. Fall back to old `ClusterConfig` fields when new structs are absent
|
||||
(backward compatibility for existing `cluster.bin` files)
|
||||
|
||||
### Phase 4: Remove superseded tunables from `ClusterConfig`
|
||||
|
||||
1. Remove `heartbeatInterval`, `inactiveHeartbeatCount`, `fetchJobInterval`,
|
||||
`netioTimeout`, `vcpus` from `ClusterConfig`
|
||||
2. Remove `jobmgr`, `workermgr`, `qmgr`, `resmgr`, `xstrings` from
|
||||
`ClusterConfig`
|
||||
|
||||
### Phase 5 (future): Peer list replication
|
||||
|
||||
1. Define `OPCode_peersUpdate` / `OPCode_peersUpdateAck` for peer list
|
||||
replication with joint-consensus semantics
|
||||
2. Cluster membership changes become a dedicated operation with its own
|
||||
quorum protocol
|
||||
|
||||
## Effort and Risk
|
||||
|
||||
| Phase | Effort | Risk | Justification |
|
||||
|-------|--------|------|---------------|
|
||||
| 1 — Remove port + candidates | XS (½d) | Low | No transition needed; dead code removal + schema field removal |
|
||||
| 2 — Schema split | S (1d) | Low | Additive capnp change; old fields preserved |
|
||||
| 3 — Read from new structs | M (2-3d) | Medium | Touches cluster_init paths; backward compat needed |
|
||||
| 4 — Remove superseded tunables | S (1d) | Low | Mechanical removal after Phase 3 is validated |
|
||||
| 5 — Peer replication | L (1-2wk) | High | New protocol; joint consensus; extensive testing |
|
||||
|
||||
## References
|
||||
|
||||
- `src/schema/controller.capnp:680-700` — `ClusterConfig` definition
|
||||
- `src/schema/controller.capnp:253` — `ConfigData.clusterconfig @2`
|
||||
- `src/controller/main.c:178-213` — `--peers` CLI parsing into `peers_cfg`
|
||||
- `src/controller/cluster.c:854-880` — `ha_config_t` hardcoded defaults + priority derivation
|
||||
- `src/controller/cluster.c:945-977` — port derivation from candidates vs config blob
|
||||
- `src/controller/cluster.c:803-808` — internal manager construction
|
||||
- `src/controller/ha.h:134` — `ha_config_t.heartbeat_interval_ms` default via `ha_init`
|
||||
- `src/controller/startup_validation.c:133-163` — heartbeat config validation
|
||||
- `src/include/clusterconfig.h` — `xce_cluster_config_t` (separate peers struct)
|
||||
408
src/docs/archive/job_peek.md
Normal file
408
src/docs/archive/job_peek.md
Normal file
|
|
@ -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 会挂。
|
||||
Loading…
Add table
Add a link
Reference in a new issue