feat(repository): add inventory integration and cache management

cmd/cachemount/main_linux.go

@@ -0,0 +1,177 @@
+// cachemount presents existing DingCache files as a read-only model directory.
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"github.com/hanwen/go-fuse/v2/fuse"
+	"log"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"syscall"
+
+	"dingospeed/pkg/cachemount"
+	"dingospeed/pkg/hfprojection"
+)
+
+func main() {
+	if err := run(); err != nil {
+		log.Fatal(err)
+	}
+}
+func run() error {
+	treeRoot := flag.String("tree-root", "", "discover HF and hosted models under this local cache root")
+	treeConfig := flag.String("tree-config", "", "JSON array of selected cache_root/namespace/repo/revision objects")
+	reportPath := flag.String("report", "", "write pinned version/file mapping JSON outside the mount")
+	manifest := flag.String("manifest", "", "JSON object mapping model-relative paths to absolute OLAH cache paths")
+	root := flag.String("cache-root", "", "existing HF cache root (alternative to -manifest)")
+	repo := flag.String("repo", "", "HF repository, e.g. owner/model")
+	revision := flag.String("revision", "main", "cached revision to pin")
+	flag.Parse()
+	if *treeRoot != "" || *treeConfig != "" {
+		if flag.NArg() != 1 || (*treeRoot != "" && *treeConfig != "") || *manifest != "" || *root != "" {
+			return fmt.Errorf("use exactly one of -tree-root or -tree-config and one mountpoint")
+		}
+		var selections []cachemount.Selection
+		if *treeRoot != "" {
+			var err error
+			selections, err = cachemount.Discover(*treeRoot)
+			if err != nil {
+				return err
+			}
+		} else {
+			b, err := os.ReadFile(*treeConfig)
+			if err != nil {
+				return err
+			}
+			if err = json.Unmarshal(b, &selections); err != nil {
+				return err
+			}
+		}
+		if len(selections) == 0 {
+			return fmt.Errorf("no model versions found or selected")
+		}
+		mountpoint, err := filepath.Abs(flag.Arg(0))
+		if err != nil {
+			return err
+		}
+		if *reportPath != "" {
+			reportAbs, err := filepath.Abs(*reportPath)
+			if err != nil {
+				return err
+			}
+			rel, err := filepath.Rel(mountpoint, reportAbs)
+			if err != nil {
+				return err
+			}
+			if rel == "." || (!filepath.IsAbs(rel) && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))) {
+				return fmt.Errorf("report must be outside mountpoint")
+			}
+		}
+		snapshot, versions, err := cachemount.PrepareTree(selections, mountpoint)
+		if err != nil {
+			return err
+		}
+		defer snapshot.Close()
+		server, err := cachemount.Mount(snapshot, mountpoint)
+		if err != nil {
+			return err
+		}
+		b, err := json.MarshalIndent(versions, "", "  ")
+		if err != nil {
+			server.Unmount()
+			return err
+		}
+		if *reportPath != "" {
+			f, err := os.OpenFile(*reportPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
+			if err != nil {
+				server.Unmount()
+				return err
+			}
+			_, err = f.Write(b)
+			closeErr := f.Close()
+			if err != nil {
+				server.Unmount()
+				return err
+			}
+			if closeErr != nil {
+				server.Unmount()
+				return closeErr
+			}
+		}
+		for _, v := range versions {
+			log.Printf("%s %s (%d files) %s", v.Status, v.SourceFilePath, len(v.Files), v.Error)
+		}
+		return serve(server)
+	}
+	if flag.NArg() != 1 || (*manifest == "") == (*root == "") {
+		return fmt.Errorf("usage: cachemount (-manifest files.json | -cache-root ROOT -repo OWNER/MODEL [-revision REV]) MOUNTPOINT")
+	}
+	sources := map[string]string{}
+	expected := map[string]int64{}
+	if *manifest != "" {
+		b, err := os.ReadFile(*manifest)
+		if err != nil {
+			return err
+		}
+		if err = json.Unmarshal(b, &sources); err != nil {
+			return err
+		}
+	} else {
+		m, paths, err := (hfprojection.Reader{Root: *root}).MountSources("models", *repo, *revision)
+		if err != nil {
+			return err
+		}
+		sources = paths
+		for _, f := range m.Files {
+			expected[f.Path] = f.Size
+		}
+		log.Printf("pinning HF commit %s", m.Commit)
+	}
+	for name, p := range sources {
+		if !filepath.IsAbs(p) {
+			return fmt.Errorf("%s: backing path must be absolute", name)
+		}
+	}
+	snapshot, err := cachemount.Pin(sources)
+	if err != nil {
+		return err
+	}
+	defer snapshot.Close()
+	for name, size := range expected {
+		if size >= 0 && snapshot.Files[name].Size != size {
+			return fmt.Errorf("%s: cache size differs from repository manifest", name)
+		}
+	}
+	server, err := cachemount.Mount(snapshot, flag.Arg(0))
+	if err != nil {
+		return err
+	}
+	log.Printf("mounted %d files read-only at %s; no payload copied", len(sources), flag.Arg(0))
+	return serve(server)
+}
+
+func serve(server *fuse.Server) error {
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
+	defer signal.Stop(sig)
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		for {
+			select {
+			case <-sig:
+				if err := server.Unmount(); err != nil {
+					log.Printf("unmount failed (stop readers and run fusermount3 -u): %v", err)
+				}
+			case <-done:
+				return
+			}
+		}
+	}()
+	server.Wait()
+	return nil
+}

