diff --git a/README.ko.md b/README.ko.md index 37e692e..6e88b91 100644 --- a/README.ko.md +++ b/README.ko.md @@ -14,25 +14,36 @@ RNGD + ATOM 노드에서 Rebellions ATOM용 Llama-3.1-8B 빌드 묶음 하나가 들어 있기 때문입니다. FuriosaAI RNGD는 HF 체크포인트와 바이트가 같은 가중치를 세 벌 보관하고 있었습니다. 내용 기반 분할은 고정 블록으로는 전혀 보이지 않던 ATOM 변형 간 99.3% 공유를 찾아냅니다([측정](docs/ko/artifact-formats.md)). -Onggi(옹기)는 서로 다른 바이트를 한 번만 저장하고, pod가 요청하기 전에 필요한 노드로 옮겨 두며, 어떤 엔진에도 일반 -파일로 넘겨줍니다. +Onggi(옹기)는 서로 다른 바이트를 한 번만 저장하고, 복원한 결과물을 엔진에 일반 파일로 넘깁니다. 현재 CSI 플러그인은 +마운트 요청 시 복원하며, 배치 결정에 따른 자동 프리페치는 후속 구현입니다.

Onggi 구조: 사용자는 Kubernetes API로 조작하고, onggi-controller가 컴파일 팜·프리페치·GC·웹훅을 맡으며, 영속 계층은 교체 가능한 fs·s3·oci 백엔드, 캐시 계층은 노드 agent가 없는 청크를 받아 peer끼리 나누고 엔진 pod에 마운트한다. rm-core, rm-accel, rm-controller, llm-d, npu-train, lmd-top, Optimus Prime과 연결된다

-**흐름:** ① 컴파일 Job(또는 `onggi push`)이 결과물을 청크로 나눠 영속 계층에 없는 청크만 올리고 ② Kubernetes API의 Ref를 +**목표 구조:** ① 컴파일 Job(또는 `onggi push`)이 결과물을 청크로 나눠 영속 계층에 없는 청크만 올리고 ② Kubernetes API의 Ref를 커밋합니다. ③ pod가 결과물을 마운트하면 그 노드의 agent가 없는 청크를 peer나 영속 계층에서 받아 모든 해시를 검증한 뒤 엔진에 -일반 파일로 넘깁니다. ai-rm·llm-d의 배치 결정과 확장 신호가 들어오면 pod가 생기기 전에 미리 받기 시작합니다. +일반 파일로 넘깁니다. Kubernetes Ref API, 컴파일 컨트롤러, ai-rm·llm-d의 배치·확장 신호에 따른 자동 프리페치는 후속 구현입니다. ## 하는 일 | | | |---|---| | **한 번만 저장** | 내용 기반 청크, 결과물별 매니페스트, 가중치는 배열 방식별로 한 벌 | -| **미리 배치** | 노드별 캐시 계층, HRW 배치와 peer 간 전송; 배치·확장 신호로 프리페치 | +| **노드 데이터 재사용** | 노드 캐시와 해시 검증을 거치는 peer 전송; 배치·확장 신호에 따른 프리페치는 계획 단계 | | **파일로 전달** | CSI 볼륨: kubelet이 준비된 결과물을 마운트; vLLM, vllm-rbln, furiosa-llm, rngd-pd, Optimus Prime 수정 불필요 | -| **선언형 빌드** | `ArtifactSet` CRD가 컴파일 팜을 구동; 게시할 때 중복 제거 | +| **선언형 빌드(계획)** | `ArtifactSet` CRD와 컴파일 팜; 현재는 기존 빌드를 `onggi push`로 게시 | | **멀티 노드** | 영속 계층은 기존 이레이저 코딩 오브젝트 스토어; 캐시는 노드 장애를 전제로 설계 | +## 준비된 변형과 NPU 델타 로딩 + +Onggi는 미리 컴파일한 변형과 가중치를 기존 청크와 함께 일반 파일로 복원합니다. 실제로 배치·시퀀스 길이에 맞는 변형을 +선택하고, 필요하면 처리 중인 요청을 비운 뒤 장치 메모리를 갱신하는 주체는 [rngd-pd](https://github.com/ai-computing/rngd-pd) +같은 서빙 런타임입니다. 파일 델타만으로 NPU 메모리가 바뀌거나 두 그래프가 상주 가중치를 공유할 수 있는 것은 아닙니다. + +빌드마다 원본 체크포인트 식별자, 컴파일러·SDK, 장치, 가중치 배열 방식, 지원 배치·시퀀스 길이를 기록해야 합니다. +배치나 길이가 바뀌어도 호환되는 사전 컴파일 그래프가 필요합니다. RNGD 외부 가중치와 ATOM 장치 입력 가중치는 런타임 지원이 +다르며, 가중치를 내장한 ATOM 그래프에는 해당 컴파일 결과물이 필요합니다. 자동 호환성 선택·예열·배치 연동은 후속 작업입니다. +[연계 설계](docs/ko/integration.md)를 참고하세요. + ## 빠른 시작 ```bash @@ -118,8 +129,9 @@ prefill과 decode는 같은 가중치를 다른 순서로 담고 있습니다. ## ai-rm 구성요소 -Onggi는 이종 AI 자원관리 스택 [ai-rm](https://github.com/ai-computing/ai-rm)의 구성요소이며 -[rm-core](https://github.com/ai-computing/rm-core) 계약으로 연결됩니다. 운영 콘솔은 [lmd-top](https://github.com/ai-computing/lmd-top)입니다. +Onggi는 이종 AI 자원관리 스택 [ai-rm](https://github.com/ai-computing/ai-rm)의 구성요소입니다. +[rm-core](https://github.com/ai-computing/rm-core) 아티팩트 계약과 [lmd-top](https://github.com/ai-computing/lmd-top) +운영 콘솔 연동은 계획 단계이며, 현재 인터페이스는 CLI와 CSI 볼륨입니다. ## 기여 diff --git a/README.md b/README.md index abbfb92..b0cc0d6 100644 --- a/README.md +++ b/README.md @@ -15,26 +15,39 @@ weights, and FuriosaAI RNGD keeps three copies of weights that are byte-identica chunking finds 99.3 % shared bytes between ATOM variants that fixed-size blocks miss entirely ([measurements](docs/artifact-formats.md)). -Onggi (옹기, the Korean jar that stores and ferments food) keeps each distinct byte once, moves artifacts to the nodes -that need them before a pod asks, and hands them to any engine as plain files. +Onggi (옹기, the Korean jar that stores and ferments food) keeps each distinct byte once and hands restored artifacts +to engines as plain files. The CSI plugin restores them on demand; placement-driven prefetch remains planned.

