解决冲突；storage相关数据库定义

agent/internal/http/hub_io.go

@@ -4,17 +4,18 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"time"
+
 	"github.com/gin-gonic/gin"
 	"github.com/inhies/go-bytesize"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
-	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+	"gitlink.org.cn/cloudream/common/sdks/storage/cdsapi"
 	"gitlink.org.cn/cloudream/common/utils/serder"
-	"io"
-	"io/ioutil"
-	"net/http"
-	"time"
 )
 
 type IOService struct {
@@ -28,7 +29,7 @@ func (s *Server) IOSvc() *IOService {
 }
 
 func (s *IOService) GetStream(ctx *gin.Context) {
-	var req cdssdk.GetStreamReq
+	var req cdsapi.GetStreamReq
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		logger.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
@@ -126,7 +127,7 @@ func (s *IOService) SendStream(ctx *gin.Context) {
 	//planID := ctx.PostForm("plan_id")
 	//varID := ctx.PostForm("var_id")
 
-	var req cdssdk.SendStreamReq
+	var req cdsapi.SendStreamReq
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		logger.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
@@ -201,9 +202,9 @@ func (s *IOService) ExecuteIOPlan(ctx *gin.Context) {
 		return
 	}
 	println("Received body: %s", string(bodyBytes))
-	ctx.Request.Body = ioutil.NopCloser(bytes.NewBuffer(bodyBytes)) // Reset body for subsequent reads
+	ctx.Request.Body = io.NopCloser(bytes.NewBuffer(bodyBytes)) // Reset body for subsequent reads
 
-	var req cdssdk.ExecuteIOPlanReq
+	var req cdsapi.ExecuteIOPlanReq
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		logger.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
@@ -240,7 +241,7 @@ func (s *IOService) ExecuteIOPlan(ctx *gin.Context) {
 }
 
 func (s *IOService) SendVar(ctx *gin.Context) {
-	var req cdssdk.SendVarReq
+	var req cdsapi.SendVarReq
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		logger.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
@@ -268,7 +269,7 @@ func (s *IOService) SendVar(ctx *gin.Context) {
 }
 
 func (s *IOService) GetVar(ctx *gin.Context) {
-	var req cdssdk.GetVarReq
+	var req cdsapi.GetVarReq
 	if err := ctx.ShouldBindJSON(&req); err != nil {
 		logger.Warnf("binding body: %s", err.Error())
 		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))

agent/internal/http/server.go

@@ -3,7 +3,7 @@ package http
 import (
 	"github.com/gin-gonic/gin"
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
-	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+	"gitlink.org.cn/cloudream/common/sdks/storage/cdsapi"
 )
 
 type Server struct {
@@ -38,9 +38,9 @@ func (s *Server) Serve() error {
 }
 
 func (s *Server) initRouters() {
-	s.engine.GET(cdssdk.GetStreamPath, s.IOSvc().GetStream)
-	s.engine.POST(cdssdk.SendStreamPath, s.IOSvc().SendStream)
-	s.engine.POST(cdssdk.ExecuteIOPlanPath, s.IOSvc().ExecuteIOPlan)
-	s.engine.POST(cdssdk.SendVarPath, s.IOSvc().SendVar)
-	s.engine.GET(cdssdk.GetVarPath, s.IOSvc().GetVar)
+	s.engine.GET(cdsapi.GetStreamPath, s.IOSvc().GetStream)
+	s.engine.POST(cdsapi.SendStreamPath, s.IOSvc().SendStream)
+	s.engine.POST(cdsapi.ExecuteIOPlanPath, s.IOSvc().ExecuteIOPlan)
+	s.engine.POST(cdsapi.SendVarPath, s.IOSvc().SendVar)
+	s.engine.GET(cdsapi.GetVarPath, s.IOSvc().GetVar)
 }

common/pkgs/db2/db2.go

@@ -22,3 +22,17 @@ func NewDB(cfg *config.Config) (*DB, error) {
 		db: mydb,
 	}, nil
 }
+
+func (s *DB) DoTx(do func(tx SQLContext) error) error {
+	return s.db.Transaction(func(tx *gorm.DB) error {
+		return do(SQLContext{tx})
+	})
+}
+
+type SQLContext struct {
+	*gorm.DB
+}
+
+func (d *DB) DefCtx() SQLContext {
+	return SQLContext{d.db}
+}

common/pkgs/db2/node.go

@@ -1,8 +1,9 @@
 package db2
 
 import (
-	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
 	"time"
+
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
 )
 
 type NodeDB struct {
@@ -13,24 +14,24 @@ func (nodeDB *DB) Node() *NodeDB {
 	return &NodeDB{DB: nodeDB}
 }
 
-func (nodeDB *NodeDB) GetAllNodes() ([]cdssdk.Node, error) {
+func (*NodeDB) GetAllNodes(ctx SQLContext) ([]cdssdk.Node, error) {
 	var ret []cdssdk.Node
 
-	err := nodeDB.DB.db.Table("node").Find(&ret).Error
+	err := ctx.Table("node").Find(&ret).Error
 	return ret, err
 }
 
-func (nodeDB *NodeDB) GetByID(nodeID cdssdk.NodeID) (cdssdk.Node, error) {
+func (*NodeDB) GetByID(ctx SQLContext, nodeID cdssdk.NodeID) (cdssdk.Node, error) {
 	var ret cdssdk.Node
-	err := nodeDB.DB.db.Table("node").Where("NodeID = ?", nodeID).Find(&ret).Error
+	err := ctx.Table("node").Where("NodeID = ?", nodeID).Find(&ret).Error
 
 	return ret, err
 }
 
 // GetUserNodes 根据用户id查询可用node
-func (nodeDB *NodeDB) GetUserNodes(userID cdssdk.UserID) ([]cdssdk.Node, error) {
+func (*NodeDB) GetUserNodes(ctx SQLContext, userID cdssdk.UserID) ([]cdssdk.Node, error) {
 	var nodes []cdssdk.Node
-	err := nodeDB.DB.db.
+	err := ctx.
 		Table("Node").
 		Select("Node.*").
 		Joins("JOIN UserNode ON UserNode.NodeID = Node.NodeID").
@@ -40,8 +41,8 @@ func (nodeDB *NodeDB) GetUserNodes(userID cdssdk.UserID) ([]cdssdk.Node, error)
 }
 
 // UpdateState 更新状态，并且设置上次上报时间为现在
-func (nodeDB *NodeDB) UpdateState(nodeID cdssdk.NodeID, state string) error {
-	err := nodeDB.DB.db.
+func (*NodeDB) UpdateState(ctx SQLContext, nodeID cdssdk.NodeID, state string) error {
+	err := ctx.
 		Model(&cdssdk.Node{}).
 		Where("NodeID = ?", nodeID).
 		Updates(map[string]interface{}{

common/pkgs/db2/shard_storage.go

@@ -0,0 +1,19 @@
+package db2
+
+import (
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+)
+
+type ShardStorageDB struct {
+	*DB
+}
+
+func (db *DB) ShardStorage() *ShardStorageDB {
+	return &ShardStorageDB{DB: db}
+}
+
+func (*ShardStorageDB) GetByStorageID(ctx SQLContext, stgID cdssdk.StorageID) (cdssdk.ShardStorage, error) {
+	var ret cdssdk.ShardStorage
+	err := ctx.Table("ShardStorage").First(&ret, stgID).Error
+	return ret, err
+}

common/pkgs/db2/shared_storage.go

@@ -0,0 +1,19 @@
+package db2
+
+import (
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+)
+
+type SharedStorageDB struct {
+	*DB
+}
+
+func (db *DB) SharedStorage() *SharedStorageDB {
+	return &SharedStorageDB{DB: db}
+}
+
+func (*SharedStorageDB) GetByStorageID(ctx SQLContext, stgID cdssdk.StorageID) (cdssdk.SharedStorage, error) {
+	var ret cdssdk.SharedStorage
+	err := ctx.Table("SharedStorage").First(&ret, stgID).Error
+	return ret, err
+}

common/pkgs/db2/storage.go

@@ -0,0 +1,62 @@
+package db2
+
+import (
+	"fmt"
+
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
+)
+
+type StorageDB struct {
+	*DB
+}
+
+func (db *DB) Storage() *StorageDB {
+	return &StorageDB{DB: db}
+}
+
+func (db *StorageDB) GetByID(ctx SQLContext, stgID cdssdk.StorageID) (model.Storage, error) {
+	var stg model.Storage
+	err := ctx.Table("Storage").First(&stg, stgID).Error
+	return stg, err
+}
+
+func (db *StorageDB) BatchGetAllStorageIDs(ctx SQLContext, start int, count int) ([]cdssdk.StorageID, error) {
+	var ret []cdssdk.StorageID
+	err := ctx.Table("Storage").Select("StorageID").Find(ret).Limit(count).Offset(start).Error
+	return ret, err
+}
+
+func (db *StorageDB) IsAvailable(ctx SQLContext, userID cdssdk.UserID, storageID cdssdk.StorageID) (bool, error) {
+	rows, err := ctx.Table("Storage").Select("Storage.StorageID").
+		Joins("inner join UserStorage on Storage.StorageID = UserStorage.StorageID").
+		Where("UserID = ? and StorageID = ?", userID, storageID).Rows()
+	if err != nil {
+		return false, fmt.Errorf("execute sql: %w", err)
+	}
+	defer rows.Close()
+
+	return rows.Next(), nil
+}
+
+func (db *StorageDB) GetUserStorage(ctx SQLContext, userID cdssdk.UserID, storageID cdssdk.StorageID) (model.Storage, error) {
+	var stg model.Storage
+	err := ctx.Table("Storage").Select("Storage.*").
+		Joins("inner join UserStorage on Storage.StorageID = UserStorage.StorageID").
+		Where("UserID = ? and StorageID = ?", userID, storageID).First(&stg).Error
+
+	return stg, err
+}
+
+func (db *StorageDB) GetUserStorageByName(ctx SQLContext, userID cdssdk.UserID, name string) (model.Storage, error) {
+	var stg model.Storage
+	err := ctx.Table("Storage").Select("Storage.*").
+		Joins("inner join UserStorage on Storage.StorageID = UserStorage.StorageID").
+		Where("UserID = ? and Name = ?", userID, name).First(&stg).Error
+
+	return stg, err
+}
+
+// func (db *StorageDB) ChangeState(ctx SQLContext, storageID cdssdk.StorageID, state string) error {
+// 	return ctx.Table("Storage").Where("StorageID = ?", storageID).Update("State", state).Error
+// }

coordinator/internal/cmd/migrate.go

@@ -39,6 +39,12 @@ func migrate(configPath string) {
 		os.Exit(1)
 	}
 
+	err = db.AutoMigrate(&cdssdk.Node{})
+	if err != nil {
+		fmt.Printf("migratting model Node: %v\n", err)
+		os.Exit(1)
+	}
+
 	err = db.AutoMigrate(&cdssdk.Storage{})
 	if err != nil {
 		fmt.Printf("migratting model Storage: %v\n", err)