cmd/main.go

@@ -43,7 +43,7 @@ func init() {
 
 func newApp(s *server.HTTPServer, uploadServer *server.UploadServer, uploadCleanupServer *server.UploadCleanupServer, schedulerServer *server.SchedulerServer) *app.App {
 	app := app.New(app.ID(id), app.Name(Name), app.Version(Version),
-		app.Server(s, uploadServer, uploadCleanupServer, schedulerServer))
+		app.Server(s, uploadServer, uploadCleanupServer, schedulerServer, server.NewStorageProbeServer(config.SysConfig)))
 	return app
 }
 

config/config-backup.yaml

@@ -63,7 +63,7 @@ diskClean:
     enabled: true
     cacheSizeLimit: 41781441855488  #38T
     cacheCleanStrategy: "LRU"  #LRU,FIFO,LARGE_FIRST
-    collectTimePeriod: 1  #定期检测磁盘使用量时间周期，单位小时（H）
+    collectTimePeriod: 24  #定期检测磁盘使用量时间周期，单位小时（H）
     instanceID: mas  # 唯一标识符
 
 dynamicProxy:

config/config-offline.yaml

@@ -57,7 +57,7 @@ diskClean:
     enabled: true
     cacheSizeLimit: 41781441855488  #38T
     cacheCleanStrategy: "LRU"  #LRU,FIFO,LARGE_FIRST
-    collectTimePeriod: 1 #定期检测磁盘使用量时间周期，单位小时（H）
+    collectTimePeriod: 24 #定期检测磁盘使用量时间周期，单位小时（H）
     instanceID: mas  # 唯一标识符
 
 dynamicProxy:

config/config.yaml

@@ -79,7 +79,7 @@ diskClean:
     enabled: false             #是否启用磁盘清理
     cacheSizeLimit: 41781441855488  #38T 41781441855488
     cacheCleanStrategy: "LRU"  #LRU,FIFO,LARGE_FIRST
-    collectTimePeriod: 1  #定期检测磁盘使用量时间周期，单位小时（H）
+    collectTimePeriod: 24  #定期检测磁盘使用量时间周期，单位小时（H）
 
 dynamicProxy:
     enabled: false    #是否启用动态代理，当hfNetLoc配置的地址访问异常时，会自动切换到bpHfNetLoc。
@@ -114,7 +114,7 @@ proxyPool:
     #       url: http://10.220.70.213:8121
     #       weight: 1
 
-modelscope:
+modelscope: # 可省略整个配置段；以下为程序默认值，按需覆盖
     officialBaseURL: https://www.modelscope.cn  # ModelScope官方基础地址
     chunkSize: 8388608  # 8MB分块，16*1024*1024的数值结果
     maxRetry: 5  # 超时重试次数

config/prometheus/cache_task_alerts.yml

@@ -0,0 +1,22 @@
+groups:
+  - name: cache-task-lifecycle
+    rules:
+      - alert: DingoCacheTaskStoppingStalled
+        expr: dingospeed_cache_transition_oldest_seconds{state=~"pausing|canceling"} > 30
+        for: 1m
+        labels:
+          severity: warning
+        annotations:
+          summary: Cache task has not completed its stop barrier
+      - alert: DingoCacheActivityAfterStop
+        expr: increase(dingospeed_cache_activity_after_stop_total[5m]) > 0
+        labels:
+          severity: critical
+        annotations:
+          summary: Cache activity occurred after confirmed stop
+      - alert: DingoCacheSnapshotPersistenceFailure
+        expr: increase(dingospeed_cache_snapshot_write_failures_total[5m]) > 0
+        labels:
+          severity: warning
+        annotations:
+          summary: Cache lifecycle state could not be saved

config/prometheus/dependency_alerts.yml

@@ -0,0 +1,19 @@
+groups:
+  - name: dingospeed-dependency-observation
+    rules:
+      - alert: DingoSpeedDependencyDegraded
+        expr: dingospeed_dependency_state == 3
+        for: 30s
+        labels:
+          severity: warning
+        annotations:
+          summary: 'Dependency capability affected: {{ $labels.dependency }}'
+          description: 'Instance {{ $labels.instance }} has sustained failed operations affecting {{ $labels.impact }}. This does not establish whole-node or mount failure; do not automatically restart.'
+      - alert: DingoSpeedDependencyUnresolved
+        expr: dingospeed_dependency_unresolved == 1
+        for: 2m
+        labels:
+          severity: warning
+        annotations:
+          summary: 'Dependency failure lacks recovery evidence: {{ $labels.dependency }}'
+          description: 'Instance {{ $labels.instance }} has an unresolved observation affecting {{ $labels.impact }}. Repeated failures or absence of further traffic may explain this; inspect evidence before changing request policy.'