Onggi architecture: users act through the Kubernetes API; onggi-controller runs the compile farm, prefetch, GC and a webhook; a durable tier behind pluggable fs, s3 and oci backends; a cache tier of node agents that fetch missing chunks, share them peer to peer and mount artifacts into engine pods; integration points with rm-core, rm-accel, rm-controller, llm-d, npu-train, lmd-top and Optimus Prime

-**How it fits together:** ① a compile Job (or `onggi push`) chunks an artifact and uploads only the chunks the durable +**Target architecture:** ① a compile Job (or `onggi push`) chunks an artifact and uploads only the chunks the durable tier does not have, then ② commits a Ref in the Kubernetes API. ③ When a pod mounts the artifact, the node's agent fetches the missing chunks from peers or the durable tier, verifies every hash and hands the engine plain files. -Placement decisions and scale-up signals from ai-rm and llm-d start the fetch before the pod exists. +The Kubernetes Ref API, compile controller and fetches triggered by ai-rm/llm-d placement or scale-up signals are future work. ## What it does | | | |---|---| | **Store once** | content-defined chunks, manifests per artifact, weights stored once per weight layout | -| **Place ahead** | per-node cache tier with rendezvous (HRW) placement and peer-to-peer transfer; prefetch from placement and autoscaling signals | +| **Reuse node data** | node cache and hash-checked peer transfer; placement/autoscaling prefetch is planned | | **Deliver as files** | CSI volume: kubelet mounts a ready artifact; vLLM, vllm-rbln, furiosa-llm, rngd-pd and Optimus Prime need no changes | -| **Build declaratively** | `ArtifactSet` CRDs drive a compile farm; results are deduplicated on publish | +| **Build declaratively (planned)** | `ArtifactSet` CRDs and a compile farm; today, publish existing builds with `onggi push` | | **Stay multi-node** | durable tier on an existing erasure-coded object store; caches survive node loss by design | +## Prepared variants and NPU delta loading + +Onggi restores precompiled variants and weights, including their unchanged chunks, as ordinary files. A serving +runtime such as [rngd-pd](https://github.com/ai-computing/rngd-pd) selects a compatible prepared batch/sequence +variant, drains active requests when required, and performs any supported device-memory update. Onggi's file +delta does not itself update NPU memory or establish that two compiled graphs can share resident weights. + +Record the source checkpoint identity, compiler/SDK, device, weight layout and supported batch/sequence shapes +with each build. Changing batch or sequence length still requires a compatible precompiled graph. RNGD external +weights and ATOM device-input weights have different runtime capabilities; ATOM graphs with embedded weights +require their corresponding compiled artifacts. Automated compatibility selection, prewarm and placement hooks +remain integration work; see [integration](docs/integration.md). + ## Quick start ```bash @@ -122,9 +135,9 @@ holds the store's index (compact arrays) for dedup and peaks at 1.9 GiB by defau ## Part of ai-rm -Onggi is a component of [ai-rm](https://github.com/ai-computing/ai-rm), the heterogeneous AI resource management stack, -and connects through [rm-core](https://github.com/ai-computing/rm-core) contracts. [lmd-top](https://github.com/ai-computing/lmd-top) -is its operator console. +Onggi is a component of [ai-rm](https://github.com/ai-computing/ai-rm), the heterogeneous AI resource management stack. +Its [rm-core](https://github.com/ai-computing/rm-core) artifact contract and +[lmd-top](https://github.com/ai-computing/lmd-top) operator-console integration are planned; the current interfaces are the CLI and CSI volumes. ## Contributing diff --git a/ai-rm.yaml b/ai-rm.yaml index f47dbcd..f2b2e71 100644 --- a/ai-rm.yaml +++ b/ai-rm.yaml @@ -1,8 +1,8 @@ name: onggi protocol: "0.1" layer: l2-memory -kind: tool # P0: a CLI; the controller, node agent and CSI driver follow in P2-P3 (docs/design.md) -summary: Content-addressed store for model weights and accelerator-compiled artifacts (P0 dedup-report) +kind: tool # CLI plus CSI node plugin; controller/compile farm remain planned +summary: Content-addressed artifact store with node cache, peers, delta apply and read-only CSI volumes (P2) maturity: experimental hardware: [RNGD, ATOM, A100] # reads their artifacts; runs on any Linux/Windows host without an accelerator install: @@ -11,7 +11,7 @@ test: - go vet ./... - go test ./... deploy: - target: none # P0 runs where the files are; cluster components come with P2 + target: none # CSI examples require site image/store settings; no automatic deployment manifests: null image: null config: diff --git a/cmd/onggi/csi.go b/cmd/onggi/csi.go index b95976d..3c86d4e 100644 --- a/cmd/onggi/csi.go +++ b/cmd/onggi/csi.go @@ -32,6 +32,7 @@ func csiNode(args []string, stderr io.Writer) int { dataDir := fs.String("data-dir", "/var/lib/onggi", "node directory for materialized artifacts and leases") cacheDir := fs.String("cache", "", "node chunk cache (default /cache; none: no cache)") workers := fs.Int("workers", 16, "ranges read at once per pull") + pulls := fs.Int("pulls", 1, "distinct artifacts materialized at once per node") evictAfter := fs.Duration("evict-after", 24*time.Hour, "evict materialized artifacts no volume used for this long (0: never)") gcEvery := fs.Duration("gc-every", time.Hour, "how often to evict and collect the node cache (0: never)") keepUsed := fs.Duration("keep-used", 7*24*time.Hour, "the node cache keeps what artifacts used within this period need") @@ -45,6 +46,10 @@ func csiNode(args []string, stderr io.Writer) int { fmt.Fprintln(stderr, "usage: onggi csi --store S [--endpoint unix:///csi/csi.sock] [--node-id N] [--data-dir DIR] [--cache DIR] [--peer-listen :7070] [--peers-dns H:P]") return 2 } + if *pulls < 1 { + fmt.Fprintln(stderr, "--pulls must be at least 1") + return 2 + } logger := log.New(stderr, "", log.LstdFlags|log.Lmicroseconds) if *nodeID == "" { logger.Print("no node id: pass --node-id or set NODE_NAME") @@ -94,7 +99,7 @@ func csiNode(args []string, stderr io.Writer) int { return append(append([]store.ChunkSource(nil), static...), d.Peers(ctx)...) } } - m := &csi.Materializer{Root: *dataDir, Pull: csi.StorePullerPeers(s, openCache, peerList, tmp, *workers), Logf: logger.Printf} + m := &csi.Materializer{Root: *dataDir, Pull: csi.StorePullerPeers(s, openCache, peerList, tmp, *workers), Logf: logger.Printf, MaxPulls: *pulls} if err := m.Init(); err != nil { logger.Print("data dir: ", err) return 1 diff --git a/deploy/csi/README.md b/deploy/csi/README.md index e90fa9c..0f88cce 100644 --- a/deploy/csi/README.md +++ b/deploy/csi/README.md @@ -34,9 +34,18 @@ On `NodePublishVolume` the plugin: The pod stays in `ContainerCreating` until this returns. kubelet gives up on a call after about two minutes and retries it. The pull keeps running in the meantime, and the retry joins it. +Distinct artifacts are materialized one at a time by default (`--pulls 1`). Increase `--pulls` only with +enough node memory: each pull has its own planner and up to `--workers` concurrent range reads. This bounds +the number of large working sets; `GOMEMLIMIT` alone is a soft GC target, not a process memory cap. + `NodeUnpublishVolume` unmounts and drops the lease. A ref moved to a new manifest gets a new directory; pods that already run keep the old one. +A failed volume-record write rolls back its newly created lease. Reusing a volume ID with a different +manifest or target is rejected; unpublish it first. A process crash between the lease and volume-record +writes can still leave a conservative extra lease, so restart does not automatically remove such leases +without checking whether their mounts are in use. + ## Eviction (#33) Every `--gc-every` (default 1 h) the plugin: diff --git a/docs/validation-2026-10-23.md b/docs/validation-2026-10-23.md new file mode 100644 index 0000000..2bbde32 --- /dev/null +++ b/docs/validation-2026-10-23.md @@ -0,0 +1,72 @@ +# CSI safety validation — 2026-10-23 + +Base: `b63d5022ae1586916d92fae37dcf9cdf51e58f77`. Linux: Ubuntu 24.04 under WSL, amd64. + +## Regression evidence + +- Failed volume-record writes previously left one lease behind; the regression failed with `failed acquisition left 1 leases`. +- Rebinding a volume ID to a different manifest/target was accepted; the regression now checks rejection and preservation of the original volume. +- Three distinct manifests previously entered their pullers together. The regression now checks default/explicit concurrency limits of one and two, while the existing suite covers single-flight retries and callers that time out. + +## Verification + +Go **1.24.13**, matching the module's minimum version, and VersityGW **v1.8.0** bound to loopback with temporary test credentials/data: + +```sh +export ONGGI_TEST_S3='s3://onggitest?endpoint=127.0.0.1:7070&insecure=1' +# Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY to the local server's test credentials. +go vet ./... +go test -race -count=1 -json ./... +go build -o onggi ./cmd/onggi +``` + +Result: **113 test cases passed**, including subtests; no test case skipped or failed. Three packages contain no tests. +The suite also passed on Go 1.27.3 without the S3 server, with nine S3 subtests skipped in that earlier run. +`gofmt` and `git diff --check` passed. + +An additional CLI round trip pushed two 4 MiB synthetic artifacts to S3, repeated a push, restored the base, +then pulled the changed artifact using `--base`/`--base-dir`. Both restores were byte-identical. The delta +pull reused **4,119,698 bytes** and fetched **70,733 bytes**. These are integrity checks, not NPU performance measurements. + +## Real Linux mount lifecycle + +Source `ca0226464d209f3680ec0d1efcf7a637300cbffc` plus +[`TestLinuxMountLifecycle`](../pkg/csi/mount_integration_linux_test.go) was checked in a disposable +Kubernetes pod on `cpu-2`, Linux `7.0.0-38-generic` amd64, using `golang:1.24` (Go **1.24.13**). +The pod was privileged, with an `emptyDir` workspace and no service account token, host paths, +host PID/network namespace, or mount propagation. No CSI driver registration or DaemonSet was installed. + +Run this test only in a disposable mount namespace with `CAP_SYS_ADMIN`: + +```sh +ONGGI_TEST_MOUNTS=1 go test -race ./pkg/csi -run '^TestLinuxMountLifecycle$' -count=1 -v +``` + +Without `ONGGI_TEST_MOUNTS=1`, the Linux test skips. With it, an unavailable mount capability fails +the test. The test uses the real filesystem store, materializer, `NodePublishVolume`, +`NodeUnpublishVolume`, eviction, and Linux `NewMounter`; it does not use the fake mounter. + +The real mount test passed: concurrent publishes pulled the artifact once, held two leases, and +exposed byte-identical files sharing an inode. Both targets and the source self-bind had +`readonly,nosuid,nodev` flags, and writes failed with `EROFS`. Publish and unpublish retries succeeded. +A remaining lease prevented eviction; after the final unpublish, eviction removed the source +self-bind, artifact, and lease directory. Cleanup also checks `/proc/self/mountinfo` for test mounts +before temporary-directory removal. + +The full CSI package also passed with `ONGGI_TEST_MOUNTS=1 go test -race ./pkg/csi -count=1 -v`: +**22 top-level tests passed**, none skipped, in 10.794 s. A negative control using +`setpriv --bounding-set=-sys_admin env ONGGI_TEST_MOUNTS=1 go test ./pkg/csi -run '^TestLinuxMountLifecycle$' -count=1 -v` +failed with `operation not permitted` and exit 1, confirming that an explicitly enabled test cannot +silently skip without mount privileges. `gofmt` and `git diff --check` passed. The final container +mount table contained **zero** integration-test mounts. + +After checking resource UIDs, the test pod (`857e8cb0-beee-4be2-a30a-af5308b7cf62`) and namespace +`onggi-verify-20261023-8c6a12` (`79a84246-b6fb-4a9c-8c30-2365c48e15f2`) were deleted; the namespace +was confirmed absent. The cluster's CSI registrations remained NFS and SMB only. + +## Limits + +No live CSI driver rollout, kubelet registration, host mount propagation, actual model, or NPU-memory +update was tested. The isolated Linux test covers the node service and real mounts within one container. +The concurrency limit bounds active materializations, not absolute RSS. A process crash between lease and +volume-record writes can conservatively retain an extra lease; automatic crash reconciliation remains unimplemented. diff --git a/pkg/csi/acquire_failure_test.go b/pkg/csi/acquire_failure_test.go new file mode 100644 index 0000000..b48ec29 --- /dev/null +++ b/pkg/csi/acquire_failure_test.go @@ -0,0 +1,83 @@ +package csi + +import ( + "os" + "path/filepath" + "testing" + + "github.com/ai-computing/onggi/pkg/digest" +) + +func TestAcquireRecordFailureDoesNotLeakLease(t *testing.T) { + m := &Materializer{Root: t.TempDir()} + if err := m.Init(); err != nil { + t.Fatal(err) + } + d := digest.Of([]byte("artifact")) + // Fail the volume-record write after the artifact lease has been written. + blocked := filepath.Join(m.Root, "volumes", "volume.json.tmp") + if err := os.Mkdir(blocked, 0o755); err != nil { + t.Fatal(err) + } + if err := m.Acquire("volume", d, "/target"); err == nil { + t.Fatal("want record failure") + } + if n, err := m.Users(d); err != nil || n != 0 { + t.Fatalf("failed acquisition left %d leases: %v", n, err) + } + if err := os.Remove(blocked); err != nil { + t.Fatal(err) + } + if err := m.Acquire("volume", d, "/target"); err != nil { + t.Fatal(err) + } + if _, n, ok, err := m.Release("volume"); err != nil || !ok || n != 0 { + t.Fatalf("retry/release: %d %v %v", n, ok, err) + } +} + +func TestAcquireRejectsChangingPublishedIdentity(t *testing.T) { + m := &Materializer{Root: t.TempDir()} + if err := m.Init(); err != nil { + t.Fatal(err) + } + a, b := digest.Of([]byte("a")), digest.Of([]byte("b")) + if err := m.Acquire("volume", a, "/target"); err != nil { + t.Fatal(err) + } + for _, v := range []struct { + d digest.Digest + target string + }{{b, "/target"}, {a, "/other"}} { + if err := m.Acquire("volume", v.d, v.target); err == nil { + t.Fatal("changed volume identity accepted") + } + } + if n, _ := m.Users(a); n != 1 { + t.Fatalf("original lease lost: %d", n) + } + if n, _ := m.Users(b); n != 0 { + t.Fatalf("foreign lease created: %d", n) + } + if got, _, ok, err := m.Release("volume"); err != nil || !ok || got != a { + t.Fatalf("original volume changed: %v %v %v", got, ok, err) + } +} + +func TestAcquireRetryFailurePreservesExistingLease(t *testing.T) { + m := &Materializer{Root: t.TempDir()} + if err := m.Init(); err != nil { + t.Fatal(err) + } + d := digest.Of([]byte("artifact")) + if err := m.Acquire("volume", d, "/target"); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(filepath.Join(m.Root, "volumes", "volume.json.tmp"), 0o755); err != nil { + t.Fatal(err) + } + _ = m.Acquire("volume", d, "/target") + if n, err := m.Users(d); err != nil || n != 1 { + t.Fatalf("existing lease removed: %d %v", n, err) + } +} diff --git a/pkg/csi/materializer.go b/pkg/csi/materializer.go index 3717fda..59b86e9 100644 --- a/pkg/csi/materializer.go +++ b/pkg/csi/materializer.go @@ -81,9 +81,13 @@ type Materializer struct { RebaseIdle time.Duration // Unmount removes the read-only self-bind publishing put on an artifact directory (nil: none). Unmount func(dir string) error + // MaxPulls bounds distinct artifacts materialized concurrently (0: one). + // Set before the first Ensure; retries for one digest still share one pull. + MaxPulls int mu sync.Mutex inflight map[digest.Digest]*flight + slots chan struct{} } type flight struct { @@ -111,6 +115,9 @@ func (m *Materializer) Dir(d digest.Digest) string { // Init creates the layout and removes pulls that a previous process left unfinished. Only one // materializer may use Root at a time. func (m *Materializer) Init() error { + if m.MaxPulls < 0 { + return fmt.Errorf("MaxPulls must not be negative") + } for _, d := range []string{"artifacts", "tmp", "volumes", "leases"} { if err := os.MkdirAll(filepath.Join(m.Root, d), 0o755); err != nil { return err @@ -144,13 +151,16 @@ func (m *Materializer) Ensure(ctx context.Context, d digest.Digest) (string, err } if m.inflight == nil { m.inflight = map[digest.Digest]*flight{} + m.slots = make(chan struct{}, max(1, m.MaxPulls)) } f, ok := m.inflight[d] if !ok { f = &flight{done: make(chan struct{})} m.inflight[d] = f go func() { + m.slots <- struct{}{} f.err = m.materialize(d) + <-m.slots m.mu.Lock() delete(m.inflight, d) m.mu.Unlock() @@ -239,15 +249,41 @@ func (m *Materializer) Acquire(id string, d digest.Digest, target string) error m.mu.Lock() defer m.mu.Unlock() name := volumeName(id) + vf := filepath.Join(m.Root, "volumes", name+".json") + if data, err := os.ReadFile(vf); err == nil { + var v Volume + if err := json.Unmarshal(data, &v); err != nil { + return fmt.Errorf("volume %s: %w", id, err) + } + if v.Manifest != d.String() || v.Target != target { + return fmt.Errorf("volume %s already uses %s at %s", id, v.Manifest, v.Target) + } + } else if !errors.Is(err, os.ErrNotExist) { + return err + } ldir := filepath.Join(m.Root, "leases", d.Hex()) if err := os.MkdirAll(ldir, 0o755); err != nil { return err } - if err := writeAtomic(filepath.Join(ldir, name), []byte(target+"\n")); err != nil { + lease := filepath.Join(ldir, name) + _, statErr := os.Stat(lease) + if statErr != nil && !errors.Is(statErr, os.ErrNotExist) { + return statErr + } + if err := writeAtomic(lease, []byte(target+"\n")); err != nil { return err } data, _ := json.Marshal(Volume{Manifest: d.String(), Target: target}) - return writeAtomic(filepath.Join(m.Root, "volumes", name+".json"), data) + // Lease first so a process crash cannot leave a published artifact unprotected. + if err := writeAtomic(vf, data); err != nil { + if errors.Is(statErr, os.ErrNotExist) { + if removeErr := os.Remove(lease); removeErr != nil { + return errors.Join(err, fmt.Errorf("rollback lease: %w", removeErr)) + } + } + return err + } + return nil } // Release drops volume id's lease and returns the artifact it used and how many volumes still @@ -308,6 +344,7 @@ func writeAtomic(path string, data []byte) error { if err := os.WriteFile(tmp, data, 0o644); err != nil { return err } + defer os.Remove(tmp) return os.Rename(tmp, path) } diff --git a/pkg/csi/mount_integration_linux_test.go b/pkg/csi/mount_integration_linux_test.go new file mode 100644 index 0000000..611946f --- /dev/null +++ b/pkg/csi/mount_integration_linux_test.go @@ -0,0 +1,151 @@ +//go:build linux + +package csi + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + csipb "github.com/container-storage-interface/spec/lib/go/csi" + "golang.org/x/sys/unix" +) + +// Run only in a disposable mount namespace with CAP_SYS_ADMIN, never against host paths. +func TestLinuxMountLifecycle(t *testing.T) { + if os.Getenv("ONGGI_TEST_MOUNTS") != "1" { + t.Skip("set ONGGI_TEST_MOUNTS=1 in a disposable mount namespace with CAP_SYS_ADMIN") + } + n, f, pulls, _ := newNode(t) + n.Mounter = NewMounter() + dir := n.M.Dir(f.a) + pods := t.TempDir() + targets := []string{filepath.Join(pods, "pod-1", "mount"), filepath.Join(pods, "pod-2", "mount")} + paths := []string{targets[0], targets[1], dir} + t.Cleanup(func() { + // Targets must leave before their shared read-only source and TempDir cleanup. + for _, p := range paths { + mounted, err := n.Mounter.IsMounted(p) + if err != nil { + t.Errorf("cleanup check %s: %v", p, err) + continue + } + if mounted { + if err := n.Mounter.Unmount(p); err != nil { + t.Errorf("cleanup unmount %s: %v", p, err) + } + } + } + file, err := os.Open("/proc/self/mountinfo") + if err != nil { + t.Error(err) + return + } + defer file.Close() + points, err := mountPoints(file) + if err != nil { + t.Error(err) + } + for _, p := range points { + if p == n.M.Root || strings.HasPrefix(p, n.M.Root+"/") || p == pods || strings.HasPrefix(p, pods+"/") { + t.Errorf("test mount remains: %s", p) + } + } + }) + ctx := context.Background() + ids := []string{"linux-volume-1", "linux-volume-2"} + var wg sync.WaitGroup + errs := make([]error, len(targets)) + for i, target := range targets { + wg.Add(1) + go func(i int, target string) { + defer wg.Done() + _, errs[i] = n.NodePublishVolume(ctx, publishReq(ids[i], target, "a")) + }(i, target) + } + wg.Wait() + for _, err := range errs { + if err != nil { + t.Fatalf("real Linux publish (requires CAP_SYS_ADMIN): %v", err) + } + } + if len(pulls.pulls) != 1 { + t.Fatalf("%d store pulls, want 1", len(pulls.pulls)) + } + if users, err := n.M.Users(f.a); err != nil || users != 2 { + t.Fatalf("shared leases: %d, %v", users, err) + } + if _, err := n.NodePublishVolume(ctx, publishReq(ids[0], targets[0], "a")); err != nil { + t.Fatalf("publish retry: %v", err) + } + for _, p := range paths { + mounted, err := n.Mounter.IsMounted(p) + if err != nil || !mounted { + t.Fatalf("%s mounted: %v, %v", p, mounted, err) + } + var st unix.Statfs_t + if err := unix.Statfs(p, &st); err != nil { + t.Fatal(err) + } + want := int64(unix.ST_RDONLY | unix.ST_NOSUID | unix.ST_NODEV) + if int64(st.Flags)&want != want { + t.Fatalf("%s flags %#x lack readonly/nosuid/nodev", p, st.Flags) + } + if err := os.WriteFile(filepath.Join(p, "must-not-write"), []byte("x"), 0o600); !errors.Is(err, unix.EROFS) { + t.Fatalf("write %s: %v, want EROFS", p, err) + } + sameTree(t, f.srcA, p) + } + a, err := os.Stat(filepath.Join(targets[0], "model.bin")) + if err != nil { + t.Fatal(err) + } + b, err := os.Stat(filepath.Join(targets[1], "model.bin")) + if err != nil || !os.SameFile(a, b) { + t.Fatalf("shared inode: %v", err) + } + t.Log("one real store pull; two leases; identical shared inode; source and targets readonly/nosuid/nodev; writes EROFS") + for i, target := range targets { + for retry := 0; retry < 2; retry++ { + if _, err := n.NodeUnpublishVolume(ctx, &csipb.NodeUnpublishVolumeRequest{VolumeId: ids[i], TargetPath: target}); err != nil { + t.Fatalf("unpublish %d retry %d: %v", i, retry, err) + } + } + if _, err := os.Stat(target); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("unpublished target remains: %v", err) + } + if users, err := n.M.Users(f.a); err != nil || users != 1-i { + t.Fatalf("remaining leases: %d, %v", users, err) + } + if i == 0 { + sameTree(t, f.srcA, targets[1]) + if gone, err := n.M.Evict(time.Now().Add(2*time.Hour), time.Hour, n.Mounter.Unmount); err != nil || len(gone) != 0 { + t.Fatalf("evicted leased artifact: %v, %v", gone, err) + } + } + } + if records, err := os.ReadDir(filepath.Join(n.M.Root, "volumes")); err != nil || len(records) != 0 { + t.Fatalf("unpublished volume records remain: %v, %v", records, err) + } + if mounted, err := n.Mounter.IsMounted(dir); err != nil || !mounted { + t.Fatalf("idle source self-bind: %v, %v", mounted, err) + } + gone, err := n.M.Evict(time.Now().Add(2*time.Hour), time.Hour, n.Mounter.Unmount) + if err != nil || len(gone) != 1 || gone[0] != f.a { + t.Fatalf("eviction: %v, %v", gone, err) + } + for _, p := range []string{dir, filepath.Join(n.M.Root, "leases", f.a.Hex())} { + if _, err := os.Stat(p); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("evicted path remains %s: %v", p, err) + } + } + if mounted, err := n.Mounter.IsMounted(dir); err != nil || mounted { + t.Fatalf("evicted self-bind remains: %v, %v", mounted, err) + } + t.Log("publish/unpublish retries succeeded; live lease blocked eviction; final eviction removed self-bind, artifact and lease directory") +} diff --git a/pkg/csi/pull_limit_test.go b/pkg/csi/pull_limit_test.go new file mode 100644 index 0000000..1c50c5e --- /dev/null +++ b/pkg/csi/pull_limit_test.go @@ -0,0 +1,55 @@ +package csi + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/ai-computing/onggi/pkg/digest" + "github.com/ai-computing/onggi/pkg/store" +) + +func TestEnsureBoundsDistinctPulls(t *testing.T) { + for _, limit := range []int{0, 1, 2} { + t.Run(fmt.Sprint(limit), func(t *testing.T) { checkPullLimit(t, limit) }) + } +} + +func checkPullLimit(t *testing.T, limit int) { + t.Helper() + entered := make(chan digest.Digest, 3) + gate := make(chan struct{}) + m := &Materializer{Root: t.TempDir(), MaxPulls: limit, Pull: func(_ context.Context, d digest.Digest, _ string, _ *Offer) (store.PullStats, error) { + entered <- d + <-gate + return store.PullStats{}, nil + }} + if err := m.Init(); err != nil { + t.Fatal(err) + } + var wg sync.WaitGroup + defer func() { close(gate); wg.Wait() }() + for i := byte(0); i < 3; i++ { + wg.Add(1) + go func(i byte) { + defer wg.Done() + if _, err := m.Ensure(context.Background(), digest.Of([]byte{i})); err != nil { + t.Error(err) + } + }(i) + } + for i := 0; i < max(1, limit); i++ { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("configured number of pulls did not start") + } + } + select { + case <-entered: + t.Fatal("multiple distinct pulls run without a concurrency budget") + case <-time.After(100 * time.Millisecond): + } +}