package main // 挂载管理(docs/design/21-knowledge-sources.md §4.4、§4.5): // 平台用 sync 下发完整的登记列表;挂载进程挂上缺少的、卸载多余的、更新有变化的。 // 挂载表变化时(设备插拔)立即重新计算状态,不需要定时检查。 import ( "encoding/json" "fmt" "log" "os" "os/exec" "path/filepath" "regexp" "sort" "sync" "time" "github.com/hanwen/go-fuse/v2/fuse" "golang.org/x/sys/unix" ) var idPattern = regexp.MustCompile(`^kb_[0-9a-z]{1,64}$`) type mountSpec struct { ID string `json:"id"` Kind string `json:"kind"` Local *localSource `json:"local,omitempty"` } type mountStatus struct { ID string `json:"id"` State string `json:"state"` Message *string `json:"message"` Since string `json:"since"` // 当前的实际路径(本机目录) RealPath string `json:"realPath,omitempty"` } type mount struct { spec mountSpec point string src *source server *fuse.Server state string message string since time.Time realPath string } type manager struct { mu sync.Mutex sys system roots []string mntDir string stateFile string mounts map[string]*mount lastMiss time.Time // mountFn 便于测试替换 mountFn func(point, name string, src *source) (*fuse.Server, error) } func newManager(sys system, roots []string, mntDir, stateFile string) *manager { return &manager{sys: sys, roots: roots, mntDir: mntDir, stateFile: stateFile, mounts: map[string]*mount{}, mountFn: mountFS} } func (m *manager) setState(mt *mount, state, msg string) { if mt.state != state || mt.message != msg { if mt.state != "" { log.Printf("%s:%s → %s %s", mt.spec.ID, mt.state, state, msg) } mt.state, mt.message, mt.since = state, msg, time.Now() } } // resolve 按登记找到源目录并检查,返回真实路径与设备号、inode 号 func (m *manager) resolve(spec mountSpec) (string, uint64, uint64, *checkError) { if spec.Kind != "local" || spec.Local == nil { return "", 0, 0, &checkError{"mount_failed", "暂不支持这种资料库类型"} } p, cerr := locate(m.sys, *spec.Local) if cerr != nil { return "", 0, 0, cerr } real, cerr := checkSource(p, m.roots) if cerr != nil { return "", 0, 0, cerr } how := unix.OpenHow{Flags: unix.O_PATH | unix.O_DIRECTORY | unix.O_CLOEXEC, Resolve: unix.RESOLVE_NO_SYMLINKS | unix.RESOLVE_NO_MAGICLINKS} fd, err := unix.Openat2(unix.AT_FDCWD, real, &how) if err != nil { return "", 0, 0, &checkError{"path_missing", "无法打开目录:" + real} } defer unix.Close(fd) var st unix.Stat_t if err := unix.Fstat(fd, &st); err != nil { return "", 0, 0, &checkError{"path_missing", "无法读取目录:" + real} } return real, st.Dev, st.Ino, nil } // refreshLocked 重新计算一个挂载的状态 func (m *manager) refreshLocked(mt *mount) { real, dev, ino, cerr := m.resolve(mt.spec) if cerr != nil { mt.src.close() mt.realPath = "" m.setState(mt, cerr.State, cerr.Message) return } mt.src.set(real, dev, ino) mt.realPath = real if mt.server == nil { m.setState(mt, "mount_failed", "挂载失败") return } m.setState(mt, "ok", "") } func (m *manager) refreshAll() { m.mu.Lock() defer m.mu.Unlock() for _, mt := range m.mounts { m.refreshLocked(mt) } } // onMiss 资料库不可用时有人访问:可能是设备刚接上(如自动挂载),限速重新检查 func (m *manager) onMiss() { m.mu.Lock() if time.Since(m.lastMiss) < 2*time.Second { m.mu.Unlock() return } m.lastMiss = time.Now() m.mu.Unlock() go m.refreshAll() } func (m *manager) mountLocked(mt *mount) { if mt.server != nil { return } if err := os.MkdirAll(mt.point, 0o755); err != nil { log.Printf("%s:创建挂载点失败:%v", mt.spec.ID, err) return } srv, err := m.mountFn(mt.point, "kb:"+mt.spec.ID, mt.src) if err != nil { log.Printf("%s:挂载失败:%v", mt.spec.ID, err) return } mt.server = srv } func (m *manager) unmountLocked(mt *mount) { if mt.server != nil { if err := mt.server.Unmount(); err != nil { // 有文件正被打开:延迟卸载 _ = exec.Command("fusermount3", "-uz", mt.point).Run() } mt.server = nil } mt.src.close() _ = os.Remove(mt.point) } // sync 按完整列表挂载、卸载、更新 func (m *manager) sync(specs []mountSpec, save bool) ([]mountStatus, error) { want := map[string]mountSpec{} for _, s := range specs { if !idPattern.MatchString(s.ID) { return nil, fmt.Errorf("资料库 ID 格式不正确:%q", s.ID) } if s.Kind == "local" && (s.Local == nil || !filepath.IsAbs(s.Local.Path)) { return nil, fmt.Errorf("%s:缺少本机目录", s.ID) } want[s.ID] = s } m.mu.Lock() for id, mt := range m.mounts { if _, ok := want[id]; !ok { m.unmountLocked(mt) delete(m.mounts, id) log.Printf("%s:已卸载", id) } } for id, s := range want { mt := m.mounts[id] if mt == nil { mt = &mount{spec: s, point: filepath.Join(m.mntDir, id), src: newSource()} mt.src.onMiss = m.onMiss m.mounts[id] = mt log.Printf("%s:登记 %s", id, describeSpec(s)) } else { mt.spec = s } m.mountLocked(mt) m.refreshLocked(mt) } m.mu.Unlock() if save { if err := m.saveState(specs); err != nil { log.Printf("保存 state.json 失败:%v", err) } } return m.status(), nil } func describeSpec(s mountSpec) string { if s.Local != nil { if s.Local.UUID != "" { return fmt.Sprintf("%s(UUID %s,设备内 %s)", s.Local.Path, s.Local.UUID, s.Local.RelPath) } return s.Local.Path } return s.Kind } func (m *manager) status() []mountStatus { m.mu.Lock() defer m.mu.Unlock() out := make([]mountStatus, 0, len(m.mounts)) for id, mt := range m.mounts { st := mountStatus{ID: id, State: mt.state, Since: mt.since.UTC().Format(time.RFC3339), RealPath: mt.realPath} if mt.message != "" { msg := mt.message st.Message = &msg } out = append(out, st) } sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) return out } func (m *manager) unmountAll() { m.mu.Lock() defer m.mu.Unlock() for id, mt := range m.mounts { m.unmountLocked(mt) delete(m.mounts, id) } } type stateFile struct { Mounts []mountSpec `json:"mounts"` } func (m *manager) saveState(specs []mountSpec) error { b, err := json.MarshalIndent(stateFile{Mounts: specs}, "", " ") if err != nil { return err } tmp := m.stateFile + ".tmp" if err := os.WriteFile(tmp, b, 0o600); err != nil { return err } return os.Rename(tmp, m.stateFile) } func (m *manager) loadState() []mountSpec { b, err := os.ReadFile(m.stateFile) if err != nil { return nil } var s stateFile if json.Unmarshal(b, &s) != nil { log.Printf("state.json 格式不正确,忽略") return nil } return s.Mounts } // cleanupStale 启动时清理 mnt/ 下残留的挂载(上次进程退出后失效)和多余的空目录 func (m *manager) cleanupStale() { for _, e := range m.sys.mountinfo() { if filepath.Dir(e.Point) == m.mntDir { if err := exec.Command("fusermount3", "-uz", e.Point).Run(); err != nil { log.Printf("清理残留挂载 %s 失败:%v", e.Point, err) } else { log.Printf("已清理残留挂载 %s", e.Point) } } } ents, _ := os.ReadDir(m.mntDir) for _, e := range ents { if idPattern.MatchString(e.Name()) { _ = os.Remove(filepath.Join(m.mntDir, e.Name())) } } } // watchMountinfo 挂载表变化时(POLLPRI)重新计算状态 func (m *manager) watchMountinfo(stop <-chan struct{}) { f, err := os.Open("/proc/self/mountinfo") if err != nil { log.Printf("无法监听挂载表:%v", err) return } defer f.Close() buf := make([]byte, 64*1024) drain := func() { _, _ = f.Seek(0, 0) for { n, err := f.Read(buf) if n == 0 || err != nil { return } } } drain() fds := []unix.PollFd{{Fd: int32(f.Fd()), Events: unix.POLLPRI}} for { select { case <-stop: return default: } n, err := unix.Poll(fds, 1000) if err != nil && err != unix.EINTR { log.Printf("监听挂载表出错:%v", err) time.Sleep(time.Second) continue } if n <= 0 || fds[0].Revents&(unix.POLLPRI|unix.POLLERR) == 0 { continue } drain() // 设备插入时往往连续变化几次,稍等一下再统一计算 time.Sleep(300 * time.Millisecond) drain() m.refreshAll() } }