|
|
@@ -132,25 +132,106 @@ func (l *CreateServiceLogic) CreateService(req *types.CreateServiceReq) (*types.
|
|
|
return resp, nil
|
|
|
}
|
|
|
|
|
|
-// notifyFuseBrush 下发一键刷库到外挂程序(svc.Ip:svc.Port)。
|
|
|
+// notifyFuseBrush 下发一键刷库到外挂程序(svc.Ip:svc.Port)的 /api/migrate 接口。
|
|
|
// 刷库语义:外挂读取保护目录存量明文 → 按服务密钥加密写入 <保护目录>-cipher
|
|
|
-// → 清空保护目录原明文(数据已转移到密文目录),为挂载的空目录校验扫清障碍。
|
|
|
-// 调用时机:创建流程中、挂载之前(目录非空时挂载会被空目录校验拒绝)。
|
|
|
-// TODO: 外挂程序侧的刷库接口(建议 POST /api/brush,请求体 {mount,ikey} 与挂载同源)
|
|
|
-// 开发完成后,在此填入真实下发实现(HTTP 客户端复用包级 fuseMountClient)。
|
|
|
-// 当前返回固定 error,由调用方软处理成黄色警告,绝不阻塞服务创建。
|
|
|
+// → 校验通过后清空保护目录原明文(数据已转移到密文目录,磁盘上只留密文一份),
|
|
|
+// 完成加密语义闭环,同时为挂载的空目录校验扫清障碍。
|
|
|
+// 调用时机:创建流程中、挂载之前(先刷后挂编排,见 CreateService 第4步)。
|
|
|
+// 实际下发逻辑抽成包级 dispatchFuseBrush,与手动补刷(BrushServiceLogic)共用。
|
|
|
func (l *CreateServiceLogic) notifyFuseBrush(svc *model.TService, dirs []string) error {
|
|
|
+ return dispatchFuseBrush(l.ctx, l.svcCtx, svc, dirs)
|
|
|
+}
|
|
|
+
|
|
|
+// doFuseRequest 发送发往外挂程序的 HTTP 请求,统一携带逐服务的 X-TOKEN 鉴权头。
|
|
|
+// X-TOKEN 的值 = 服务表 t_service 的 iden(接入标识)字段;外挂程序拿它与
|
|
|
+// 自己环境变量 TOKEN 配的值恒定时间比对(见 FuseDemo 的 NewAuthInterceptor),
|
|
|
+// 因此每台外挂机的 TOKEN 要配成"接在该机上的服务"的 iden 值。
|
|
|
+// client 按场景传:挂载/卸载/目录浏览用 fuseMountClient(10s),刷库用 fuseBrushClient(10min)。
|
|
|
+// 之所以不用 client.Post/client.Get 快捷方式:它们构造的请求没法逐个加 Header,
|
|
|
+// 必须先 http.NewRequest 再 client.Do。
|
|
|
+func doFuseRequest(client *http.Client, iden, method, url string, body io.Reader, contentType string) (*http.Response, error) {
|
|
|
+ req, err := http.NewRequest(method, url, body)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ if contentType != "" {
|
|
|
+ req.Header.Set("Content-Type", contentType)
|
|
|
+ }
|
|
|
+ req.Header.Set("X-TOKEN", iden)
|
|
|
+ return client.Do(req)
|
|
|
+}
|
|
|
+
|
|
|
+// fuseBrushClient 刷库专用HTTP客户端:刷库是重操作(全量加密可能数分钟),
|
|
|
+// 不能复用挂载的10s短超时,单独放宽到10分钟。
|
|
|
+var fuseBrushClient = &http.Client{Timeout: 10 * time.Minute}
|
|
|
+
|
|
|
+// dispatchFuseBrush 下发刷库到外挂(包级函数,创建流程与手动补刷共用同一实现)。
|
|
|
+// 逐保护目录调用 POST /api/migrate,请求体:
|
|
|
+//
|
|
|
+// { "backing": "<目录>-cipher", "mount": ["<目录>"], "overwrite": true, "ikey": "<密钥内容>" }
|
|
|
+//
|
|
|
+// ikey 取自 t_key.content(按服务 keyIden 查询),与挂载下发同源,保证刷库密钥=挂载密钥。
|
|
|
+// 外挂约定:部分文件失败也返回 HTTP 200,成败靠响应体的 failed 计数判断。
|
|
|
+// 注:创建流程的自动刷库无需挂载中校验——创建时序是先刷后挂,刷库时必然未挂载;
|
|
|
+// 手动补刷的挂载中拦截由调用方 BrushServiceLogic 负责。
|
|
|
+func dispatchFuseBrush(ctx context.Context, svcCtx *svc.ServiceContext, svc *model.TService, dirs []string) error {
|
|
|
if len(dirs) == 0 {
|
|
|
- return nil // 没配保护目录,无库可刷
|
|
|
+ return fmt.Errorf("没配保护目录,无库可刷")
|
|
|
}
|
|
|
- if svc.Ip.String == "" || svc.Port == 0 { // port 已是 int64,0 视为未配置
|
|
|
+ ip, port := svc.Ip.String, svc.Port // port 是 int64
|
|
|
+ if ip == "" || port == 0 {
|
|
|
return fmt.Errorf("外挂服务 IP/端口为空,无法下发刷库")
|
|
|
}
|
|
|
- // TODO: 逐目录调用外挂程序刷库接口(与 notifyFuseMount 同款循环模式)
|
|
|
- return fmt.Errorf("外挂程序刷库接口尚未实装,本轮未执行刷库(存量数据场景挂载会被空目录校验拒绝)")
|
|
|
+ keyIden := svc.KeyIden.String
|
|
|
+ if keyIden == "" {
|
|
|
+ return fmt.Errorf("服务未配置密钥标识(keyIden),无法下发刷库")
|
|
|
+ }
|
|
|
+ key, err := svcCtx.KeyModel.FindOneByIden(ctx, keyIden)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("查询密钥(%s)失败: %w", keyIden, err)
|
|
|
+ }
|
|
|
+ ikey := key.Content.String
|
|
|
+ if ikey == "" {
|
|
|
+ return fmt.Errorf("密钥(%s)内容为空,无法下发刷库", keyIden)
|
|
|
+ }
|
|
|
+ baseURL := fmt.Sprintf("http://%s:%d/api/migrate", ip, port)
|
|
|
+ for _, dir := range dirs {
|
|
|
+ dirLinuxPath := path.Clean(dir)
|
|
|
+ payload := map[string]interface{}{
|
|
|
+ "backing": dirLinuxPath + "-cipher", // 密文输出目录
|
|
|
+ "mount": []string{dirLinuxPath}, // 明文源目录(接口定义是数组,单目录传单元素)
|
|
|
+ "overwrite": true, // 补刷时目标已存在则覆盖,保证刷库可重复执行
|
|
|
+ "ikey": ikey,
|
|
|
+ }
|
|
|
+ body, err := json.Marshal(payload)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("序列化刷库请求失败: %w", err)
|
|
|
+ }
|
|
|
+ resp, err := doFuseRequest(fuseBrushClient, svc.Iden.String, http.MethodPost, baseURL, bytes.NewReader(body), "application/json")
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", baseURL, err)
|
|
|
+ }
|
|
|
+ // 先把响应体整体读出再Close,保证Keep-Alive连接可复用
|
|
|
+ respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
|
|
|
+ resp.Body.Close()
|
|
|
+ if resp.StatusCode >= http.StatusBadRequest {
|
|
|
+ return fmt.Errorf("刷库目录 %s 失败,外挂程序返回 HTTP %d: %s", dir, resp.StatusCode, string(respBody))
|
|
|
+ }
|
|
|
+ var result struct { // 匿名结构体:只用一次的临时JSON模型,不必单独定义类型
|
|
|
+ Total int `json:"total"`
|
|
|
+ Failed int `json:"failed"`
|
|
|
+ }
|
|
|
+ if err := json.Unmarshal(respBody, &result); err != nil {
|
|
|
+ return fmt.Errorf("解析刷库响应失败(目录 %s): %w", dir, err)
|
|
|
+ }
|
|
|
+ if result.Failed > 0 {
|
|
|
+ return fmt.Errorf("刷库目录 %s: %d/%d 个文件失败", dir, result.Failed, result.Total)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return nil
|
|
|
}
|
|
|
|
|
|
-// fuseMountClient 复用的 HTTP 客户端,10s 超时防止外挂程序无响应拖死创建接口
|
|
|
+// fuseMountClient 复用的 HTTP 客户端,10s 超时防止外挂程序无响应拖死创建接口。
|
|
|
var fuseMountClient = &http.Client{Timeout: 10 * time.Second}
|
|
|
|
|
|
// notifyFuseMount 落库成功后回调外挂程序(svc.Ip:svc.Port)的 /api/fuseMount 接口,
|
|
|
@@ -197,7 +278,7 @@ func (l *CreateServiceLogic) notifyFuseMount(svc *model.TService, dirs []string)
|
|
|
return fmt.Errorf("序列化挂载请求失败: %w", err)
|
|
|
}
|
|
|
|
|
|
- resp, err := fuseMountClient.Post(baseURL, "application/json", bytes.NewReader(body))
|
|
|
+ resp, err := doFuseRequest(fuseMountClient, svc.Iden.String, http.MethodPost, baseURL, bytes.NewReader(body), "application/json")
|
|
|
if err != nil {
|
|
|
return fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", baseURL, err)
|
|
|
}
|
