|
|
@@ -0,0 +1,589 @@
|
|
|
+package main
|
|
|
+
|
|
|
+// 远程存储（docs/design/21-knowledge-sources.md §6）：每个资料库一个 rclone 子进程，只读挂到 mounter/remote/<id>（内部，只有本用户能访问），
|
|
|
+// 再由本进程的文件系统（隐藏文件过滤、只读）提供给 knowledge/mnt/<id>。
|
|
|
+// 连接参数和凭据通过子进程的环境变量传给 rclone，不放在命令行里（命令行在进程列表中所有人可见）。
|
|
|
+
|
|
|
+import (
|
|
|
+ "bytes"
|
|
|
+ "context"
|
|
|
+ "fmt"
|
|
|
+ "log"
|
|
|
+ "net/url"
|
|
|
+ "os"
|
|
|
+ "os/exec"
|
|
|
+ "path/filepath"
|
|
|
+ "regexp"
|
|
|
+ "sort"
|
|
|
+ "strings"
|
|
|
+ "sync"
|
|
|
+ "syscall"
|
|
|
+ "time"
|
|
|
+)
|
|
|
+
|
|
|
+type remoteSpec struct {
|
|
|
+ Type string `json:"type"`
|
|
|
+ Params map[string]string `json:"params"`
|
|
|
+ Secrets map[string]string `json:"secrets"`
|
|
|
+}
|
|
|
+
|
|
|
+// allowLocalRemote 仅测试使用：rclone 的 local 后端，不需要网络
|
|
|
+var allowLocalRemote = false
|
|
|
+
|
|
|
+var (
|
|
|
+ bucketPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,254}$`)
|
|
|
+ s3Providers = map[string]bool{"Minio": true, "AWS": true, "Alibaba": true, "TencentCOS": true, "HuaweiOBS": true, "Qiniu": true, "Ceph": true, "Cloudflare": true, "Other": true}
|
|
|
+ webdavVendors = map[string]bool{"other": true, "nextcloud": true, "owncloud": true, "sharepoint": true, "sharepoint-ntlm": true}
|
|
|
+)
|
|
|
+
|
|
|
+type rcloneEnv struct {
|
|
|
+ env []string // RCLONE_CONFIG_KB_*
|
|
|
+ path string // kb:bucket/prefix
|
|
|
+}
|
|
|
+
|
|
|
+func clean(v string) (string, error) {
|
|
|
+ v = strings.TrimSpace(v)
|
|
|
+ if strings.ContainsAny(v, "\x00\r\n") {
|
|
|
+ return "", fmt.Errorf("参数中不能有换行")
|
|
|
+ }
|
|
|
+ return v, nil
|
|
|
+}
|
|
|
+
|
|
|
+func httpURL(v, name string) (string, error) {
|
|
|
+ u, err := url.Parse(v)
|
|
|
+ if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
|
|
|
+ return "", fmt.Errorf("%s必须是 http 或 https 地址", name)
|
|
|
+ }
|
|
|
+ if u.User != nil {
|
|
|
+ return "", fmt.Errorf("%s中不能包含用户名或密码", name)
|
|
|
+ }
|
|
|
+ return v, nil
|
|
|
+}
|
|
|
+
|
|
|
+// buildRemote 按存储类型生成 rclone 的环境变量与远程路径
|
|
|
+func buildRemote(r remoteSpec, rclone string) (*rcloneEnv, error) {
|
|
|
+ p := map[string]string{}
|
|
|
+ for k, v := range r.Params {
|
|
|
+ c, err := clean(v)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ p[k] = c
|
|
|
+ }
|
|
|
+ sec := func(k string) (string, error) {
|
|
|
+ v := r.Secrets[k]
|
|
|
+ if strings.ContainsAny(v, "\x00\r\n") {
|
|
|
+ return "", fmt.Errorf("凭据中不能有换行")
|
|
|
+ }
|
|
|
+ return v, nil
|
|
|
+ }
|
|
|
+ out := &rcloneEnv{}
|
|
|
+ add := func(k, v string) { out.env = append(out.env, "RCLONE_CONFIG_KB_"+k+"="+v) }
|
|
|
+ switch r.Type {
|
|
|
+ case "s3":
|
|
|
+ provider := p["provider"]
|
|
|
+ if provider == "" {
|
|
|
+ provider = "Other"
|
|
|
+ }
|
|
|
+ if !s3Providers[provider] {
|
|
|
+ return nil, fmt.Errorf("不支持的 S3 服务商：%s", provider)
|
|
|
+ }
|
|
|
+ if p["endpoint"] == "" && provider != "AWS" {
|
|
|
+ return nil, fmt.Errorf("请填写服务地址（endpoint）")
|
|
|
+ }
|
|
|
+ if p["endpoint"] != "" {
|
|
|
+ if _, err := httpURL(p["endpoint"], "服务地址"); err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if !bucketPattern.MatchString(p["bucket"]) {
|
|
|
+ return nil, fmt.Errorf("存储桶名称不正确")
|
|
|
+ }
|
|
|
+ ak, err := sec("accessKeyId")
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ sk, err := sec("secretAccessKey")
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ if ak == "" || sk == "" {
|
|
|
+ return nil, fmt.Errorf("请填写 Access Key 和 Secret Key")
|
|
|
+ }
|
|
|
+ add("TYPE", "s3")
|
|
|
+ add("PROVIDER", provider)
|
|
|
+ add("ENV_AUTH", "false")
|
|
|
+ add("ACCESS_KEY_ID", ak)
|
|
|
+ add("SECRET_ACCESS_KEY", sk)
|
|
|
+ add("NO_CHECK_BUCKET", "true")
|
|
|
+ if p["endpoint"] != "" {
|
|
|
+ add("ENDPOINT", p["endpoint"])
|
|
|
+ }
|
|
|
+ if p["region"] != "" {
|
|
|
+ add("REGION", p["region"])
|
|
|
+ }
|
|
|
+ prefix := strings.Trim(p["prefix"], "/")
|
|
|
+ if strings.Contains("/"+prefix+"/", "/../") {
|
|
|
+ return nil, fmt.Errorf("路径前缀不能包含 ..")
|
|
|
+ }
|
|
|
+ out.path = "kb:" + p["bucket"]
|
|
|
+ if prefix != "" {
|
|
|
+ out.path += "/" + prefix
|
|
|
+ }
|
|
|
+ case "webdav":
|
|
|
+ u, err := httpURL(p["url"], "地址")
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ vendor := p["vendor"]
|
|
|
+ if vendor == "" {
|
|
|
+ vendor = "other"
|
|
|
+ }
|
|
|
+ if !webdavVendors[vendor] {
|
|
|
+ return nil, fmt.Errorf("不支持的 WebDAV 服务类型：%s", vendor)
|
|
|
+ }
|
|
|
+ add("TYPE", "webdav")
|
|
|
+ add("URL", u)
|
|
|
+ add("VENDOR", vendor)
|
|
|
+ if p["user"] != "" {
|
|
|
+ add("USER", p["user"])
|
|
|
+ }
|
|
|
+ pass, err := sec("password")
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ if pass != "" {
|
|
|
+ ob, err := obscure(rclone, pass)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ add("PASS", ob)
|
|
|
+ }
|
|
|
+ out.path = "kb:"
|
|
|
+ case "local":
|
|
|
+ if !allowLocalRemote {
|
|
|
+ return nil, fmt.Errorf("不支持的存储类型：local")
|
|
|
+ }
|
|
|
+ add("TYPE", "local")
|
|
|
+ out.path = "kb:" + p["path"]
|
|
|
+ default:
|
|
|
+ return nil, fmt.Errorf("不支持的存储类型：%s", r.Type)
|
|
|
+ }
|
|
|
+ return out, nil
|
|
|
+}
|
|
|
+
|
|
|
+// obscure rclone 的 WebDAV 密码需要先混淆；明文从标准输入传入，不出现在命令行
|
|
|
+func obscure(rclone, pass string) (string, error) {
|
|
|
+ cmd := exec.Command(rclone, "obscure", "-")
|
|
|
+ cmd.Env = []string{"PATH=/usr/bin:/bin"}
|
|
|
+ cmd.Stdin = strings.NewReader(pass)
|
|
|
+ out, err := cmd.Output()
|
|
|
+ if err != nil {
|
|
|
+ return "", fmt.Errorf("处理密码失败：%v", err)
|
|
|
+ }
|
|
|
+ return strings.TrimSpace(string(out)), nil
|
|
|
+}
|
|
|
+
|
|
|
+// classify 根据 rclone 的错误输出判断状态
|
|
|
+func classify(stderr string) (state, msg string) {
|
|
|
+ low := strings.ToLower(stderr)
|
|
|
+ detail := lastError(stderr)
|
|
|
+ switch {
|
|
|
+ case containsAny(low, "signaturedoesnotmatch", "invalidaccesskeyid", "accessdenied", "access denied", "statuscode: 401", "statuscode: 403", "401 unauthorized", "403 forbidden", "authentication", "unauthorized"):
|
|
|
+ return "auth_failed", "认证失败或没有权限：" + detail
|
|
|
+ case containsAny(low, "nosuchbucket", "bucket does not exist", "directory not found", "404 not found", "statuscode: 404"):
|
|
|
+ return "path_missing", "存储桶或路径不存在：" + detail
|
|
|
+ case containsAny(low, "connection refused", "no such host", "i/o timeout", "timeout", "network is unreachable", "tls:", "x509", "connection reset", "unexpected eof"):
|
|
|
+ return "unreachable", "无法连接：" + detail
|
|
|
+ }
|
|
|
+ if detail == "" {
|
|
|
+ detail = "未知错误"
|
|
|
+ }
|
|
|
+ return "unreachable", detail
|
|
|
+}
|
|
|
+
|
|
|
+var (
|
|
|
+ apiErr = regexp.MustCompile(`api error ([A-Za-z]+): ([^\n]+)`)
|
|
|
+ netErr = regexp.MustCompile(`(dial tcp [^\n]*|lookup [^\n]*|tls: [^\n]*|x509: [^\n]*)`)
|
|
|
+)
|
|
|
+
|
|
|
+func lastError(s string) string {
|
|
|
+ if m := apiErr.FindAllStringSubmatch(s, -1); len(m) > 0 {
|
|
|
+ last := m[len(m)-1]
|
|
|
+ return strings.TrimSpace(last[1] + "：" + last[2])
|
|
|
+ }
|
|
|
+ if m := netErr.FindAllString(s, -1); len(m) > 0 {
|
|
|
+ return strings.TrimSpace(m[len(m)-1])
|
|
|
+ }
|
|
|
+ lines := strings.Split(strings.TrimSpace(s), "\n")
|
|
|
+ for i := len(lines) - 1; i >= 0; i-- {
|
|
|
+ l := lines[i]
|
|
|
+ if strings.Contains(l, "ERROR") || strings.Contains(l, "CRITICAL") || strings.Contains(l, "Failed") {
|
|
|
+ if j := strings.Index(l, ": "); j >= 0 && strings.Contains(l[:j], "/") {
|
|
|
+ l = l[j+2:]
|
|
|
+ }
|
|
|
+ if len(l) > 300 {
|
|
|
+ l = l[:300] + "…"
|
|
|
+ }
|
|
|
+ return strings.TrimSpace(l)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return ""
|
|
|
+}
|
|
|
+
|
|
|
+func containsAny(s string, subs ...string) bool {
|
|
|
+ for _, x := range subs {
|
|
|
+ if strings.Contains(s, x) {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false
|
|
|
+}
|
|
|
+
|
|
|
+// rcloneRunner 运行 rclone 的公共参数
|
|
|
+type rcloneRunner struct {
|
|
|
+ bin string
|
|
|
+ config string // 空的配置文件，避免读取部署用户自己的 rclone.conf
|
|
|
+ home string
|
|
|
+}
|
|
|
+
|
|
|
+func (r *rcloneRunner) command(ctx context.Context, env *rcloneEnv, args ...string) *exec.Cmd {
|
|
|
+ cmd := exec.CommandContext(ctx, r.bin, append(args, "--config", r.config)...)
|
|
|
+ cmd.Env = append([]string{"PATH=/usr/bin:/bin", "HOME=" + r.home}, env.env...)
|
|
|
+ return cmd
|
|
|
+}
|
|
|
+
|
|
|
+type probeResult struct {
|
|
|
+ State string `json:"state"`
|
|
|
+ Message string `json:"message"`
|
|
|
+ Entries []string `json:"entries"`
|
|
|
+}
|
|
|
+
|
|
|
+// probe 列出远程根目录（带超时），用于"测试连接"与状态检查
|
|
|
+func (r *rcloneRunner) probe(parent context.Context, env *rcloneEnv, timeout time.Duration) probeResult {
|
|
|
+ ctx, cancel := context.WithTimeout(parent, timeout)
|
|
|
+ defer cancel()
|
|
|
+ cmd := r.command(ctx, env, "lsf", env.path, "--max-depth", "1", "--low-level-retries", "1", "--retries", "1", "--contimeout", "5s", "--timeout", "10s", "--log-level", "ERROR")
|
|
|
+ var out, errb bytes.Buffer
|
|
|
+ cmd.Stdout, cmd.Stderr = &limitWriter{w: &out, n: 64 << 10}, &limitWriter{w: &errb, n: 64 << 10}
|
|
|
+ err := cmd.Run()
|
|
|
+ if ctx.Err() == context.DeadlineExceeded {
|
|
|
+ return probeResult{State: "unreachable", Message: "连接超时"}
|
|
|
+ }
|
|
|
+ if ctx.Err() != nil {
|
|
|
+ return probeResult{State: "pending", Message: "已取消"}
|
|
|
+ }
|
|
|
+ // rclone lsf 出错时退出码可能为 0，以错误输出为准
|
|
|
+ if err != nil || strings.Contains(errb.String(), "ERROR") {
|
|
|
+ st, msg := classify(errb.String())
|
|
|
+ return probeResult{State: st, Message: msg}
|
|
|
+ }
|
|
|
+ var entries []string
|
|
|
+ for _, l := range strings.Split(strings.TrimSpace(out.String()), "\n") {
|
|
|
+ if l != "" && !strings.HasPrefix(l, ".") && len(entries) < 50 {
|
|
|
+ entries = append(entries, l)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return probeResult{State: "ok", Entries: entries}
|
|
|
+}
|
|
|
+
|
|
|
+type limitWriter struct {
|
|
|
+ w *bytes.Buffer
|
|
|
+ n int
|
|
|
+}
|
|
|
+
|
|
|
+func (l *limitWriter) Write(p []byte) (int, error) {
|
|
|
+ if rest := l.n - l.w.Len(); rest > 0 {
|
|
|
+ if len(p) > rest {
|
|
|
+ l.w.Write(p[:rest])
|
|
|
+ } else {
|
|
|
+ l.w.Write(p)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return len(p), nil
|
|
|
+}
|
|
|
+
|
|
|
+// ringLog 保留 rclone 最近的错误输出
|
|
|
+type ringLog struct {
|
|
|
+ mu sync.Mutex
|
|
|
+ buf []string
|
|
|
+}
|
|
|
+
|
|
|
+func (r *ringLog) Write(p []byte) (int, error) {
|
|
|
+ r.mu.Lock()
|
|
|
+ defer r.mu.Unlock()
|
|
|
+ for _, l := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
|
|
|
+ if l == "" {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ r.buf = append(r.buf, l)
|
|
|
+ if len(r.buf) > 50 {
|
|
|
+ r.buf = r.buf[len(r.buf)-50:]
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return len(p), nil
|
|
|
+}
|
|
|
+
|
|
|
+func (r *ringLog) String() string {
|
|
|
+ r.mu.Lock()
|
|
|
+ defer r.mu.Unlock()
|
|
|
+ return strings.Join(r.buf, "\n")
|
|
|
+}
|
|
|
+
|
|
|
+func (r *ringLog) reset() {
|
|
|
+ r.mu.Lock()
|
|
|
+ r.buf = nil
|
|
|
+ r.mu.Unlock()
|
|
|
+}
|
|
|
+
|
|
|
+// remoteProc 一个远程存储的 rclone 挂载及其状态
|
|
|
+type remoteProc struct {
|
|
|
+ id string
|
|
|
+ point string // 内部挂载点
|
|
|
+ cache string
|
|
|
+ runner *rcloneRunner
|
|
|
+ sys system
|
|
|
+ env *rcloneEnv
|
|
|
+ key string // 配置的指纹，变化时重启
|
|
|
+
|
|
|
+ mu sync.Mutex
|
|
|
+ mounted bool
|
|
|
+ state string
|
|
|
+ message string
|
|
|
+ probedAt time.Time
|
|
|
+ ready chan struct{} // 首次得出状态后关闭
|
|
|
+ readyOnce sync.Once
|
|
|
+ stop chan struct{}
|
|
|
+ done chan struct{}
|
|
|
+ cmd *exec.Cmd
|
|
|
+ logs ringLog
|
|
|
+ onChange func()
|
|
|
+}
|
|
|
+
|
|
|
+func (p *remoteProc) status() (mounted bool, state, msg string) {
|
|
|
+ p.mu.Lock()
|
|
|
+ defer p.mu.Unlock()
|
|
|
+ return p.mounted, p.state, p.message
|
|
|
+}
|
|
|
+
|
|
|
+func (p *remoteProc) set(mounted bool, state, msg string) {
|
|
|
+ p.mu.Lock()
|
|
|
+ changed := p.mounted != mounted || p.state != state || p.message != msg
|
|
|
+ p.mounted, p.state, p.message = mounted, state, msg
|
|
|
+ p.mu.Unlock()
|
|
|
+ if state != "pending" {
|
|
|
+ p.readyOnce.Do(func() { close(p.ready) })
|
|
|
+ }
|
|
|
+ if changed && p.onChange != nil {
|
|
|
+ p.onChange()
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// isMounted 内部挂载点上确实是 rclone 的挂载（而不是下面的空目录）
|
|
|
+func (p *remoteProc) isMounted() bool {
|
|
|
+ for _, e := range p.sys.mountinfo() {
|
|
|
+ if e.Point == p.point && strings.HasPrefix(e.FSType, "fuse.rclone") {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false
|
|
|
+}
|
|
|
+
|
|
|
+// run 启动并看护 rclone：退出时按退避间隔重启
|
|
|
+func (p *remoteProc) run() {
|
|
|
+ defer close(p.done)
|
|
|
+ backoff := 5 * time.Second
|
|
|
+ for {
|
|
|
+ started := time.Now()
|
|
|
+ p.runOnce()
|
|
|
+ select {
|
|
|
+ case <-p.stop:
|
|
|
+ return
|
|
|
+ default:
|
|
|
+ }
|
|
|
+ if time.Since(started) > time.Minute {
|
|
|
+ backoff = 5 * time.Second
|
|
|
+ }
|
|
|
+ select {
|
|
|
+ case <-p.stop:
|
|
|
+ return
|
|
|
+ case <-time.After(backoff):
|
|
|
+ }
|
|
|
+ if backoff < 5*time.Minute {
|
|
|
+ backoff *= 2
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (p *remoteProc) runOnce() {
|
|
|
+ _ = exec.Command("fusermount3", "-uz", p.point).Run()
|
|
|
+ if err := os.MkdirAll(p.point, 0o700); err != nil {
|
|
|
+ p.set(false, "mount_failed", "创建内部挂载点失败："+err.Error())
|
|
|
+ return
|
|
|
+ }
|
|
|
+ _ = os.MkdirAll(p.cache, 0o700)
|
|
|
+ p.logs.reset()
|
|
|
+ cmd := p.runner.command(context.Background(), p.env, "mount", p.env.path, p.point,
|
|
|
+ "--read-only", "--vfs-cache-mode", "full", "--cache-dir", p.cache,
|
|
|
+ "--vfs-cache-max-size", "1G", "--vfs-cache-max-age", "1h", "--dir-cache-time", "1m",
|
|
|
+ "--low-level-retries", "3", "--contimeout", "10s", "--timeout", "60s", "--log-level", "NOTICE")
|
|
|
+ cmd.Stdout, cmd.Stderr = &p.logs, &p.logs
|
|
|
+ cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGTERM}
|
|
|
+ if err := cmd.Start(); err != nil {
|
|
|
+ p.set(false, "mount_failed", "启动 rclone 失败："+err.Error())
|
|
|
+ return
|
|
|
+ }
|
|
|
+ p.mu.Lock()
|
|
|
+ p.cmd = cmd
|
|
|
+ p.mu.Unlock()
|
|
|
+ exited := make(chan error, 1)
|
|
|
+ go func() { exited <- cmd.Wait() }()
|
|
|
+
|
|
|
+ // 等挂载出现
|
|
|
+ deadline := time.After(30 * time.Second)
|
|
|
+ tick := time.NewTicker(200 * time.Millisecond)
|
|
|
+ defer tick.Stop()
|
|
|
+wait:
|
|
|
+ for {
|
|
|
+ select {
|
|
|
+ case err := <-exited:
|
|
|
+ st, msg := classify(p.logs.String())
|
|
|
+ if msg == "未知错误" || msg == "" {
|
|
|
+ msg = fmt.Sprintf("rclone 退出：%v", err)
|
|
|
+ }
|
|
|
+ log.Printf("%s：rclone 退出：%s", p.id, lastError(p.logs.String()))
|
|
|
+ p.set(false, st, msg)
|
|
|
+ return
|
|
|
+ case <-p.stop:
|
|
|
+ p.terminate(exited)
|
|
|
+ return
|
|
|
+ case <-deadline:
|
|
|
+ log.Printf("%s：rclone 30 秒内没有完成挂载", p.id)
|
|
|
+ p.set(false, "unreachable", "挂载超时")
|
|
|
+ p.terminate(exited)
|
|
|
+ return
|
|
|
+ case <-tick.C:
|
|
|
+ if p.isMounted() {
|
|
|
+ break wait
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ log.Printf("%s：rclone 已挂载 %s", p.id, p.env.path)
|
|
|
+ p.set(true, "pending", "正在连接")
|
|
|
+ p.checkNow()
|
|
|
+
|
|
|
+ // 运行期间：未连通时每分钟检查一次，正常时每 5 分钟
|
|
|
+ for {
|
|
|
+ _, state, _ := p.status()
|
|
|
+ every := 5 * time.Minute
|
|
|
+ if state != "ok" {
|
|
|
+ every = time.Minute
|
|
|
+ }
|
|
|
+ select {
|
|
|
+ case err := <-exited:
|
|
|
+ st, msg := "unreachable", fmt.Sprintf("rclone 意外退出（%v），稍后自动重启", err)
|
|
|
+ if detail := lastError(p.logs.String()); detail != "" {
|
|
|
+ st, msg = classify(p.logs.String())
|
|
|
+ }
|
|
|
+ log.Printf("%s：rclone 意外退出（%v）：%s", p.id, err, lastError(p.logs.String()))
|
|
|
+ p.set(false, st, msg)
|
|
|
+ _ = exec.Command("fusermount3", "-uz", p.point).Run()
|
|
|
+ return
|
|
|
+ case <-p.stop:
|
|
|
+ p.terminate(exited)
|
|
|
+ return
|
|
|
+ case <-time.After(every):
|
|
|
+ p.checkNow()
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// checkNow 列远程根目录，得出连接状态
|
|
|
+func (p *remoteProc) checkNow() {
|
|
|
+ ctx, cancel := context.WithCancel(context.Background())
|
|
|
+ defer cancel()
|
|
|
+ go func() {
|
|
|
+ select {
|
|
|
+ case <-p.stop:
|
|
|
+ cancel()
|
|
|
+ case <-ctx.Done():
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ r := p.runner.probe(ctx, p.env, 20*time.Second)
|
|
|
+ if r.State == "pending" {
|
|
|
+ return
|
|
|
+ }
|
|
|
+ if r.State == "ok" {
|
|
|
+ // 之前的错误已经过去，rclone 之后退出时不再据此判断
|
|
|
+ p.logs.reset()
|
|
|
+ }
|
|
|
+ p.mu.Lock()
|
|
|
+ p.probedAt = time.Now()
|
|
|
+ mounted := p.mounted
|
|
|
+ p.mu.Unlock()
|
|
|
+ p.set(mounted, r.State, r.Message)
|
|
|
+}
|
|
|
+
|
|
|
+// checkIfStale 状态查询时：超过 30 秒没检查过的，在后台重新检查
|
|
|
+func (p *remoteProc) checkIfStale() {
|
|
|
+ p.mu.Lock()
|
|
|
+ stale := p.mounted && time.Since(p.probedAt) > 30*time.Second
|
|
|
+ if stale {
|
|
|
+ p.probedAt = time.Now()
|
|
|
+ }
|
|
|
+ p.mu.Unlock()
|
|
|
+ if stale {
|
|
|
+ go p.checkNow()
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (p *remoteProc) terminate(exited chan error) {
|
|
|
+ p.mu.Lock()
|
|
|
+ cmd := p.cmd
|
|
|
+ p.mu.Unlock()
|
|
|
+ if cmd != nil && cmd.Process != nil {
|
|
|
+ _ = cmd.Process.Signal(syscall.SIGTERM)
|
|
|
+ select {
|
|
|
+ case <-exited:
|
|
|
+ case <-time.After(5 * time.Second):
|
|
|
+ _ = cmd.Process.Kill()
|
|
|
+ <-exited
|
|
|
+ }
|
|
|
+ }
|
|
|
+ _ = exec.Command("fusermount3", "-uz", p.point).Run()
|
|
|
+}
|
|
|
+
|
|
|
+func newRemoteProc(id, remoteDir, cacheDir string, runner *rcloneRunner, sys system, env *rcloneEnv, key string, onChange func()) *remoteProc {
|
|
|
+ return &remoteProc{
|
|
|
+ id: id, point: filepath.Join(remoteDir, id), cache: filepath.Join(cacheDir, id), runner: runner, sys: sys, env: env, key: key,
|
|
|
+ state: "pending", message: "正在连接", ready: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), onChange: onChange,
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (p *remoteProc) start() { go p.run() }
|
|
|
+
|
|
|
+// shutdown 停止 rclone、卸载、删除缓存
|
|
|
+func (p *remoteProc) shutdown() {
|
|
|
+ close(p.stop)
|
|
|
+ <-p.done
|
|
|
+ _ = os.Remove(p.point)
|
|
|
+ _ = os.RemoveAll(p.cache)
|
|
|
+}
|
|
|
+
|
|
|
+// remoteKey 配置指纹（含凭据），用于判断是否需要重启 rclone
|
|
|
+func remoteKey(r *remoteSpec) string {
|
|
|
+ var b strings.Builder
|
|
|
+ b.WriteString(r.Type)
|
|
|
+ for _, m := range []map[string]string{r.Params, r.Secrets} {
|
|
|
+ keys := make([]string, 0, len(m))
|
|
|
+ for k := range m {
|
|
|
+ keys = append(keys, k)
|
|
|
+ }
|
|
|
+ sort.Strings(keys)
|
|
|
+ for _, k := range keys {
|
|
|
+ fmt.Fprintf(&b, "\x00%s=%s", k, m[k])
|
|
|
+ }
|
|
|
+ b.WriteString("\x01")
|
|
|
+ }
|
|
|
+ return b.String()
|
|
|
+}
|
Jerry Liang