|
@@ -4,9 +4,14 @@
|
|
|
package logic
|
|
package logic
|
|
|
|
|
|
|
|
import (
|
|
import (
|
|
|
|
|
+ "bytes"
|
|
|
"context"
|
|
"context"
|
|
|
|
|
+ "encoding/json"
|
|
|
"fmt"
|
|
"fmt"
|
|
|
|
|
+ "io"
|
|
|
|
|
+ "net/http"
|
|
|
"strings"
|
|
"strings"
|
|
|
|
|
+ "time"
|
|
|
|
|
|
|
|
"udeServer/internal/model"
|
|
"udeServer/internal/model"
|
|
|
"udeServer/internal/svc"
|
|
"udeServer/internal/svc"
|
|
@@ -32,6 +37,7 @@ func NewCreateServiceLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Cre
|
|
|
|
|
|
|
|
func (l *CreateServiceLogic) CreateService(req *types.CreateServiceReq) (*types.ServiceResp, error) {
|
|
func (l *CreateServiceLogic) CreateService(req *types.CreateServiceReq) (*types.ServiceResp, error) {
|
|
|
var svcId int64
|
|
var svcId int64
|
|
|
|
|
+ var dirList []string // 落库成功的保护目录(trim 后非空),事务提交后用于通知外挂程序挂载
|
|
|
err := l.svcCtx.DB.Transact(func(session sqlx.Session) error {
|
|
err := l.svcCtx.DB.Transact(func(session sqlx.Session) error {
|
|
|
// 把 model 绑到事务 session 上,后续所有写操作都走同一个连接/事务
|
|
// 把 model 绑到事务 session 上,后续所有写操作都走同一个连接/事务
|
|
|
svcModel := l.svcCtx.ServiceModel.WithSession(session)
|
|
svcModel := l.svcCtx.ServiceModel.WithSession(session)
|
|
@@ -62,6 +68,7 @@ func (l *CreateServiceLogic) CreateService(req *types.CreateServiceReq) (*types.
|
|
|
if path == "" {
|
|
if path == "" {
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
|
|
+ dirList = append(dirList, path)
|
|
|
_, err := dirModel.Insert(l.ctx, &model.TDirectory{
|
|
_, err := dirModel.Insert(l.ctx, &model.TDirectory{
|
|
|
ServiceId: nullInt64(svcId),
|
|
ServiceId: nullInt64(svcId),
|
|
|
Directory: nullString(path),
|
|
Directory: nullString(path),
|
|
@@ -83,5 +90,56 @@ func (l *CreateServiceLogic) CreateService(req *types.CreateServiceReq) (*types.
|
|
|
return nil, err
|
|
return nil, err
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ // 3. 落库成功后,通知外挂程序(req.Ip:req.Port)执行 fuse 挂载。
|
|
|
|
|
+ // 外挂程序未启动/网络不通时挂载会失败:服务数据已落库不回滚,仅把错误抛给前端提示。
|
|
|
|
|
+ if err := l.notifyFuseMount(created, dirList); err != nil {
|
|
|
|
|
+ l.Errorf("service created(id=%d), but fuse mount notify failed: %v", svcId, err)
|
|
|
|
|
+ return toServiceResp(created), fmt.Errorf("服务已创建,但通知外挂程序挂载失败: %v", err)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
return toServiceResp(created), nil
|
|
return toServiceResp(created), nil
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+// fuseMountClient 复用的 HTTP 客户端,10s 超时防止外挂程序无响应拖死创建接口
|
|
|
|
|
+var fuseMountClient = &http.Client{Timeout: 10 * time.Second}
|
|
|
|
|
+
|
|
|
|
|
+// notifyFuseMount 落库成功后回调外挂程序(svc.Ip:svc.Port)的 /api/fuseMount 接口,
|
|
|
|
|
+// 由外挂程序对保护目录执行 fuse 挂载。请求体为 JSON:
|
|
|
|
|
+//
|
|
|
|
|
+// POST http://{ip}:{port}/api/fuseMount
|
|
|
|
|
+// { "iden": "...", "keyIden": "...", "keyAlgo": "...", "directories": ["/data/fuse", ...] }
|
|
|
|
|
+func (l *CreateServiceLogic) notifyFuseMount(svc *model.TService, dirs []string) error {
|
|
|
|
|
+ ip, port := svc.Ip.String, svc.Port.String
|
|
|
|
|
+ if ip == "" || port == "" {
|
|
|
|
|
+ // 没配外挂服务地址:只落库,跳过挂载通知
|
|
|
|
|
+ l.Infof("service %d has empty ip/port, skip fuse mount notify", svc.Id)
|
|
|
|
|
+ return nil
|
|
|
|
|
+ }
|
|
|
|
|
+ if len(dirs) == 0 {
|
|
|
|
|
+ return nil // 没有保护目录,无需通知挂载
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ url := fmt.Sprintf("http://%s:%s/api/fuseMount", ip, port)
|
|
|
|
|
+ payload := map[string]interface{}{
|
|
|
|
|
+ "iden": svc.Iden.String,
|
|
|
|
|
+ "keyIden": svc.KeyIden.String,
|
|
|
|
|
+ "keyAlgo": svc.KeyAlgo.String,
|
|
|
|
|
+ "directories": dirs,
|
|
|
|
|
+ }
|
|
|
|
|
+ body, err := json.Marshal(payload)
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ return fmt.Errorf("序列化挂载请求失败: %w", err)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ resp, err := fuseMountClient.Post(url, "application/json", bytes.NewReader(body))
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ return fmt.Errorf("请求 %s 失败(外挂程序未启动?): %w", url, err)
|
|
|
|
|
+ }
|
|
|
|
|
+ defer resp.Body.Close()
|
|
|
|
|
+
|
|
|
|
|
+ if resp.StatusCode >= http.StatusBadRequest {
|
|
|
|
|
+ snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
|
|
|
|
|
+ return fmt.Errorf("外挂程序返回 HTTP %d: %s", resp.StatusCode, string(snippet))
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|