public/saasapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

支持dataspace

Browse Source
This commit is contained in:
algotao
2025-05-21 17:06:32 +08:00
parent e7d4aa27f1
commit b8c2a3a7d4
7 changed files with 265 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/saastool/params.go
View File

@@ -69,6 +69,10 @@ func paramClear(fs *flag.FlagSet) *bool {
return fs.Bool("clear", false, "Clear all data before write")
}
func paramDataSpaceId(fs *flag.FlagSet) *string {
return fs.String("ds", "", "Data space ID")
}
// ParseByteSize 解析字节大小字符串为字节数
func ParseByteSize(sizeStr string) (uint64, error) {
sizeStr = strings.TrimSpace(sizeStr)

23
cmd/saastool/read.go
View File

@@ -19,6 +19,7 @@ const (
type readParams struct {
cfg *Config
appid string
ds string
userids []string
saasHttp *saashttp.SaasClient
}
@@ -27,6 +28,7 @@ func RunRead(args ...string) error {
fs := flag.NewFlagSet("read", flag.ExitOnError)
cfgFile := paramConfig(fs)
appid := paramAppid(fs)
ds := paramDataSpaceId(fs)
userids := paramUserids(fs)
if err := fs.Parse(args); err != nil {
@@ -37,11 +39,16 @@ func RunRead(args ...string) error {
// 切割字符串
idsSlice := strings.Split(*userids, ",")
if fs.NArg() > 0 || len(idsSlice) == 0 || (len(idsSlice) == 1 && idsSlice[0] == "") || len(idsSlice) > getIdsMax {
if fs.NArg() > 0 || len(idsSlice) == 0 || (len(idsSlice) == 1 && idsSlice[0] == "") || len(idsSlice) > getIdsMax || len(*ds) == 0 {
fs.PrintDefaults()
return nil
}
if strings.ToLower(*ds) == "openid" && len(*appid) == 0 {
fmt.Fprintln(os.Stderr, "appid must be set when data space is openid")
return nil
}
cfg, err := LoadConfigFile(*cfgFile)
if err != nil {
fmt.Fprintln(os.Stderr, "load config file error", "err", err)
@@ -52,6 +59,7 @@ func RunRead(args ...string) error {
cfg: cfg,
userids: idsSlice,
appid: *appid,
ds: *ds,
saasHttp: &saashttp.SaasClient{
Client: &http.Client{},
ApiUrls: &cfg.ApiUrls,
@@ -63,16 +71,17 @@ func RunRead(args ...string) error {
}
func doRead(readParams readParams) error {
read := &saasapi.Read{
DataspaceId: readParams.ds,
Appid: readParams.appid,
}
saasReq := &saasapi.SaasReq{
Cmd: &saasapi.SaasReq_Read{
Read: &saasapi.Read{},
Read: read,
},
}
if readParams.appid != "" {
saasReq.UseridType = saasapi.UserIdType_OPENID
saasReq.Appid = readParams.appid
}
saasReadItems := []*saasapi.ReadItem{}
for _, userid := range readParams.userids {
saasReadItems = append(saasReadItems, &saasapi.ReadItem{
@@ -80,7 +89,7 @@ func doRead(readParams readParams) error {
})
}
saasReq.Cmd.(*saasapi.SaasReq_Read).Read.ReadItems = saasReadItems
read.ReadItems = saasReadItems
res, err := readParams.saasHttp.Read(saasReq)

9
cmd/saastool/task_create.go
View File

@@ -13,7 +13,6 @@ import (
type createTaskParams struct {
hashFile string
appid string
task *saasapi.Task
saasHttp *saashttp.SaasClient
}
@@ -21,9 +20,7 @@ type createTaskParams struct {
func RunTaskCreate(args ...string) error {
fs := flag.NewFlagSet("create", flag.ExitOnError)
cfgFile := paramConfig(fs)
// sourcePath := paramSourceConvertedPath(fs)
hashFile := paramInputHashFile(fs)
appid := paramAppid(fs)
if err := fs.Parse(args); err != nil {
fmt.Fprintln(os.Stderr, "command line parse error", "err", err)
@@ -43,7 +40,6 @@ func RunTaskCreate(args ...string) error {
createTaskParams := createTaskParams{
hashFile: *hashFile,
appid: *appid,
saasHttp: &saashttp.SaasClient{
Client: &http.Client{},
ApiUrls: &cfg.ApiUrls,
@@ -72,11 +68,6 @@ func doTaskCreate(createTaskParams createTaskParams) error {
},
}
if createTaskParams.appid != "" {
saasReq.UseridType = saasapi.UserIdType_OPENID
saasReq.Appid = createTaskParams.appid
}
res, err := createTaskParams.saasHttp.TaskCreate(saasReq)
if err != nil {

12
cmd/saastool/task_make.go
View File

@@ -9,6 +9,7 @@ import (
"path"
"runtime"
"sort"
"strings"
"sync"
"e.coding.net/rta/public/saasapi"
@@ -49,17 +50,24 @@ func RunTaskMake(args ...string) error {
hashFile := paramOutputHashFile(fs)
blockSize := paramBlockSize(fs)
desc := paramTaskDesc(fs)
appid := paramAppid(fs)
ds := paramDataSpaceId(fs)
if err := fs.Parse(args); err != nil {
fmt.Fprintln(os.Stderr, "command line parse error", "err", err)
return err
}
if fs.NArg() > 0 || len(*sourcePath) == 0 || len(*hashFile) == 0 {
if fs.NArg() > 0 || len(*sourcePath) == 0 || len(*hashFile) == 0 || len(*ds) == 0 {
fs.PrintDefaults()
return nil
}
if strings.ToLower(*ds) == "openid" && len(*appid) == 0 {
fmt.Fprintln(os.Stderr, "appid must be set when data space is openid")
return nil
}
blockSizeNum, err := ParseByteSize(*blockSize)
if err != nil {
fmt.Fprintln(os.Stderr, "Error parsing block size", "err", err)
@@ -79,6 +87,8 @@ func RunTaskMake(args ...string) error {
task: &saasapi.Task{
TaskBlockSize: blockSizeNum,
TaskDescription: *desc,
Appid: *appid,
DataspaceId: *ds,
},
}
return doMakeHash(makeTaskParams)

50
cmd/saastool/write.go
View File

@@ -7,6 +7,7 @@ import (
"net/http"
"os"
"path"
"strings"
"e.coding.net/rta/public/saasapi"
"e.coding.net/rta/public/saasapi/pkg/saashttp"
@@ -17,6 +18,7 @@ type writeParams struct {
cfg *Config
sourcePath string
appid string
ds string
batchSize uint
clear bool
saasHttp *saashttp.SaasClient
@@ -27,6 +29,7 @@ func RunWrite(args ...string) error {
cfgFile := paramConfig(fs)
sourcePath := paramSourcePath(fs)
appid := paramAppid(fs)
ds := paramDataSpaceId(fs)
batchSize := paramBatchSize(fs)
clear := paramClear(fs)
@@ -35,11 +38,16 @@ func RunWrite(args ...string) error {
return err
}
if fs.NArg() > 0 || len(*sourcePath) == 0 {
if fs.NArg() > 0 || len(*sourcePath) == 0 || len(*ds) == 0 {
fs.PrintDefaults()
return nil
}
if strings.ToLower(*ds) == "openid" && len(*appid) == 0 {
fmt.Fprintln(os.Stderr, "appid must be set when data space is openid")
return nil
}
cfg, err := LoadConfigFile(*cfgFile)
if err != nil {
fmt.Fprintln(os.Stderr, "load config file error", "err", err)
@@ -50,6 +58,7 @@ func RunWrite(args ...string) error {
cfg: cfg,
sourcePath: *sourcePath,
appid: *appid,
ds: *ds,
batchSize: *batchSize,
clear: *clear,
saasHttp: &saashttp.SaasClient{
@@ -107,9 +116,9 @@ func doLoadFileToWrite(writeParams writeParams) error {
saasWriteItems := []*saasapi.WriteItem{}
succ := uint32(0)
succTotal := uint32(0)
total := uint32(0)
errCount := 0
errTotal := 0
total := 0
for scaner.Scan() {
total++
line := scaner.Text()
@@ -125,45 +134,42 @@ func doLoadFileToWrite(writeParams writeParams) error {
saasWriteItems = append(saasWriteItems, saasWriteItem)
if len(saasWriteItems) == int(writeParams.batchSize) {
if succ, _, err = submitWrite(writeParams, saasWriteItems); err != nil {
if errCount, err = submitWrite(writeParams, saasWriteItems); err != nil {
return err
}
succTotal += succ
fmt.Printf("[%v] batch_succ = %v, succ_total = %v, total_processed = %v\n", writeParams.sourcePath, succ, succTotal, total)
errTotal += errCount
fmt.Printf("[%v] err_batch = %v, err_total = %v, total_processed = %v\n", writeParams.sourcePath, errCount, errTotal, total)
saasWriteItems = saasWriteItems[:0]
}
}
if len(saasWriteItems) > 0 {
if succ, _, err = submitWrite(writeParams, saasWriteItems); err != nil {
if errCount, err = submitWrite(writeParams, saasWriteItems); err != nil {
return err
}
succTotal += succ
fmt.Printf("[%v] batch_succ = %v, succ_total = %v, total_processed = %v\n", writeParams.sourcePath, succ, succTotal, total)
errTotal += errCount
fmt.Printf("[%v] err_batch = %v, err_total = %v, total_processed = %v\n", writeParams.sourcePath, errCount, errTotal, total)
}
return nil
}
func submitWrite(writeParams writeParams, saasWriteCmds []*saasapi.WriteItem) (succ, total uint32, err error) {
func submitWrite(writeParams writeParams, saasWriteCmds []*saasapi.WriteItem) (errcount int, err error) {
write := &saasapi.Write{
DataspaceId: writeParams.ds,
Appid: writeParams.appid,
IsClearAllFirst: writeParams.clear,
}
saasReq := &saasapi.SaasReq{
Cmd: &saasapi.SaasReq_Write{
Write: &saasapi.Write{
IsClearAllFirst: writeParams.clear,
},
Write: write,
},
}
if writeParams.appid != "" {
saasReq.UseridType = saasapi.UserIdType_OPENID
saasReq.Appid = writeParams.appid
}
write.WriteItems = saasWriteCmds
saasReq.Cmd.(*saasapi.SaasReq_Write).Write.WriteItems = saasWriteCmds
total = uint32(len(saasWriteCmds))
res, err := writeParams.saasHttp.Write(saasReq)
if err != nil {
@@ -177,5 +183,5 @@ func submitWrite(writeParams writeParams, saasWriteCmds []*saasapi.WriteItem) (s
return
}
return res.GetWriteRes().GetSuccCmdCount(), total, nil
return len(res.GetWriteRes().GetFailedUserid()), nil
}