config/prometheus/storage_probe_alerts.yml

@@ -0,0 +1,19 @@
+groups:
+  - name: dingospeed-storage-probes
+    rules:
+      - alert: DingoSpeedStorageProbeBlocked
+        expr: dingospeed_storage_probe_blocked == 1
+        for: 30s
+        labels:
+          severity: warning
+        annotations:
+          summary: 'Storage probe still blocked: {{ $labels.operation }}'
+          description: 'Instance {{ $labels.instance }} has a timed-out dedicated probe still running. Its slot remains occupied; no replacement is launched. This does not establish whole-mount failure.'
+      - alert: DingoSpeedStorageProbeUnresolved
+        expr: dingospeed_storage_probe_unresolved == 1
+        for: 2m
+        labels:
+          severity: warning
+        annotations:
+          summary: 'Storage probe lacks recovery evidence: {{ $labels.operation }}'
+          description: 'Instance {{ $labels.instance }} has unresolved dedicated probe evidence. Compare business dependency metrics; probe success cannot clear business file errors.'

config/prometheus/tests/storage_probe.test.yml

@@ -0,0 +1,54 @@
+rule_files:
+  - ../storage_probe_alerts.yml
+evaluation_interval: 5s
+tests:
+  - name: blocked_read_late_return_and_verified_recovery
+    interval: 5s
+    input_series:
+      - series: 'dingospeed_storage_probe_blocked{job="dingospeed",instance="node-a",operation="read"}'
+        values: '1x29 0x7'
+      - series: 'dingospeed_storage_probe_unresolved{job="dingospeed",instance="node-a",operation="read"}'
+        values: '1x35 0'
+      - series: 'dingospeed_storage_probe_blocked{job="dingospeed",instance="node-a",operation="write"}'
+        values: '0x36'
+      - series: 'dingospeed_storage_probe_unresolved{job="dingospeed",instance="node-a",operation="write"}'
+        values: '0x36'
+      - series: 'dingospeed_storage_probe_blocked{job="dingospeed",instance="node-b",operation="read"}'
+        values: '0x36'
+      - series: 'dingospeed_storage_probe_unresolved{job="dingospeed",instance="node-b",operation="read"}'
+        values: '0x36'
+    alert_rule_test:
+      - eval_time: 25s
+        alertname: DingoSpeedStorageProbeBlocked
+        exp_alerts: []
+      - eval_time: 30s
+        alertname: DingoSpeedStorageProbeBlocked
+        exp_alerts:
+          - &blocked
+            exp_labels: {job: dingospeed, instance: node-a, operation: read, severity: warning}
+            exp_annotations:
+              summary: 'Storage probe still blocked: read'
+              description: 'Instance node-a has a timed-out dedicated probe still running. Its slot remains occupied; no replacement is launched. This does not establish whole-mount failure.'
+      - eval_time: 145s
+        alertname: DingoSpeedStorageProbeBlocked
+        exp_alerts: [*blocked]
+      - eval_time: 150s
+        alertname: DingoSpeedStorageProbeBlocked
+        exp_alerts: []
+      - eval_time: 115s
+        alertname: DingoSpeedStorageProbeUnresolved
+        exp_alerts: []
+      - eval_time: 120s
+        alertname: DingoSpeedStorageProbeUnresolved
+        exp_alerts:
+          - &unresolved
+            exp_labels: {job: dingospeed, instance: node-a, operation: read, severity: warning}
+            exp_annotations:
+              summary: 'Storage probe lacks recovery evidence: read'
+              description: 'Instance node-a has unresolved dedicated probe evidence. Compare business dependency metrics; probe success cannot clear business file errors.'
+      - eval_time: 150s
+        alertname: DingoSpeedStorageProbeUnresolved
+        exp_alerts: [*unresolved]
+      - eval_time: 180s
+        alertname: DingoSpeedStorageProbeUnresolved
+        exp_alerts: []

deploy/cachemount-sidecar/Dockerfile

@@ -0,0 +1,18 @@
+# Build context: this directory (source snapshot included).
+FROM golang:1.24 AS build
+WORKDIR /src
+COPY source/ ./
+RUN CGO_ENABLED=0 go build -trimpath -o /out/cachemount ./cmd/cachemount
+
+FROM debian:bookworm-slim
+RUN apt-get update && apt-get install -y --no-install-recommends python3 fuse3 ca-certificates \
+    && rm -rf /var/lib/apt/lists/* \
+    && groupadd -g 65532 nonroot \
+    && useradd -u 65532 -g 65532 -d /home/nonroot nonroot \
+    && mkdir -p /run/cachemount /home/nonroot \
+    && chown -R 65532:65532 /run/cachemount /home/nonroot
+COPY --from=build /out/cachemount /usr/local/bin/cachemount
+COPY source/sidecar.py /opt/cachemount/sidecar.py
+USER 65532:65532
+ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
+ENTRYPOINT ["python3", "/opt/cachemount/sidecar.py"]